package leasetest

import (
	"crypto/sha256"
	"encoding/base64"
	"encoding/hex"
	"encoding/xml"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"sort"
	"strconv"
	"strings"
	"sync/atomic"
	"time"

	"github.com/carlosframework/platform/internal/lease"
)

type xmlVersion struct {
	XMLName      xml.Name `xml:"Version"`
	Key          string
	VersionId    string
	IsLatest     bool
	LastModified string
	ETag         string
	Size         int64
}

type xmlMarker struct {
	XMLName      xml.Name `xml:"DeleteMarker"`
	Key          string
	VersionId    string
	IsLatest     bool
	LastModified string
}

type listVersionsResult struct {
	XMLName             xml.Name `xml:"ListVersionsResult"`
	Name                string
	Prefix              string
	KeyMarker           string
	VersionIdMarker     string
	NextKeyMarker       string `xml:",omitempty"`
	NextVersionIdMarker string `xml:",omitempty"`
	MaxKeys             int
	IsTruncated         bool
	Entries             []any
}

func s3err(w http.ResponseWriter, status int, code string) { s3errMsg(w, status, code, code) }

func s3errMsg(w http.ResponseWriter, status int, code, msg string) {
	w.Header().Set("Content-Type", "application/xml")
	w.WriteHeader(status)
	io.WriteString(w, "<Error><Code>"+code+"</Code><Message>"+msg+"</Message></Error>")
}

// readValidatedBody reads a PUT's body and checks it as real S3 does: its
// length must equal the declared Content-Length (when one is declared), and,
// when x-amz-content-sha256 is a hex digest, its sha256 must equal it. Any
// other value of that header (UNSIGNED-PAYLOAD, a streaming marker) or none
// is not checked. On a mismatch it answers 400 with S3's code and reports
// false, and the caller stores nothing.
func readValidatedBody(w http.ResponseWriter, r *http.Request) ([]byte, bool) {
	body, err := io.ReadAll(r.Body)
	if err != nil || (r.ContentLength >= 0 && int64(len(body)) != r.ContentLength) {
		s3errMsg(w, http.StatusBadRequest, "IncompleteBody", "You did not provide the number of bytes specified by the Content-Length HTTP header.")
		return nil, false
	}
	if want := r.Header.Get("x-amz-content-sha256"); isSHA256Hex(want) {
		sum := sha256.Sum256(body)
		if !strings.EqualFold(want, hex.EncodeToString(sum[:])) {
			s3errMsg(w, http.StatusBadRequest, "XAmzContentSHA256Mismatch", "The provided 'x-amz-content-sha256' header does not match what was computed.")
			return nil, false
		}
	}
	return body, true
}

func isSHA256Hex(s string) bool {
	if len(s) != 2*sha256.Size {
		return false
	}
	_, err := hex.DecodeString(s)
	return err == nil
}

// copyMarkerMsg is S3's answer to a COPY naming a delete marker's version.
const copyMarkerMsg = "The source of a copy request may not specifically refer to a delete marker by version id."

// writeErr maps a model error onto the S3 response real S3 gives.
func writeErr(w http.ResponseWriter, err error) {
	switch {
	case errors.Is(err, lease.ErrMirrorNotFound):
		s3err(w, http.StatusNotFound, "NoSuchVersion")
	case errors.Is(err, lease.ErrIsDeleteMarker):
		w.Header().Set("x-amz-delete-marker", "true")
		s3err(w, http.StatusMethodNotAllowed, "MethodNotAllowed")
	case errors.Is(err, lease.ErrDestinationExists), errors.Is(err, lease.ErrExists):
		s3err(w, http.StatusPreconditionFailed, "PreconditionFailed")
	default:
		s3err(w, http.StatusInternalServerError, "InternalError")
	}
}

// NewS3Server speaks the subset of the S3 wire protocol VersionedStore uses,
// and ListObjectsV2 (serveListV2), path-style, over f, holding f.mu for each request. The second result
// counts requests served. With serves non-empty, the endpoint answers only
// for those request buckets (a regional endpoint); any other bucket gets
// 301 PermanentRedirect, as S3 answers a wrong-region request. A copy's
// source bucket may be any bucket in f.
func NewS3Server(f *VersionedBucket, serves ...string) (http.Handler, func() int) {
	var n atomic.Int64
	allowed := map[string]bool{}
	for _, b := range serves {
		allowed[b] = true
	}
	h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		n.Add(1)
		bucket, rawKey, _ := strings.Cut(strings.TrimPrefix(r.URL.EscapedPath(), "/"), "/")
		key, err := url.PathUnescape(rawKey)
		if err != nil {
			s3err(w, http.StatusBadRequest, "InvalidURI")
			return
		}
		q := r.URL.Query()
		if len(allowed) > 0 && !allowed[bucket] {
			s3err(w, http.StatusMovedPermanently, "PermanentRedirect")
			return
		}
		var body []byte
		if r.Method == http.MethodPut { // every PUT, a copy included, is validated before it is applied
			var ok bool
			if body, ok = readValidatedBody(w, r); !ok {
				return
			}
		}
		f.mu.Lock()
		defer f.mu.Unlock()
		if err := f.known(bucket); err != nil {
			s3err(w, http.StatusNotFound, "NoSuchBucket")
			return
		}
		switch {
		case r.Method == http.MethodGet && key == "" && q.Has("versions"):
			f.serveList(w, bucket, q)
		case r.Method == http.MethodGet && key == "" && q.Get("list-type") == "2":
			f.serveListV2(w, bucket, q)
		case r.Method == http.MethodGet && key != "":
			versionID := q.Get("versionId")
			if versionID == "" { // no version named: the current one
				cur := f.buckets[bucket][key]
				if len(cur) == 0 || cur[0].deleteMarker {
					s3err(w, http.StatusNotFound, "NoSuchKey")
					return
				}
				versionID = cur[0].id
			}
			body, err := f.read(bucket, key, versionID)
			if err != nil {
				writeErr(w, err)
				return
			}
			if v, ok := f.find(bucket, key, versionID); ok {
				w.Header().Set("ETag", `"`+v.etag+`"`)
			}
			if hdr := r.Header.Get("Range"); hdr != "" {
				span, ok := strings.CutPrefix(hdr, "bytes=")
				start, end, dash := strings.Cut(span, "-")
				offset, startErr := strconv.ParseInt(start, 10, 64)
				length := int64(-1)
				if end != "" {
					last, endErr := strconv.ParseInt(end, 10, 64)
					if endErr != nil || last < offset || last-offset == 1<<63-1 {
						s3err(w, http.StatusRequestedRangeNotSatisfiable, "InvalidRange")
						return
					}
					length = last - offset + 1
				}
				total := len(body)
				body, err = versionRange(body, offset, length)
				if !ok || !dash || startErr != nil || err != nil {
					s3err(w, http.StatusRequestedRangeNotSatisfiable, "InvalidRange")
					return
				}
				w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", offset, offset+int64(len(body))-1, total))
				w.Header().Set("Content-Length", strconv.Itoa(len(body)))
				w.WriteHeader(http.StatusPartialContent)
			}
			w.Write(body)
		case r.Method == http.MethodPut && r.Header.Get("x-amz-copy-source") != "":
			f.serveCopy(w, r, bucket, key)
		case r.Method == http.MethodPut:
			if match := r.Header.Get("If-Match"); match != "" {
				current := f.buckets[bucket][key]
				if len(current) == 0 || current[0].deleteMarker || current[0].etag != strings.Trim(match, `"`) {
					s3err(w, http.StatusPreconditionFailed, "PreconditionFailed")
					return
				}
			}
			var id string
			var err error
			if r.Header.Get("If-None-Match") == "*" {
				id, err = f.putIfAbsentLocked(bucket, key, body)
			} else {
				id, err = f.putLocked(bucket, key, body) // a new version, as S3 with versioning on
			}
			if err != nil {
				writeErr(w, err)
				return
			}
			w.Header().Set("x-amz-version-id", id)
			w.Header().Set("ETag", `"`+md5hex(body)+`"`)
		case r.Method == http.MethodDelete && key != "" && q.Get("versionId") != "":
			if err := f.deleteLocked(bucket, key, q.Get("versionId")); err != nil {
				writeErr(w, err)
				return
			}
			w.WriteHeader(http.StatusNoContent)
		case r.Method == http.MethodDelete && key != "":
			if match := r.Header.Get("If-Match"); match != "" { // as S3 enforces it on DeleteObject
				current := f.buckets[bucket][key]
				if len(current) == 0 || current[0].deleteMarker {
					s3err(w, http.StatusNotFound, "NoSuchKey")
					return
				}
				if current[0].etag != strings.Trim(match, `"`) {
					s3err(w, http.StatusPreconditionFailed, "PreconditionFailed")
					return
				}
			}
			id, err := f.bareDeleteLocked(bucket, key) // versioning on: a delete marker
			if err != nil {
				writeErr(w, err)
				return
			}
			w.Header().Set("x-amz-delete-marker", "true")
			w.Header().Set("x-amz-version-id", id)
			w.WriteHeader(http.StatusNoContent)
		default:
			s3err(w, http.StatusMethodNotAllowed, "MethodNotAllowed")
		}
	})
	return h, func() int { return int(n.Load()) }
}

func (f *VersionedBucket) serveList(w http.ResponseWriter, bucket string, q url.Values) {
	if err := f.fail("list", q.Get("prefix")); err != nil {
		writeErr(w, err)
		return
	}
	all := f.listAll(bucket, q.Get("prefix"))
	km, vm := q.Get("key-marker"), q.Get("version-id-marker")
	start := 0
	if km != "" {
		start = len(all) // a marker not found resumes at the end
		for i, v := range all {
			if v.Key == km && v.VersionID == vm {
				start = i + 1
				break
			}
		}
	}
	size := f.PageSize
	if size <= 0 {
		size = 1000
	}
	page := (start + size - 1) / size // pages already served before this one
	vs, more, err := f.pageOf(all[start:], 0)
	if f.FailListAfterPages > 0 && page >= f.FailListAfterPages {
		err = errors.New("leasetest: injected listing failure")
	}
	if err != nil {
		writeErr(w, err)
		return
	}
	res := listVersionsResult{Name: bucket, Prefix: q.Get("prefix"), KeyMarker: km, VersionIdMarker: vm, MaxKeys: size, IsTruncated: more}
	for _, v := range vs {
		lm := v.LastModified.UTC().Format(time.RFC3339)
		if v.DeleteMarker {
			res.Entries = append(res.Entries, xmlMarker{Key: v.Key, VersionId: v.VersionID, IsLatest: v.IsLatest, LastModified: lm})
		} else {
			res.Entries = append(res.Entries, xmlVersion{Key: v.Key, VersionId: v.VersionID, IsLatest: v.IsLatest, LastModified: lm, ETag: `"` + v.ETag + `"`, Size: v.Size})
		}
	}
	if more && len(vs) > 0 {
		res.NextKeyMarker, res.NextVersionIdMarker = vs[len(vs)-1].Key, vs[len(vs)-1].VersionID
	}
	w.Header().Set("Content-Type", "application/xml")
	xml.NewEncoder(w).Encode(res)
}

type xmlObject struct {
	Key          string
	LastModified string
	ETag         string
	Size         int64
	StorageClass string
}

type xmlCommonPrefix struct {
	Prefix string
}

type listBucketResult struct {
	XMLName               xml.Name `xml:"ListBucketResult"`
	Name                  string
	Prefix                string
	Delimiter             string `xml:",omitempty"`
	MaxKeys               int
	KeyCount              int
	IsTruncated           bool
	ContinuationToken     string `xml:",omitempty"`
	NextContinuationToken string `xml:",omitempty"`
	StartAfter            string `xml:",omitempty"`
	Contents              []xmlObject
	CommonPrefixes        []xmlCommonPrefix
}

// serveListV2 is ListObjectsV2, the subset litestream's replica client
// (its restore lists with a delimiter) and lease.NewS3Mirror page through:
// the current objects under prefix in key order (an object whose newest
// version is a delete marker is not listed), a delimiter rolling every key
// that holds it after the prefix into one common prefix, max-keys (at most
// 1000, and at most PageSize when that is set lower; keys and common
// prefixes both count) and continuation-token pagination. The token is
// opaque to the client; here it is the last key or common prefix of the
// previous page, and a page resumes after it (after every key under it,
// for a common prefix). start-after is honoured on the first page.
func (f *VersionedBucket) serveListV2(w http.ResponseWriter, bucket string, q url.Values) {
	prefix, delim := q.Get("prefix"), q.Get("delimiter")
	if err := f.fail("list", prefix); err != nil {
		writeErr(w, err)
		return
	}
	size := 1000
	if f.PageSize > 0 && f.PageSize < size {
		size = f.PageSize
	}
	if raw := q.Get("max-keys"); raw != "" {
		n, err := strconv.Atoi(raw)
		if err != nil || n < 0 {
			s3err(w, http.StatusBadRequest, "InvalidArgument")
			return
		}
		size = min(size, n)
	}
	after := q.Get("start-after")
	token := q.Get("continuation-token")
	if token != "" {
		last, err := base64.StdEncoding.DecodeString(token)
		if err != nil || len(last) == 0 {
			s3errMsg(w, http.StatusBadRequest, "InvalidArgument", "The continuation token provided is incorrect")
			return
		}
		after = string(last)
	}
	// entries are the page candidates in key order: a key, or a common
	// prefix (isPrefix) standing for every key under it.
	type entry struct {
		name     string
		isPrefix bool
	}
	var keys []string
	for k, vs := range f.buckets[bucket] {
		if strings.HasPrefix(k, prefix) && len(vs) > 0 && !vs[0].deleteMarker {
			keys = append(keys, k)
		}
	}
	sort.Strings(keys)
	var entries []entry
	for _, k := range keys {
		e := entry{name: k}
		if delim != "" {
			if i := strings.Index(k[len(prefix):], delim); i >= 0 {
				e = entry{name: k[:len(prefix)+i+len(delim)], isPrefix: true}
			}
		}
		if n := len(entries); n > 0 && entries[n-1] == e {
			continue
		}
		// A common prefix that ended the last page is that page's token
		// itself, and so is every key rolled into it.
		if e.name <= after {
			continue
		}
		entries = append(entries, e)
	}
	res := listBucketResult{Name: bucket, Prefix: prefix, Delimiter: delim, MaxKeys: size, ContinuationToken: token}
	if token == "" {
		res.StartAfter = q.Get("start-after")
	}
	page := entries
	if len(page) > size {
		page = page[:size]
		res.IsTruncated = size > 0
	}
	for _, e := range page {
		if e.isPrefix {
			res.CommonPrefixes = append(res.CommonPrefixes, xmlCommonPrefix{Prefix: e.name})
			continue
		}
		v := f.buckets[bucket][e.name][0]
		res.Contents = append(res.Contents, xmlObject{Key: e.name, LastModified: v.modified.UTC().Format(time.RFC3339), ETag: `"` + v.etag + `"`, Size: int64(len(v.body)), StorageClass: "STANDARD"})
	}
	res.KeyCount = len(page)
	if res.IsTruncated && len(page) > 0 {
		res.NextContinuationToken = base64.StdEncoding.EncodeToString([]byte(page[len(page)-1].name))
	}
	w.Header().Set("Content-Type", "application/xml")
	xml.NewEncoder(w).Encode(res)
}

func (f *VersionedBucket) serveCopy(w http.ResponseWriter, r *http.Request, bucket, key string) {
	src := strings.TrimPrefix(r.Header.Get("x-amz-copy-source"), "/")
	srcPath, rawQ, _ := strings.Cut(src, "?")
	if strings.ContainsAny(srcPath, "+? ") {
		s3err(w, http.StatusBadRequest, "InvalidArgument") // AWS's encoder escapes these; a raw one is a client bug
		return
	}
	sq, err := url.ParseQuery(rawQ)
	if err != nil {
		s3err(w, http.StatusBadRequest, "InvalidArgument")
		return
	}
	unesc, err := url.PathUnescape(srcPath)
	if err != nil {
		s3err(w, http.StatusBadRequest, "InvalidArgument")
		return
	}
	sb, sk, _ := strings.Cut(unesc, "/")
	srcVersion := sq.Get("versionId")
	if srcVersion == "" { // no version named: the source's current one
		cur := f.buckets[sb][sk]
		if len(cur) == 0 || cur[0].deleteMarker {
			s3err(w, http.StatusNotFound, "NoSuchKey")
			return
		}
		srcVersion = cur[0].id
	}
	if im := r.Header.Get("x-amz-copy-source-if-match"); im != "" {
		if v, ok := f.find(sb, sk, srcVersion); ok && strings.Trim(im, `"`) != v.etag {
			s3err(w, http.StatusPreconditionFailed, "PreconditionFailed")
			return
		}
	}
	id, etag, err := f.copyLocked(sb, sk, srcVersion, bucket, key, r.Header.Get("If-None-Match") == "*")
	if errors.Is(err, lease.ErrIsDeleteMarker) {
		s3errMsg(w, http.StatusBadRequest, "InvalidRequest", copyMarkerMsg) // a COPY is 400, not GET's 405
		return
	}
	if err != nil {
		writeErr(w, err)
		return
	}
	w.Header().Set("x-amz-version-id", id)
	w.Header().Set("Content-Type", "application/xml")
	io.WriteString(w, `<CopyObjectResult><ETag>"`+etag+`"</ETag><LastModified>2026-10-05T00:00:00Z</LastModified></CopyObjectResult>`)
}