// Store — the deployment bucket behind the release plane: S3 in production,
// a plain directory in tests. CAS discipline (If-Match / If-None-Match)
// ported from Eleven internal/idalloc/idalloc.go (the litestream leaser's
// conditional-write pattern); not-found folding from Eleven s3blobstore.go.
package release

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"io"
	"io/fs"
	"net/http"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"sync"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	awsconfig "github.com/aws/aws-sdk-go-v2/config"
	awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/aws/smithy-go"
	"github.com/carlosframework/platform/internal/rootaws"
	"github.com/carlosframework/platform/internal/safefile"

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

// ErrCASConflict is a conditional Put that lost: the object already existed
// (ifMatch "") or moved on since the etag was read (ifMatch stale). Callers
// re-read and retry; they never overwrite blind.
var ErrCASConflict = errors.New("store: compare-and-swap conflict")

// Store is the deployment bucket. Get's not-found is os.ErrNotExist-wrapped
// so callers use errors.Is; Put with ifMatch "" is create-only
// (If-None-Match: *), non-empty is replace-only (If-Match: <etag>), and a
// lost race is ErrCASConflict either way. Delete is idempotent: deleting a
// missing key is not an error.
type Store interface {
	Get(ctx context.Context, key string) ([]byte, string, error)                      // body, etag
	Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) // new etag
	List(ctx context.Context, prefix string) ([]string, error)
	// ListDirs returns the immediate child "directories" of prefix — the
	// delimiter-aware listing, so a caller that only wants the names one
	// level down does not pay for every object beneath them. Each result is
	// a FULL prefix ending in "/" (S3 CommonPrefixes semantics), never a
	// bare name. A prefix that does not end in "/" is treated as though it
	// did, so the two implementations answer identically. Objects at the
	// level itself (files, not directories) are never included.
	ListDirs(ctx context.Context, prefix string) ([]string, error)
	Delete(ctx context.Context, key string) error
}

// ---- prefixStore -------------------------------------------------------------

// prefixStore rebases every key a caller works with onto a fixed base before
// it ever reaches the underlying store — the actual enforcement point for a
// namespacing env var like CARLOS_STATUS_PREFIX (customer-fleets plan, Task
// 5: a fleet box's own heartbeat, written through one of these with base
// "fleets/<name>/", lands at "fleets/<name>/status/boxes/<label>.json"
// instead of the shared "status/boxes/<label>.json" every platform box also
// writes — outside the exact prefix status.BuildFleet's loadBoxes lists, and
// so invisible to the platform's own fleet.json by construction, with no
// change needed to the status package itself). List/ListDirs strip the base
// back off their results, so a key one of these hands back still round-trips
// through the SAME wrapper's Get/Put without doubling the prefix.
//
// Exported from internal/release (rather than living next to its first
// caller in internal/host) because it is a property of Store itself, not of
// any one reader/writer of one: the console's fleet page (customer-fleets
// plan, Task 9) needs to read a fleet box's snapshots through this same
// rebasing, and a console-side package cannot reach an unexported type in
// internal/host.
type prefixStore struct {
	Store
	base string // always "" or ends in "/"
}

// BucketListDirs is the optional half of Store that says ListDirs answers in
// the bucket's own shape: ListDirs("apps/") is the account directories, and
// ListDirs("apps/<acct>/") that account's app directories, exactly as an S3
// delimiter listing of the deployment bucket would be. A reader may then walk
// the tree by its directories instead of listing every key under it.
//
// Optional, because not every Store has that shape. The fleet door scopes a
// box to its own account and answers ListDirs("apps/") with that account's
// APP directories, and DirStore lists by walking its whole root, so walking
// it per app is slower than one List. A Store that does not say so is read
// the old way.
type BucketListDirs interface {
	BucketShapedListDirs() bool
}

// HasBucketListDirs reports whether s promises bucket-shaped ListDirs.
func HasBucketListDirs(s Store) bool {
	b, ok := s.(BucketListDirs)
	return ok && b.BucketShapedListDirs()
}

// BucketShapedListDirs: rebasing keeps the shape of the store underneath.
func (p prefixStore) BucketShapedListDirs() bool { return HasBucketListDirs(p.Store) }

// NewPrefixStore wraps s so every key is rebased under prefix. An empty
// prefix returns s UNCHANGED (not a prefixStore with an empty base) — every
// deployment that never sets the namespacing env var (every platform box
// today) must get back the exact same store, a pure no-op, verifiable by
// interface equality.
func NewPrefixStore(s Store, prefix string) Store {
	if prefix == "" {
		return s
	}
	if !strings.HasSuffix(prefix, "/") {
		prefix += "/"
	}
	return prefixStore{Store: s, base: prefix}
}

func (p prefixStore) full(key string) string { return p.base + key }

func (p prefixStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	return p.Store.Get(ctx, p.full(key))
}

// Open is written out, not promoted: the embedded field is the interface,
// and a promoted Open would forward the UNREBASED key past the prefix.
func (p prefixStore) Open(ctx context.Context, key string) (io.ReadCloser, string, error) {
	return OpenObject(ctx, p.Store, p.full(key))
}

func (p prefixStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	return p.Store.Put(ctx, p.full(key), body, ifMatch)
}

// PutFrom is written out, not promoted, for the reason Open is: a promoted
// method would forward the UNREBASED key.
func (p prefixStore) PutFrom(ctx context.Context, key string, body io.ReadSeeker, size int64, ifMatch string) (string, error) {
	return PutObjectFrom(ctx, p.Store, p.full(key), body, size, ifMatch)
}

func (p prefixStore) Delete(ctx context.Context, key string) error {
	return p.Store.Delete(ctx, p.full(key))
}

// List rebases the query prefix the same way Get/Put rebase a key, then
// strips its own base back off every result — the underlying store's keys
// always start with p.base (that's what querying under p.full(prefix)
// guarantees), so TrimPrefix here exactly undoes full() above, and the
// returned keys round-trip through this wrapper's Get unchanged.
func (p prefixStore) List(ctx context.Context, prefix string) ([]string, error) {
	keys, err := p.Store.List(ctx, p.full(prefix))
	if err != nil {
		return nil, err
	}
	out := make([]string, len(keys))
	for i, k := range keys {
		out[i] = strings.TrimPrefix(k, p.base)
	}
	return out, nil
}

// ListDirs mirrors List's rebase-then-strip, for the delimiter-aware listing.
func (p prefixStore) ListDirs(ctx context.Context, prefix string) ([]string, error) {
	dirs, err := p.Store.ListDirs(ctx, p.full(prefix))
	if err != nil {
		return nil, err
	}
	out := make([]string, len(dirs))
	for i, d := range dirs {
		out[i] = strings.TrimPrefix(d, p.base)
	}
	return out, nil
}

// A prefixStore deliberately does NOT forward ListMod: it embeds the Store
// INTERFACE, so only Store's methods are promoted, and a ModLister
// type-assert on a wrapped store fails — callers fall back to List, which
// this wrapper does rebase. No prefix-store caller ranks by recency today;
// forward it here (rebase + strip, like List) before any does.

// KeyMod pairs a listed key with the object's last-modified stamp.
type KeyMod struct {
	Key string
	Mod time.Time
}

// ModLister is the optional Store extension for listings that carry each
// object's last-modified time — information both real stores already have in
// hand (S3 returns it with every ListObjectsV2 entry; DirStore has the file
// mtime) and List throws away. It exists so a caller ranking objects by
// recency can decide WHICH bodies to read before reading any: the app page's
// release list was O(every version ever shipped) in bucket GETs without it.
// Optional rather than a Store method so the many test fakes that implement
// Store stay valid; callers type-assert and fall back to List.
type ModLister interface {
	ListMod(ctx context.Context, prefix string) ([]KeyMod, error)
}

// Both real stores answer ListMod; only fakes fall back.
var (
	_ ModLister = (*DirStore)(nil)
	_ ModLister = (*S3Store)(nil)
)

// ConditionalGetter is the optional conditional read: GetIfChanged is Get
// unless the object still carries etag, in which case it returns
// nil, etag, false, nil without the body (S3 answers 304). An empty etag
// behaves as Get. An absent object returns an os.ErrNotExist-wrapped error.
// Optional and type-asserted, like ModLister: a store without it is read the
// old way.
type ConditionalGetter interface {
	GetIfChanged(ctx context.Context, key, etag string) (body []byte, newETag string, changed bool, err error)
}

// KeyETag pairs a listed key with its ETag.
type KeyETag struct {
	Key  string
	ETag string
}

// ETagLister is the optional listing that carries each object's ETag, sorted
// by key, in the same form Get returns it (S3 keeps its quoting, DirStore its
// content hash), so a listed ETag can be compared with one read earlier.
type ETagLister interface {
	ListETags(ctx context.Context, prefix string) ([]KeyETag, error)
}

// Both real stores and the prefix wrapper answer both; only fakes fall back.
var (
	_ ConditionalGetter = (*DirStore)(nil)
	_ ConditionalGetter = (*S3Store)(nil)
	_ ConditionalGetter = prefixStore{}
	_ ETagLister        = (*DirStore)(nil)
	_ ETagLister        = (*S3Store)(nil)
	_ ETagLister        = prefixStore{}
)

// GetIfChanged rebases the key, like Get. A raw store without the extension
// answers errors.ErrUnsupported, which callers read as "unknown".
func (p prefixStore) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error) {
	if g, ok := p.Store.(ConditionalGetter); ok {
		return g.GetIfChanged(ctx, p.full(key), etag)
	}
	return nil, "", false, errors.ErrUnsupported
}

// ListETags rebases the query prefix and strips the base back off, like List.
func (p prefixStore) ListETags(ctx context.Context, prefix string) ([]KeyETag, error) {
	l, ok := p.Store.(ETagLister)
	if !ok {
		return nil, errors.ErrUnsupported
	}
	got, err := l.ListETags(ctx, p.full(prefix))
	if err != nil {
		return nil, err
	}
	for i := range got {
		got[i].Key = strings.TrimPrefix(got[i].Key, p.base)
	}
	return got, nil
}

// KeySize is one object and its byte size.
type KeySize struct {
	Key  string
	Size int64
}

// SizeLister is the optional sized-listing extension the economics storage
// sweep needs. Like ModLister it is NOT part of Store and NOT forwarded by
// NewPrefixStore — callers type-assert and degrade ("unswept") when the
// backing store cannot answer.
type SizeLister interface {
	ListSizes(ctx context.Context, prefix string) ([]KeySize, error)
}

// Both real stores answer ListSizes; only fakes fall back.
var (
	_ SizeLister = (*DirStore)(nil)
	_ SizeLister = (*S3Store)(nil)
)

// Opener is the optional streaming read. Store.Get returns the whole object
// in memory, which is fine for a pointer and wrong for a platform binary on
// a box under MemoryHigh=70%. Implemented by every production store; a test
// fake that only has Get still works through OpenObject. The etag may be ""
// (DirStore cannot know it without reading the file).
type Opener interface {
	Open(ctx context.Context, key string) (io.ReadCloser, string, error)
}

var (
	_ Opener = (*DirStore)(nil)

	_ Opener = (*S3Store)(nil)
	_ Opener = prefixStore{}
)

// PutStreamer is the optional streaming write: Store.Put takes the whole
// body as []byte, which is wrong for a 130 MB GeoIP database on a box
// under MemoryHigh. Implemented by every production store; a fake with only
// Put still works through PutObjectFrom for bodies up to PutFromFallbackMax.
// ifMatch has Put's exact semantics ("" = create-only).
type PutStreamer interface {
	PutFrom(ctx context.Context, key string, body io.ReadSeeker, size int64, ifMatch string) (string, error)
}

var (
	_ PutStreamer = (*DirStore)(nil)
	_ PutStreamer = (*S3Store)(nil)
	_ PutStreamer = prefixStore{}
)

// PutFromFallbackMax bounds how much PutObjectFrom will buffer for a store
// with no PutFrom.
const PutFromFallbackMax = 64 << 20

// ErrTooLargeForPut is PutObjectFrom refusing to buffer a large body.
var ErrTooLargeForPut = errors.New("store cannot stream and the body is too large to buffer")

// PutObjectFrom streams body into s at key.
func PutObjectFrom(ctx context.Context, s Store, key string, body io.ReadSeeker, size int64, ifMatch string) (string, error) {
	if ps, ok := s.(PutStreamer); ok {
		return ps.PutFrom(ctx, key, body, size, ifMatch)
	}
	if size > PutFromFallbackMax {
		return "", fmt.Errorf("%s: %w", key, ErrTooLargeForPut)
	}
	b, err := io.ReadAll(io.LimitReader(body, size+1))
	if err != nil {
		return "", err
	}
	if int64(len(b)) != size {
		return "", fmt.Errorf("%s: body is %d bytes, want %d", key, len(b), size)
	}
	return s.Put(ctx, key, b, ifMatch)
}

type streamingReadKey struct{}

// OpenObjectStreaming refuses Get fallbacks, including inside wrappers that
// forward this context through OpenObject. Callers must still bound their reads.
func OpenObjectStreaming(ctx context.Context, s Store, key string) (io.ReadCloser, string, error) {
	return OpenObject(context.WithValue(ctx, streamingReadKey{}, true), s, key)
}

// OpenObject streams key from s, buffering through Get when s cannot stream.
func OpenObject(ctx context.Context, s Store, key string) (io.ReadCloser, string, error) {
	if o, ok := s.(Opener); ok {
		return o.Open(ctx, key)
	}
	if required, _ := ctx.Value(streamingReadKey{}).(bool); required {
		return nil, "", errors.ErrUnsupported
	}
	b, etag, err := s.Get(ctx, key)
	if err != nil {
		return nil, "", err
	}
	return io.NopCloser(bytes.NewReader(b)), etag, nil
}

// PageLister is the optional BOUNDED listing. List, ListMod and ListSizes all
// return every key under a prefix, which is right for a sweep that wants the
// whole picture and wrong for a queue drain that wants the oldest few.
//
// The scan queue is the caller. It caps how many items it handles per app per
// tick, and without a bounded listing it would enumerate a flooded app's whole
// queue every 30 seconds regardless of that cap — paying for the listing it
// deliberately declined to act on.
//
// Optional and type-asserted, like ModLister and SizeLister above, NOT a Store
// method: Store has eleven implementations across the real stores, the prefix
// wrapper, fleetapi and seven test fakes, and a new method would break all of
// them for one caller's benefit. Callers degrade to List.
//
// startAfter is EXCLUSIVE and keys come back sorted, so paging is
// "startAfter = the last key I saw". limit <= 0 means no bound.
type PageLister interface {
	ListPage(ctx context.Context, prefix, startAfter string, limit int) ([]string, error)
}

// Both real stores answer ListPage; only fakes fall back.
var (
	_ PageLister = (*DirStore)(nil)
	_ PageLister = (*S3Store)(nil)
)

// ---- DirStore ---------------------------------------------------------------

// DirStore keeps objects as files under a root directory — the test double
// and the single-box no-AWS deployment shape. The etag is the hex sha256 of
// the content (content-derived, so it survives process restarts with no
// sidecar files). CAS is enforced under one process-wide mutex plus
// temp-file + rename writes: single-process fidelity, which is what the
// tests and a one-box deployment need.
type DirStore struct {
	root        string
	mu          sync.Mutex
	changeIndex bool // EnableChangeIndex: bump control/changes.json after each mutation
}

func NewDirStore(root string) *DirStore { return &DirStore{root: root} }

// path maps a bucket key onto the root dir, refusing traversal — a key is
// bucket data, never a filesystem instruction.
func (d *DirStore) path(key string) (string, error) {
	if key == "" || strings.HasPrefix(key, "/") {
		return "", fmt.Errorf("dirstore: invalid key %q", key)
	}
	for _, seg := range strings.Split(key, "/") {
		if seg == "" || seg == "." || seg == ".." {
			return "", fmt.Errorf("dirstore: invalid key %q", key)
		}
	}
	return filepath.Join(d.root, filepath.FromSlash(key)), nil
}

// hashFile is etagOf over a file, streamed.
func hashFile(p string) (string, error) {
	f, err := safefile.OpenFile(p, os.O_RDONLY, 0)
	if err != nil {
		return "", err
	}
	defer f.Close()
	h := sha256.New()
	if _, err := io.Copy(h, f); err != nil {
		return "", err
	}
	return hex.EncodeToString(h.Sum(nil)), nil
}

func etagOf(body []byte) string {
	sum := sha256.Sum256(body)
	return hex.EncodeToString(sum[:])
}

func (d *DirStore) Get(_ context.Context, key string) ([]byte, string, error) {
	p, err := d.path(key)
	if err != nil {
		return nil, "", err
	}
	d.mu.Lock()
	defer d.mu.Unlock()
	b, err := safefile.ReadFile(p)
	if err != nil {
		if errors.Is(err, fs.ErrNotExist) {
			return nil, "", fmt.Errorf("dirstore: %s: %w", key, os.ErrNotExist)
		}
		return nil, "", err
	}
	return b, etagOf(b), nil
}

func (d *DirStore) Open(_ context.Context, key string) (io.ReadCloser, string, error) {
	p, err := d.path(key)
	if err != nil {
		return nil, "", err
	}
	f, err := safefile.OpenFile(p, os.O_RDONLY, 0)
	if err != nil {
		if errors.Is(err, os.ErrNotExist) {
			return nil, "", fmt.Errorf("dirstore: %s: %w", key, os.ErrNotExist)
		}
		return nil, "", err
	}
	return f, "", nil
}

// EnableChangeIndex makes every successful Put, PutFrom, Delete and
// DeleteIfMatch inside a change scope bump control/changes.json afterwards
// (design §3.3). Call it before the store is shared; it is not synchronised.
func (d *DirStore) EnableChangeIndex() { d.changeIndex = true }

func (d *DirStore) bump(ctx context.Context, key string) {
	bumpAfterMutation(ctx, d.changeIndex, key, d.Get, d.putRaw)
}

func (d *DirStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	etag, err := d.putRaw(ctx, key, body, ifMatch)
	if err == nil {
		d.bump(ctx, key)
	}
	return etag, err
}

func (d *DirStore) putRaw(_ context.Context, key string, body []byte, ifMatch string) (string, error) {
	p, err := d.path(key)
	if err != nil {
		return "", err
	}
	d.mu.Lock()
	defer d.mu.Unlock()
	// The CAS gate, mirroring S3's semantics exactly: "" = create-only,
	// non-empty = the object must still carry that etag. Stat and a
	// streamed hash, as in PutFrom, so an oversized object at the key is
	// never read into memory.
	_, err = os.Stat(p)
	exists := err == nil
	if err != nil && !errors.Is(err, fs.ErrNotExist) {
		return "", err
	}
	if ifMatch == "" && exists {
		return "", fmt.Errorf("dirstore: %s exists: %w", key, ErrCASConflict)
	}
	if ifMatch != "" {
		cur := ""
		if exists {
			if cur, err = hashFile(p); err != nil {
				return "", err
			}
		}
		if !exists || cur != ifMatch {
			return "", fmt.Errorf("dirstore: %s moved on: %w", key, ErrCASConflict)
		}
	}
	if err := safefile.MkdirAll(filepath.Dir(p), 0o755); err != nil {
		return "", err
	}
	if err := safefile.WriteFile(p, body, 0o644); err != nil {
		return "", err
	}
	return etagOf(body), nil
}

// PutFrom is Put streamed: the body goes to a temp file (hashed as it
// goes, so the etag is the same content sha256 Put would give) and is
// renamed into place under d.mu after the same CAS gate.
func (d *DirStore) PutFrom(ctx context.Context, key string, body io.ReadSeeker, size int64, ifMatch string) (string, error) {
	etag, err := d.putFromRaw(ctx, key, body, size, ifMatch)
	if err == nil {
		d.bump(ctx, key)
	}
	return etag, err
}

func (d *DirStore) putFromRaw(_ context.Context, key string, body io.ReadSeeker, size int64, ifMatch string) (string, error) {
	p, err := d.path(key)
	if err != nil {
		return "", err
	}
	if err := safefile.MkdirAll(filepath.Dir(p), 0o755); err != nil {
		return "", err
	}
	d.mu.Lock()
	defer d.mu.Unlock()
	// The CAS gate without reading the existing object into memory (it may
	// be 1 GiB): Stat for create-only, a streamed hash for replace-only.
	_, err = os.Stat(p)
	exists := err == nil
	if err != nil && !errors.Is(err, fs.ErrNotExist) {
		return "", err
	}
	if ifMatch == "" && exists {
		return "", fmt.Errorf("dirstore: %s exists: %w", key, ErrCASConflict)
	}
	if ifMatch != "" {
		cur := ""
		if exists {
			if cur, err = hashFile(p); err != nil {
				return "", err
			}
		}
		if !exists || cur != ifMatch {
			return "", fmt.Errorf("dirstore: %s moved on: %w", key, ErrCASConflict)
		}
	}
	h := sha256.New()
	err = safefile.Write(p, 0o644, func(w io.Writer) error {
		n, err := io.Copy(io.MultiWriter(w, h), io.LimitReader(body, size+1))
		if err != nil {
			return err
		}
		if n != size {
			return fmt.Errorf("dirstore: %s: body is %d bytes, want %d", key, n, size)
		}
		return nil
	})
	if err != nil {
		return "", err
	}
	return hex.EncodeToString(h.Sum(nil)), nil
}

// List is ListMod minus the mtimes — one walk owns the prefix filter and the
// .tmp exclusion, so the two listings cannot drift.
func (d *DirStore) List(ctx context.Context, prefix string) ([]string, error) {
	kms, err := d.ListMod(ctx, prefix)
	out := make([]string, len(kms))
	for i, km := range kms {
		out[i] = km.Key
	}
	return out, err
}

// ListPage is the bounded listing. It leans on ListMod being key-sorted, so
// "after" and "limit" are a filter and a slice rather than a second walk.
func (d *DirStore) ListPage(ctx context.Context, prefix, startAfter string, limit int) ([]string, error) {
	kms, err := d.ListMod(ctx, prefix)
	if err != nil {
		return nil, err
	}
	out := make([]string, 0, len(kms))
	for _, km := range kms {
		if km.Key <= startAfter {
			continue
		}
		out = append(out, km.Key)
		if limit > 0 && len(out) == limit {
			break
		}
	}
	return out, nil
}

// ListMod is the listing with each file's mtime alongside, key-sorted.
func (d *DirStore) ListMod(_ context.Context, prefix string) ([]KeyMod, error) {
	d.mu.Lock()
	defer d.mu.Unlock()
	var out []KeyMod
	err := filepath.WalkDir(d.root, func(p string, de fs.DirEntry, err error) error {
		if err != nil {
			if errors.Is(err, fs.ErrNotExist) {
				return nil // an empty store lists empty, like an empty bucket
			}
			return err
		}
		if de.IsDir() && strings.HasPrefix(de.Name(), ".carlos-write-") {
			return filepath.SkipDir
		}
		if de.IsDir() || strings.HasSuffix(p, ".tmp") {
			return nil
		}
		rel, err := filepath.Rel(d.root, p)
		if err != nil {
			return err
		}
		key := filepath.ToSlash(rel)
		if !strings.HasPrefix(key, prefix) {
			return nil
		}
		info, err := de.Info()
		if err != nil {
			// Vanished between readdir and stat — another process's delete
			// racing this walk (in-process writers all hold d.mu). A file
			// that is gone is not listed; it does not break the listing.
			// Untestable deterministically, so pinned by comment alone.
			if errors.Is(err, fs.ErrNotExist) {
				return nil
			}
			return err
		}
		out = append(out, KeyMod{Key: key, Mod: info.ModTime()})
		return nil
	})
	sort.Slice(out, func(i, j int) bool { return out[i].Key < out[j].Key })
	return out, err
}

// GetIfChanged is Get, except an object whose content hash is etag is not
// returned.
func (d *DirStore) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error) {
	b, e, err := d.Get(ctx, key)
	if err != nil {
		return nil, "", false, err
	}
	if etag != "" && e == etag {
		return nil, etag, false, nil
	}
	return b, e, true, nil
}

// ListETags is ListMod's walk with each file's content hash alongside,
// key-sorted. The hash is the one Get returns.
func (d *DirStore) ListETags(_ context.Context, prefix string) ([]KeyETag, error) {
	d.mu.Lock()
	defer d.mu.Unlock()
	var out []KeyETag
	err := filepath.WalkDir(d.root, func(p string, de fs.DirEntry, err error) error {
		if err != nil {
			if errors.Is(err, fs.ErrNotExist) {
				return nil // an empty store lists empty, like an empty bucket
			}
			return err
		}
		if de.IsDir() && strings.HasPrefix(de.Name(), ".carlos-write-") {
			return filepath.SkipDir
		}
		if de.IsDir() || strings.HasSuffix(p, ".tmp") {
			return nil
		}
		rel, err := filepath.Rel(d.root, p)
		if err != nil {
			return err
		}
		key := filepath.ToSlash(rel)
		if !strings.HasPrefix(key, prefix) {
			return nil
		}
		etag, err := hashFile(p)
		if err != nil {
			if errors.Is(err, fs.ErrNotExist) {
				return nil // deleted by another process mid-walk: not listed
			}
			return err
		}
		out = append(out, KeyETag{Key: key, ETag: etag})
		return nil
	})
	sort.Slice(out, func(i, j int) bool { return out[i].Key < out[j].Key })
	return out, err
}

// ListSizes mirrors ListMod's walk, collecting each file's byte size instead
// of its mtime.
func (d *DirStore) ListSizes(_ context.Context, prefix string) ([]KeySize, error) {
	d.mu.Lock()
	defer d.mu.Unlock()
	var out []KeySize
	err := filepath.WalkDir(d.root, func(p string, de fs.DirEntry, err error) error {
		if err != nil {
			if errors.Is(err, fs.ErrNotExist) {
				return nil // an empty store lists empty, like an empty bucket
			}
			return err
		}
		if de.IsDir() && strings.HasPrefix(de.Name(), ".carlos-write-") {
			return filepath.SkipDir
		}
		if de.IsDir() || strings.HasSuffix(p, ".tmp") {
			return nil
		}
		rel, err := filepath.Rel(d.root, p)
		if err != nil {
			return err
		}
		key := filepath.ToSlash(rel)
		if !strings.HasPrefix(key, prefix) {
			return nil
		}
		info, err := de.Info()
		if err != nil {
			// Vanished between readdir and stat — another process's delete
			// racing this walk (in-process writers all hold d.mu). A file
			// that is gone is not listed; it does not break the listing.
			// Untestable deterministically, so pinned by comment alone.
			if errors.Is(err, fs.ErrNotExist) {
				return nil
			}
			return err
		}
		out = append(out, KeySize{Key: key, Size: info.Size()})
		return nil
	})
	sort.Slice(out, func(i, j int) bool { return out[i].Key < out[j].Key })
	return out, err
}

// ListDirs is DirStore's delimiter-aware listing: one directory read, no
// walk. A missing directory lists empty, like a bucket prefix with nothing
// under it. Only directories are returned, so a file sitting at this level
// (ledgers.json among them) is never mistaken for a child prefix.
func (d *DirStore) ListDirs(_ context.Context, prefix string) ([]string, error) {
	if prefix != "" && !strings.HasSuffix(prefix, "/") {
		prefix += "/"
	}
	dir := d.root
	if trimmed := strings.TrimSuffix(prefix, "/"); trimmed != "" {
		p, err := d.path(trimmed)
		if err != nil {
			return nil, err
		}
		dir = p
	}
	d.mu.Lock()
	defer d.mu.Unlock()
	ents, err := os.ReadDir(dir)
	if err != nil {
		if errors.Is(err, fs.ErrNotExist) {
			return nil, nil
		}
		return nil, err
	}
	var out []string
	for _, e := range ents {
		if !e.IsDir() || strings.HasPrefix(e.Name(), ".carlos-write-") {
			continue
		}
		out = append(out, prefix+e.Name()+"/")
	}
	sort.Strings(out)
	return out, nil
}

func (d *DirStore) Delete(ctx context.Context, key string) error {
	err := d.deleteRaw(ctx, key)
	if err == nil {
		d.bump(ctx, key)
	}
	return err
}

func (d *DirStore) deleteRaw(_ context.Context, key string) error {
	p, err := d.path(key)
	if err != nil {
		return err
	}
	d.mu.Lock()
	defer d.mu.Unlock()
	if err := safefile.Remove(p); err != nil {
		if errors.Is(err, os.ErrNotExist) {
			return nil // idempotent: missing key is not an error
		}
		return err
	}
	return nil
}

// ---- S3Store ----------------------------------------------------------------

// S3Store is the production bucket via aws-sdk-go-v2, with the conditional
// headers doing the CAS work server-side. Compiles always; exercised live
// only when CARLOS_TEST_S3_BUCKET is set (see release_test.go).
type S3Store struct {
	Client *awss3.Client
	Bucket string

	changeIndex bool // EnableChangeIndex: bump control/changes.json after each mutation
}

// NewS3Store uses the standard AWS chain for non-root callers. Root never loads
// shared profiles; rootaws admits only safe static off-cloud credentials files.
// Everything that makes a non-AWS store work — CARLOS_S3_ENDPOINT, path-style
// addressing, when-required checksums, and any per-provider quirk — now lives
// in internal/s3compat, so this store and the hibernation lease cannot drift
// apart against the same bucket. See that package for the quirk table.
func NewS3Store(ctx context.Context, bucket string) (*S3Store, error) {
	return newS3Store(ctx, bucket)
}

// NewS3StoreIn is NewS3Store with the bucket's region pinned: a box's or a
// console's data bucket is named with its region beside it
// (CARLOS_HIBERNATE_REGION), and the default chain's region may be
// another. An empty region is NewS3Store.
func NewS3StoreIn(ctx context.Context, bucket, region string) (*S3Store, error) {
	if region == "" {
		return newS3Store(ctx, bucket)
	}
	return newS3Store(ctx, bucket, awsconfig.WithRegion(region))
}

func newS3Store(ctx context.Context, bucket string, opts ...func(*awsconfig.LoadOptions) error) (*S3Store, error) {
	cfg, err := rootaws.LoadDefaultConfig(ctx, opts...)
	if err != nil {
		return nil, fmt.Errorf("s3store: load aws config: %w", err)
	}
	client := awss3.NewFromConfig(cfg, s3compat.Options(), func(o *awss3.Options) {
		// s3compat sets these too, but only under a custom endpoint. This
		// store has always set them on EVERY client including stock AWS, and
		// the flagship runs on stock AWS — so the unconditional form stays
		// here rather than becoming a silent behaviour change on the one
		// deployment that cannot be tested from a compatible store.
		o.RequestChecksumCalculation = aws.RequestChecksumCalculationWhenRequired
		o.ResponseChecksumValidation = aws.ResponseChecksumValidationWhenRequired
	})
	return &S3Store{Client: client, Bucket: bucket}, nil
}

// s3NotFound folds the shapes a missing object answers with (NoSuchKey on
// GET, NotFound on HEAD, bare 404s from S3-compatibles) into one test.
func s3NotFound(err error) bool {
	var apiErr smithy.APIError
	if errors.As(err, &apiErr) {
		switch apiErr.ErrorCode() {
		case "NoSuchKey", "NotFound", "404":
			return true
		}
	}
	return false
}

// s3CASConflict reports whether an error is a conditional-write loss: 412
// (If-Match / If-None-Match precondition failed) or 409 (S3's
// ConditionalRequestConflict when two conditional writes race). Matched by
// both the S3 error code and the raw HTTP status, so an S3-compatible that
// only sets one of them still classifies correctly.
func s3CASConflict(err error) bool {
	var apiErr smithy.APIError
	if errors.As(err, &apiErr) {
		switch apiErr.ErrorCode() {
		case "PreconditionFailed", "ConditionalRequestConflict":
			return true
		}
	}
	var httpErr interface{ HTTPStatusCode() int }
	if errors.As(err, &httpErr) {
		switch httpErr.HTTPStatusCode() {
		case http.StatusPreconditionFailed, http.StatusConflict:
			return true
		}
	}
	return false
}

func (s *S3Store) Get(ctx context.Context, key string) ([]byte, string, error) {
	out, err := s.Client.GetObject(ctx, &awss3.GetObjectInput{Bucket: &s.Bucket, Key: &key})
	if err != nil {
		if s3NotFound(err) {
			return nil, "", fmt.Errorf("s3store: %s: %w", key, os.ErrNotExist)
		}
		return nil, "", err
	}
	defer out.Body.Close()
	b, err := io.ReadAll(out.Body)
	if err != nil {
		return nil, "", err
	}
	return b, aws.ToString(out.ETag), nil
}

func (s *S3Store) Open(ctx context.Context, key string) (io.ReadCloser, string, error) {
	out, err := s.Client.GetObject(ctx, &awss3.GetObjectInput{Bucket: &s.Bucket, Key: &key})
	if err != nil {
		if s3NotFound(err) {
			return nil, "", fmt.Errorf("s3store: %s: %w", key, os.ErrNotExist)
		}
		return nil, "", err
	}
	return out.Body, aws.ToString(out.ETag), nil
}

// EnableChangeIndex makes every successful Put, PutFrom, Delete and
// DeleteIfMatch inside a change scope bump control/changes.json afterwards
// (design §3.3). Call it before the store is shared; it is not synchronised.
func (s *S3Store) EnableChangeIndex() { s.changeIndex = true }

func (s *S3Store) bump(ctx context.Context, key string) {
	bumpAfterMutation(ctx, s.changeIndex, key, s.Get, s.putRaw)
}

// NewDeploymentS3Store is NewS3Store for the deployment bucket: the returned
// store bumps the change index after each mutation. Stores over any other
// bucket keep using NewS3Store.
func NewDeploymentS3Store(ctx context.Context, bucket string) (*S3Store, error) {
	s, err := NewS3Store(ctx, bucket)
	if err != nil {
		return nil, err
	}
	s.EnableChangeIndex()
	return s, nil
}

// NewDeploymentS3StoreIn is NewS3StoreIn with the change index enabled.
func NewDeploymentS3StoreIn(ctx context.Context, bucket, region string) (*S3Store, error) {
	s, err := NewS3StoreIn(ctx, bucket, region)
	if err != nil {
		return nil, err
	}
	s.EnableChangeIndex()
	return s, nil
}

func (s *S3Store) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	etag, err := s.putRaw(ctx, key, body, ifMatch)
	if err == nil {
		s.bump(ctx, key)
	}
	return etag, err
}

func (s *S3Store) putRaw(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	in := &awss3.PutObjectInput{
		Bucket:        &s.Bucket,
		Key:           &key,
		Body:          bytes.NewReader(body),
		ContentLength: aws.Int64(int64(len(body))),
	}
	if ifMatch == "" {
		in.IfNoneMatch = aws.String("*") // create-only: 412 if someone created it first
	} else {
		in.IfMatch = aws.String(ifMatch) // replace-only: 412 if the object moved on
	}
	out, err := s.Client.PutObject(ctx, in)
	if err != nil {
		// A replace-only write whose object has been DELETED under it comes
		// back 404/NoSuchKey, not 412: there is no etag left to fail the
		// precondition against. That is still a lost conditional write — the
		// caller decided this body from bytes that no longer exist — so it
		// classifies as a CAS conflict rather than escaping as an opaque
		// transport error, which is what made a concurrently-deleted instance
		// record answer 502 ("could not update — try again", forever) instead
		// of the 409 the door already handles. Scoped to the ifMatch branch on
		// purpose: on a create-only write (IfNoneMatch:*) a 404 is not an
		// expected answer at all and must not be masked into a conflict.
		if s3CASConflict(err) || (ifMatch != "" && s3NotFound(err)) {
			return "", fmt.Errorf("s3store: %s: %w", key, ErrCASConflict)
		}
		return "", err
	}
	return aws.ToString(out.ETag), nil
}

// PutFrom is Put with a streamed, seekable body and an explicit length.
func (s *S3Store) PutFrom(ctx context.Context, key string, body io.ReadSeeker, size int64, ifMatch string) (string, error) {
	etag, err := s.putFromRaw(ctx, key, body, size, ifMatch)
	if err == nil {
		s.bump(ctx, key)
	}
	return etag, err
}

func (s *S3Store) putFromRaw(ctx context.Context, key string, body io.ReadSeeker, size int64, ifMatch string) (string, error) {
	in := &awss3.PutObjectInput{
		Bucket:        &s.Bucket,
		Key:           &key,
		Body:          body,
		ContentLength: aws.Int64(size),
	}
	if ifMatch == "" {
		in.IfNoneMatch = aws.String("*")
	} else {
		in.IfMatch = aws.String(ifMatch)
	}
	out, err := s.Client.PutObject(ctx, in)
	if err != nil {
		if s3CASConflict(err) || (ifMatch != "" && s3NotFound(err)) {
			return "", fmt.Errorf("s3store: %s: %w", key, ErrCASConflict)
		}
		return "", err
	}
	return aws.ToString(out.ETag), nil
}

// List is ListMod minus the stamps — one paginator loop serves both.
func (s *S3Store) List(ctx context.Context, prefix string) ([]string, error) {
	kms, err := s.ListMod(ctx, prefix)
	if err != nil {
		return nil, err
	}
	out := make([]string, len(kms))
	for i, km := range kms {
		out[i] = km.Key
	}
	return out, nil
}

// ListPage is the bounded listing, and unlike its siblings it does NOT
// paginate: one request, at most limit keys, which is the whole point.
//
// S3 returns keys in lexicographic order and StartAfter is exclusive, so this
// maps onto the API directly.
func (s *S3Store) ListPage(ctx context.Context, prefix, startAfter string, limit int) ([]string, error) {
	in := &awss3.ListObjectsV2Input{Bucket: &s.Bucket, Prefix: &prefix}
	if startAfter != "" {
		in.StartAfter = &startAfter
	}
	if limit <= 0 {
		// "No bound" has to MEAN no bound. One ListObjectsV2 returns at most
		// a thousand keys and reports IsTruncated; a version of this that
		// ignored that would answer a caller asking for everything with the
		// first page and no hint there was more — and DirStore, which really
		// does return everything, would disagree with it silently. The
		// bounded case below is deliberately still one request: that is the
		// whole point of the interface.
		var keys []string
		p := awss3.NewListObjectsV2Paginator(s.Client, in)
		for p.HasMorePages() {
			page, err := p.NextPage(ctx)
			if err != nil {
				return nil, err
			}
			for _, o := range page.Contents {
				keys = append(keys, aws.ToString(o.Key))
			}
		}
		return keys, nil
	}
	n := int32(limit)
	in.MaxKeys = &n
	out, err := s.Client.ListObjectsV2(ctx, in)
	if err != nil {
		return nil, err
	}
	keys := make([]string, 0, len(out.Contents))
	for _, o := range out.Contents {
		keys = append(keys, aws.ToString(o.Key))
	}
	return keys, nil
}

// ListMod is the listing keeping the LastModified S3 already sends with
// every entry.
func (s *S3Store) ListMod(ctx context.Context, prefix string) ([]KeyMod, error) {
	var out []KeyMod
	p := awss3.NewListObjectsV2Paginator(s.Client, &awss3.ListObjectsV2Input{
		Bucket: &s.Bucket,
		Prefix: &prefix,
	})
	for p.HasMorePages() {
		page, err := p.NextPage(ctx)
		if err != nil {
			return nil, err
		}
		for _, o := range page.Contents {
			out = append(out, KeyMod{Key: aws.ToString(o.Key), Mod: aws.ToTime(o.LastModified)})
		}
	}
	return out, nil
}

// s3NotModified recognises the 304 a conditional GET answers with. The SDK
// surfaces it as an error: by HTTP status, or by the NotModified code an
// S3-compatible may set instead.
func s3NotModified(err error) bool {
	var httpErr interface{ HTTPStatusCode() int }
	if errors.As(err, &httpErr) && httpErr.HTTPStatusCode() == http.StatusNotModified {
		return true
	}
	var apiErr smithy.APIError
	if errors.As(err, &apiErr) {
		switch apiErr.ErrorCode() {
		case "NotModified", "304":
			return true
		}
	}
	return false
}

// GetIfChanged sends If-None-Match, so an unchanged object costs a 304 and no
// body.
func (s *S3Store) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error) {
	in := &awss3.GetObjectInput{Bucket: &s.Bucket, Key: &key}
	if etag != "" {
		in.IfNoneMatch = &etag
	}
	out, err := s.Client.GetObject(ctx, in)
	if err != nil {
		if etag != "" && s3NotModified(err) {
			return nil, etag, false, nil
		}
		if s3NotFound(err) {
			return nil, "", false, fmt.Errorf("s3store: %s: %w", key, os.ErrNotExist)
		}
		return nil, "", false, err
	}
	defer out.Body.Close()
	b, err := io.ReadAll(out.Body)
	if err != nil {
		return nil, "", false, err
	}
	return b, aws.ToString(out.ETag), true, nil
}

// ListETags pages ListObjectsV2 and keeps each ETag exactly as the listing
// sends it, which is the form GetObject reports.
func (s *S3Store) ListETags(ctx context.Context, prefix string) ([]KeyETag, error) {
	var out []KeyETag
	p := awss3.NewListObjectsV2Paginator(s.Client, &awss3.ListObjectsV2Input{
		Bucket: &s.Bucket,
		Prefix: &prefix,
	})
	for p.HasMorePages() {
		page, err := p.NextPage(ctx)
		if err != nil {
			return nil, err
		}
		for _, o := range page.Contents {
			out = append(out, KeyETag{Key: aws.ToString(o.Key), ETag: aws.ToString(o.ETag)})
		}
	}
	return out, nil
}

// ListSizes mirrors ListMod's paginator loop, collecting each object's byte
// size instead of its LastModified stamp.
func (s *S3Store) ListSizes(ctx context.Context, prefix string) ([]KeySize, error) {
	var out []KeySize
	p := awss3.NewListObjectsV2Paginator(s.Client, &awss3.ListObjectsV2Input{
		Bucket: &s.Bucket,
		Prefix: &prefix,
	})
	for p.HasMorePages() {
		page, err := p.NextPage(ctx)
		if err != nil {
			return nil, err
		}
		for _, o := range page.Contents {
			out = append(out, KeySize{Key: aws.ToString(o.Key), Size: aws.ToInt64(o.Size)})
		}
	}
	return out, nil
}

// BucketShapedListDirs: S3Store IS the deployment bucket, and its ListDirs
// is the delimiter listing of it.
func (s *S3Store) BucketShapedListDirs() bool { return true }

// ListDirs is the Delimiter:"/" listing: S3 rolls everything below one level
// up into a CommonPrefix, so the response is bounded by the number of child
// prefixes rather than by the total object count beneath them. Objects AT
// this level come back in Contents, which this deliberately ignores.
func (s *S3Store) ListDirs(ctx context.Context, prefix string) ([]string, error) {
	if prefix != "" && !strings.HasSuffix(prefix, "/") {
		prefix += "/"
	}
	var out []string
	p := awss3.NewListObjectsV2Paginator(s.Client, &awss3.ListObjectsV2Input{
		Bucket:    &s.Bucket,
		Prefix:    &prefix,
		Delimiter: aws.String("/"),
	})
	for p.HasMorePages() {
		page, err := p.NextPage(ctx)
		if err != nil {
			return nil, err
		}
		for _, cp := range page.CommonPrefixes {
			out = append(out, aws.ToString(cp.Prefix))
		}
	}
	sort.Strings(out)
	return out, nil
}

func (s *S3Store) Delete(ctx context.Context, key string) error {
	err := s.deleteRaw(ctx, key)
	if err == nil {
		s.bump(ctx, key)
	}
	return err
}

func (s *S3Store) deleteRaw(ctx context.Context, key string) error {
	_, err := s.Client.DeleteObject(ctx, &awss3.DeleteObjectInput{
		Bucket: aws.String(s.Bucket),
		Key:    aws.String(key),
	})
	if err != nil {
		return err
	}
	return nil
}