package lease

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"sync"
	"testing"
)

// fakeMirror is an in-memory MirrorStore that keeps S3's ETag rules, plus
// an op log so ordering tests can assert the sequence. Every object carries
// a body (its content's identity). A single-part object's ETag IS its body
// (S3's MD5), a multipart one's is body+"-27", and a copy always lands
// single-part: the destination's ETag is the body, whatever the source's
// was. The fakes before #175 kept the source ETag on copy, and hid the bug
// that made every multipart object look changed on every round.
//
// A round runs its copies side by side, so every method holds mu, and the
// hooks (beforeCopy, onCopy) run without it: a hook may call the fake.
// workers is the round's concurrency (MirrorConcurrency); zero is the
// production default.
type fakeMirror struct {
	mu      sync.Mutex
	buckets map[string]map[string]fakeObject
	ops     []string
	failOn  string
	// beforeCopy runs at the top of Copy, before the precondition check:
	// the window between a round's listing and its copy.
	beforeCopy func(srcKey string)
	// onCopy runs after the precondition check and before the object
	// lands; ctx is the copy's own. An error it returns is the copy's.
	onCopy func(ctx context.Context, srcKey string) error
	// onGet runs at the top of Get, without the lock.
	onGet   func(key string)
	failPut bool
	failGet bool
	puts    int
	gets    int
	workers int
	// inFlight and peak count the copies running at once.
	inFlight, peak int
}

type fakeObject struct {
	MirrorObject
	body string
}

func newFakeMirror() *fakeMirror { return &fakeMirror{buckets: map[string]map[string]fakeObject{}} }

func (f *fakeMirror) MirrorWorkers() int {
	if f.workers == 0 {
		return DefaultMirrorWorkers
	}
	return f.workers
}

func (f *fakeMirror) put(bucket, key, etag string) { f.putSized(bucket, key, etag, 0) }

// putSized stores a single-part object: its ETag is its body.
func (f *fakeMirror) putSized(bucket, key, etag string, size int64) {
	f.store(bucket, key, fakeObject{MirrorObject{ETag: etag, Size: size}, etag})
}

// putMultipart stores a multipart upload of body (litestream's uploader
// for anything over one 5 MiB part).
func (f *fakeMirror) putMultipart(bucket, key, body string, size int64) {
	f.store(bucket, key, fakeObject{MirrorObject{ETag: body + "-27", Size: size}, body})
}

func (f *fakeMirror) store(bucket, key string, o fakeObject) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.storeLocked(bucket, key, o)
}

func (f *fakeMirror) storeLocked(bucket, key string, o fakeObject) {
	if f.buckets[bucket] == nil {
		f.buckets[bucket] = map[string]fakeObject{}
	}
	f.buckets[bucket][key] = o
}

// remove deletes one object, as a test's own hand does.
func (f *fakeMirror) remove(bucket, key string) {
	f.mu.Lock()
	defer f.mu.Unlock()
	delete(f.buckets[bucket], key)
}

// object is one stored object, as a test reads it.
func (f *fakeMirror) object(bucket, key string) (fakeObject, bool) {
	f.mu.Lock()
	defer f.mu.Unlock()
	o, ok := f.buckets[bucket][key]
	return o, ok
}

func (f *fakeMirror) List(_ context.Context, bucket, prefix string) (map[string]MirrorObject, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	out := map[string]MirrorObject{}
	for k, v := range f.buckets[bucket] {
		if strings.HasPrefix(k, prefix) {
			out[k] = v.MirrorObject
		}
	}
	return out, nil
}

func (f *fakeMirror) Copy(ctx context.Context, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (string, error) {
	f.mu.Lock()
	failOn, before, on := f.failOn, f.beforeCopy, f.onCopy
	f.inFlight++
	if f.inFlight > f.peak {
		f.peak = f.inFlight
	}
	f.mu.Unlock()
	defer func() {
		f.mu.Lock()
		f.inFlight--
		f.mu.Unlock()
	}()
	if failOn == srcKey {
		return "", fmt.Errorf("injected copy failure on %s", srcKey)
	}
	if before != nil {
		before(srcKey)
	}
	f.mu.Lock()
	o, ok := f.buckets[srcBucket][srcKey]
	f.mu.Unlock()
	if !ok {
		return "", fmt.Errorf("no such key %s", srcKey)
	}
	if srcETag != "" && o.ETag != srcETag {
		return "", ErrCopySourceChanged
	}
	if on != nil {
		if err := on(ctx, srcKey); err != nil {
			return "", err
		}
	}
	f.mu.Lock()
	defer f.mu.Unlock()
	f.ops = append(f.ops, "copy "+srcKey+" -> "+dstBucket+"/"+dstKey)
	f.storeLocked(dstBucket, dstKey, fakeObject{MirrorObject{ETag: o.body, Size: o.Size}, o.body})
	return o.body, nil
}

func (f *fakeMirror) Delete(_ context.Context, bucket, key string) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.ops = append(f.ops, "delete "+bucket+"/"+key)
	delete(f.buckets[bucket], key)
	return nil
}

func (f *fakeMirror) Get(_ context.Context, bucket, key string) ([]byte, error) {
	f.mu.Lock()
	on := f.onGet
	f.mu.Unlock()
	if on != nil {
		on(key)
	}
	f.mu.Lock()
	defer f.mu.Unlock()
	f.gets++
	if f.failGet {
		return nil, fmt.Errorf("injected get failure on %s", key)
	}
	o, ok := f.buckets[bucket][key]
	if !ok {
		return nil, ErrMirrorNotFound
	}
	return []byte(o.body), nil
}

func (f *fakeMirror) Put(_ context.Context, bucket, key string, body []byte) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	if f.failPut {
		return fmt.Errorf("injected put failure on %s", key)
	}
	f.puts++
	f.ops = append(f.ops, "put "+bucket+"/"+key)
	f.storeLocked(bucket, key, fakeObject{MirrorObject{ETag: fmt.Sprint("put-", f.puts), Size: int64(len(body))}, string(body)})
	return nil
}

func TestMirrorPrefix(t *testing.T) {
	ctx := context.Background()
	f := newFakeMirror()
	f.put("src", "instances/h/generations/a/snapshot", "e1")
	f.put("src", "instances/h/generations/a/wal/000", "e2")

	// Initial round: full copy.
	r, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{})
	if err != nil || r.Changed != 2 {
		t.Fatalf("initial round = %+v, %v; want 2 copies", r, err)
	}
	// Identical: zero.
	if r, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{}); err != nil || r.Changed != 0 {
		t.Fatalf("caught-up round = %+v, %v; want 0", r, err)
	}
	// Delta: one changed etag, one new, one pruned at source.
	f.put("src", "instances/h/generations/a/wal/000", "e2b")
	f.put("src", "instances/h/generations/a/wal/001", "e3")
	delete(f.buckets["src"], "instances/h/generations/a/snapshot")
	r, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{})
	if err != nil || r.Changed != 3 {
		t.Fatalf("delta round = %+v, %v; want 2 copies + 1 delete", r, err)
	}
	got, _ := f.List(ctx, "dst", "moves/h/")
	want := map[string]string{
		"moves/h/generations/a/wal/000": "e2b",
		"moves/h/generations/a/wal/001": "e3",
	}
	if len(got) != len(want) {
		t.Fatalf("staging = %v, want %v", got, want)
	}
	for k, v := range want {
		if got[k].ETag != v {
			t.Fatalf("staging[%s] = %q, want %q", k, got[k].ETag, v)
		}
	}
	// A copy failure surfaces with the partial count.
	f.put("src", "instances/h/generations/a/wal/002", "e4")
	f.failOn = "instances/h/generations/a/wal/002"
	if _, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{}); err == nil {
		t.Fatal("injected failure did not surface")
	}
}

// Prefix boundaries: a slashless prefix (activator.HibernatePrefix's shape)
// is normalized to a directory boundary — no fused keys, and a sibling host
// whose name extends this one's is never swept in.
func TestMirrorPrefixNormalizesBoundaries(t *testing.T) {
	ctx := context.Background()
	f := newFakeMirror()
	f.put("src", "instances/h/generations/a/snapshot", "e1")
	f.put("src", "instances/h2/generations/a/snapshot", "eX") // the sibling
	r, err := MirrorPrefix(ctx, f, "src", "instances/h", "dst", "moves/h", Receipts{})
	if err != nil || r.Changed != 1 {
		t.Fatalf("round = %+v, %v; want exactly the one h object", r, err)
	}
	got, _ := f.List(ctx, "dst", "")
	if _, ok := got["moves/h/generations/a/snapshot"]; !ok || len(got) != 1 {
		t.Fatalf("staging keys = %v, want the one properly-joined key", got)
	}
}

// lock.json is coordination state: never copied, never deleted, and its
// churn (the renewer rewrites it every TTL) must not keep changed above 0.
func TestMirrorPrefixSkipsTheLease(t *testing.T) {
	ctx := context.Background()
	f := newFakeMirror()
	f.put("src", "instances/h/lock.json", "renewal-1")
	f.put("src", "instances/h/generations/a/snapshot", "e1")
	f.put("dst", "moves/h/lock.json", "old-foreign-lease")
	if r, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{}); err != nil || r.Changed != 1 {
		t.Fatalf("round = %+v, %v; want only the snapshot to move", r, err)
	}
	f.put("src", "instances/h/lock.json", "renewal-2") // the churn
	if r, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{}); err != nil || r.Changed != 0 {
		t.Fatalf("churn round = %+v, %v; want caught-up despite the renewal", r, err)
	}
	got, _ := f.List(ctx, "dst", "moves/h/")
	if got["moves/h/lock.json"].ETag != "old-foreign-lease" || len(got) != 2 {
		t.Fatalf("staging = %v; the dst lease must be untouched", got)
	}
}

// A round reports the bytes it copied: the listed source size of every
// object it copied, and nothing for an object it skipped as unchanged, a
// delete, or the lease. The move worker's readiness bar is this number.
func TestMirrorPrefixCountsBytesCopied(t *testing.T) {
	ctx := context.Background()
	f := newFakeMirror()
	f.putSized("src", "instances/h/generations/a/snapshot", "e1", 1000)
	f.putSized("src", "instances/h/generations/a/wal/000", "e2", 30)
	f.putSized("src", "instances/h/lock.json", "renewal-1", 500)

	r, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{})
	if err != nil || r.Changed != 2 || r.BytesCopied != 1030 {
		t.Fatalf("initial round = %+v, %v; want 2 changed, 1030 bytes", r, err)
	}

	// Unchanged objects are skipped and cost nothing; the lease's churn
	// costs nothing; a rewritten object costs its NEW size; a new one its
	// size; a pruned one is a change but no bytes.
	f.putSized("src", "instances/h/lock.json", "renewal-2", 500)
	f.putSized("src", "instances/h/generations/a/wal/000", "e2b", 70)
	f.putSized("src", "instances/h/generations/a/wal/001", "e3", 5)
	f.putSized("dst", "moves/h/generations/old/snapshot", "eOld", 9999)
	r, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{})
	if err != nil || r.Changed != 3 || r.BytesCopied != 75 {
		t.Fatalf("delta round = %+v, %v; want 3 changed (2 copies + 1 delete), 75 bytes", r, err)
	}

	r, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{})
	if err != nil || r.Changed != 0 || r.BytesCopied != 0 {
		t.Fatalf("caught-up round = %+v, %v; want nothing", r, err)
	}

	// A failed round still reports what it copied before the failure.
	f.putSized("src", "instances/h/generations/a/wal/002", "e4", 11)
	f.putSized("src", "instances/h/generations/a/wal/003", "e5", 13)
	f.failOn = "instances/h/generations/a/wal/003"
	r, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", Receipts{})
	if err == nil || r.Changed != 1 || r.BytesCopied != 11 {
		t.Fatalf("failed round = %+v, %v; want an error with the partial 1 change, 11 bytes", r, err)
	}
}

var testReceipts = Receipts{Bucket: "src", Key: "move-receipts/h.json"}

func mirrorRound(t *testing.T, f *fakeMirror, rc Receipts) MirrorRound {
	t.Helper()
	r, err := MirrorPrefix(context.Background(), f, "src", "instances/h/", "dst", "moves/h/", rc)
	if err != nil {
		t.Fatalf("round: %v", err)
	}
	return r
}

func readReceipts(t *testing.T, f *fakeMirror) map[string]copyReceipt {
	t.Helper()
	body, err := f.Get(context.Background(), testReceipts.Bucket, testReceipts.Key)
	if err != nil {
		t.Fatalf("receipts: %v", err)
	}
	var b receiptBook
	if err := json.Unmarshal(body, &b); err != nil {
		t.Fatalf("receipts: %v", err)
	}
	return b.Objects
}

// #175 as production saw it: a multipart source's copy lands with a
// different ETag, so ETag equality alone recopies every multipart object
// on every round. Receipts let an unchanged one be copied once.
func TestMirrorPrefixReceiptsSkipAnUnchangedMultipartObject(t *testing.T) {
	const big = 138 << 20
	fill := func(f *fakeMirror) {
		f.putMultipart("src", "instances/h/ltx/1/a.ltx", "level1", big)
		f.putSized("src", "instances/h/ltx/0/b.ltx", "l0", 4096)
	}

	// Without receipts: the bug, kept visible.
	f := newFakeMirror()
	fill(f)
	mirrorRound(t, f, Receipts{})
	if r := mirrorRound(t, f, Receipts{}); r.Changed != 1 || r.BytesCopied != big {
		t.Fatalf("without receipts the unchanged multipart object should recopy: %+v", r)
	}

	f = newFakeMirror()
	fill(f)
	if r := mirrorRound(t, f, testReceipts); r.Changed != 2 || r.BytesCopied != big+4096 {
		t.Fatalf("first round = %+v, want both copied", r)
	}
	if got := readReceipts(t, f); len(got) != 1 || got["ltx/1/a.ltx"] != (copyReceipt{SrcETag: "level1-27", SrcSize: big, DstETag: "level1"}) {
		t.Fatalf("receipts = %+v, want exactly the multipart object's", got)
	}
	puts := f.puts
	for i := 0; i < 3; i++ {
		// MirrorRound gained a map (Versions), so it is no longer comparable
		// with ==; this is the same "nothing happened" assertion, field by field.
		if r := mirrorRound(t, f, testReceipts); r.Changed != 0 || r.BytesCopied != 0 || r.Deferred != 0 || r.BytesDeferred != 0 || r.Versions != nil {
			t.Fatalf("unchanged round %d = %+v, want nothing copied", i, r)
		}
	}
	if f.puts != puts {
		t.Fatalf("receipts rewritten on rounds that changed nothing (%d puts, want %d)", f.puts, puts)
	}
	// Receipts are never data: nothing beside the two objects is staged.
	if staged, _ := f.List(context.Background(), "dst", ""); len(staged) != 2 {
		t.Fatalf("staging = %v, want only the two data objects", staged)
	}
}

// Single-part sources keep the plain ETag skip: no receipts are written,
// so the existing behaviour and request count do not change.
func TestMirrorPrefixSinglePartWritesNoReceipts(t *testing.T) {
	f := newFakeMirror()
	f.putSized("src", "instances/h/ltx/0/a.ltx", "a", 100)
	mirrorRound(t, f, testReceipts)
	f.putSized("src", "instances/h/ltx/0/b.ltx", "b", 100)
	mirrorRound(t, f, testReceipts)
	if r := mirrorRound(t, f, testReceipts); r.Changed != 0 {
		t.Fatalf("caught-up round = %+v", r)
	}
	if f.puts != 0 {
		t.Fatalf("%d receipt writes for single-part objects, want none", f.puts)
	}
}

func TestMirrorPrefixReceiptRecopiesWhenTheSourceChanges(t *testing.T) {
	f := newFakeMirror()
	f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v1", 100<<20)
	mirrorRound(t, f, testReceipts)
	// Compaction rewrote it under the same name.
	f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v2", 101<<20)
	if r := mirrorRound(t, f, testReceipts); r.Changed != 1 || r.BytesCopied != 101<<20 {
		t.Fatalf("rewritten source = %+v, want it recopied", r)
	}
	if got := f.buckets["dst"]["moves/h/ltx/1/a.ltx"].body; got != "v2" {
		t.Fatalf("staged body = %q, want v2", got)
	}
	// The same size with different content is still a change: never
	// skip on size alone.
	f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v3", 101<<20)
	if r := mirrorRound(t, f, testReceipts); r.Changed != 1 {
		t.Fatalf("same-size rewrite = %+v, want it recopied", r)
	}
	if r := mirrorRound(t, f, testReceipts); r.Changed != 0 {
		t.Fatalf("after the recopy = %+v, want the new receipt to hold", r)
	}
}

func TestMirrorPrefixReceiptRecopiesWhenTheDestinationChanges(t *testing.T) {
	for _, tc := range []struct {
		name  string
		touch func(f *fakeMirror)
	}{
		{"overwritten", func(f *fakeMirror) { f.putSized("dst", "moves/h/ltx/1/a.ltx", "intruder", 100<<20) }},
		{"deleted", func(f *fakeMirror) { delete(f.buckets["dst"], "moves/h/ltx/1/a.ltx") }},
	} {
		t.Run(tc.name, func(t *testing.T) {
			f := newFakeMirror()
			f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v1", 100<<20)
			mirrorRound(t, f, testReceipts)
			tc.touch(f)
			if r := mirrorRound(t, f, testReceipts); r.Changed != 1 || r.BytesCopied != 100<<20 {
				t.Fatalf("destination %s behind the receipt = %+v, want a recopy", tc.name, r)
			}
			if got := f.buckets["dst"]["moves/h/ltx/1/a.ltx"].body; got != "v1" {
				t.Fatalf("staged body = %q, want v1", got)
			}
		})
	}
}

// A crash between the copy and the receipt write costs one extra copy on
// the next round, never a wrong skip.
func TestMirrorPrefixLostReceiptWriteRecopiesOnce(t *testing.T) {
	f := newFakeMirror()
	f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v1", 100<<20)
	f.failPut = true
	r, err := MirrorPrefix(context.Background(), f, "src", "instances/h/", "dst", "moves/h/", testReceipts)
	if err == nil || r.BytesCopied != 100<<20 {
		t.Fatalf("round with a failed receipt write = %+v, %v; want the copy reported and the error surfaced", r, err)
	}
	f.failPut = false
	if r := mirrorRound(t, f, testReceipts); r.Changed != 1 || r.BytesCopied != 100<<20 {
		t.Fatalf("next round = %+v, want exactly one recopy", r)
	}
	if r := mirrorRound(t, f, testReceipts); r.Changed != 0 {
		t.Fatalf("round after = %+v, want the receipt to hold", r)
	}
}

// A source overwritten between the listing and the copy fails the
// conditional copy. The round records no receipt for it and carries on;
// the next round copies the new version.
func TestMirrorPrefixSourceChangedMidRoundIsDeferred(t *testing.T) {
	f := newFakeMirror()
	f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v1", 100<<20)
	f.putSized("src", "instances/h/ltx/0/b.ltx", "b", 10)
	var once sync.Once
	f.beforeCopy = func(k string) {
		if k == "instances/h/ltx/1/a.ltx" {
			once.Do(func() { f.putMultipart("src", k, "v2", 120<<20) })
		}
	}
	r := mirrorRound(t, f, testReceipts)
	if r.Changed != 2 || r.Deferred != 1 || r.BytesDeferred != 100<<20 || r.BytesCopied != 10 {
		t.Fatalf("round = %+v, want b copied and a deferred at its listed size", r)
	}
	if r.DeltaBytes() != 100<<20+10 {
		t.Fatalf("DeltaBytes = %d", r.DeltaBytes())
	}
	if _, ok := f.buckets["dst"]["moves/h/ltx/1/a.ltx"]; ok {
		t.Fatal("a deferred copy landed anyway")
	}
	if _, err := f.Get(context.Background(), testReceipts.Bucket, testReceipts.Key); err == nil {
		if _, ok := readReceipts(t, f)["ltx/1/a.ltx"]; ok {
			t.Fatal("a receipt was recorded for a copy that failed its precondition")
		}
	}
	if r := mirrorRound(t, f, testReceipts); r.Changed != 1 || r.BytesCopied != 120<<20 || r.Deferred != 0 {
		t.Fatalf("next round = %+v, want the new version copied", r)
	}
	if got := f.buckets["dst"]["moves/h/ltx/1/a.ltx"].body; got != "v2" {
		t.Fatalf("staged body = %q, want v2", got)
	}
}

// Receipts name the pair they were written for. Another pair's receipts,
// even with matching keys and ETags, are not evidence about this one.
func TestMirrorPrefixIgnoresAnotherPairsReceipts(t *testing.T) {
	ctx := context.Background()
	f := newFakeMirror()
	f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v1", 100<<20)
	mirrorRound(t, f, testReceipts)
	// Another destination that happens to hold a lookalike object.
	f.putSized("dst", "other/h/ltx/1/a.ltx", "v1", 100<<20)
	r, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "other/h/", testReceipts)
	if err != nil || r.Changed != 1 {
		t.Fatalf("round toward another destination = %+v, %v; want a copy", r, err)
	}
}

// Unreadable receipts are no receipts: everything recopies once, and the
// book is rewritten.
func TestMirrorPrefixCorruptReceiptsRecopy(t *testing.T) {
	f := newFakeMirror()
	f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v1", 100<<20)
	mirrorRound(t, f, testReceipts)
	f.store(testReceipts.Bucket, testReceipts.Key, fakeObject{body: "{not json"})
	if r := mirrorRound(t, f, testReceipts); r.Changed != 1 {
		t.Fatalf("round over corrupt receipts = %+v, want a recopy", r)
	}
	if r := mirrorRound(t, f, testReceipts); r.Changed != 0 {
		t.Fatalf("round after = %+v, want the rewritten receipts to hold", r)
	}
}

// A source object retention pruned takes its receipt with it.
func TestMirrorPrefixPrunesReceiptsWithTheirObjects(t *testing.T) {
	f := newFakeMirror()
	f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v1", 100<<20)
	f.putMultipart("src", "instances/h/ltx/1/b.ltx", "w1", 100<<20)
	mirrorRound(t, f, testReceipts)
	delete(f.buckets["src"], "instances/h/ltx/1/a.ltx")
	if r := mirrorRound(t, f, testReceipts); r.Changed != 1 || r.BytesCopied != 0 {
		t.Fatalf("prune round = %+v, want one delete", r)
	}
	if got := readReceipts(t, f); len(got) != 1 || got["ltx/1/b.ltx"].SrcETag != "w1-27" {
		t.Fatalf("receipts = %+v, want only b's", got)
	}
}

// Receipts that cannot be READ (a failure that is not "missing") fail the
// round: guessing "no receipts" would quietly bring back the recopy loop,
// and saving over a book this round never read would destroy it.
func TestMirrorPrefixReceiptReadFailureFailsTheRound(t *testing.T) {
	f := newFakeMirror()
	f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v1", 100<<20)
	mirrorRound(t, f, testReceipts)
	before := readReceipts(t, f)
	f.putMultipart("src", "instances/h/ltx/1/b.ltx", "w1", 100<<20)
	f.failGet = true
	if _, err := MirrorPrefix(context.Background(), f, "src", "instances/h/", "dst", "moves/h/", testReceipts); err == nil {
		t.Fatal("an unreadable receipts object did not fail the round")
	}
	f.failGet = false
	if got := readReceipts(t, f); len(got) != len(before) || got["ltx/1/a.ltx"] != before["ltx/1/a.ltx"] {
		t.Fatalf("receipts after a failed read = %+v, want them untouched (%+v)", got, before)
	}
}

// fakeVersionedMirror is fakeMirror plus version ids on copies and puts,
// the S3 behaviour MirrorPrefix records for the seal (moves spec §7.3).
type fakeVersionedMirror struct {
	*fakeMirror
	n int
}

func (f *fakeVersionedMirror) CopyVersioned(ctx context.Context, sb, sk, se, db, dk string) (string, string, error) {
	etag, err := f.fakeMirror.Copy(ctx, sb, sk, se, db, dk)
	if err != nil {
		return "", "", err
	}
	return etag, f.next(), nil
}

func (f *fakeVersionedMirror) PutVersioned(ctx context.Context, b, k string, body []byte) (string, error) {
	if err := f.fakeMirror.Put(ctx, b, k, body); err != nil {
		return "", err
	}
	return f.next(), nil
}

// next is the next version id; copies run side by side, so under mu.
func (f *fakeVersionedMirror) next() string {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.n++
	return fmt.Sprintf("v%d", f.n)
}

func TestMirrorPrefixRecordsDestinationVersions(t *testing.T) {
	f := &fakeVersionedMirror{fakeMirror: newFakeMirror()}
	f.put("src", "instances/h/0000/a.ltx", "e1")
	f.put("src", "instances/h/0009/s.ltx", "e2")
	rc := Receipts{Bucket: "src", Key: "move-receipts/h/mv.json"}
	r1, err := MirrorPrefix(context.Background(), f, "src", "instances/h/", "dst", "moves/h/mv/replica/", rc)
	if err != nil {
		t.Fatal(err)
	}
	if r1.Versions["0000/a.ltx"] == "" || r1.Versions["0009/s.ltx"] == "" {
		t.Fatalf("versions after round 1: %v", r1.Versions)
	}
	// Round 2 copies nothing; the versions come from the receipts.
	r2, err := MirrorPrefix(context.Background(), f, "src", "instances/h/", "dst", "moves/h/mv/replica/", rc)
	if err != nil {
		t.Fatal(err)
	}
	if r2.BytesCopied != 0 || r2.Versions["0000/a.ltx"] != r1.Versions["0000/a.ltx"] || len(r2.Versions) != 2 {
		t.Fatalf("round 2: %+v", r2)
	}
	// A crash between the copy and the receipt write: the next round sees
	// the destination's ETag equal to the source's but no receipt (or a
	// receipt for an older version of the key). It recopies rather than
	// sealing a version it cannot vouch for.
	f.putSized("src", "instances/h/0000/a.ltx", "e1-new", 64)
	f.n++ // a copy whose receipt was never persisted
	f.store("dst", "moves/h/mv/replica/0000/a.ltx", fakeObject{MirrorObject{ETag: "e1-new", Size: 64}, "e1-new"})
	r4, err := MirrorPrefix(context.Background(), f, "src", "instances/h/", "dst", "moves/h/mv/replica/", rc)
	if err != nil || r4.Versions["0000/a.ltx"] == r1.Versions["0000/a.ltx"] || r4.Changed == 0 || r4.BytesCopied != 64 {
		t.Fatalf("stale receipt sealed: %+v %v", r4, err)
	}
	// The plain store's second identical round copies nothing (the
	// existing TestMirrorPrefix contract, kept): see the next block.
	// A plain MirrorStore records no versions and keeps today's rule: an
	// equal ETag skips the copy with no receipt needed.
	g := newFakeMirror()
	g.putSized("src", "instances/h/0000/a.ltx", "e1", 64)
	r3, err := MirrorPrefix(context.Background(), g, "src", "instances/h/", "dst", "moves/h/mv/replica/", rc)
	if err != nil || r3.Versions != nil || r3.BytesCopied != 64 {
		t.Fatalf("plain store round 1: %+v %v", r3, err)
	}
	r3b, err := MirrorPrefix(context.Background(), g, "src", "instances/h/", "dst", "moves/h/mv/replica/", rc)
	if err != nil || r3b.Changed != 0 || r3b.BytesCopied != 0 {
		t.Fatalf("plain store recopied an unchanged object: %+v %v", r3b, err)
	}
}

// A multipart source's copy lands with a destination ETag that differs from
// the source's, so the receipt-match branch (not the equal-ETag one) must
// also vouch for the version.
func TestVersionedMirrorRecordsMultipartVersions(t *testing.T) {
	ctx := context.Background()
	f := &fakeVersionedMirror{fakeMirror: newFakeMirror()}
	f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v1", 100<<20)
	f.putMultipart("src", "instances/h/ltx/1/b.ltx", "w1", 50<<20)
	rc := testReceipts
	round := func() MirrorRound {
		t.Helper()
		r, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", rc)
		if err != nil {
			t.Fatal(err)
		}
		return r
	}
	r1 := round()
	if r1.Changed != 2 || r1.Versions["ltx/1/a.ltx"] == "" || r1.Versions["ltx/1/b.ltx"] == "" || len(r1.Versions) != 2 {
		t.Fatalf("round 1: %+v", r1)
	}
	if d := f.buckets["dst"]["moves/h/ltx/1/a.ltx"]; d.ETag == "v1-27" {
		t.Fatalf("fake did not model the multipart ETag change: %v", d.ETag)
	}
	r2 := round()
	if r2.Changed != 0 || r2.BytesCopied != 0 || len(r2.Versions) != 2 ||
		r2.Versions["ltx/1/a.ltx"] != r1.Versions["ltx/1/a.ltx"] || r2.Versions["ltx/1/b.ltx"] != r1.Versions["ltx/1/b.ltx"] {
		t.Fatalf("round 2: %+v (round 1 versions %v)", r2, r1.Versions)
	}
	// A receipt hand-written without dst_version_id (the plain path, or
	// before this change) is recopied exactly once, then carries one.
	book := readReceipts(t, f.fakeMirror)
	rcpt := book["ltx/1/a.ltx"]
	if rcpt.DstVersionID == "" {
		t.Fatalf("receipt carries no version: %+v", rcpt)
	}
	rcpt.DstVersionID = ""
	book["ltx/1/a.ltx"] = rcpt
	body, err := json.Marshal(receiptBook{V: 1, SrcBucket: "src", SrcPrefix: "instances/h/", DstBucket: "dst", DstPrefix: "moves/h/", Objects: book})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.fakeMirror.Put(ctx, rc.Bucket, rc.Key, body); err != nil {
		t.Fatal(err)
	}
	r3 := round()
	if r3.Changed != 1 || r3.BytesCopied != 100<<20 || r3.Versions["ltx/1/a.ltx"] == "" || r3.Versions["ltx/1/a.ltx"] == r1.Versions["ltx/1/a.ltx"] {
		t.Fatalf("round 3 (versionless receipt): %+v", r3)
	}
	r4 := round()
	if r4.Changed != 0 || r4.Versions["ltx/1/a.ltx"] != r3.Versions["ltx/1/a.ltx"] || len(r4.Versions) != 2 {
		t.Fatalf("round 4: %+v", r4)
	}
}

// Final review finding 4: a receipt that names the version "null" (a copy
// into a versioning-suspended bucket, recorded before the staging mirror
// refused it) vouches for nothing, since a later write overwrites that
// version. The round recopies the object rather than skip it or report the
// null version, and a copier that answers "null" has no version recorded.
func TestVersionedMirrorRecopiesANullVersionReceipt(t *testing.T) {
	ctx := context.Background()
	f := &fakeVersionedMirror{fakeMirror: newFakeMirror()}
	f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v1", 100<<20)
	rc := testReceipts
	round := func() MirrorRound {
		t.Helper()
		r, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", rc)
		if err != nil {
			t.Fatal(err)
		}
		return r
	}
	r1 := round()
	book := readReceipts(t, f.fakeMirror)
	rcpt := book["ltx/1/a.ltx"]
	rcpt.DstVersionID = "null"
	book["ltx/1/a.ltx"] = rcpt
	body, err := json.Marshal(receiptBook{V: 1, SrcBucket: "src", SrcPrefix: "instances/h/", DstBucket: "dst", DstPrefix: "moves/h/", Objects: book})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.fakeMirror.Put(ctx, rc.Bucket, rc.Key, body); err != nil {
		t.Fatal(err)
	}
	r2 := round()
	if r2.Changed != 1 || r2.BytesCopied != 100<<20 || r2.Versions["ltx/1/a.ltx"] == "null" || r2.Versions["ltx/1/a.ltx"] == r1.Versions["ltx/1/a.ltx"] {
		t.Fatalf("round after a null-version receipt: %+v (round 1 %v)", r2, r1.Versions)
	}
	n := &nullVersionMirror{fakeMirror: newFakeMirror()}
	n.putMultipart("src", "instances/h/ltx/1/a.ltx", "v1", 100<<20)
	r, err := MirrorPrefix(ctx, n, "src", "instances/h/", "dst", "moves/h/", rc)
	if err != nil {
		t.Fatal(err)
	}
	if v, ok := r.Versions["ltx/1/a.ltx"]; ok {
		t.Fatalf("a null version was recorded: %q", v)
	}
	if got := readReceipts(t, n.fakeMirror)["ltx/1/a.ltx"]; got.DstVersionID != "" {
		t.Fatalf("a receipt carries the null version: %+v", got)
	}
}

// nullVersionMirror is a copier into a versioning-suspended bucket.
type nullVersionMirror struct{ *fakeMirror }

func (f *nullVersionMirror) CopyVersioned(ctx context.Context, sb, sk, se, db, dk string) (string, string, error) {
	etag, err := f.fakeMirror.Copy(ctx, sb, sk, se, db, dk)
	return etag, "null", err
}

// Version collection is opt-in: the production s3Mirror must never enter
// the receipt-required mode, since an unversioned destination can never
// supply a version receipt and every unchanged object would recopy forever
// (round-4 finding 7).
func TestPlainS3MirrorRecordsNoVersions(t *testing.T) {
	if _, ok := interface{}(&s3Mirror{}).(VersionedCopier); ok {
		t.Fatal("s3Mirror implements VersionedCopier")
	}
	if _, ok := interface{}(&s3Mirror{}).(VersionedPutter); ok {
		t.Fatal("s3Mirror implements VersionedPutter")
	}
}

// MirrorPrefixKeep mirrors only the source keys keep admits: a refused key
// is never copied, and a destination copy of it is deleted as if the source
// no longer had it. A nil keep is MirrorPrefix.
func TestMirrorPrefixKeepMirrorsOnlyKeptKeys(t *testing.T) {
	ctx := context.Background()
	f := newFakeMirror()
	f.put("src", "loc/0000/a.ltx", "a")
	f.put("src", "loc/0001/b.ltx", "b")
	f.put("src", "loc/g-old/0000/c.ltx", "c")
	f.put("src", "loc/generation.json", "g")
	f.put("dst", "lane/g-old/0000/c.ltx", "c") // a stale copy of a refused key
	keep := func(rel string) bool { return strings.Count(rel, "/") == 1 && !strings.HasPrefix(rel, "g-") }
	round, err := MirrorPrefixKeep(ctx, f, "src", "loc", "dst", "lane", Receipts{}, keep)
	if err != nil {
		t.Fatal(err)
	}
	got, _ := f.List(ctx, "dst", "lane/")
	if len(got) != 2 || got["lane/0000/a.ltx"].ETag != "a" || got["lane/0001/b.ltx"].ETag != "b" {
		t.Fatalf("staged %v, want only the two kept level files", got)
	}
	if round.Changed != 3 {
		t.Fatalf("round %+v: want two copies and one delete", round)
	}
	// nil keep: MirrorPrefix exactly, every key.
	if _, err := MirrorPrefixKeep(ctx, f, "src", "loc", "dst", "all", Receipts{}, nil); err != nil {
		t.Fatal(err)
	}
	if all, _ := f.List(ctx, "dst", "all/"); len(all) != 4 {
		t.Fatalf("nil keep staged %d keys, want 4", len(all))
	}
}