package console

import (
	"context"
	"errors"
	"fmt"
	"net/http/httptest"
	"os"
	"path/filepath"
	"slices"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/dbdoor/dbdoortest"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/lease/leasetest"
	"github.com/carlosframework/platform/internal/objstore"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/seeds"
)

// finishFixture is the --from fixture with private storage on and one data
// bucket, as production has it: seeds/ and every instance's replica in the
// same store, so the leftover-replica rail and the finish see one bucket.
// The host's lease is read by the lease package itself (lease.LockState),
// so it is served from leasetest's S3 server; leases answers for it.
func finishFixture(t *testing.T) fromFixture {
	t.Helper()
	f := newFromFixture(t, 3)
	f.data = release.NewDirStore(f.root)
	f.fx.srv.cfg.SeedData = f.data
	enableInstances(t, f.fx, fromApp, fromManifest(true))
	leases := leasetest.NewVersionedBucket()
	leases.AddBucket(f.fx.srv.cfg.HibernateBucket)
	handler, _ := leasetest.NewS3Server(leases)
	srv := httptest.NewServer(handler)
	t.Cleanup(srv.Close)
	t.Setenv("AWS_ACCESS_KEY_ID", "test-key")
	t.Setenv("AWS_SECRET_ACCESS_KEY", "test-secret")
	t.Setenv("AWS_SESSION_TOKEN", "")
	t.Setenv("AWS_EC2_METADATA_DISABLED", "true")
	t.Setenv("AWS_CONFIG_FILE", t.TempDir()+"/absent")
	t.Setenv("AWS_SHARED_CREDENTIALS_FILE", t.TempDir()+"/absent")
	t.Setenv("CARLOS_S3_ENDPOINT", srv.URL)
	f.fx.srv.cfg.HibernateRegion = "test-region"
	finishLeases[f.fx.srv] = leases
	t.Cleanup(func() { delete(finishLeases, f.fx.srv) })
	return f
}

// finishLeases is each finish fixture's lease bucket.
var finishLeases = map[*Server]*leasetest.VersionedBucket{}

// putLease writes host's lock.json as the litestream leaser does, expiring
// at expires.
func (f fromFixture) putLease(t *testing.T, host string, expires time.Time) {
	t.Helper()
	finishLeases[f.fx.srv].Put(f.fx.srv.cfg.HibernateBucket, "instances/"+host+"/lock.json",
		[]byte(`{"generation":1,"expires_at":"`+expires.UTC().Format(time.RFC3339Nano)+`","owner":"box-a"}`))
}

// putReplicaKeys writes n stand-in LTX objects under host's replica prefix.
func (f fromFixture) putReplicaKeys(t *testing.T, host string, n int) []string {
	t.Helper()
	var keys []string
	for i := 1; i <= n; i++ {
		k := fmt.Sprintf("instances/%s/0000/%016x-%016x.ltx", host, i, i)
		if _, err := f.data.Put(context.Background(), k, []byte("x"), ""); err != nil {
			t.Fatal(err)
		}
		keys = append(keys, k)
	}
	return keys
}

// rollBack walks a seed the way a box does once it has been aborted: the
// abort, then the box's CAS to rolled-back.
func (f fromFixture) rollBack(t *testing.T, host, id string) {
	t.Helper()
	ctx := context.Background()
	for _, step := range []func(*provision.Record) error{provision.SeedAbort(id), provision.SeedRolledBack(id)} {
		if _, err := provision.SetRecordSeed(ctx, f.fx.rel, "bab", fromApp, host, step); err != nil {
			t.Fatal(err)
		}
	}
}

// seedRecord writes a record of shop on host in env whose seed is id, of
// kind, in state.
func (f fromFixture) seedRecord(t *testing.T, host, env, id, kind, state string) {
	t.Helper()
	rec := provision.Record{Schema: 1, Host: host, Slug: "canary", App: fromApp, Account: "bab", Channel: "stable",
		Environment: env, CreatedAt: time.Now().UTC(),
		Seed: &provision.SeedSource{ID: id, Kind: kind, State: state, By: "bob@keymail.dev", At: time.Now().UTC()}}
	if err := provision.CreateRecord(context.Background(), f.fx.rel, rec); err != nil {
		t.Fatal(err)
	}
}

// markedStore writes a canary store holding a grant, made by seed (""
// for a store no seed made).
func (f fromFixture) markedStore(t *testing.T, seed string) {
	t.Helper()
	at := time.Now().UTC().Add(-time.Hour)
	putCanaryStoreOf(t, f.fx, fromApp, &objstore.Record{
		Declared: &objstore.Declared{EnvPrefix: "S3", At: at, By: "bob@keymail.dev", Seed: seed},
		Grant:    &objstore.GrantBlock{Backend: "mem", Bucket: "carlos-bab-shop-canary", Source: objstore.GrantSourceAccountApproval, At: at, By: "operator@keymail.dev"},
	})
}

func (f fromFixture) canaryStore(t *testing.T) *objstore.Record {
	t.Helper()
	rec, _, err := objstore.GetIn(context.Background(), f.fx.rel, "bab", fromApp, "canary")
	if err != nil {
		t.Fatal(err)
	}
	return rec
}

func (f fromFixture) rawKey(t *testing.T, s release.Store, key string) string {
	t.Helper()
	b, _, err := s.Get(context.Background(), key)
	if err != nil {
		t.Fatalf("read %s: %v", key, err)
	}
	return string(b)
}

func (f fromFixture) requireRecordGone(t *testing.T, host string) {
	t.Helper()
	if rec, err := f.record(t, host); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("record %s after the finish = %+v, %v; want it gone", host, rec, err)
	}
}

func newSeedID(t *testing.T) string {
	t.Helper()
	id, err := seeds.NewID()
	if err != nil {
		t.Fatal(err)
	}
	return id
}

func replicaKeys(t *testing.T, f fromFixture, host string) []string {
	t.Helper()
	keys, err := f.data.List(context.Background(), "instances/"+host+"/")
	if err != nil {
		t.Fatal(err)
	}
	return keys
}

// The whole finish for a --from seed that failed after its first upload:
// its own store torn down for real, its snapshot and its replica deleted,
// and the record last. The host is then free for the same copy again: no
// leftover store for §4 to refuse, no replica for the #199 rail to refuse.
func TestFinishSeedTearsDownAndFreesTheHost(t *testing.T) {
	f := finishFixture(t)
	rec := f.copy(t, fromBody)
	requireReady(t, rec)
	id := rec.Seed.ID
	store := f.canaryStore(t)
	if store == nil || store.Declared == nil || store.Declared.Seed != id {
		t.Fatalf("canary store = %+v; want the copy's own", store)
	}
	bucket := storeBucketName(store)
	if _, ok := f.fake.Buckets[bucket]; !ok {
		t.Fatalf("bucket %s was never made", bucket)
	}
	// A box placed the copy and uploaded its first replica before the
	// abort landed (a lost publish answer, Plan 5 T5).
	dbdoortest.Replicate(t, filepath.Join(f.root, "instances", fromHost), 2)
	if held, err := f.fx.srv.hostHoldsReplicaData(context.Background(), fromHost); err != nil || !held {
		t.Fatalf("replica before the finish: %v, %v", held, err)
	}
	sourceReplica := replicaKeys(t, f, fromSource)
	f.rollBack(t, fromHost, id)

	if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil {
		t.Fatal(err)
	}
	if got := f.canaryStore(t); got != nil {
		t.Fatalf("store record after the finish = %+v; want it gone", got)
	}
	if _, ok := f.fake.Buckets[bucket]; ok {
		t.Fatalf("bucket %s survived the teardown", bucket)
	}
	if _, _, err := f.data.Get(context.Background(), seeds.SnapshotKey(fromHost, id)); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("snapshot after the finish: %v; want it gone", err)
	}
	if keys := replicaKeys(t, f, fromHost); len(keys) != 0 {
		t.Fatalf("replica after the finish: %v", keys)
	}
	if held, err := f.fx.srv.hostHoldsReplicaData(context.Background(), fromHost); err != nil || held {
		t.Fatalf("the rail after the finish: %v, %v", held, err)
	}
	if got := replicaKeys(t, f, fromSource); !slices.Equal(got, sourceReplica) {
		t.Fatalf("the source's replica changed:\n before %v\n after  %v", sourceReplica, got)
	}
	f.requireRecordGone(t, fromHost)

	// Idempotent: a second finish (the sweep after an inline one) is done.
	if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil {
		t.Fatalf("a second finish: %v", err)
	}
	again := f.copy(t, fromBody)
	requireReady(t, again)
	if again.Seed.ID == id {
		t.Fatal("the new copy reused the old seed id")
	}
	if store := f.canaryStore(t); store == nil || store.Declared == nil || store.Declared.Seed != again.Seed.ID {
		t.Fatalf("the new copy's store = %+v; want one marked with %s", store, again.Seed.ID)
	}
}

// A teardown that fails leaves the record rolled-back and holding the host,
// so the next finish (the sweep's) resumes and completes.
func TestFinishSeedResumesAfterAFailedTeardown(t *testing.T) {
	f := finishFixture(t)
	rec := f.copy(t, fromBody)
	requireReady(t, rec)
	id := rec.Seed.ID
	f.rollBack(t, fromHost, id)
	f.fake.FailNext["DeleteBucket"] = errors.New("the backend is down")

	if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err == nil {
		t.Fatal("a failed teardown answered nil")
	}
	if got, err := f.record(t, fromHost); err != nil || got.Seed == nil || got.Seed.State != seeds.StateRolledBack {
		t.Fatalf("record after a failed teardown = %+v, %v; want it kept, rolled-back", got, err)
	}
	if f.canaryStore(t) == nil {
		t.Fatal("the store record went before its estate")
	}
	if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil {
		t.Fatalf("the resumed finish: %v", err)
	}
	if got := f.canaryStore(t); got != nil {
		t.Fatalf("store record after the resumed finish = %+v", got)
	}
	f.requireRecordGone(t, fromHost)
}

// Only a store this seed made is torn down. Any other store is left as it
// was, and the record is still deleted.
func TestFinishSeedLeavesAStoreItDidNotMake(t *testing.T) {
	for name, seed := range map[string]string{
		"another seed's": "sd_bbbbbbbbbbbbbbbbbbbbbbbbbb",
		"no seed's":      "",
	} {
		t.Run(name, func(t *testing.T) {
			f := finishFixture(t)
			id := newSeedID(t)
			f.seedRecord(t, fromHost, "canary", id, seeds.KindImport, seeds.StateRolledBack)
			f.markedStore(t, seed)
			key := objstore.KeyIn("bab", fromApp, "canary")
			before := f.rawKey(t, f.fx.rel, key)
			if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil {
				t.Fatal(err)
			}
			if after := f.rawKey(t, f.fx.rel, key); after != before {
				t.Fatalf("the store record changed:\n before %s\n after  %s", before, after)
			}
			f.requireRecordGone(t, fromHost)
		})
	}
}

// A record the finish does not own is left alone, with the store its seed
// made: another seed, a seed not yet rolled back, no seed at all.
func TestFinishSeedLeavesARecordItDoesNotOwn(t *testing.T) {
	id := "sd_aaaaaaaaaaaaaaaaaaaaaaaaaa"
	for name, c := range map[string]struct{ id, state string }{
		"another seed":       {"sd_bbbbbbbbbbbbbbbbbbbbbbbbbb", seeds.StateRolledBack},
		"aborted, not yet":   {id, seeds.StateAborted},
		"ready":              {id, seeds.StateReady},
		"placed":             {id, seeds.StatePlaced},
		"seedless (no seed)": {"", ""},
	} {
		t.Run(name, func(t *testing.T) {
			f := finishFixture(t)
			if c.id == "" {
				if err := provision.CreateRecord(context.Background(), f.fx.rel, provision.Record{Schema: 1, Host: fromHost, Slug: "canary",
					App: fromApp, Account: "bab", Channel: "stable", Environment: "canary", CreatedAt: time.Now().UTC()}); err != nil {
					t.Fatal(err)
				}
			} else {
				f.seedRecord(t, fromHost, "canary", c.id, seeds.KindReplica, c.state)
			}
			f.markedStore(t, id)
			if _, err := f.data.Put(context.Background(), "instances/"+fromHost+"/0000/0000000000000001-0000000000000001.ltx", []byte("x"), ""); err != nil {
				t.Fatal(err)
			}
			recKey := provision.RecordKey("bab", fromApp, fromHost)
			storeKey := objstore.KeyIn("bab", fromApp, "canary")
			recBefore, storeBefore := f.rawKey(t, f.fx.rel, recKey), f.rawKey(t, f.fx.rel, storeKey)
			if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil {
				t.Fatal(err)
			}
			if after := f.rawKey(t, f.fx.rel, recKey); after != recBefore {
				t.Fatalf("the record changed:\n before %s\n after  %s", recBefore, after)
			}
			if after := f.rawKey(t, f.fx.rel, storeKey); after != storeBefore {
				t.Fatalf("the store record changed:\n before %s\n after  %s", storeBefore, after)
			}
			if keys := replicaKeys(t, f, fromHost); len(keys) != 1 {
				t.Fatalf("replica = %v; want it untouched", keys)
			}
		})
	}
}

// The finish decides under the store's walk lock (preflight M9). A grant
// by another writer that held the lock while the finish queued, and took
// the store over (clearing its seed mark), wins: the finish re-reads under
// the lock and leaves the store standing.
func TestFinishSeedReReadsTheStoreUnderItsLock(t *testing.T) {
	f := finishFixture(t)
	id := newSeedID(t)
	f.seedRecord(t, fromHost, "canary", id, seeds.KindImport, seeds.StateRolledBack)
	f.markedStore(t, id)
	unlock := f.fx.srv.storeWalks.lock("bab", fromApp, "canary")
	done := make(chan error, 1)
	go func() { done <- f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id) }()
	deadline := time.Now().Add(3 * time.Second)
	for {
		f.fx.srv.storeWalks.mu.Lock()
		refs := 0
		if e := f.fx.srv.storeWalks.m[storeWalkKey("bab", fromApp, "canary")]; e != nil {
			refs = e.refs
		}
		f.fx.srv.storeWalks.mu.Unlock()
		if refs == 2 {
			break
		}
		if time.Now().After(deadline) {
			unlock()
			t.Fatalf("the finish never queued on the store lock: refs=%d", refs)
		}
		time.Sleep(time.Millisecond)
	}
	store, etag, err := objstore.GetIn(context.Background(), f.fx.rel, "bab", fromApp, "canary")
	if err != nil {
		t.Fatal(err)
	}
	store.Declared.Seed = ""
	if _, err := objstore.PutIn(context.Background(), f.fx.rel, "bab", fromApp, "canary", store, etag); err != nil {
		t.Fatal(err)
	}
	key := objstore.KeyIn("bab", fromApp, "canary")
	taken := f.rawKey(t, f.fx.rel, key)
	unlock()
	if err := <-done; err != nil {
		t.Fatal(err)
	}
	if after := f.rawKey(t, f.fx.rel, key); after != taken {
		t.Fatalf("a store taken over under the lock was touched:\n before %s\n after  %s", taken, after)
	}
	f.requireRecordGone(t, fromHost)
}

// Defence in depth (preflight M10): a store this seed made is not torn down
// while another instance names its environment. The record still goes:
// the store now serves the other instance, and an operator decides.
func TestFinishSeedSparesAStoreAnotherInstanceUses(t *testing.T) {
	f := finishFixture(t)
	id := newSeedID(t)
	f.seedRecord(t, fromHost, "canary", id, seeds.KindImport, seeds.StateRolledBack)
	if err := provision.CreateRecord(context.Background(), f.fx.rel, provision.Record{Schema: 1, Host: "other.shop.example.com", Slug: "other",
		App: fromApp, Account: "bab", Channel: "stable", Environment: "canary", CreatedAt: time.Now().UTC()}); err != nil {
		t.Fatal(err)
	}
	f.markedStore(t, id)
	key := objstore.KeyIn("bab", fromApp, "canary")
	before := f.rawKey(t, f.fx.rel, key)
	if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil {
		t.Fatal(err)
	}
	if after := f.rawKey(t, f.fx.rel, key); after != before {
		t.Fatalf("a shared store was touched:\n before %s\n after  %s", before, after)
	}
	f.requireRecordGone(t, fromHost)
}

// A record of the app that will not decode may name the environment: the
// finish cannot know it does not, so the store is spared as for a sharer.
// The sweep runs this finish unattended (Plan 8 M7).
func TestFinishSeedSparesAStoreWhenARecordIsUnreadable(t *testing.T) {
	f := finishFixture(t)
	id := newSeedID(t)
	f.seedRecord(t, fromHost, "canary", id, seeds.KindImport, seeds.StateRolledBack)
	if _, err := f.fx.rel.Put(context.Background(), provision.RecordKey("bab", fromApp, "other.shop.example.com"), []byte("{not json"), ""); err != nil {
		t.Fatal(err)
	}
	f.markedStore(t, id)
	key := objstore.KeyIn("bab", fromApp, "canary")
	before := f.rawKey(t, f.fx.rel, key)
	if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil {
		t.Fatal(err)
	}
	if after := f.rawKey(t, f.fx.rel, key); after != before {
		t.Fatalf("a store an unreadable record may use was touched:\n before %s\n after  %s", before, after)
	}
}

// failingRecordList fails every listing of the app's records.
type failingRecordList struct {
	release.Store
	prefix string
}

func (s failingRecordList) DeleteIfMatch(ctx context.Context, key, etag string) error {
	return release.DeleteIfMatch(ctx, s.Store, key, etag)
}

func (s failingRecordList) List(ctx context.Context, prefix string) ([]string, error) {
	if prefix == s.prefix {
		return nil, errors.New("the bucket is unreachable")
	}
	return s.Store.List(ctx, prefix)
}

// When the records cannot be listed the finish cannot tell whether the
// store is shared, so it does nothing and answers an error for the sweep
// to retry.
func TestFinishSeedRetriesWhenTheRecordsCannotBeListed(t *testing.T) {
	f := finishFixture(t)
	id := newSeedID(t)
	f.seedRecord(t, fromHost, "canary", id, seeds.KindImport, seeds.StateRolledBack)
	f.markedStore(t, id)
	rel := f.fx.srv.cfg.Releases
	f.fx.srv.cfg.Releases = failingRecordList{Store: rel, prefix: provision.RecordsPrefix("bab", fromApp)}
	if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err == nil {
		t.Fatal("a failed listing answered nil")
	}
	if f.canaryStore(t) == nil {
		t.Fatal("the store record went on a failed listing")
	}
	if _, err := f.record(t, fromHost); err != nil {
		t.Fatalf("the record went on a failed listing: %v", err)
	}
}

// A store this seed made with no backend to tear it down with is an error,
// and the record keeps holding the host.
func TestFinishSeedNeedsAStorageBackend(t *testing.T) {
	f := finishFixture(t)
	id := newSeedID(t)
	f.seedRecord(t, fromHost, "canary", id, seeds.KindImport, seeds.StateRolledBack)
	f.markedStore(t, id)
	f.fx.srv.cfg.StoreProvisioner = nil
	if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err == nil {
		t.Fatal("no backend answered nil")
	}
	if _, err := f.record(t, fromHost); err != nil {
		t.Fatalf("the record went with its store still standing: %v", err)
	}
}

// An import's finish deletes its snapshot and never the host's replica:
// only a --from seed's replica is known to be its own.
func TestFinishSeedLeavesAnImportsReplica(t *testing.T) {
	f := finishFixture(t)
	id := newSeedID(t)
	f.seedRecord(t, fromHost, "canary", id, seeds.KindImport, seeds.StateRolledBack)
	ctx := context.Background()
	if _, err := f.data.Put(ctx, seeds.SnapshotKey(fromHost, id), []byte("snapshot"), ""); err != nil {
		t.Fatal(err)
	}
	dbdoortest.Replicate(t, filepath.Join(f.root, "instances", fromHost), 2)
	before := replicaKeys(t, f, fromHost)
	if err := f.fx.srv.finishSeed(ctx, "bab", fromApp, fromHost, id); err != nil {
		t.Fatal(err)
	}
	if _, _, err := f.data.Get(ctx, seeds.SnapshotKey(fromHost, id)); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("snapshot after the finish: %v; want it gone", err)
	}
	if after := replicaKeys(t, f, fromHost); len(before) == 0 || !slices.Equal(after, before) {
		t.Fatalf("an import's finish touched the replica:\n before %v\n after  %v", before, after)
	}
	f.requireRecordGone(t, fromHost)
}

// A seed id that fails seeds.Check builds no key and deletes nothing
// (preflight M8): every bad id would map to the one shared sentinel.
func TestFinishSeedChecksTheSeedBeforeAnyKey(t *testing.T) {
	f := finishFixture(t)
	ctx := context.Background()
	sentinel := seeds.SnapshotKey(fromHost, "sd_bad")
	if _, err := f.data.Put(ctx, sentinel, []byte("someone else's"), ""); err != nil {
		t.Fatal(err)
	}
	// A rolled-back record carrying the bad id: without the check, the
	// finish would match it and delete the sentinel.
	f.seedRecord(t, fromHost, "canary", "sd_bad", seeds.KindImport, seeds.StateRolledBack)
	for _, c := range []struct{ host, id string }{{fromHost, "sd_bad"}, {"../shop", newSeedID(t)}} {
		if err := f.fx.srv.finishSeed(ctx, "bab", fromApp, c.host, c.id); !errors.Is(err, seeds.ErrBadID) && !errors.Is(err, seeds.ErrBadHost) {
			t.Fatalf("finishSeed(%q, %q) = %v; want a seeds.Check error", c.host, c.id, err)
		}
	}
	if _, _, err := f.data.Get(ctx, sentinel); err != nil {
		t.Fatalf("the sentinel key was deleted: %v", err)
	}
}

// No record: nothing to finish, and no error.
func TestFinishSeedWithNoRecord(t *testing.T) {
	f := finishFixture(t)
	if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, newSeedID(t)); err != nil {
		t.Fatal(err)
	}
}

// rolledBackFrom writes a --from record rolled back on host, with replica
// keys under its prefix, and answers its seed id and those keys.
func (f fromFixture) rolledBackFrom(t *testing.T, host string) (string, []string) {
	t.Helper()
	id := newSeedID(t)
	f.seedRecord(t, host, "canary", id, seeds.KindReplica, seeds.StateRolledBack)
	return id, f.putReplicaKeys(t, host, 3)
}

// Review I1 (b): a lease that is present and unexpired by the lease
// package's own rule blocks the replica step. Nothing is deleted, the
// error is retryable and the record stays rolled-back. An absent lease, or
// one expired past its margin (a box that died), does not block.
func TestFinishSeedWaitsForTheHostsLease(t *testing.T) {
	// Expiries are relative to the moment each case writes its lease: the
	// margin is seconds, and earlier cases take most of one.
	for name, c := range map[string]struct {
		expiresIn *time.Duration
		blocked   bool
	}{
		"held":                {expiresIn: ptr(time.Minute), blocked: true},
		"just expired":        {expiresIn: ptr(-lease.RenewMargin / 2), blocked: true},
		"expired past margin": {expiresIn: ptr(-2 * lease.RenewMargin)},
		"absent":              {},
	} {
		t.Run(name, func(t *testing.T) {
			f := finishFixture(t)
			id, keys := f.rolledBackFrom(t, fromHost)
			if c.expiresIn != nil {
				f.putLease(t, fromHost, time.Now().Add(*c.expiresIn))
			}
			err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id)
			if !c.blocked {
				if err != nil {
					t.Fatal(err)
				}
				if got := replicaKeys(t, f, fromHost); len(got) != 0 {
					t.Fatalf("replica after the finish: %v", got)
				}
				f.requireRecordGone(t, fromHost)
				return
			}
			if err == nil {
				t.Fatal("a held lease answered nil")
			}
			if got := replicaKeys(t, f, fromHost); !slices.Equal(got, keys) {
				t.Fatalf("replica under a held lease:\n before %v\n after  %v", keys, got)
			}
			if got, err := f.record(t, fromHost); err != nil || got.Seed.State != seeds.StateRolledBack {
				t.Fatalf("record under a held lease = %+v, %v; want it kept, rolled-back", got, err)
			}
		})
	}
}

func ptr[T any](v T) *T { return &v }

// recreatingData runs recreate once: before the first Delete whose key
// onDelete matches, or after the first List whose prefix onList matches.
type recreatingData struct {
	release.Store
	onDelete, onList func(key string) bool
	recreate         func()
	done             bool
}

func (d *recreatingData) fire(match func(string) bool, key string) {
	if !d.done && match != nil && match(key) {
		d.done = true
		d.recreate()
	}
}

func (d *recreatingData) Delete(ctx context.Context, key string) error {
	d.fire(d.onDelete, key)
	return d.Store.Delete(ctx, key)
}

func (d *recreatingData) List(ctx context.Context, prefix string) ([]string, error) {
	keys, err := d.Store.List(ctx, prefix)
	d.fire(d.onList, prefix)
	return keys, err
}

// recreateHost deletes host's record and writes a new, seedless instance
// of it, as another finish and then a new create would.
func (f fromFixture) recreateHost(t *testing.T, host string) func() {
	return func() {
		ctx := context.Background()
		if err := provision.DeleteRecord(ctx, f.fx.rel, "bab", fromApp, host); err != nil {
			t.Error(err)
		}
		if err := provision.CreateRecord(ctx, f.fx.rel, provision.Record{Schema: 1, Host: host, Slug: "canary",
			App: fromApp, Account: "bab", Channel: "stable", Environment: "canary", CreatedAt: time.Now().UTC()}); err != nil {
			t.Error(err)
		}
	}
}

// Review I2: the replica delete is fenced on the record. A record deleted
// and created again for the same host, between the store teardown and the
// replica step, or while the replica is listed, keeps every replica key:
// they may be the new instance's. The finish answers an error (the
// record it was finishing is not gone in the sense that matters: the host
// has a record), and the new record is untouched.
func TestFinishSeedNeverDeletesANewInstancesReplica(t *testing.T) {
	for name, wrap := range map[string]func(f fromFixture) *recreatingData{
		"before the replica step": func(f fromFixture) *recreatingData {
			return &recreatingData{Store: f.data, recreate: f.recreateHost(t, fromHost),
				onDelete: func(key string) bool { return strings.HasPrefix(key, seeds.Prefix) }}
		},
		"while it is listed": func(f fromFixture) *recreatingData {
			return &recreatingData{Store: f.data, recreate: f.recreateHost(t, fromHost),
				onList: func(prefix string) bool { return strings.HasPrefix(prefix, "instances/") }}
		},
	} {
		t.Run(name, func(t *testing.T) {
			f := finishFixture(t)
			id, keys := f.rolledBackFrom(t, fromHost)
			d := wrap(f)
			f.fx.srv.cfg.SeedData = d
			if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err == nil {
				t.Fatal("a re-created host answered nil")
			}
			if !d.done {
				t.Fatal("the record was never re-created: the test did not reach its window")
			}
			if got := replicaKeys(t, f, fromHost); !slices.Equal(got, keys) {
				t.Fatalf("the new instance's replica:\n before %v\n after  %v", keys, got)
			}
			if got, err := f.record(t, fromHost); err != nil || got.Seed != nil {
				t.Fatalf("the new record = %+v, %v; want it untouched", got, err)
			}
			// The sweep's next finish of the old seed leaves the new
			// instance alone and is done.
			f.fx.srv.cfg.SeedData = f.data
			if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil {
				t.Fatal(err)
			}
			if got := replicaKeys(t, f, fromHost); !slices.Equal(got, keys) {
				t.Fatalf("the retry touched the new instance's replica: %v", got)
			}
		})
	}
}

// A record rewritten in place while the replica is listed (an ordinary
// writer, the same seed still rolled-back) stops the deletes too: the
// finish only deletes under the record it read. The next finish resumes.
func TestFinishSeedStopsWhenTheRecordChangesMidReplica(t *testing.T) {
	f := finishFixture(t)
	id, keys := f.rolledBackFrom(t, fromHost)
	key := provision.RecordKey("bab", fromApp, fromHost)
	f.fx.srv.cfg.SeedData = &recreatingData{Store: f.data,
		onList: func(prefix string) bool { return strings.HasPrefix(prefix, "instances/") },
		recreate: func() {
			ctx := context.Background()
			raw, etag, err := f.fx.rel.Get(ctx, key)
			if err == nil {
				_, err = f.fx.rel.Put(ctx, key, append(raw, '\n'), etag)
			}
			if err != nil {
				t.Error(err)
			}
		}}
	if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err == nil {
		t.Fatal("a record changed mid-replica answered nil")
	}
	if got := replicaKeys(t, f, fromHost); !slices.Equal(got, keys) {
		t.Fatalf("replica after the record changed:\n before %v\n after  %v", keys, got)
	}
	f.fx.srv.cfg.SeedData = f.data
	if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil {
		t.Fatal(err)
	}
	if got := replicaKeys(t, f, fromHost); len(got) != 0 {
		t.Fatalf("replica after the resumed finish: %v", got)
	}
	f.requireRecordGone(t, fromHost)
}

// A record deleted (not re-created) during the replica step was finished
// by another finish: nil, and nothing more is deleted.
func TestFinishSeedStopsWhenAnotherFinishDeletedTheRecord(t *testing.T) {
	f := finishFixture(t)
	id, keys := f.rolledBackFrom(t, fromHost)
	f.fx.srv.cfg.SeedData = &recreatingData{Store: f.data,
		onList: func(prefix string) bool { return strings.HasPrefix(prefix, "instances/") },
		recreate: func() {
			if err := provision.DeleteRecord(context.Background(), f.fx.rel, "bab", fromApp, fromHost); err != nil {
				t.Error(err)
			}
		}}
	if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil {
		t.Fatal(err)
	}
	if got := replicaKeys(t, f, fromHost); !slices.Equal(got, keys) {
		t.Fatalf("replica after the record went: %v", got)
	}
}

// The prefix stops at the host: a host that extends this one's name keeps
// its replica.
func TestFinishSeedReplicaPrefixStopsAtTheHost(t *testing.T) {
	f := finishFixture(t)
	id, _ := f.rolledBackFrom(t, fromHost)
	sibling := f.putReplicaKeys(t, fromHost+".au", 2)
	if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil {
		t.Fatal(err)
	}
	if got := replicaKeys(t, f, fromHost); len(got) != 0 {
		t.Fatalf("replica after the finish: %v", got)
	}
	if got := replicaKeys(t, f, fromHost+".au"); !slices.Equal(got, sibling) {
		t.Fatalf("the sibling host's replica:\n before %v\n after  %v", sibling, got)
	}
}

// A replica outside this console's data bucket (another region) is not
// known to be the seed's, and is left alone; the record still goes.
func TestFinishSeedLeavesAReplicaOutsideItsBucket(t *testing.T) {
	f := finishFixture(t)
	id := newSeedID(t)
	rec := provision.Record{Schema: 1, Host: fromHost, Slug: "canary", App: fromApp, Account: "bab", Channel: "stable",
		Environment: "canary", Region: "elsewhere", CreatedAt: time.Now().UTC(),
		Seed: &provision.SeedSource{ID: id, Kind: seeds.KindReplica, State: seeds.StateRolledBack, By: "bob@keymail.dev", At: time.Now().UTC()}}
	if err := provision.CreateRecord(context.Background(), f.fx.rel, rec); err != nil {
		t.Fatal(err)
	}
	keys := f.putReplicaKeys(t, fromHost, 2)
	if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil {
		t.Fatal(err)
	}
	if got := replicaKeys(t, f, fromHost); !slices.Equal(got, keys) {
		t.Fatalf("a replica outside this bucket was touched: %v", got)
	}
	f.requireRecordGone(t, fromHost)
}

// More instances than one listing holds: the finish cannot tell whether
// the store is shared, so it does nothing and answers an error.
func TestFinishSeedRefusesATruncatedListing(t *testing.T) {
	f := finishFixture(t)
	id := newSeedID(t)
	f.seedRecord(t, fromHost, "canary", id, seeds.KindImport, seeds.StateRolledBack)
	f.markedStore(t, id)
	for i := range instanceListMax {
		if err := provision.CreateRecord(context.Background(), f.fx.rel, provision.Record{Schema: 1, Host: fmt.Sprintf("i%d.shop.example.com", i),
			Slug: fmt.Sprintf("i%d", i), App: fromApp, Account: "bab", Channel: "stable", CreatedAt: time.Now().UTC()}); err != nil {
			t.Fatal(err)
		}
	}
	if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err == nil {
		t.Fatal("a truncated listing answered nil")
	}
	if f.canaryStore(t) == nil {
		t.Fatal("the store went on a truncated listing")
	}
	if _, err := f.record(t, fromHost); err != nil {
		t.Fatalf("the record went on a truncated listing: %v", err)
	}
}

// touchOnDelete rewrites the record once (an ordinary writer's update)
// just before the finish's conditional delete, so that delete loses.
type touchOnDelete struct {
	release.Store
	key     string
	touched bool
}

func (s *touchOnDelete) DeleteIfMatch(ctx context.Context, key, etag string) error {
	if key == s.key && !s.touched {
		s.touched = true
		raw, cur, err := s.Store.Get(ctx, key)
		if err != nil {
			return err
		}
		if _, err := s.Store.Put(ctx, key, append(raw, '\n'), cur); err != nil {
			return err
		}
	}
	return release.DeleteIfMatch(ctx, s.Store, key, etag)
}

// A record changed between the finish's read and its delete is read again
// once, and the finish completes.
func TestFinishSeedRetriesALostRecordDelete(t *testing.T) {
	f := finishFixture(t)
	id := newSeedID(t)
	f.seedRecord(t, fromHost, "canary", id, seeds.KindImport, seeds.StateRolledBack)
	rel := &touchOnDelete{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, fromHost)}
	f.fx.srv.cfg.Releases = rel
	if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil {
		t.Fatal(err)
	}
	if !rel.touched {
		t.Fatal("the record was never touched: the test did not reach its window")
	}
	f.requireRecordGone(t, fromHost)
}

// A console with no data bucket refuses before anything is destroyed: the
// store this seed made still stands, with the record.
func TestFinishSeedNeedsADataBucketBeforeTheTeardown(t *testing.T) {
	f := finishFixture(t)
	id := newSeedID(t)
	f.seedRecord(t, fromHost, "canary", id, seeds.KindImport, seeds.StateRolledBack)
	f.markedStore(t, id)
	f.fx.srv.cfg.SeedData = nil
	if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err == nil {
		t.Fatal("no data bucket answered nil")
	}
	if f.canaryStore(t) == nil {
		t.Fatal("the store was torn down by a finish that could not complete")
	}
	if _, err := f.record(t, fromHost); err != nil {
		t.Fatalf("the record went: %v", err)
	}
}

// A console with host claims armed finishes no seed.
func TestFinishSeedRefusesAClaimConsole(t *testing.T) {
	f := finishFixture(t)
	id := newSeedID(t)
	f.seedRecord(t, fromHost, "canary", id, seeds.KindImport, seeds.StateRolledBack)
	f.markedStore(t, id)
	f.fx.srv.cfg.HostClaim = true
	if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err == nil {
		t.Fatal("a claim console answered nil")
	}
	if f.canaryStore(t) == nil {
		t.Fatal("a claim console tore the store down")
	}
	if _, err := f.record(t, fromHost); err != nil {
		t.Fatalf("a claim console deleted the record: %v", err)
	}
}