// Package leasetest holds test doubles for internal/lease. VersionedBucket
// is the reference model of a versioned S3 bucket that every plan of the
// host lifecycle claim tests against (spec §13).
package leasetest

import (
	"bytes"
	"context"
	"crypto/md5"
	"encoding/hex"
	"fmt"
	"io"
	"sort"
	"strings"
	"sync"
	"time"

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

type version struct {
	id           string
	body         []byte
	etag         string
	deleteMarker bool
	modified     time.Time
}

// VersionedBucket is an in-memory, versioned, multi-bucket S3 model.
type VersionedBucket struct {
	mu      sync.Mutex
	buckets map[string]map[string][]version // bucket -> key -> versions, newest first
	next    int
	clock   time.Time

	PageSize           int                        // listing page size; default 1000
	FailListAfterPages int                        // fail a listing once this many pages have been served; 0 never
	FailOn             func(op, key string) error // op: list|get|copy|delete|put
}

func NewVersionedBucket() *VersionedBucket {
	return &VersionedBucket{buckets: map[string]map[string][]version{}, PageSize: 1000,
		clock: time.Date(2026, 10, 5, 0, 0, 0, 0, time.UTC)}
}

func md5hex(b []byte) string { s := md5.Sum(b); return hex.EncodeToString(s[:]) }

func (f *VersionedBucket) AddBucket(name string) {
	f.mu.Lock()
	defer f.mu.Unlock()
	if f.buckets[name] == nil {
		f.buckets[name] = map[string][]version{}
	}
}

// --- locked helpers: callers hold f.mu ---

func (f *VersionedBucket) fail(op, key string) error {
	if f.FailOn != nil {
		return f.FailOn(op, key)
	}
	return nil
}

func (f *VersionedBucket) known(bucket string) error {
	if _, ok := f.buckets[bucket]; !ok {
		return fmt.Errorf("leasetest: no bucket %q", bucket)
	}
	return nil
}

func (f *VersionedBucket) push(bucket, key string, v version) string {
	f.next++
	f.clock = f.clock.Add(time.Second)
	v.id, v.modified = fmt.Sprintf("v%06d", f.next), f.clock
	f.buckets[bucket][key] = append([]version{v}, f.buckets[bucket][key]...)
	return v.id
}

func (f *VersionedBucket) hasCurrent(bucket, key string) bool {
	vs := f.buckets[bucket][key]
	return len(vs) > 0 && !vs[0].deleteMarker
}

func (f *VersionedBucket) find(bucket, key, id string) (version, bool) {
	for _, v := range f.buckets[bucket][key] {
		if v.id == id {
			return v, true
		}
	}
	return version{}, false
}

// listAll is the whole ordered listing under prefix (key order, newest
// first within a key).
func (f *VersionedBucket) listAll(bucket, prefix string) []lease.ObjectVersion {
	o := f.buckets[bucket]
	keys := make([]string, 0, len(o))
	for k := range o {
		if strings.HasPrefix(k, prefix) {
			keys = append(keys, k)
		}
	}
	sort.Strings(keys)
	var all []lease.ObjectVersion
	for _, k := range keys {
		for i, v := range o[k] {
			ov := lease.ObjectVersion{Key: k, VersionID: v.id, IsLatest: i == 0, DeleteMarker: v.deleteMarker, LastModified: v.modified}
			if !v.deleteMarker {
				ov.ETag, ov.Size = v.etag, int64(len(v.body))
			}
			all = append(all, ov)
		}
	}
	return all
}

// pageOf returns page number page (0-based) of all, whether more follow,
// and the failure FailListAfterPages injects.
func (f *VersionedBucket) pageOf(all []lease.ObjectVersion, page int) ([]lease.ObjectVersion, bool, error) {
	if f.FailListAfterPages > 0 && page >= f.FailListAfterPages {
		return nil, false, fmt.Errorf("leasetest: injected listing failure on page %d", page+1)
	}
	size := f.PageSize
	if size <= 0 {
		size = 1000
	}
	start := page * size
	if start > len(all) {
		start = len(all)
	}
	end := start + size
	if end >= len(all) {
		return all[start:], false, nil
	}
	return all[start:end], true, nil
}

// --- exported helpers for tests ---

func (f *VersionedBucket) Put(bucket, key string, body []byte) string {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.push(bucket, key, version{body: append([]byte(nil), body...), etag: md5hex(body)})
}

func (f *VersionedBucket) PutMultipart(bucket, key string, body []byte) string {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.push(bucket, key, version{body: append([]byte(nil), body...), etag: md5hex(body) + "-2"})
}

func (f *VersionedBucket) BareDelete(bucket, key string) string {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.push(bucket, key, version{deleteMarker: true})
}

func (f *VersionedBucket) LatestID(bucket, key string) string {
	f.mu.Lock()
	defer f.mu.Unlock()
	if vs := f.buckets[bucket][key]; len(vs) > 0 {
		return vs[0].id
	}
	return ""
}

func (f *VersionedBucket) Current(bucket, key string) ([]byte, bool) {
	f.mu.Lock()
	defer f.mu.Unlock()
	vs := f.buckets[bucket][key]
	if len(vs) == 0 || vs[0].deleteMarker {
		return nil, false
	}
	return vs[0].body, true
}

// Snapshot renders the whole state, for before/after comparison.
func (f *VersionedBucket) Snapshot() string {
	f.mu.Lock()
	defer f.mu.Unlock()
	var b strings.Builder
	names := make([]string, 0, len(f.buckets))
	for n := range f.buckets {
		names = append(names, n)
	}
	sort.Strings(names)
	for _, n := range names {
		for _, v := range f.listAll(n, "") {
			fmt.Fprintf(&b, "%s/%s@%s:%v:%s\n", n, v.Key, v.VersionID, v.DeleteMarker, v.ETag)
		}
	}
	return b.String()
}

// --- lease.VersionedStore ---

func (f *VersionedBucket) ListVersions(ctx context.Context, bucket, prefix string) ([]lease.ObjectVersion, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	if err := f.known(bucket); err != nil {
		return nil, err
	}
	if err := f.fail("list", prefix); err != nil {
		return nil, err
	}
	all := f.listAll(bucket, prefix)
	var out []lease.ObjectVersion
	for page := 0; ; page++ {
		vs, more, err := f.pageOf(all, page)
		if err != nil {
			return nil, err
		}
		out = append(out, vs...)
		if !more {
			return out, nil
		}
	}
}

func (f *VersionedBucket) read(bucket, key, versionID string) ([]byte, error) {
	if versionID == "" {
		return nil, lease.ErrNoVersionID
	}
	if err := f.known(bucket); err != nil {
		return nil, err
	}
	if err := f.fail("get", key); err != nil {
		return nil, err
	}
	v, ok := f.find(bucket, key, versionID)
	if !ok {
		return nil, lease.ErrMirrorNotFound
	}
	if v.deleteMarker {
		return nil, lease.ErrIsDeleteMarker
	}
	return append([]byte(nil), v.body...), nil
}

func (f *VersionedBucket) GetVersion(ctx context.Context, bucket, key, versionID string) ([]byte, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.read(bucket, key, versionID)
}

func (f *VersionedBucket) OpenVersion(ctx context.Context, bucket, key, versionID string) (io.ReadCloser, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	b, err := f.read(bucket, key, versionID)
	if err != nil {
		return nil, err
	}
	return io.NopCloser(bytes.NewReader(b)), nil
}

func versionRange(body []byte, offset, length int64) ([]byte, error) {
	if offset < 0 || length == 0 || (length > 0 && offset > (1<<63-1)-(length-1)) || offset >= int64(len(body)) {
		return nil, fmt.Errorf("leasetest: invalid version range")
	}
	n := int64(len(body)) - offset
	if length > 0 && length < n {
		n = length
	}
	return body[offset : offset+n], nil
}

func (f *VersionedBucket) OpenVersionRange(ctx context.Context, bucket, key, versionID string, offset, length int64) (io.ReadCloser, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	b, err := f.read(bucket, key, versionID)
	if err != nil {
		return nil, err
	}
	b, err = versionRange(b, offset, length)
	if err != nil {
		return nil, err
	}
	return io.NopCloser(bytes.NewReader(b)), nil
}

func (f *VersionedBucket) copyLocked(srcBucket, srcKey, srcVersionID, dstBucket, dstKey string, createOnly bool) (string, string, error) {
	if srcVersionID == "" {
		return "", "", lease.ErrNoVersionID
	}
	if err := f.known(srcBucket); err != nil {
		return "", "", err
	}
	if err := f.known(dstBucket); err != nil {
		return "", "", err
	}
	if err := f.fail("copy", srcKey); err != nil {
		return "", "", err
	}
	v, ok := f.find(srcBucket, srcKey, srcVersionID)
	if !ok {
		return "", "", lease.ErrMirrorNotFound
	}
	if v.deleteMarker {
		return "", "", lease.ErrIsDeleteMarker
	}
	if createOnly && f.hasCurrent(dstBucket, dstKey) {
		return "", "", lease.ErrDestinationExists
	}
	etag := md5hex(v.body) // a copy is single-part (§13 row 9)
	return f.push(dstBucket, dstKey, version{body: append([]byte(nil), v.body...), etag: etag}), etag, nil
}

func (f *VersionedBucket) CopyVersion(ctx context.Context, srcBucket, srcKey, srcVersionID, dstBucket, dstKey string, createOnly bool) (string, string, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.copyLocked(srcBucket, srcKey, srcVersionID, dstBucket, dstKey, createOnly)
}

func (f *VersionedBucket) deleteLocked(bucket, key, versionID string) error {
	if versionID == "" {
		return lease.ErrNoVersionID
	}
	if err := f.known(bucket); err != nil {
		return err
	}
	if err := f.fail("delete", key); err != nil {
		return err
	}
	vs := f.buckets[bucket][key]
	for i, v := range vs {
		if v.id == versionID {
			rest := append(vs[:i:i], vs[i+1:]...)
			if len(rest) == 0 {
				delete(f.buckets[bucket], key)
			} else {
				f.buckets[bucket][key] = rest
			}
			return nil
		}
	}
	return nil
}

func (f *VersionedBucket) DeleteVersion(ctx context.Context, bucket, key, versionID string) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.deleteLocked(bucket, key, versionID)
}

func (f *VersionedBucket) putIfAbsentLocked(bucket, key string, body []byte) (string, error) {
	if err := f.known(bucket); err != nil {
		return "", err
	}
	if err := f.fail("put", key); err != nil {
		return "", err
	}
	if f.hasCurrent(bucket, key) {
		return "", lease.ErrExists
	}
	return f.push(bucket, key, version{body: append([]byte(nil), body...), etag: md5hex(body)}), nil
}

// putLocked is a plain PUT: it stores a new version beside the existing
// ones, as S3 with versioning on does. putIfAbsentLocked is the create-only
// variant (If-None-Match: *).
func (f *VersionedBucket) putLocked(bucket, key string, body []byte) (string, error) {
	if err := f.known(bucket); err != nil {
		return "", err
	}
	if err := f.fail("put", key); err != nil {
		return "", err
	}
	return f.push(bucket, key, version{body: append([]byte(nil), body...), etag: md5hex(body)}), nil
}

// bareDeleteLocked is a DELETE naming no version: S3 with versioning on
// adds a delete marker and removes nothing.
func (f *VersionedBucket) bareDeleteLocked(bucket, key string) (string, error) {
	if err := f.known(bucket); err != nil {
		return "", err
	}
	if err := f.fail("delete", key); err != nil {
		return "", err
	}
	return f.push(bucket, key, version{deleteMarker: true}), nil
}

func (f *VersionedBucket) PutIfAbsent(ctx context.Context, bucket, key string, body []byte) (string, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.putIfAbsentLocked(bucket, key, body)
}

var _ lease.VersionedStore = (*VersionedBucket)(nil)

var _ lease.VersionRangeOpener = (*VersionedBucket)(nil)