package lease

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/url"
	"sort"
	"strings"
	"sync"

	awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
	s3types "github.com/aws/aws-sdk-go-v2/service/s3/types"
	"github.com/aws/smithy-go"
)

// The staging mirror (issue #331, instance-move spec): a residence move
// copies the instance's REPLICA prefix — the litestream generations the
// activator already maintains — never the database itself. MirrorPrefix is
// the one delta round; the move worker calls it repeatedly until a round
// copies little enough that the cut's own round is cheap, and once more
// after the child is down.
//
// The interface exists for the tests (an in-memory fake) and because the
// destination bucket lives in ANOTHER region: the implementation holds one
// client per side.

// MirrorStore is the S3 actions a delta round needs. List answers
// key→object under a prefix (keys RELATIVE to the bucket root, not the
// prefix). Copy is a server-side CopyObject, conditional on srcETag when it
// is set: a source that no longer carries that ETag answers
// ErrCopySourceChanged and copies nothing. It returns the ETag the
// destination object landed with. Delete removes one key. Get and Put read
// and write one small whole object (the copy receipts); Get answers
// ErrMirrorNotFound for a missing key.
type MirrorStore interface {
	List(ctx context.Context, bucket, prefix string) (map[string]MirrorObject, error)
	Copy(ctx context.Context, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (dstETag string, err error)
	Delete(ctx context.Context, bucket, key string) error
	Get(ctx context.Context, bucket, key string) ([]byte, error)
	Put(ctx context.Context, bucket, key string, body []byte) error
}

// VersionedCopier is the optional half of MirrorStore a versioned bucket
// supports: the copy's response carries the destination's version id,
// which a move's seal records (moves spec §7.3, "seal by version"). The
// source never lists the target's staging, so the response is the only
// place the id can come from. Only the staging mirror (Task 6,
// lease.NewStagingMirror, through Task 14's guardedMirror) implements it;
// the ordinary s3Mirror and every plain MirrorStore record no versions and
// MirrorRound.Versions stays empty, so production mirroring (NewS3Mirror,
// the legacy worker) is unchanged (round-4 finding 7).
type VersionedCopier interface {
	CopyVersioned(ctx context.Context, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (etag, versionID string, err error)
}

// VersionedPutter is the same for Put: the archive, the timers snapshot
// and move.json are sealed by the version their PUT returned.
type VersionedPutter interface {
	PutVersioned(ctx context.Context, bucket, key string, body []byte) (versionID string, err error)
}

// ErrCopySourceChanged is a conditional Copy whose source was overwritten
// after the round listed it. The round defers that object to the next
// round rather than failing: a source that keeps changing is a busy
// source, not a broken one.
var ErrCopySourceChanged = errors.New("mirror: copy source changed since it was listed")

// ErrMirrorNotFound is Get's answer for a key that does not exist.
var ErrMirrorNotFound = errors.New("mirror: no such object")

// MirrorObject is one listed object: the ETag the sameness test compares,
// and the size a round's byte accounting adds up.
type MirrorObject struct {
	ETag string
	Size int64
}

// MirrorRound is what one delta round did.
type MirrorRound struct {
	// Changed counts objects copied, deferred or deleted; 0 means the
	// sides were identical when listed.
	Changed int
	// BytesCopied sums the listed SOURCE size of every object copied.
	BytesCopied int64
	// Deferred counts objects whose conditional copy found the source
	// overwritten since the listing (ErrCopySourceChanged); BytesDeferred
	// sums their listed sizes. They are still to copy, so a round that
	// has to be byte-final (the cut's) must not count as done while
	// Deferred > 0.
	Deferred      int
	BytesDeferred int64
	// Versions maps a relative key to the destination version id of the
	// copy that produced its current destination object, for every current
	// source key whose version is known (a copy this round, or a receipt
	// that still vouches for the destination). Non-nil only when the store
	// is a VersionedCopier; a plain MirrorStore records none.
	Versions map[string]string
}

// DeltaBytes is how much the round found to move: copied plus deferred.
// The move worker's readiness bar measures this. An awake instance never
// reaches Changed == 0 (litestream writes new files and compaction
// rewrites old ones between any two listings), but the bytes a round had
// to move bound the work left for the cut's round.
func (r MirrorRound) DeltaBytes() int64 { return r.BytesCopied + r.BytesDeferred }

// Receipts names the object where a mirror keeps its copy receipts. The
// zero value keeps none, and every object whose ETags differ is copied.
//
// Why receipts exist (#175): S3's CopyObject always writes a single-part
// object, so copying a MULTIPART source (litestream uploads anything over
// one 5 MiB part that way) lands a destination whose ETag differs from the
// source's. ETag equality then reports every multipart object as changed
// on every round, forever: production recopied six 138 MB level files a
// round. A receipt records what one copy did: the source ETag and size it
// copied, and the ETag the destination landed with. An object is skipped
// when the source still lists that ETag and size AND the destination still
// lists that destination ETag, and copied otherwise.
//
// The receipts object must live where the mirroring box can READ it. A
// move's destination is push-only to the source box (it may list, put and
// delete under moves/*, never get), so the move worker keeps them in the
// SOURCE's own bucket, outside every prefix a restore or a promotion
// reads. They are never data.
type Receipts struct {
	Bucket, Key string
}

// receiptBook is the receipts object's shape. It names the pair it
// describes, and a book written for another pair is not evidence about
// this one.
type receiptBook struct {
	V         int                    `json:"v"`
	SrcBucket string                 `json:"src_bucket"`
	SrcPrefix string                 `json:"src_prefix"`
	DstBucket string                 `json:"dst_bucket"`
	DstPrefix string                 `json:"dst_prefix"`
	Objects   map[string]copyReceipt `json:"objects"` // relative key → receipt
}

// copyReceipt is one completed copy whose destination ETag differs from
// its source's, or (on a versioned store) one whose version id is known.
// Otherwise, where the ETags already match, the plain equality test needs
// no help.
type copyReceipt struct {
	SrcETag string `json:"src_etag"`
	SrcSize int64  `json:"src_size"`
	DstETag string `json:"dst_etag"`
	// DstVersionID is the destination version the copy returned. Written
	// only by a VersionedCopier; a versioned round skips an object only
	// when a receipt carrying one still vouches for it.
	DstVersionID string `json:"dst_version_id,omitempty"`
}

// MirrorPrefix makes dstBucket/dstPrefix an exact mirror of
// srcBucket/srcPrefix: copy keys that are new or changed, delete keys the
// source no longer has (litestream retention must hold in staging too: a
// restore that replays pruned WAL segments is a restore nobody tested).
// Returns what the round did; on error, what it did before the failure.
//
// "Unchanged" is exact, never a guess from size: either the two listed
// ETags are equal (a single-part source, whose copy keeps its ETag), or a
// receipt (see Receipts) proves the destination is this mirror's copy of
// exactly the source version now listed.
//
// Crash safety: a receipt is written only AFTER the copy it describes
// returned, carrying the ETag that copy returned, once per round at the
// end. A crash before that write loses receipts, and the next round copies
// those objects once more. It can never produce a wrong skip, because a
// skip needs the destination to still list the ETag the copy produced.
func MirrorPrefix(ctx context.Context, m MirrorStore, srcBucket, srcPrefix, dstBucket, dstPrefix string, rc Receipts) (MirrorRound, error) {
	return MirrorPrefixKeep(ctx, m, srcBucket, srcPrefix, dstBucket, dstPrefix, rc, nil)
}

// MirrorPrefixKeep is MirrorPrefix over the source keys keep admits: keep
// is asked each source key relative to srcPrefix, and a key it refuses is
// treated as absent, so it is never copied and a destination copy of it is
// deleted. A move's lane (moves.ReplicaSource) mirrors one generation's
// LTX files and nothing else this way. A nil keep admits every key, which
// is MirrorPrefix.
func MirrorPrefixKeep(ctx context.Context, m MirrorStore, srcBucket, srcPrefix, dstBucket, dstPrefix string, rc Receipts, keep func(rel string) bool) (MirrorRound, error) {
	// Both prefixes are DIRECTORY boundaries, normalized here to end in
	// "/": activator.HibernatePrefix carries none (path.Join), MovesPrefix
	// carries one, and relative-key math over a slashless prefix would both
	// mint fused keys (instances/hgenerations/…) and let a listing bleed
	// into a sibling host whose name merely extends this one's.
	if !strings.HasSuffix(srcPrefix, "/") {
		srcPrefix += "/"
	}
	if !strings.HasSuffix(dstPrefix, "/") {
		dstPrefix += "/"
	}
	src, err := m.List(ctx, srcBucket, srcPrefix)
	if err != nil {
		return MirrorRound{}, fmt.Errorf("mirror: list source %s/%s: %w", srcBucket, srcPrefix, err)
	}
	if keep != nil {
		for k := range src {
			if !keep(k[len(srcPrefix):]) {
				delete(src, k)
			}
		}
	}
	dst, err := m.List(ctx, dstBucket, dstPrefix)
	if err != nil {
		return MirrorRound{}, fmt.Errorf("mirror: list staging %s/%s: %w", dstBucket, dstPrefix, err)
	}
	book := receiptLedger{m: m, rc: rc, want: receiptBook{V: 1, SrcBucket: srcBucket, SrcPrefix: srcPrefix, DstBucket: dstBucket, DstPrefix: dstPrefix}}
	// Deterministic order: a failed round resumes the same way everywhere.
	rel := make([]string, 0, len(src))
	for k := range src {
		rel = append(rel, k[len(srcPrefix):])
	}
	sort.Strings(rel)
	var round MirrorRound
	// Version collection is decided once, by the store: a VersionedCopier
	// is in receipt-required mode, any other store keeps the plain rules.
	vc, versioned := m.(VersionedCopier)
	if versioned {
		round.Versions = map[string]string{}
	}
	// The plan: which objects this round copies. Deciding a skip reads only
	// the two listings and the receipts, so it is done here, in key order,
	// before any copy starts; the copies then run side by side.
	var jobs []string
	for _, r := range rel {
		// The lease's lock.json is COORDINATION state, not data (live
		// rehearsal catch, 2026-08-29): the renewer rewrites it on every
		// TTL renewal, so mirroring it would put a change in every round
		// for any awake instance — and promoting it would hand the target a
		// stale foreign lease. Skipped on both sides: never copied, and
		// the destination's own copy (if any) is never deleted.
		if r == leaseLockName {
			continue
		}
		s := src[srcPrefix+r]
		d, staged := dst[dstPrefix+r]
		if staged && d.ETag == s.ETag && !versioned {
			continue
		}
		if staged {
			rcpt, ok, err := book.get(ctx, r)
			if err != nil {
				return round, err
			}
			// A versioned round must be able to name the version it would
			// seal, so a matching receipt has to carry one; without it the
			// object is recopied (a fresh version id; the stale version
			// stays unreferenced).
			// A receipt naming "null" (UsableVersionID) is no better than
			// none: a later write overwrote that version.
			if ok && rcpt.SrcETag == s.ETag && rcpt.SrcSize == s.Size && rcpt.DstETag == d.ETag && d.Size == s.Size && (!versioned || UsableVersionID(rcpt.DstVersionID)) {
				if versioned {
					round.Versions[r] = rcpt.DstVersionID
				}
				continue
			}
		}
		jobs = append(jobs, r)
	}
	// The copies, up to mirrorWorkers(m) at once (#175 offline-move
	// speedup: one server-side copy at a time took six minutes for a 1.6 GB
	// replica while the instance was offline). Every update to round and
	// to the book is made under mu, in whatever order the copies finish;
	// neither depends on that order. The first failure stops the round: no
	// further copy starts, the ones in flight are cancelled, and every copy
	// that did land keeps its receipt (the best-effort save below).
	//
	// No network request is ever made under mu (Astra round 1, finding 3):
	// a receipts read that stalled there would hold every other worker,
	// the first failure's included, so the round could not stop. The book
	// is read here, before any copy, whenever this round copies anything
	// and keeps receipts; set and drop then touch memory only. (A round
	// that copies nothing still reads no receipts.)
	if len(jobs) > 0 {
		if err := book.load(ctx); err != nil {
			return round, err
		}
	}
	var mu sync.Mutex
	err = forEachParallel(ctx, mirrorWorkers(m), len(jobs), func(cctx context.Context, i int) error {
		r := jobs[i]
		sk, dk := srcPrefix+r, dstPrefix+r
		s := src[sk]
		var etag, ver string
		var err error
		if versioned {
			etag, ver, err = vc.CopyVersioned(cctx, srcBucket, sk, s.ETag, dstBucket, dk)
		} else {
			etag, err = m.Copy(cctx, srcBucket, sk, s.ETag, dstBucket, dk)
		}
		mu.Lock()
		defer mu.Unlock()
		if errors.Is(err, ErrCopySourceChanged) {
			book.drop(r)
			round.Changed++
			round.Deferred++
			round.BytesDeferred += s.Size
			return nil
		}
		if err != nil {
			return fmt.Errorf("mirror: copy %s: %w", sk, err)
		}
		if !UsableVersionID(ver) {
			ver = "" // never recorded or sealed by
		}
		round.Changed++
		round.BytesCopied += s.Size
		if ver != "" {
			round.Versions[r] = ver
		}
		if ver != "" || (etag != "" && etag != s.ETag) {
			dstETag := etag
			if dstETag == "" {
				dstETag = s.ETag
			}
			// The book was read before the copies began, so this is a
			// write to memory, never a request under mu.
			return book.set(ctx, r, copyReceipt{SrcETag: s.ETag, SrcSize: s.Size, DstETag: dstETag, DstVersionID: ver})
		}
		book.drop(r)
		return nil
	})
	if err != nil {
		book.save(ctx) // best effort: what already landed keeps its receipts
		return round, err
	}
	// The deletes, only once every copy has landed (as before), and side
	// by side among themselves: their order never mattered.
	var gone []string
	for dk := range dst {
		r := dk[len(dstPrefix):]
		if r == leaseLockName {
			continue
		}
		if _, ok := src[srcPrefix+r]; !ok {
			gone = append(gone, r)
		}
	}
	sort.Strings(gone)
	err = forEachParallel(ctx, mirrorWorkers(m), len(gone), func(cctx context.Context, i int) error {
		r := gone[i]
		if err := m.Delete(cctx, dstBucket, dstPrefix+r); err != nil {
			return fmt.Errorf("mirror: delete %s: %w", dstPrefix+r, err)
		}
		mu.Lock()
		defer mu.Unlock()
		book.drop(r)
		round.Changed++
		return nil
	})
	if err != nil {
		book.save(ctx)
		return round, err
	}
	book.prune(func(r string) bool { _, ok := src[srcPrefix+r]; return ok })
	if err := book.save(ctx); err != nil {
		return round, err
	}
	return round, nil
}

// DefaultMirrorWorkers is how many copies (and, after them, deletes) one
// delta round runs at once. Each is one server-side request, so the round's
// wall time is the slowest request of each batch rather than their sum.
const DefaultMirrorWorkers = 8

// MirrorConcurrency is the optional half of a MirrorStore that sets how many
// requests a round runs at once. A store that does not implement it gets
// DefaultMirrorWorkers; an answer below one is one (a strictly serial
// round, as every round was before). Tests use it; production stores do not.
type MirrorConcurrency interface {
	MirrorWorkers() int
}

func mirrorWorkers(m MirrorStore) int {
	if c, ok := m.(MirrorConcurrency); ok {
		if n := c.MirrorWorkers(); n >= 1 {
			return n
		}
		return 1
	}
	return DefaultMirrorWorkers
}

// forEachParallel calls do for every index below n, on up to workers
// goroutines, starting them in index order. The first error do returns
// stops it: no further index starts, the context handed to the calls still
// in flight is cancelled, and once they have returned that first error is
// the answer (any error a cancelled call returns after it is dropped). A ctx
// that ends before every index has started is its own error. Calls that
// were in flight when it stopped have returned by the time it does, so the
// caller's state is final.
func forEachParallel(ctx context.Context, workers, n int, do func(ctx context.Context, i int) error) error {
	if n == 0 {
		return nil
	}
	if workers < 1 {
		workers = 1
	}
	if workers > n {
		workers = n
	}
	cctx, cancel := context.WithCancel(ctx)
	defer cancel()
	var (
		mu    sync.Mutex
		first error
		wg    sync.WaitGroup
	)
	next := make(chan int)
	for w := 0; w < workers; w++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for i := range next {
				if err := do(cctx, i); err != nil {
					mu.Lock()
					if first == nil {
						first = err
						cancel()
					}
					mu.Unlock()
				}
			}
		}()
	}
	started := 0
feed:
	for ; started < n; started++ {
		// Checked first: a select with both cases ready picks at random,
		// and a stopped round must not start another copy.
		if cctx.Err() != nil {
			break
		}
		select {
		case next <- started:
		case <-cctx.Done():
			break feed
		}
	}
	close(next)
	wg.Wait()
	if first != nil {
		return first
	}
	if started < n {
		if err := ctx.Err(); err != nil {
			return fmt.Errorf("mirror: %w", err)
		}
	}
	return nil
}

// receiptLedger is one round's view of the receipts object. It reads the
// object lazily, on the first object whose ETags differ, so a round over
// single-part objects only costs no request it did not cost before.
type receiptLedger struct {
	m      MirrorStore
	rc     Receipts
	want   receiptBook // the pair this round mirrors; Objects unused
	loaded bool
	book   map[string]copyReceipt
	dirty  bool
}

func (l *receiptLedger) load(ctx context.Context) error {
	if l.loaded || l.rc.Key == "" {
		return nil
	}
	body, err := l.m.Get(ctx, l.rc.Bucket, l.rc.Key)
	if err != nil && !errors.Is(err, ErrMirrorNotFound) {
		// Not loaded: a failed read must never be saved over as if the
		// book were empty.
		return fmt.Errorf("mirror: read receipts %s/%s: %w", l.rc.Bucket, l.rc.Key, err)
	}
	l.loaded = true
	l.book = map[string]copyReceipt{}
	if err != nil {
		return nil // no receipts yet
	}
	var b receiptBook
	if json.Unmarshal(body, &b) != nil || b.V != 1 || b.SrcBucket != l.want.SrcBucket || b.SrcPrefix != l.want.SrcPrefix ||
		b.DstBucket != l.want.DstBucket || b.DstPrefix != l.want.DstPrefix {
		// Unreadable, or written for another pair: no evidence at all.
		// Rewritten at the end of the round.
		l.dirty = true
		return nil
	}
	for k, v := range b.Objects {
		l.book[k] = v
	}
	return nil
}

func (l *receiptLedger) get(ctx context.Context, rel string) (copyReceipt, bool, error) {
	if err := l.load(ctx); err != nil {
		return copyReceipt{}, false, err
	}
	r, ok := l.book[rel]
	return r, ok, nil
}

// set records a completed copy. It reads the book first if this round
// has not (a copy to a key the destination lacked needs no lookup), so the
// write at the end of the round keeps every other receipt.
func (l *receiptLedger) set(ctx context.Context, rel string, r copyReceipt) error {
	if l.rc.Key == "" {
		return nil
	}
	if err := l.load(ctx); err != nil {
		return err
	}
	if old, ok := l.book[rel]; !ok || old != r {
		l.book[rel] = r
		l.dirty = true
	}
	return nil
}

// drop forgets rel's receipt. Only a book already read can hold one.
func (l *receiptLedger) drop(rel string) {
	if !l.loaded {
		return
	}
	if _, ok := l.book[rel]; ok {
		delete(l.book, rel)
		l.dirty = true
	}
}

func (l *receiptLedger) prune(keep func(rel string) bool) {
	if !l.loaded {
		return
	}
	for rel := range l.book {
		if !keep(rel) {
			delete(l.book, rel)
			l.dirty = true
		}
	}
}

func (l *receiptLedger) save(ctx context.Context) error {
	if !l.dirty || l.rc.Key == "" {
		return nil
	}
	b := l.want
	b.Objects = l.book
	body, err := json.Marshal(b)
	if err != nil {
		return err
	}
	if err := l.m.Put(ctx, l.rc.Bucket, l.rc.Key, body); err != nil {
		return fmt.Errorf("mirror: write receipts %s/%s: %w", l.rc.Bucket, l.rc.Key, err)
	}
	l.dirty = false
	return nil
}

// leaseLockName is the coordination file this package's own leaser keeps
// beside every replica. MirrorPrefix's exclusion of it is lease POLICY, not
// generic mirroring — which is why the mirror lives in this package.
const leaseLockName = "lock.json"

// DeletePrefix removes every object under prefix (the lease lock included:
// a deleted staging area holds nothing back). The two callers are abort's
// box half (errors surface — the abort retries) and the activator's
// post-restore cleanup (best-effort at the call site).
func DeletePrefix(ctx context.Context, m MirrorStore, bucket, prefix string) error {
	objs, err := m.List(ctx, bucket, prefix)
	if err != nil {
		return err
	}
	for k := range objs {
		if err := m.Delete(ctx, bucket, k); err != nil {
			return err
		}
	}
	return nil
}

// s3Mirror implements MirrorStore over two regional clients: reads and the
// copy call go to the DESTINATION region's client (CopyObject executes in
// the destination bucket's region and pulls from the source), source
// listings to the source's.
type s3Mirror struct {
	src, dst *awss3.Client
	// byBucket memoizes which client answered for a bucket, so only the
	// FIRST List of each bucket pays the wrong-region probe — and a later
	// transient error on the right client is not retried against a region
	// that can only answer with a redirect.
	mu       sync.Mutex
	byBucket map[string]*awss3.Client
}

// NewS3Mirror builds the two-region pair with the same credential chain and
// endpoint compatibility every other S3 consumer here uses (endpoint.go's
// newLeaserClient).
func NewS3Mirror(ctx context.Context, srcRegion, dstRegion string) (MirrorStore, error) {
	src, _, err := newLeaserClient(ctx, srcRegion)
	if err != nil {
		return nil, err
	}
	dst, _, err := newLeaserClient(ctx, dstRegion)
	if err != nil {
		return nil, err
	}
	return &s3Mirror{src: src, dst: dst, byBucket: map[string]*awss3.Client{}}, nil
}

func (m *s3Mirror) list(ctx context.Context, c *awss3.Client, bucket, prefix string) (map[string]MirrorObject, error) {
	out := map[string]MirrorObject{}
	pager := awss3.NewListObjectsV2Paginator(c, &awss3.ListObjectsV2Input{Bucket: &bucket, Prefix: &prefix})
	for pager.HasMorePages() {
		resp, err := pager.NextPage(ctx)
		if err != nil {
			return nil, err
		}
		for _, o := range resp.Contents {
			if o.Key == nil {
				continue
			}
			var obj MirrorObject
			if o.ETag != nil {
				obj.ETag = normalizeETag(*o.ETag)
			}
			if o.Size != nil {
				obj.Size = *o.Size
			}
			out[*o.Key] = obj
		}
	}
	return out, nil
}

// List routes by memoized bucket→client binding: the interface's caller
// never says which region a bucket lives in, so the FIRST touch of a bucket
// probes source-then-destination (a wrong-region List is an error, so the
// probe is unambiguous) and the winner is remembered.
func (m *s3Mirror) List(ctx context.Context, bucket, prefix string) (map[string]MirrorObject, error) {
	m.mu.Lock()
	c := m.byBucket[bucket]
	m.mu.Unlock()
	if c != nil {
		return m.list(ctx, c, bucket, prefix)
	}
	out, err := m.list(ctx, m.src, bucket, prefix)
	c = m.src
	if err != nil {
		if out, err = m.list(ctx, m.dst, bucket, prefix); err != nil {
			return nil, err
		}
		c = m.dst
	}
	m.mu.Lock()
	m.byBucket[bucket] = c
	m.mu.Unlock()
	return out, nil
}

// normalizeETag strips the quotes S3 wraps an ETag in, so a listing and a
// CopyObject result compare equal when they name the same object: a
// receipt whose quoting differed from the next listing's would never match,
// and the recopy-forever bug would come back silently.
func normalizeETag(et string) string { return strings.Trim(et, `"`) }

// clientFor is the client List bound bucket to, or fallback if this pair
// has never listed it.
func (m *s3Mirror) clientFor(bucket string, fallback *awss3.Client) *awss3.Client {
	m.mu.Lock()
	defer m.mu.Unlock()
	if c := m.byBucket[bucket]; c != nil {
		return c
	}
	return fallback
}

func (m *s3Mirror) Copy(ctx context.Context, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (string, error) {
	source := url.PathEscape(srcBucket + "/" + srcKey)
	in := &awss3.CopyObjectInput{Bucket: &dstBucket, Key: &dstKey, CopySource: &source}
	if srcETag != "" {
		quoted := `"` + normalizeETag(srcETag) + `"`
		in.CopySourceIfMatch = &quoted
	}
	out, err := m.dst.CopyObject(ctx, in)
	if err != nil {
		var ae smithy.APIError
		if srcETag != "" && errors.As(err, &ae) && ae.ErrorCode() == "PreconditionFailed" {
			return "", ErrCopySourceChanged
		}
		return "", err
	}
	if out.CopyObjectResult == nil || out.CopyObjectResult.ETag == nil {
		return "", nil // no receipt possible; the next round recopies
	}
	return normalizeETag(*out.CopyObjectResult.ETag), nil
}

// Delete goes to the client List bound the bucket to: the receipts object
// lives in the SOURCE's bucket, which the destination's client cannot
// reach. A bucket this pair never listed (an abort after an agent restart
// lists only the target) tries the destination client, as before, then
// the source's; if both fail, the destination's error is the one reported.
func (m *s3Mirror) Delete(ctx context.Context, bucket, key string) error {
	in := &awss3.DeleteObjectInput{Bucket: &bucket, Key: &key}
	if c := m.clientFor(bucket, nil); c != nil {
		_, err := c.DeleteObject(ctx, in)
		return err
	}
	_, err := m.dst.DeleteObject(ctx, in)
	if err != nil && m.src != m.dst {
		if _, err2 := m.src.DeleteObject(ctx, in); err2 == nil {
			return nil
		}
	}
	return err
}

// Get and Put serve the receipts object, which lives in the source's own
// bucket: the bound client, else the source's.
func (m *s3Mirror) Get(ctx context.Context, bucket, key string) ([]byte, error) {
	out, err := m.clientFor(bucket, m.src).GetObject(ctx, &awss3.GetObjectInput{Bucket: &bucket, Key: &key})
	if err != nil {
		var nsk *s3types.NoSuchKey
		if errors.As(err, &nsk) {
			return nil, ErrMirrorNotFound
		}
		return nil, err
	}
	defer out.Body.Close()
	return io.ReadAll(out.Body)
}

func (m *s3Mirror) Put(ctx context.Context, bucket, key string, body []byte) error {
	ct := "application/json"
	_, err := m.clientFor(bucket, m.src).PutObject(ctx, &awss3.PutObjectInput{
		Bucket: &bucket, Key: &key, Body: bytes.NewReader(body), ContentType: &ct,
	})
	return err
}