package console

import (
	"bytes"
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"errors"
	"log"
	"net/http"
	"net/url"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/objstore"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/seeds"
)

const (
	importHost = "old.shop.example.com"
	importAddr = "127.0.0.1:8401"
)

var importSHA = strings.Repeat("a", 64)

// newImportFixture is Plan 5's --from fixture: the app shop, a ready
// one-box home region, a seeds floor, a scratch pool and a data bucket.
func newImportFixture(t *testing.T) fromFixture {
	t.Helper()
	return newFromFixture(t, 0)
}

func importBody(host, sha string, n int64, replaces string) string {
	body, _ := json.Marshal(map[string]any{"host": host, "import": map[string]any{"sha256": sha, "bytes": n, "replaces": replaces}})
	return string(body)
}

func beginImport(t *testing.T, f fromFixture, host, sha string, n int64) (int, map[string]any) {
	t.Helper()
	return f.create(t, importBody(host, sha, n, importAddr))
}

// requireBegun: a 201 holding a 43-character token, and an awaiting-upload
// import record that keeps only the token's hash.
func (f fromFixture) requireBegun(t *testing.T, code int, out map[string]any, host string) (*provision.Record, string) {
	t.Helper()
	if code != http.StatusCreated {
		t.Fatalf("begin = %d %v", code, out)
	}
	token := str(t, out, "verify_token")
	if len(token) != 43 {
		t.Fatalf("verify token %q", token)
	}
	rec, err := f.record(t, host)
	if err != nil || rec.Seed == nil || rec.Seed.Kind != seeds.KindImport || rec.Seed.State != seeds.StateAwaitingUpload ||
		!seeds.ValidID(rec.Seed.ID) || rec.Seed.VerifyHash != seeds.HashToken(token) || !seeds.TokenMatches(rec.Seed.VerifyHash, token) {
		t.Fatalf("record = %+v, %v", rec, err)
	}
	return rec, token
}

func TestImportBegin(t *testing.T) {
	f := newImportFixture(t)
	var logs bytes.Buffer
	log.SetOutput(&logs)
	defer log.SetOutput(os.Stderr)
	code, out := f.create(t, `{"host":"old.shop.example.com","message":"moving the shop","import":{"sha256":"`+importSHA+`","bytes":4096,"replaces":"127.0.0.1:8401"}}`)
	rec, token := f.requireBegun(t, code, out, importHost)
	if rec.Seed.SHA256 != importSHA || rec.Seed.Bytes != 4096 || rec.Seed.Replaces != importAddr || rec.Seed.By != "alice@keymail.dev" ||
		rec.Seed.From != "" || rec.Seed.TXID != "" || rec.Seed.Reason != "" || rec.Seed.At.IsZero() {
		t.Fatalf("record seed = %+v", rec.Seed)
	}
	if rec.Environment != "" || rec.Slug != "old" || rec.Channel != "stable" || rec.Region != "" {
		t.Fatalf("record = %+v; want the ordinary create's fields", rec)
	}
	seed, _ := out["seed"].(map[string]any)
	if out["host"] != importHost || seed == nil || seed["id"] != rec.Seed.ID || seed["state"] != seeds.StateAwaitingUpload {
		t.Fatalf("201 body = %v", out)
	}
	if _, ok := out["private_store_bucket"]; ok {
		t.Fatalf("201 body names a store on an app without private storage: %v", out)
	}

	// The token is answered once, here, and kept nowhere: not in the
	// bucket, not in the log, not in the audit row.
	raw, _, err := f.fx.rel.Get(context.Background(), provision.RecordKey("bab", fromApp, importHost))
	if err != nil || strings.Contains(string(raw), token) {
		t.Fatalf("the token itself reached the bucket (%v)", err)
	}
	if strings.Contains(logs.String(), token) {
		t.Fatal("the token reached the log")
	}
	events, err := f.fx.st.Events(OpsAccountID, fromApp, 50)
	if err != nil {
		t.Fatal(err)
	}
	var rows []string
	for _, e := range events {
		if strings.Contains(e.Detail+e.Message, token) {
			t.Errorf("the token reached an audit row: %+v", e)
		}
		switch e.Action {
		case "instance-seed":
			rows = append(rows, e.Detail+" | "+e.Message+" | "+e.Address)
		case "instance-create":
			t.Errorf("an import also wrote an instance-create row: %+v", e)
		}
	}
	want := "began an import of old.shop.example.com over 127.0.0.1:8401 (aaaaaaaaaaaa, 4096 bytes) | moving the shop | alice@keymail.dev"
	if len(rows) != 1 || rows[0] != want {
		t.Fatalf("instance-seed rows = %q, want [%q]", rows, want)
	}

	// A second begin for the same host: the record holds it.
	code, out = beginImport(t, f, importHost, importSHA, 4096)
	if code != http.StatusConflict || str(t, out, "error") != "instance old.shop.example.com already exists" {
		t.Fatalf("second begin = %d %v, want 409", code, out)
	}
	if again, err := f.record(t, importHost); err != nil || again.Seed.ID != rec.Seed.ID || again.Seed.VerifyHash != rec.Seed.VerifyHash {
		t.Fatalf("the second begin changed the record: %+v, %v", again, err)
	}
}

// Each begin mints its own seed id and token.
func TestImportBeginMintsFreshTokens(t *testing.T) {
	f := newImportFixture(t)
	code, out := beginImport(t, f, importHost, importSHA, 10)
	a, ta := f.requireBegun(t, code, out, importHost)
	code, out = beginImport(t, f, "new.shop.example.com", importSHA, 10)
	b, tb := f.requireBegun(t, code, out, "new.shop.example.com")
	if a.Seed.ID == b.Seed.ID || ta == tb || a.Seed.VerifyHash == b.Seed.VerifyHash {
		t.Fatalf("two begins share a seed or a token: %+v %+v", a.Seed, b.Seed)
	}
}

func TestImportBeginRefusals(t *testing.T) {
	f := newImportFixture(t)
	const ordinary = "a seeded instance is an ordinary instance. Drop --kind, --addr and --role."
	const oneDoor = "a create takes from or import, not both."
	const size = "bytes must be more than 0 and at most 4 GiB"
	const hash = "sha256 must be 64 lowercase hex characters"
	const replaces = "replaces must name the address the import takes over, as host:port"
	for _, c := range []struct {
		body string
		code int
		want string
	}{
		{importBody("a.shop.example.com", importSHA, 0, "127.0.0.1:1"), http.StatusBadRequest, size},
		{importBody("a.shop.example.com", importSHA, -1, "127.0.0.1:1"), http.StatusBadRequest, size},
		{importBody("a.shop.example.com", importSHA, 4294967297, "127.0.0.1:1"), http.StatusRequestEntityTooLarge, size},
		{importBody("a.shop.example.com", "ABC", 10, "127.0.0.1:1"), http.StatusBadRequest, hash},
		{importBody("a.shop.example.com", strings.Repeat("A", 64), 10, "127.0.0.1:1"), http.StatusBadRequest, hash},
		{importBody("a.shop.example.com", strings.Repeat("a", 63), 10, "127.0.0.1:1"), http.StatusBadRequest, hash},
		{importBody("a.shop.example.com", strings.Repeat("g", 64), 10, "127.0.0.1:1"), http.StatusBadRequest, hash},
		{importBody("a.shop.example.com", importSHA, 10, "nope"), http.StatusBadRequest, replaces},
		{importBody("a.shop.example.com", importSHA, 10, ":8401"), http.StatusBadRequest, replaces},
		{importBody("a.shop.example.com", importSHA, 10, "127.0.0.1:http"), http.StatusBadRequest, replaces},
		{importBody("a.shop.example.com", importSHA, 10, "127.0.0.1:0"), http.StatusBadRequest, replaces},
		{importBody("a.shop.example.com", importSHA, 10, "127.0.0.1:65536"), http.StatusBadRequest, replaces},
		{importBody("a.shop.example.com", importSHA, 10, ""), http.StatusBadRequest, replaces},
		{`{"host":"a.shop.example.com","kind":"service","import":{"sha256":"` + importSHA + `","bytes":10,"replaces":"127.0.0.1:1"}}`, http.StatusBadRequest, ordinary},
		{`{"host":"a.shop.example.com","addrs":["127.0.0.1:9"],"import":{"sha256":"` + importSHA + `","bytes":10,"replaces":"127.0.0.1:1"}}`, http.StatusBadRequest, ordinary},
		// One seed door per create.
		{`{"host":"a.shop.example.com","from":{"host":"main.shop.example.com"},"import":{"sha256":"` + importSHA + `","bytes":10,"replaces":"127.0.0.1:1"}}`, http.StatusBadRequest, oneDoor},
	} {
		if code, out := f.create(t, c.body); code != c.code || str(t, out, "error") != c.want {
			t.Errorf("%s = %d %v, want %d %q", c.body, code, out, c.code, c.want)
		}
		f.requireNoRecord(t, "a.shop.example.com")
	}
	// An addr with a host name and the largest size are taken.
	code, out := f.create(t, importBody(importHost, importSHA, 4294967296, "legacy.internal:8401"))
	f.requireBegun(t, code, out, importHost)
}

func TestImportBeginNeedsOneBox(t *testing.T) {
	f := newImportFixture(t)
	seedReadyBox(t, f.fx, "kass-2", "", 1, "v0.27.0", "v0.27.0", "nested", false)
	code, out := beginImport(t, f, importHost, importSHA, 10)
	if code != http.StatusConflict || str(t, out, "error") != "the home region has more than one box. An import needs a region with one box for now." {
		t.Fatalf("= %d %v", code, out)
	}
	f.requireNoRecord(t, importHost)
}

func TestImportBeginRefusedBeforeTheFloor(t *testing.T) {
	f := newImportFixture(t)
	seedReadyBox(t, f.fx, "kass-1", "", 0, "v0.27.0", "v0.27.0", "nested", false)
	code, out := beginImport(t, f, importHost, importSHA, 10)
	if code != http.StatusConflict || !strings.Contains(str(t, out, "error"), "below the seeds floor") {
		t.Fatalf("= %d %v", code, out)
	}
	f.requireNoRecord(t, importHost)
}

func TestImportBeginRefusedWithClaimsArmed(t *testing.T) {
	f := newImportFixture(t)
	f.fx.srv.cfg.HostClaim = true
	code, out := beginImport(t, f, importHost, importSHA, 10)
	if code != http.StatusConflict || str(t, out, "error") != "This console runs with claim control on. Seeded instances are not supported here yet." {
		t.Fatalf("= %d %v", code, out)
	}
	f.requireNoRecord(t, importHost)
}

// The import is homed in this console's own region: its snapshot goes to
// this region's data bucket.
func TestImportBeginRefusesAnotherRegion(t *testing.T) {
	f := newImportFixture(t)
	seedReadyBox(t, f.fx, "far-1", "us", 1, "v0.27.0", "v0.27.0", "nested", false)
	code, out := f.create(t, `{"host":"old.shop.example.com","region":"us","import":{"sha256":"`+importSHA+`","bytes":10,"replaces":"127.0.0.1:8401"}}`)
	if code != http.StatusConflict || str(t, out, "error") != "an import runs in this console's region, the home region. To run old.shop.example.com elsewhere, import it here, commit it, and then run carlos move." {
		t.Fatalf("= %d %v", code, out)
	}
	f.requireNoRecord(t, importHost)
}

func TestImportBeginNeedsADataBucket(t *testing.T) {
	f := newImportFixture(t)
	f.fx.srv.cfg.SeedData = nil
	code, out := beginImport(t, f, importHost, importSHA, 10)
	if code != http.StatusServiceUnavailable || str(t, out, "error") != "this console has no data bucket set up, so it cannot create a seeded instance." {
		t.Fatalf("= %d %v", code, out)
	}
	f.requireNoRecord(t, importHost)
}

func TestImportBeginRefusesAPooledApp(t *testing.T) {
	f := newImportFixture(t)
	m := fromManifest(false)
	m.Pool = "shared"
	enableInstances(t, f.fx, fromApp, m)
	code, out := beginImport(t, f, importHost, importSHA, 10)
	if code != http.StatusConflict || str(t, out, "error") != "shop places its instances in a pool or on a fleet, so it cannot take a seeded instance." {
		t.Fatalf("= %d %v", code, out)
	}
	f.requireNoRecord(t, importHost)
}

// A failed record write answers SD-N67 and leaves the host free.
func TestImportBeginRecordWriteFails(t *testing.T) {
	f := newImportFixture(t)
	fail := &failOnceOn{Store: f.fx.srv.cfg.Releases, key: provision.RecordKey("bab", fromApp, importHost)}
	fail.armed.Store(true)
	f.fx.srv.cfg.Releases = fail
	code, out := beginImport(t, f, importHost, importSHA, 10)
	if code != http.StatusBadGateway || str(t, out, "error") != "could not store the upload. Try again." {
		t.Fatalf("= %d %v", code, out)
	}
	f.requireNoRecord(t, importHost)
	code, out = beginImport(t, f, importHost, importSHA, 10)
	f.requireBegun(t, code, out, importHost)
}

// With private storage the import gets the derived environment and a new
// store marked with its seed, named on the 201 as the ordinary create does.
func TestImportBeginPrivateStorage(t *testing.T) {
	f := newImportFixture(t)
	enableInstances(t, f.fx, fromApp, fromManifest(true))
	code, out := f.create(t, `{"host":"old.shop.example.com","environment":"main","import":{"sha256":"`+importSHA+`","bytes":10,"replaces":"127.0.0.1:8401"}}`)
	if code != http.StatusConflict || str(t, out, "error") != "a seeded instance gets its own environment and an empty store" {
		t.Fatalf("explicit environment = %d %v", code, out)
	}
	f.requireNoRecord(t, importHost)

	code, out = beginImport(t, f, importHost, importSHA, 10)
	rec, _ := f.requireBegun(t, code, out, importHost)
	if rec.Environment != "old" {
		t.Fatalf("record environment = %q, want the host's own", rec.Environment)
	}
	store, _, err := objstore.GetIn(context.Background(), f.fx.rel, "bab", fromApp, "old")
	if err != nil || store == nil || store.Declared == nil || store.Declared.Seed != rec.Seed.ID || storeBucketName(store) == "" {
		t.Fatalf("old store = %+v, %v; want one marked with seed %s", store, err, rec.Seed.ID)
	}
	if str(t, out, "private_store_bucket") != storeBucketName(store) {
		t.Fatalf("201 private_store_bucket = %v, want %s", out["private_store_bucket"], storeBucketName(store))
	}
}

// An existing store for the derived environment is refused before the
// record is written.
func TestImportBeginRefusesATakenStore(t *testing.T) {
	f := newImportFixture(t)
	enableInstances(t, f.fx, fromApp, fromManifest(true))
	foreign := grantedStore("carlos-bab-shop-old")
	if _, err := objstore.PutIn(context.Background(), f.fx.rel, "bab", fromApp, "old", foreign, ""); err != nil {
		t.Fatal(err)
	}
	var seen *provision.Record
	f.fx.srv.cfg.Releases = &recordSpy{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, importHost), seen: &seen}
	code, out := beginImport(t, f, importHost, importSHA, 10)
	if seen != nil {
		t.Fatalf("a record was written before the refusal: %+v", seen)
	}
	if code != http.StatusConflict || str(t, out, "error") != "environment old already has a store (carlos-bab-shop-old). A seeded instance needs a new, empty one." {
		t.Fatalf("= %d %v", code, out)
	}
	f.requireNoRecord(t, importHost)
}

// The store gates that write nothing refuse before the record is written
// (ruling R3, preflight M2).
func TestImportBeginStoreGatesRefuseBeforeTheRecord(t *testing.T) {
	f := newImportFixture(t)
	enableInstances(t, f.fx, fromApp, fromManifest(true))
	f.fx.srv.cfg.SecretsPub = ""
	var seen *provision.Record
	f.fx.srv.cfg.Releases = &recordSpy{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, importHost), seen: &seen}
	code, out := beginImport(t, f, importHost, importSHA, 10)
	if code != http.StatusServiceUnavailable || str(t, out, "error") != "this console has no secrets key (CARLOS_SECRETS_PUBKEY) — it cannot deliver a store credential, so it will not mint one" {
		t.Fatalf("= %d %v", code, out)
	}
	if seen != nil {
		t.Fatalf("a record was written before the refusal: %+v", seen)
	}
	f.requireNoRecord(t, importHost)
}

func grantedStore(bucket string) *objstore.Record {
	at := time.Now().UTC().Add(-time.Hour)
	return &objstore.Record{
		Declared: &objstore.Declared{EnvPrefix: "S3", At: at, By: "bob@keymail.dev"},
		Grant:    &objstore.GrantBlock{Backend: "mem", Bucket: bucket, Source: objstore.GrantSourceAccountApproval, At: at, By: "operator@keymail.dev"},
	}
}

// A store walk refused after the record was written rolls the import back
// and finishes it before answering, so the same begin can run again at once.
func TestImportBeginStoreWalkRefusalFreesTheHost(t *testing.T) {
	f := newImportFixture(t)
	enableInstances(t, f.fx, fromApp, fromManifest(true))
	// Another environment's store already uses the name the walk derives.
	if _, err := objstore.PutIn(context.Background(), f.fx.rel, "bab", fromApp, "staging", grantedStore("carlos-bab-shop-old"), ""); err != nil {
		t.Fatal(err)
	}
	code, out := beginImport(t, f, importHost, importSHA, 10)
	if code != http.StatusConflict || str(t, out, "error") != "another store already uses this instance's storage names (app and environment names are joined with \"-\"), so the instance was not created. use a different host name" {
		t.Fatalf("= %d %v", code, out)
	}
	f.requireNoRecord(t, importHost)
	if err := f.fx.rel.Delete(context.Background(), objstore.KeyIn("bab", fromApp, "staging")); err != nil {
		t.Fatal(err)
	}
	code, out = beginImport(t, f, importHost, importSHA, 10)
	f.requireBegun(t, code, out, importHost)
}

// storeOnRecordWrite writes a foreign, granted store for env right after
// the record's create lands: the store another writer made between the
// door's check and its walk.
type storeOnRecordWrite struct {
	release.Store
	key, env string
	once     sync.Once
}

func (s *storeOnRecordWrite) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	tag, err := s.Store.Put(ctx, key, body, ifMatch)
	if err == nil && key == s.key && ifMatch == "" {
		s.once.Do(func() {
			if _, perr := objstore.PutIn(ctx, s.Store, "bab", fromApp, s.env, grantedStore("carlos-bab-shop-"+s.env), ""); perr != nil {
				panic(perr)
			}
		})
	}
	return tag, err
}

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

// The walk's own re-check under the store lock (errSeedStoreTaken) answers
// SD-N34, rolls the import back and frees the host; the other writer's
// store is left as it is.
func TestImportBeginStoreTakenUnderTheLock(t *testing.T) {
	f := newImportFixture(t)
	enableInstances(t, f.fx, fromApp, fromManifest(true))
	f.fx.srv.cfg.Releases = &storeOnRecordWrite{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, importHost), env: "old"}
	code, out := beginImport(t, f, importHost, importSHA, 10)
	if code != http.StatusConflict || str(t, out, "error") != "environment old already has a store (carlos-bab-shop-old). A seeded instance needs a new, empty one." {
		t.Fatalf("= %d %v", code, out)
	}
	f.requireNoRecord(t, importHost)
	store, _, err := objstore.GetIn(context.Background(), f.fx.rel, "bab", fromApp, "old")
	if err != nil || store == nil || store.Grant == nil || store.Grant.Bucket != "carlos-bab-shop-old" {
		t.Fatalf("the other writer's store = %+v, %v; want it kept", store, err)
	}
}

// apexOnBucket creates the app label as the store walk makes its bucket:
// a sibling app taking the host's platform apex mid-begin. It records the
// seed the store was declared for, as the walk sees it.
type apexOnBucket struct {
	*objstore.MemProvisioner
	f        fromFixture
	label    string
	declared string
}

func (p *apexOnBucket) EnsureBucket(ctx context.Context, name string) (string, error) {
	if store, _, err := objstore.GetIn(ctx, p.f.fx.rel, "bab", fromApp, p.label); err == nil && store != nil && store.Declared != nil {
		p.declared = store.Declared.Seed
	}
	if _, err := p.f.fx.st.CreateAppIn(OpsAccountID, p.label, "paul@keymail.dev", time.Now()); err != nil {
		return "", err
	}
	return p.MemProvisioner.EnsureBucket(ctx, name)
}

// Preflight m11 and M3: a begin on a private-storage app whose host is
// claimed as a sibling app's apex after the record was written. The store
// was made for this seed; the begin rolls the import back, the finish
// tears that store down and deletes the record, and the create door's own
// sentence is answered.
func TestImportBeginSiblingApexTearsDownTheStore(t *testing.T) {
	f := newImportFixture(t)
	f.fx.srv.cfg.AppsDomain = "apps.example.net"
	m := fromManifest(true)
	m.Domain = "bab.apps.example.net"
	enableInstances(t, f.fx, fromApp, m)
	p := &apexOnBucket{MemProvisioner: f.fake, f: f, label: "old"}
	f.fx.srv.cfg.StoreProvisioner = p
	var seen *provision.Record
	f.fx.srv.cfg.Releases = &recordSpy{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, "old.bab.apps.example.net"), seen: &seen}

	code, out := beginImport(t, f, "old.bab.apps.example.net", importSHA, 10)
	if code != http.StatusConflict || str(t, out, "error") != "old.bab.apps.example.net is app old's platform apex. That app was created while this instance was being written, so the instance was not kept." {
		t.Fatalf("= %d %v", code, out)
	}
	if seen == nil || seen.Seed == nil || !seeds.ValidID(seen.Seed.ID) || p.declared != seen.Seed.ID {
		t.Fatalf("the store was declared for seed %q; the record written was %+v", p.declared, seen)
	}
	f.requireNoRecord(t, "old.bab.apps.example.net")
	if store, _, err := objstore.GetIn(context.Background(), f.fx.rel, "bab", fromApp, "old"); err != nil || store != nil {
		t.Fatalf("the seed's store record = %+v, %v; want it torn down", store, err)
	}
	if _, ok := f.fake.Buckets["carlos-bab-shop-old"]; ok {
		t.Fatal("the seed's bucket survived the finish")
	}
	if len(f.fake.Buckets) != 0 {
		t.Fatalf("buckets left = %v", f.fake.Buckets)
	}
}

// recordSpy keeps the first create-only write of key's body.
type recordSpy struct {
	release.Store
	key  string
	seen **provision.Record
}

func (s *recordSpy) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if key == s.key && ifMatch == "" && *s.seen == nil {
		var rec provision.Record
		if err := json.Unmarshal(body, &rec); err == nil {
			*s.seen = &rec
		}
	}
	return s.Store.Put(ctx, key, body, ifMatch)
}

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

// Begin uploads nothing: the snapshot is the upload's (Task 2).
func TestImportBeginUploadsNothing(t *testing.T) {
	f := newImportFixture(t)
	code, out := beginImport(t, f, importHost, importSHA, 10)
	f.requireBegun(t, code, out, importHost)
	f.requireNothingUploaded(t)
	f.requireClean(t)
}

// ---- Upload (spec §5.2, preflight M5 to M8) ----

const (
	importMismatch = "the upload does not match the length and sha256 this import began with, so it was not kept."
	importNotADB   = "the upload is not a database that passes an integrity check, so it was not kept."
	importSlow     = "checking the upload took longer than 30 minutes, so it was not kept."
)

// uploadImport puts body to host's import door and returns what it
// answered.
func uploadImport(t *testing.T, f fromFixture, host string, body []byte) (int, map[string]any) {
	t.Helper()
	code, out, err := putImport(f, host, body)
	if err != nil {
		t.Fatal(err)
	}
	return code, out
}

// putImport is uploadImport for a goroutine: it returns its error. It
// names the seed host's record holds as the request is made, as the
// standalone verbs do (final review F1).
func putImport(f fromFixture, host string, body []byte) (int, map[string]any, error) {
	return putImportFor(f, host, heldSeed(f, host), body)
}

// heldSeed is the seed id host's record holds now, or "" with no record or
// no seed. It never fails a test, so goroutines may call it.
func heldSeed(f fromFixture, host string) string {
	rec, err := provision.GetRecord(context.Background(), f.fx.rel, "bab", fromApp, host)
	if err != nil || rec.Seed == nil {
		return ""
	}
	return rec.Seed.ID
}

// importDoorURL is host's import door (verb "" for the upload), naming id.
func importDoorURL(f fromFixture, host, verb, id string) string {
	u := f.url + "/" + host + "/import"
	if verb != "" {
		u += "/" + verb
	}
	return u + "?seed=" + url.QueryEscape(id)
}

// putImportFor is putImport naming seed id.
func putImportFor(f fromFixture, host, id string, body []byte) (int, map[string]any, error) {
	req, err := http.NewRequest(http.MethodPut, importDoorURL(f, host, "", id), bytes.NewReader(body))
	if err != nil {
		return 0, nil, err
	}
	req.Header.Set("Authorization", "Bearer "+f.token)
	res, err := http.DefaultClient.Do(req)
	if err != nil {
		return 0, nil, err
	}
	defer res.Body.Close()
	var out map[string]any
	_ = json.NewDecoder(res.Body).Decode(&out)
	return res.StatusCode, out, nil
}

// sqliteBytes is a small, sound SQLite database and its sha256.
func sqliteBytes(t *testing.T) ([]byte, string) {
	t.Helper()
	return sqliteBytesIn(t, "")
}

// sqliteBytesIn is sqliteBytes with a journal mode set first ("" keeps
// the default).
func sqliteBytesIn(t *testing.T, journal string) ([]byte, string) {
	t.Helper()
	p := filepath.Join(t.TempDir(), "snap.db")
	db, err := sql.Open("sqlite", p)
	if err != nil {
		t.Fatal(err)
	}
	if journal != "" {
		if _, err := db.Exec(`PRAGMA journal_mode=` + journal); err != nil {
			t.Fatal(err)
		}
	}
	if _, err := db.Exec(`CREATE TABLE t (x INTEGER); INSERT INTO t VALUES (1)`); err != nil {
		t.Fatal(err)
	}
	db.Close()
	b, err := os.ReadFile(p)
	if err != nil {
		t.Fatal(err)
	}
	return b, sha256Of(b)
}

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

// begun begins an import of host for body and returns its record.
func (f fromFixture) begun(t *testing.T, host string, body []byte) *provision.Record {
	t.Helper()
	code, out := beginImport(t, f, host, sha256Of(body), int64(len(body)))
	rec, _ := f.requireBegun(t, code, out, host)
	return rec
}

// uploaded puts body, requires the 202 that hands it to the check, and
// waits for the detached half to finish.
func (f fromFixture) uploaded(t *testing.T, host string, body []byte) *provision.Record {
	t.Helper()
	if code, out := uploadImport(t, f, host, body); code != http.StatusAccepted || len(out) != 1 || out["state"] != "checking" {
		t.Fatalf("upload = %d %v, want 202 {\"state\":\"checking\"}", code, out)
	}
	f.fx.srv.seedCopies.Wait()
	rec, err := f.record(t, host)
	if err != nil || rec.Seed == nil {
		t.Fatalf("record %s after its upload = %+v, %v", host, rec, err)
	}
	return rec
}

func requireImportState(t *testing.T, rec *provision.Record, id, state, reason string) {
	t.Helper()
	if rec.Seed == nil || rec.Seed.ID != id || rec.Seed.Kind != seeds.KindImport || rec.Seed.State != state || rec.Seed.Reason != reason {
		t.Fatalf("record %s seed = %+v; want %s %s with reason %q", rec.Host, rec.Seed, id, state, reason)
	}
}

func TestImportUploadStoresTheSnapshot(t *testing.T) {
	f := newImportFixture(t)
	body, _ := sqliteBytes(t)
	begun := f.begun(t, importHost, body)
	rec := f.uploaded(t, importHost, body)
	requireImportState(t, rec, begun.Seed.ID, seeds.StateUploaded, "")
	got, _, err := f.data.Get(context.Background(), seeds.SnapshotKey(importHost, begun.Seed.ID))
	if err != nil || !bytes.Equal(got, body) {
		t.Fatalf("snapshot = %d bytes, %v; want the upload", len(got), err)
	}
	if keys, _ := f.data.List(context.Background(), seeds.Prefix); len(keys) != 1 {
		t.Fatalf("seeds/ = %v; want the one snapshot", keys)
	}
	f.requireClean(t)
	events, err := f.fx.st.Events(OpsAccountID, fromApp, 50)
	if err != nil {
		t.Fatal(err)
	}
	var rows []string
	for _, e := range events {
		if e.Action == "import-uploaded" {
			rows = append(rows, e.Detail+" | "+e.Target+" | "+e.Address)
		}
	}
	if want := "uploaded the snapshot for old.shop.example.com | old.shop.example.com | alice@keymail.dev"; len(rows) != 1 || rows[0] != want {
		t.Fatalf("import-uploaded rows = %q, want [%q]", rows, want)
	}
}

// The check runs after the answer, detached from the request (preflight
// M6): the record waits in awaiting-upload, the spool sits in its own
// from- directory (M5), and the reservation holds off a second upload.
func TestImportUploadChecksAfterTheAnswer(t *testing.T) {
	f := newImportFixture(t)
	body, _ := sqliteBytes(t)
	begun := f.begun(t, importHost, body)
	type checking struct {
		path     string
		deadline time.Duration
	}
	entered := make(chan checking, 1)
	proceed := make(chan struct{})
	old := checkSeedDBConsole
	defer func() { checkSeedDBConsole = old }()
	checkSeedDBConsole = func(ctx context.Context, path string) error {
		d, _ := ctx.Deadline()
		entered <- checking{path, time.Until(d)}
		<-proceed
		if ctx.Err() != nil {
			return ctx.Err()
		}
		return old(ctx, path)
	}
	if code, out := uploadImport(t, f, importHost, body); code != http.StatusAccepted || out["state"] != "checking" {
		t.Fatalf("upload = %d %v", code, out)
	}
	c := <-entered
	if dir := filepath.Dir(c.path); filepath.Base(c.path) != importHost+".db" || filepath.Dir(dir) != f.fx.srv.seedScratch.dir() ||
		!strings.HasPrefix(filepath.Base(dir), seedScratchPrefix) {
		t.Errorf("spool %s; want <scratch>/%s*/%s.db", c.path, seedScratchPrefix, importHost)
	}
	if c.deadline < 29*time.Minute || c.deadline > 30*time.Minute {
		t.Errorf("the check's deadline is %v away, want 30 minutes", c.deadline)
	}
	rec, _ := f.record(t, importHost)
	requireImportState(t, rec, begun.Seed.ID, seeds.StateAwaitingUpload, "")
	if code, out := uploadImport(t, f, importHost, body); code != http.StatusTooManyRequests ||
		str(t, out, "error") != "another copy or upload is running for shop. Try again when it finishes." {
		t.Errorf("a second upload while the first is checked = %d %v, want 429", code, out)
	}
	close(proceed)
	f.fx.srv.seedCopies.Wait()
	rec, _ = f.record(t, importHost)
	requireImportState(t, rec, begun.Seed.ID, seeds.StateUploaded, "")
	f.requireClean(t)
}

// A length or hash that differs is refused before the answer, and the
// import stays awaiting-upload, so the right file can still be sent.
func TestImportUploadRefusesLengthAndHash(t *testing.T) {
	f := newImportFixture(t)
	body, _ := sqliteBytes(t)
	begun := f.begun(t, importHost, body)
	bad := append([]byte{}, body...)
	bad[len(bad)-1] ^= 0xff
	for name, b := range map[string][]byte{
		"one byte long":  append(append([]byte{}, body...), 'x'),
		"one byte short": body[:len(body)-1],
		"empty":          {},
		"wrong hash":     bad,
	} {
		if code, out := uploadImport(t, f, importHost, b); code != http.StatusUnprocessableEntity || str(t, out, "error") != importMismatch {
			t.Errorf("%s = %d %v, want 422", name, code, out)
		}
		f.fx.srv.seedCopies.Wait()
		rec, _ := f.record(t, importHost)
		requireImportState(t, rec, begun.Seed.ID, seeds.StateAwaitingUpload, "")
		f.requireClean(t)
		f.requireNothingUploaded(t)
	}
	requireImportState(t, f.uploaded(t, importHost, body), begun.Seed.ID, seeds.StateUploaded, "")
}

// A body far over its declared length is cut at bytes+1: the spool never
// holds more than that (preflight M7).
func TestImportUploadCutsALongBody(t *testing.T) {
	f := newImportFixture(t)
	body, _ := sqliteBytes(t)
	begun := f.begun(t, importHost, body)
	var spooled atomic.Int64
	spooled.Store(-1)
	defer func() { importSpoolHook = nil }()
	importSpoolHook = func(path string) {
		if st, err := os.Stat(path); err == nil {
			spooled.Store(st.Size())
		}
	}
	long := append(append([]byte{}, body...), bytes.Repeat([]byte{'x'}, 1<<20)...)
	if code, out := uploadImport(t, f, importHost, long); code != http.StatusUnprocessableEntity || str(t, out, "error") != importMismatch {
		t.Fatalf("a body 1 MiB long = %d %v, want 422", code, out)
	}
	if n := spooled.Load(); n < 0 || n > int64(len(body))+1 {
		t.Fatalf("the spool held %d bytes; want at most %d", n, len(body)+1)
	}
	rec, _ := f.record(t, importHost)
	requireImportState(t, rec, begun.Seed.ID, seeds.StateAwaitingUpload, "")
	f.requireClean(t)
	f.requireNothingUploaded(t)
}

func TestImportUploadOnlyOnce(t *testing.T) {
	f := newImportFixture(t)
	body, _ := sqliteBytes(t)
	f.begun(t, importHost, body)
	f.uploaded(t, importHost, body)
	if code, out := uploadImport(t, f, importHost, body); code != http.StatusConflict ||
		str(t, out, "error") != "old.shop.example.com is uploaded. It takes an upload only while it waits for one." {
		t.Fatalf("second upload = %d %v", code, out)
	}
	f.requireClean(t)
}

// A file of the right length and hash that is not a database is aborted,
// saying why (SD-N52 as the seed's reason).
func TestImportUploadRefusesANonDatabase(t *testing.T) {
	f := newImportFixture(t)
	junk := bytes.Repeat([]byte("j"), 100)
	begun := f.begun(t, "junk.shop.example.com", junk)
	requireImportState(t, f.uploaded(t, "junk.shop.example.com", junk), begun.Seed.ID, seeds.StateAborted, importNotADB)
	f.requireClean(t)
	f.requireNothingUploaded(t)
}

func TestImportUploadCheckDeadline(t *testing.T) {
	f := newImportFixture(t)
	body, _ := sqliteBytes(t)
	begun := f.begun(t, importHost, body)
	oldCheck, oldDeadline := checkSeedDBConsole, importCheckDeadline
	defer func() { checkSeedDBConsole, importCheckDeadline = oldCheck, oldDeadline }()
	importCheckDeadline = time.Nanosecond
	checkSeedDBConsole = func(ctx context.Context, path string) error {
		<-ctx.Done()
		return ctx.Err()
	}
	requireImportState(t, f.uploaded(t, importHost, body), begun.Seed.ID, seeds.StateAborted, importSlow)
	f.requireClean(t)
	f.requireNothingUploaded(t)
}

func TestImportUploadPutFailureAborts(t *testing.T) {
	f := newImportFixture(t)
	body, _ := sqliteBytes(t)
	begun := f.begun(t, importHost, body)
	f.fx.srv.cfg.SeedData = failingPuts{f.data}
	requireImportState(t, f.uploaded(t, importHost, body), begun.Seed.ID, seeds.StateAborted, importFailed)
	f.requireClean(t)
	f.requireNothingUploaded(t)
}

// The put is create-only: an object already at the seed's key is never
// replaced, and the upload that lost stops without aborting the import.
func TestImportUploadNeverReplacesASnapshot(t *testing.T) {
	f := newImportFixture(t)
	body, _ := sqliteBytes(t)
	begun := f.begun(t, importHost, body)
	key := seeds.SnapshotKey(importHost, begun.Seed.ID)
	if _, err := f.data.Put(context.Background(), key, []byte("another writer's snapshot"), ""); err != nil {
		t.Fatal(err)
	}
	requireImportState(t, f.uploaded(t, importHost, body), begun.Seed.ID, seeds.StateAwaitingUpload, "")
	if got := f.rawKey(t, f.data, key); got != "another writer's snapshot" {
		t.Fatalf("snapshot = %q; want it left as it was", got)
	}
	f.requireClean(t)
}

// A WAL database leaves -shm and -wal beside the spool when it is checked;
// the spool's directory goes with them (preflight M5).
func TestImportUploadOfAWALDatabase(t *testing.T) {
	f := newImportFixture(t)
	body, _ := sqliteBytesIn(t, "WAL")
	begun := f.begun(t, importHost, body)
	var sidecars []string
	old := checkSeedDBConsole
	defer func() { checkSeedDBConsole = old }()
	checkSeedDBConsole = func(ctx context.Context, path string) error {
		err := old(ctx, path)
		sidecars, _ = filepath.Glob(path + "-*")
		return err
	}
	requireImportState(t, f.uploaded(t, importHost, body), begun.Seed.ID, seeds.StateUploaded, "")
	if len(sidecars) == 0 {
		t.Fatal("the check left no sidecar files; this test no longer covers their removal")
	}
	f.requireClean(t)
}

// Two uploads of one import, the second reading the record before the
// first's last write and reserving after it (Important 1): the second is
// refused under its reservation, and the first's work stands.
func TestImportUploadSecondUploadAfterTheFirstLands(t *testing.T) {
	f := newImportFixture(t)
	body, _ := sqliteBytes(t)
	begun := f.begun(t, importHost, body)
	puts := &countingPuts{Store: f.data}
	f.fx.srv.cfg.SeedData = puts

	firstChecked := make(chan struct{})
	finishFirst := make(chan struct{})
	old := checkSeedDBConsole
	defer func() { checkSeedDBConsole = old }()
	checkSeedDBConsole = func(ctx context.Context, path string) error {
		close(firstChecked)
		<-finishFirst
		return old(ctx, path)
	}
	secondRead := make(chan struct{})
	reserveSecond := make(chan struct{})
	var calls atomic.Int32
	defer func() { importReserveHook = nil }()
	importReserveHook = func(string) {
		if calls.Add(1) == 2 {
			close(secondRead)
			<-reserveSecond
		}
	}

	if code, out := uploadImport(t, f, importHost, body); code != http.StatusAccepted {
		t.Fatalf("first upload = %d %v", code, out)
	}
	<-firstChecked
	type answer struct {
		code int
		out  map[string]any
		err  error
	}
	second := make(chan answer, 1)
	go func() {
		code, out, err := putImport(f, importHost, body)
		second <- answer{code, out, err}
	}()
	<-secondRead // the second has read awaiting-upload
	close(finishFirst)
	f.fx.srv.seedCopies.Wait() // the first is uploaded and has released
	close(reserveSecond)
	got := <-second
	if got.err != nil || got.code != http.StatusConflict ||
		str(t, got.out, "error") != "old.shop.example.com is uploaded. It takes an upload only while it waits for one." {
		t.Fatalf("second upload = %d %v %v; want 409 SD-N53", got.code, got.out, got.err)
	}
	f.fx.srv.seedCopies.Wait()
	rec, _ := f.record(t, importHost)
	requireImportState(t, rec, begun.Seed.ID, seeds.StateUploaded, "")
	if n := puts.n.Load(); n != 1 {
		t.Fatalf("%d snapshot puts; want the first upload's one", n)
	}
	if got, _, err := f.data.Get(context.Background(), seeds.SnapshotKey(importHost, begun.Seed.ID)); err != nil || !bytes.Equal(got, body) {
		t.Fatalf("snapshot = %d bytes, %v", len(got), err)
	}
	f.requireClean(t)
}

// countingPuts counts the snapshot writes that reach the bucket.
type countingPuts struct {
	release.Store
	n atomic.Int32
}

func (c *countingPuts) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	c.n.Add(1)
	return c.Store.Put(ctx, key, body, ifMatch)
}

// The import moves on while its upload is checked: the detached half puts
// nothing and aborts nothing, whether its check then fails or passes.
func TestImportUploadLeavesAnImportThatMovedOn(t *testing.T) {
	for name, c := range map[string]struct {
		moveOn  func(t *testing.T, f fromFixture, id string)
		checkOK bool
		state   string // "" for a record that is gone
	}{
		"uploaded by another writer, check fails":  {func(t *testing.T, f fromFixture, id string) { f.setSeed(t, provision.SeedUploaded(id)) }, false, seeds.StateUploaded},
		"uploaded by another writer, check passes": {func(t *testing.T, f fromFixture, id string) { f.setSeed(t, provision.SeedUploaded(id)) }, true, seeds.StateUploaded},
		"aborted by the user":                      {func(t *testing.T, f fromFixture, id string) { f.setSeed(t, provision.SeedAbort(id)) }, true, seeds.StateAborted},
		"deleted before its swap": {func(t *testing.T, f fromFixture, id string) {
			if err := f.fx.rel.Delete(context.Background(), provision.RecordKey("bab", fromApp, importHost)); err != nil {
				t.Fatal(err)
			}
		}, true, ""},
	} {
		t.Run(name, func(t *testing.T) {
			f := newImportFixture(t)
			body, _ := sqliteBytes(t)
			begun := f.begun(t, importHost, body)
			checking := make(chan struct{})
			proceed := make(chan struct{})
			old := checkSeedDBConsole
			defer func() { checkSeedDBConsole = old }()
			checkSeedDBConsole = func(ctx context.Context, path string) error {
				close(checking)
				<-proceed
				if !c.checkOK {
					return errors.New("not a database")
				}
				return old(ctx, path)
			}
			if code, out := uploadImport(t, f, importHost, body); code != http.StatusAccepted {
				t.Fatalf("upload = %d %v", code, out)
			}
			<-checking
			c.moveOn(t, f, begun.Seed.ID)
			close(proceed)
			f.fx.srv.seedCopies.Wait()
			rec, err := f.record(t, importHost)
			if c.state == "" {
				if !errors.Is(err, os.ErrNotExist) {
					t.Fatalf("record = %+v, %v; want it still gone", rec, err)
				}
			} else {
				requireImportState(t, rec, begun.Seed.ID, c.state, "")
			}
			f.requireNothingUploaded(t)
			f.requireClean(t)
		})
	}
}

func (f fromFixture) setSeed(t *testing.T, step func(*provision.Record) error) {
	t.Helper()
	if _, err := provision.SetRecordSeed(context.Background(), f.fx.rel, "bab", fromApp, importHost, step); err != nil {
		t.Fatal(err)
	}
}

// The check panics: the seed is aborted with the retry sentence and the
// console keeps serving.
func TestImportUploadPanicAborts(t *testing.T) {
	f := newImportFixture(t)
	body, _ := sqliteBytes(t)
	begun := f.begun(t, importHost, body)
	old := checkSeedDBConsole
	defer func() { checkSeedDBConsole = old }()
	checkSeedDBConsole = func(context.Context, string) error { panic("a fault in the check") }
	requireImportState(t, f.uploaded(t, importHost, body), begun.Seed.ID, seeds.StateAborted, importFailed)
	f.requireClean(t)
	f.requireNothingUploaded(t)
	checkSeedDBConsole = old
	other := f.begun(t, "new.shop.example.com", body)
	requireImportState(t, f.uploaded(t, "new.shop.example.com", body), other.Seed.ID, seeds.StateUploaded, "")
}

// The uploaded write fails: the seed is aborted with the retry sentence.
func TestImportUploadRecordWriteFailureAborts(t *testing.T) {
	f := newImportFixture(t)
	body, _ := sqliteBytes(t)
	begun := f.begun(t, importHost, body)
	rel := &failOnceOn{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, importHost)}
	f.fx.srv.cfg.Releases = rel
	old := checkSeedDBConsole
	defer func() { checkSeedDBConsole = old }()
	checkSeedDBConsole = func(ctx context.Context, path string) error {
		rel.armed.Store(true) // the next write of the record is uploaded
		return old(ctx, path)
	}
	requireImportState(t, f.uploaded(t, importHost, body), begun.Seed.ID, seeds.StateAborted, importFailed)
	f.requireClean(t)
}

func TestImportUploadRefusals(t *testing.T) {
	f := newImportFixture(t)
	body, _ := sqliteBytes(t)
	f.begun(t, importHost, body)
	for _, c := range []struct {
		name, host string
		code       int
		msg        string
	}{
		{"no record", "ghost.shop.example.com", http.StatusNotFound, "ghost.shop.example.com is not an instance of shop"},
		{"not a host", "Not_A_Host", http.StatusNotFound, "Not_A_Host is not an instance of shop"},
		{"an ordinary instance", fromSource, http.StatusConflict, "main.shop.example.com is not an import."},
	} {
		code, out := uploadImport(t, f, c.host, body)
		if code != c.code || str(t, out, "error") != c.msg {
			t.Errorf("%s = %d %v, want %d %q", c.name, code, out, c.code, c.msg)
		}
	}

	// A --from seed is not an import either.
	f.seedRecord(t, fromHost, "", newSeedID(t), seeds.KindReplica, seeds.StateReserving)
	if code, out := uploadImport(t, f, fromHost, body); code != http.StatusConflict || str(t, out, "error") != fromHost+" is not an import." {
		t.Errorf("a --from seed = %d %v", code, out)
	}

	// A seed id that cannot name a key (preflight M8): nothing is put.
	f.seedRecord(t, "bad.shop.example.com", "", "not-a-seed-id", seeds.KindImport, seeds.StateAwaitingUpload)
	if code, out := uploadImport(t, f, "bad.shop.example.com", body); code != http.StatusBadGateway || str(t, out, "error") != importFailed {
		t.Errorf("a bad seed id = %d %v", code, out)
	}
	f.requireNothingUploaded(t)

	// A claim console refuses before it reads the record.
	f.fx.srv.cfg.HostClaim = true
	if code, out := uploadImport(t, f, importHost, body); code != http.StatusConflict ||
		str(t, out, "error") != "This console runs with claim control on. Seeded instances are not supported here yet." {
		t.Errorf("claims armed = %d %v", code, out)
	}
	f.fx.srv.cfg.HostClaim = false

	// No data bucket (SD-N38).
	data := f.fx.srv.cfg.SeedData
	f.fx.srv.cfg.SeedData = nil
	if code, out := uploadImport(t, f, importHost, body); code != http.StatusServiceUnavailable ||
		str(t, out, "error") != "this console has no data bucket set up, so it cannot create a seeded instance." {
		t.Errorf("no data bucket = %d %v", code, out)
	}
	f.fx.srv.cfg.SeedData = data

	// No room in the scratch (SD-N27).
	pool := f.fx.srv.seedScratch
	f.fx.srv.seedScratch = newSeedScratch(pool.dir(), func(string) (int64, error) { return 1 << 30, nil })
	if code, out := uploadImport(t, f, importHost, body); code != http.StatusInsufficientStorage ||
		str(t, out, "error") != "the console has no room for this right now. Try again later." {
		t.Errorf("no scratch = %d %v", code, out)
	}
	f.fx.srv.seedScratch = pool
	f.fx.srv.seedCopies.Wait()
	rec, _ := f.record(t, importHost)
	if rec.Seed.State != seeds.StateAwaitingUpload {
		t.Fatalf("a refused upload moved the import to %s", rec.Seed.State)
	}
	f.requireClean(t)
	f.requireNothingUploaded(t)
}

// postImport posts verb ("stopped" or "commit") to host's import door,
// naming the seed host's record holds now.
func postImport(t *testing.T, f fromFixture, host, verb string) (int, map[string]any) {
	t.Helper()
	return postImportFor(t, f, host, heldSeed(f, host), verb)
}

// postImportFor is postImport naming seed id.
func postImportFor(t *testing.T, f fromFixture, host, id, verb string) (int, map[string]any) {
	t.Helper()
	code, _, raw := rawReq(t, http.MethodPost, importDoorURL(f, host, verb, id), f.token, "")
	var out map[string]any
	_ = json.Unmarshal([]byte(raw), &out)
	return code, out
}

// importAudit is the detail of every audit row of action on shop.
func importAudit(t *testing.T, f fromFixture, action string) []string {
	t.Helper()
	events, err := f.fx.st.Events(OpsAccountID, fromApp, 100)
	if err != nil {
		t.Fatal(err)
	}
	var rows []string
	for _, e := range events {
		if e.Action == action {
			rows = append(rows, e.Detail)
		}
	}
	return rows
}

// The whole of spec §5.3 and §5.4's commit: stopped before the upload is
// refused, stopped after it is ready and bumps the serial, commit before
// placed is refused, and commit after placed lands once and answers 200 when
// it is retried. The test plays the box for the swap and the place.
func TestImportStoppedAndCommit(t *testing.T) {
	f := newImportFixture(t)
	var logs bytes.Buffer
	log.SetOutput(&logs)
	defer log.SetOutput(os.Stderr)
	body, sha := sqliteBytes(t)
	code, out := beginImport(t, f, importHost, sha, int64(len(body)))
	begun, token := f.requireBegun(t, code, out, importHost)
	id := begun.Seed.ID
	if code, out := postImport(t, f, importHost, "stopped"); code != http.StatusConflict ||
		str(t, out, "error") != importHost+" is awaiting-upload. Mark it stopped once, after its upload." {
		t.Fatalf("stopped before upload = %d %v", code, out)
	}
	requireImportState(t, f.uploaded(t, importHost, body), id, seeds.StateUploaded, "")

	before := serial(t, f.fx)
	if code, out := postImport(t, f, importHost, "stopped"); code != http.StatusOK || len(out) != 1 || out["state"] != "ready" {
		t.Fatalf("stopped = %d %v", code, out)
	}
	if got := serial(t, f.fx); got != before+1 {
		t.Fatalf("serial after stopped = %d, want %d", got, before+1)
	}
	rec, _ := f.record(t, importHost)
	requireImportState(t, rec, id, seeds.StateReady, "")
	if rows := importAudit(t, f, "import-stopped"); len(rows) != 1 || rows[0] != "marked the old process of "+importHost+" stopped" {
		t.Fatalf("import-stopped rows = %q", rows)
	}
	// A second stopped is refused: the import is no longer uploaded.
	if code, out := postImport(t, f, importHost, "stopped"); code != http.StatusConflict ||
		str(t, out, "error") != importHost+" is ready. Mark it stopped once, after its upload." {
		t.Fatalf("second stopped = %d %v", code, out)
	}

	for _, state := range []string{seeds.StateReady, seeds.StateSwapped} {
		if state == seeds.StateSwapped {
			f.setSeed(t, provision.SeedSwapped(id))
		}
		before = serial(t, f.fx)
		if code, out := postImport(t, f, importHost, "commit"); code != http.StatusConflict ||
			str(t, out, "error") != "nothing has been verified yet. Send a request with the verify token first, then commit." {
			t.Fatalf("commit while %s = %d %v", state, code, out)
		}
		if serial(t, f.fx) != before {
			t.Fatalf("a refused commit while %s bumped the serial", state)
		}
	}
	f.setSeed(t, provision.SeedPlace(id))
	before = serial(t, f.fx)
	if code, out := postImport(t, f, importHost, "commit"); code != http.StatusOK || len(out) != 1 || out["state"] != "committed" {
		t.Fatalf("commit = %d %v", code, out)
	}
	if got := serial(t, f.fx); got != before+1 {
		t.Fatalf("serial after commit = %d, want %d", got, before+1)
	}
	rec, _ = f.record(t, importHost)
	requireImportState(t, rec, id, seeds.StateCommitted, "")
	// The retried commit changes nothing, audits nothing and bumps nothing.
	before = serial(t, f.fx)
	if code, out := postImport(t, f, importHost, "commit"); code != http.StatusOK || out["state"] != "committed" {
		t.Fatalf("a retried commit = %d %v, want 200", code, out)
	}
	if serial(t, f.fx) != before {
		t.Fatal("a retried commit bumped the serial")
	}
	if rows := importAudit(t, f, "import-committed"); len(rows) != 1 || rows[0] != "committed the import of "+importHost {
		t.Fatalf("import-committed rows = %q", rows)
	}
	// Stopped after commit is refused with the committed state.
	if code, out := postImport(t, f, importHost, "stopped"); code != http.StatusConflict ||
		str(t, out, "error") != importHost+" is committed. Mark it stopped once, after its upload." {
		t.Fatalf("stopped after commit = %d %v", code, out)
	}

	// The token never leaves begin's body: not the log, not an audit row.
	if strings.Contains(logs.String(), token) {
		t.Fatal("the token reached the log")
	}
	events, _ := f.fx.st.Events(OpsAccountID, fromApp, 100)
	for _, e := range events {
		if strings.Contains(e.Detail+e.Message, token) {
			t.Errorf("the token reached an audit row: %+v", e)
		}
	}
}

// Commit of an import that was refused, aborted or rolled back is the
// SD-N115 refusal, and writes nothing.
func TestImportCommitRefusesAnAbandonedImport(t *testing.T) {
	f := newImportFixture(t)
	for _, state := range []string{seeds.StateRefused, seeds.StateAborted, seeds.StateRolledBack} {
		host := state + ".shop.example.com"
		f.seedRecord(t, host, "", newSeedID(t), seeds.KindImport, state)
		before := serial(t, f.fx)
		if code, out := postImport(t, f, host, "commit"); code != http.StatusConflict ||
			str(t, out, "error") != host+" is "+state+", so there is nothing to commit." {
			t.Errorf("commit of a %s import = %d %v", state, code, out)
		}
		if rec, _ := f.record(t, host); rec.Seed.State != state {
			t.Errorf("commit moved a %s import to %s", state, rec.Seed.State)
		}
		if serial(t, f.fx) != before {
			t.Errorf("a refused commit of a %s import bumped the serial", state)
		}
	}
	if rows := importAudit(t, f, "import-committed"); len(rows) != 0 {
		t.Fatalf("import-committed rows = %q", rows)
	}
}

// Stopped refuses every state but uploaded, without writing the record.
func TestImportStoppedRefusesEveryOtherState(t *testing.T) {
	f := newImportFixture(t)
	for _, state := range []string{seeds.StateAwaitingUpload, seeds.StateReady, seeds.StateSwapped, seeds.StatePlaced,
		seeds.StateCommitted, seeds.StateRefused, seeds.StateAborted, seeds.StateRolledBack} {
		host := state + ".shop.example.com"
		f.seedRecord(t, host, "", newSeedID(t), seeds.KindImport, state)
		before := serial(t, f.fx)
		if code, out := postImport(t, f, host, "stopped"); code != http.StatusConflict ||
			str(t, out, "error") != host+" is "+state+". Mark it stopped once, after its upload." {
			t.Errorf("stopped while %s = %d %v", state, code, out)
		}
		if rec, _ := f.record(t, host); rec.Seed.State != state {
			t.Errorf("stopped moved a %s import to %s", state, rec.Seed.State)
		}
		if serial(t, f.fx) != before {
			t.Errorf("a refused stopped while %s bumped the serial", state)
		}
	}
}

// Both doors share the import gate: no record, a malformed host, an
// ordinary instance and a --from seed are refused as the upload refuses
// them, and a claim console refuses before it reads the record.
func TestImportStoppedAndCommitGate(t *testing.T) {
	f := newImportFixture(t)
	f.seedRecord(t, fromHost, "", newSeedID(t), seeds.KindReplica, seeds.StateReady)
	f.seedRecord(t, importHost, "", newSeedID(t), seeds.KindImport, seeds.StatePlaced)
	for _, verb := range []string{"stopped", "commit"} {
		for _, c := range []struct {
			name, host string
			code       int
			msg        string
		}{
			{"no record", "ghost.shop.example.com", http.StatusNotFound, "ghost.shop.example.com is not an instance of shop"},
			{"not a host", "Not_A_Host", http.StatusNotFound, "Not_A_Host is not an instance of shop"},
			{"an ordinary instance", fromSource, http.StatusConflict, fromSource + " is not an import."},
			{"a --from seed", fromHost, http.StatusConflict, fromHost + " is not an import."},
		} {
			if code, out := postImport(t, f, c.host, verb); code != c.code || str(t, out, "error") != c.msg {
				t.Errorf("%s %s = %d %v, want %d %q", verb, c.name, code, out, c.code, c.msg)
			}
		}
		f.fx.srv.cfg.HostClaim = true
		if code, out := postImport(t, f, importHost, verb); code != http.StatusConflict ||
			str(t, out, "error") != "This console runs with claim control on. Seeded instances are not supported here yet." {
			t.Errorf("%s with claims armed = %d %v", verb, code, out)
		}
		f.fx.srv.cfg.HostClaim = false
	}
	if rec, _ := f.record(t, importHost); rec.Seed.State != seeds.StatePlaced {
		t.Fatalf("a refused door moved the import to %s", rec.Seed.State)
	}
}

// Stopped asks the region again: a region that can no longer take the
// import leaves it uploaded.
func TestImportStoppedAsksTheRegionAgain(t *testing.T) {
	f := newImportFixture(t)
	body, _ := sqliteBytes(t)
	begun := f.begun(t, importHost, body)
	requireImportState(t, f.uploaded(t, importHost, body), begun.Seed.ID, seeds.StateUploaded, "")
	seedReadyBox(t, f.fx, "kass-2", "", 1, "v0.27.0", "v0.27.0", "nested", false)
	before := serial(t, f.fx)
	if code, out := postImport(t, f, importHost, "stopped"); code != http.StatusConflict ||
		str(t, out, "error") != "the home region has more than one box. An import needs a region with one box for now." {
		t.Fatalf("stopped = %d %v", code, out)
	}
	rec, _ := f.record(t, importHost)
	requireImportState(t, rec, begun.Seed.ID, seeds.StateUploaded, "")
	if serial(t, f.fx) != before {
		t.Fatal("a refused stopped bumped the serial")
	}
}

// movesOnWrite runs hook, once, just before the first CAS write of key: the
// write that follows loses its CAS, as it would to a concurrent request.
type movesOnWrite struct {
	release.Store
	key  string
	hook func()
	done atomic.Bool
}

func (s *movesOnWrite) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if key == s.key && ifMatch != "" && s.done.CompareAndSwap(false, true) {
		s.hook()
	}
	return s.Store.Put(ctx, key, body, ifMatch)
}

// DeleteIfMatch forwards, so a finish behind the wrapper can delete the
// record.
func (s *movesOnWrite) DeleteIfMatch(ctx context.Context, key, etag string) error {
	return release.DeleteIfMatch(ctx, s.Store, key, etag)
}

// A stopped or a commit that read the import in one state and finds it in
// another when it writes changes nothing: SD-N60, no audit row and no bump.
// It never acts on an import that has moved on (its transition is guarded
// on the seed and on its from-state).
func TestImportTransitionsLeaveAnImportThatMovedOn(t *testing.T) {
	for _, c := range []struct{ verb, action, start string }{
		{"stopped", "import-stopped", seeds.StateUploaded},
		{"commit", "import-committed", seeds.StatePlaced},
	} {
		t.Run(c.verb, func(t *testing.T) {
			f := newImportFixture(t)
			id := newSeedID(t)
			f.seedRecord(t, importHost, "", id, seeds.KindImport, c.start)
			f.fx.srv.cfg.Releases = &movesOnWrite{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, importHost), hook: func() {
				// Another request aborts the import after this one read it.
				if _, err := provision.SetRecordSeed(context.Background(), f.fx.rel, "bab", fromApp, importHost, provision.SeedAbort(id)); err != nil {
					t.Error(err)
				}
			}}
			before := serial(t, f.fx)
			code, out := postImport(t, f, importHost, c.verb)
			if code != http.StatusConflict || str(t, out, "error") !=
				importHost+" changed while this request was made. Read its state with carlos instances list and try again." {
				t.Fatalf("%s = %d %v", c.verb, code, out)
			}
			rec, _ := f.record(t, importHost)
			requireImportState(t, rec, id, seeds.StateAborted, "")
			if serial(t, f.fx) != before {
				t.Fatal("a lost transition bumped the serial")
			}
			if rows := importAudit(t, f, c.action); len(rows) != 0 {
				t.Fatalf("%s rows = %q", c.action, rows)
			}
		})
	}
}

// A record write that fails for any other reason is the retry sentence, and
// the record is left as it was.
func TestImportTransitionsWriteFailure(t *testing.T) {
	f := newImportFixture(t)
	id := newSeedID(t)
	f.seedRecord(t, importHost, "", id, seeds.KindImport, seeds.StatePlaced)
	rel := &failOnceOn{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, importHost)}
	f.fx.srv.cfg.Releases = rel
	rel.armed.Store(true)
	before := serial(t, f.fx)
	if code, out := postImport(t, f, importHost, "commit"); code != http.StatusBadGateway || str(t, out, "error") != "could not update the import of "+importHost+". Try again." {
		t.Fatalf("commit = %d %v", code, out)
	}
	rec, _ := f.record(t, importHost)
	requireImportState(t, rec, id, seeds.StatePlaced, "")
	if serial(t, f.fx) != before {
		t.Fatal("a failed commit bumped the serial")
	}
	// The retry lands.
	if code, out := postImport(t, f, importHost, "commit"); code != http.StatusOK || out["state"] != "committed" {
		t.Fatalf("retried commit = %d %v", code, out)
	}
}

// An overlapping duplicate: another request lands the same transition
// between this one's read and its write. A commit is then told what a
// sequential retry is told (200, committed), and writes no audit row and no
// serial bump. A stopped is told what a sequential second stopped is told
// (409 naming the state it is in), not that it changed under it. A hook
// that moves the import somewhere else still gets SD-N60.
func TestImportOverlappingTransitions(t *testing.T) {
	changed := func(host string) string {
		return host + " changed while this request was made. Read its state with carlos instances list and try again."
	}
	for _, c := range []struct {
		name, verb, action, start string
		hook                      func(id string) []func(string) func(*provision.Record) error
		code                      int
		msg                       string // the error, "" for a 200
		state                     string // the record's state after
	}{
		{"commit after commit", "commit", "import-committed", seeds.StatePlaced,
			func(string) []func(string) func(*provision.Record) error {
				return []func(string) func(*provision.Record) error{provision.SeedCommit}
			}, http.StatusOK, "", seeds.StateCommitted},
		{"commit after abort", "commit", "import-committed", seeds.StatePlaced,
			func(string) []func(string) func(*provision.Record) error {
				return []func(string) func(*provision.Record) error{provision.SeedAbort}
			}, http.StatusConflict, changed(importHost), seeds.StateAborted},
		{"stopped after stopped", "stopped", "import-stopped", seeds.StateUploaded,
			func(string) []func(string) func(*provision.Record) error {
				return []func(string) func(*provision.Record) error{provision.SeedStopped}
			}, http.StatusConflict, importHost + " is ready. Mark it stopped once, after its upload.", seeds.StateReady},
		{"stopped after stopped and swapped", "stopped", "import-stopped", seeds.StateUploaded,
			func(string) []func(string) func(*provision.Record) error {
				return []func(string) func(*provision.Record) error{provision.SeedStopped, provision.SeedSwapped}
			}, http.StatusConflict, importHost + " is swapped. Mark it stopped once, after its upload.", seeds.StateSwapped},
		{"stopped after abort", "stopped", "import-stopped", seeds.StateUploaded,
			func(string) []func(string) func(*provision.Record) error {
				return []func(string) func(*provision.Record) error{provision.SeedAbort}
			}, http.StatusConflict, changed(importHost), seeds.StateAborted},
	} {
		t.Run(c.name, func(t *testing.T) {
			f := newImportFixture(t)
			id := newSeedID(t)
			f.seedRecord(t, importHost, "", id, seeds.KindImport, c.start)
			steps := c.hook(id)
			f.fx.srv.cfg.Releases = &movesOnWrite{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, importHost), hook: func() {
				for _, step := range steps {
					if _, err := provision.SetRecordSeed(context.Background(), f.fx.rel, "bab", fromApp, importHost, step(id)); err != nil {
						t.Error(err)
					}
				}
			}}
			before := serial(t, f.fx)
			code, out := postImport(t, f, importHost, c.verb)
			if code != c.code || str(t, out, "error") != c.msg {
				t.Fatalf("%s = %d %v, want %d %q", c.verb, code, out, c.code, c.msg)
			}
			if c.code == http.StatusOK && (len(out) != 1 || out["state"] != c.state) {
				t.Fatalf("%s answered %v, want {state: %s}", c.verb, out, c.state)
			}
			rec, _ := f.record(t, importHost)
			requireImportState(t, rec, id, c.state, "")
			if serial(t, f.fx) != before {
				t.Fatal("the overlapping request bumped the serial")
			}
			if rows := importAudit(t, f, c.action); len(rows) != 0 {
				t.Fatalf("the overlapping request audited %s: %q", c.action, rows)
			}
		})
	}
}

// A delayed request of an import that has gone lands on nothing (final
// review F1): import A is aborted and finished, import B begins on the same
// host, and A's upload, stopped, commit and abort, arriving late, each
// answer SD-N60 and leave B as it was. A request naming no seed is refused
// the same way. B's own requests, and their idempotent repeats, answer as
// before.
func TestImportDelayedMutationsOfAnEarlierImport(t *testing.T) {
	f := newImportFixture(t)
	body, _ := sqliteBytes(t)
	a := f.begun(t, importHost, body).Seed.ID
	requireImportState(t, f.uploaded(t, importHost, body), a, seeds.StateUploaded, "")
	f.setSeed(t, provision.SeedAbort(a))
	f.setSeed(t, provision.SeedRolledBack(a))
	if code, out := postImportFor(t, f, importHost, a, "abort"); code != http.StatusOK || out["state"] != "rolled-back" {
		t.Fatalf("A's finishing abort = %d %v", code, out)
	}
	f.requireRecordGone(t, importHost)

	b := f.begun(t, importHost, body).Seed.ID
	if b == a {
		t.Fatal("B reused A's seed id")
	}
	stale := func(state string, send func(id string) (int, map[string]any)) {
		t.Helper()
		before := serial(t, f.fx)
		for _, id := range []string{a, ""} {
			if code, out := send(id); code != http.StatusConflict || str(t, out, "error") != importChanged(importHost) {
				t.Fatalf("a request naming seed %q while B is %s = %d %v, want SD-N60", id, state, code, out)
			}
			f.fx.srv.seedCopies.Wait()
			rec, _ := f.record(t, importHost)
			requireImportState(t, rec, b, state, "")
		}
		if serial(t, f.fx) != before {
			t.Fatalf("a stale request while B is %s bumped the serial", state)
		}
	}
	stale(seeds.StateAwaitingUpload, func(id string) (int, map[string]any) {
		code, out, err := putImportFor(f, importHost, id, body)
		if err != nil {
			t.Fatal(err)
		}
		return code, out
	})
	f.requireNothingUploaded(t)
	requireImportState(t, f.uploaded(t, importHost, body), b, seeds.StateUploaded, "")
	stale(seeds.StateUploaded, func(id string) (int, map[string]any) { return postImportFor(t, f, importHost, id, "stopped") })
	if code, out := postImportFor(t, f, importHost, b, "stopped"); code != http.StatusOK || out["state"] != "ready" {
		t.Fatalf("B's stopped = %d %v", code, out)
	}
	f.setSeed(t, provision.SeedSwapped(b))
	f.setSeed(t, provision.SeedPlace(b))
	for _, verb := range []string{"commit", "abort"} {
		stale(seeds.StatePlaced, func(id string) (int, map[string]any) { return postImportFor(t, f, importHost, id, verb) })
	}
	if rows := importAudit(t, f, "import-committed"); len(rows) != 0 {
		t.Fatalf("a stale commit audited: %q", rows)
	}
	for range 2 { // B's commit, then its retry
		if code, out := postImportFor(t, f, importHost, b, "commit"); code != http.StatusOK || out["state"] != "committed" {
			t.Fatalf("B's commit = %d %v", code, out)
		}
	}
	if rows := importAudit(t, f, "import-committed"); len(rows) != 1 {
		t.Fatalf("import-committed rows = %q, want B's one", rows)
	}
	// A's abort after B committed is still SD-N60, not B's SD-N56.
	if code, out := postImportFor(t, f, importHost, a, "abort"); code != http.StatusConflict || str(t, out, "error") != importChanged(importHost) {
		t.Fatalf("A's abort after B committed = %d %v", code, out)
	}
}

// ---- Abort and the delete door (spec §5.6, preflight M8, M14, M15) ----

const (
	importDeleteRetry = "could not delete the instance. Try again."
	importRunBusy     = "carlos run is busy on " + importHost + ". Wait for it to finish, then abort."
)

func importChanged(host string) string {
	return host + " changed while this request was made. Read its state with carlos instances list and try again."
}

func importCommitted(host string) string {
	return host + " is committed. Delete it with carlos instances delete instead."
}

func importAbortFirst(host string) string {
	return host + " is an import that is not committed. Abort the import instead."
}

func importAlreadyAborted(host string) string {
	return "the import of " + host + " is already aborted. Once the box puts the old route back on its next pass, delete " + host + " again."
}

// postImportErr is postImport for a goroutine: it returns its error and
// never calls t.Fatal (preflight m9).
func postImportErr(f fromFixture, host, verb string) (int, map[string]any, error) {
	req, err := http.NewRequest(http.MethodPost, importDoorURL(f, host, verb, heldSeed(f, host)), nil)
	if err != nil {
		return 0, nil, err
	}
	req.Header.Set("Authorization", "Bearer "+f.token)
	res, err := http.DefaultClient.Do(req)
	if err != nil {
		return 0, nil, err
	}
	defer res.Body.Close()
	var out map[string]any
	_ = json.NewDecoder(res.Body).Decode(&out)
	return res.StatusCode, out, nil
}

// deleteInstance sends the ordinary delete for host and returns what it
// answered.
func deleteInstance(t *testing.T, f fromFixture, host string) (int, map[string]any) {
	t.Helper()
	code, _, raw := rawReq(t, http.MethodDelete, f.url+"/"+host, f.token, "")
	var out map[string]any
	_ = json.Unmarshal([]byte(raw), &out)
	return code, out
}

// setRun writes the run series' field directly, through the public
// etag-guarded writer: run admission refuses an uncommitted import, so a
// run open on one is only reachable by a write like this (a run admitted
// before the import began, in the spec's words).
func setRun(t *testing.T, f fromFixture, host, state string) {
	t.Helper()
	rec, etag, err := provision.GetRecordETag(context.Background(), f.fx.rel, "bab", fromApp, host)
	if err != nil {
		t.Fatal(err)
	}
	rec.ActiveRun = &provision.RunState{ID: "run_1", State: state, Deadline: time.Now().Add(time.Hour)}
	if err := provision.UpdateRecord(context.Background(), f.fx.rel, *rec, etag); err != nil {
		t.Fatal(err)
	}
}

// A commit and an abort race on the record's etag, and exactly one wins
// (spec §5.6). The loser is told why: an abort that lost to the commit is
// SD-N56, a commit that lost to the abort is SD-N60 (or SD-N115 when the
// abort landed before its read). Neither is a 502.
func TestImportCommitAbortRace(t *testing.T) {
	f := newImportFixture(t)
	body, _ := sqliteBytes(t)
	begun := f.begun(t, importHost, body)
	requireImportState(t, f.uploaded(t, importHost, body), begun.Seed.ID, seeds.StateUploaded, "")
	if code, out := postImport(t, f, importHost, "stopped"); code != http.StatusOK {
		t.Fatalf("stopped = %d %v", code, out)
	}
	f.setSeed(t, provision.SeedSwapped(begun.Seed.ID))
	f.setSeed(t, provision.SeedPlace(begun.Seed.ID))
	hosts := []string{importHost}
	for i := range 8 {
		host := "race" + strconv.Itoa(i) + ".shop.example.com"
		f.seedRecord(t, host, "", newSeedID(t), seeds.KindImport, seeds.StatePlaced)
		hosts = append(hosts, host)
	}
	type answer struct {
		verb string
		code int
		out  map[string]any
		err  error
	}
	for _, host := range hosts {
		answers := make(chan answer, 2)
		for _, verb := range []string{"commit", "abort"} {
			go func(verb string) {
				code, out, err := postImportErr(f, host, verb)
				answers <- answer{verb, code, out, err}
			}(verb)
		}
		got := map[string]answer{}
		for range 2 {
			a := <-answers
			if a.err != nil {
				t.Fatalf("%s %s: %v", host, a.verb, a.err)
			}
			got[a.verb] = a
		}
		rec, err := f.record(t, host)
		if err != nil {
			t.Fatal(err)
		}
		commit, abort := got["commit"], got["abort"]
		switch rec.Seed.State {
		case seeds.StateCommitted:
			if commit.code != http.StatusOK || abort.code != http.StatusConflict || abort.out["error"] != importCommitted(host) {
				t.Fatalf("%s committed: commit %d %v, abort %d %v", host, commit.code, commit.out, abort.code, abort.out)
			}
		case seeds.StateAborted:
			lost := commit.out["error"] == importChanged(host) || commit.out["error"] == host+" is aborted, so there is nothing to commit."
			if abort.code != http.StatusOK || abort.out["state"] != seeds.StateAborted || commit.code != http.StatusConflict || !lost {
				t.Fatalf("%s aborted: commit %d %v, abort %d %v", host, commit.code, commit.out, abort.code, abort.out)
			}
		default:
			t.Fatalf("%s ended %s: commit %d %v, abort %d %v; want exactly one winner", host, rec.Seed.State, commit.code, commit.out, abort.code, abort.out)
		}
	}
	committed, aborted := importAudit(t, f, "import-committed"), importAudit(t, f, "import-aborted")
	if len(committed)+len(aborted) != len(hosts) {
		t.Fatalf("audit rows: %d committed, %d aborted, want one per host (%d)", len(committed), len(aborted), len(hosts))
	}
}

// Abort is refused while a run is open and after commit, lands from every
// state before commit, and is safe to repeat.
func TestImportAbortRefusals(t *testing.T) {
	f := newImportFixture(t)
	body, sha := sqliteBytes(t)
	code, out := beginImport(t, f, importHost, sha, int64(len(body)))
	begun, _ := f.requireBegun(t, code, out, importHost)
	id := begun.Seed.ID

	setRun(t, f, importHost, provision.RunPicked)
	before := serial(t, f.fx)
	if code, out := postImport(t, f, importHost, "abort"); code != http.StatusConflict || str(t, out, "error") != importRunBusy {
		t.Fatalf("abort with a run open = %d %v", code, out)
	}
	rec, _ := f.record(t, importHost)
	requireImportState(t, rec, id, seeds.StateAwaitingUpload, "")
	if serial(t, f.fx) != before || len(importAudit(t, f, "import-aborted")) != 0 {
		t.Fatal("a refused abort bumped the serial or wrote an audit row")
	}

	setRun(t, f, importHost, provision.RunDone)
	if code, out := postImport(t, f, importHost, "abort"); code != http.StatusOK || len(out) != 1 || out["state"] != "aborted" {
		t.Fatalf("abort = %d %v", code, out)
	}
	rec, _ = f.record(t, importHost)
	requireImportState(t, rec, id, seeds.StateAborted, "")
	if got := serial(t, f.fx); got != before+1 {
		t.Fatalf("serial after abort = %d, want %d", got, before+1)
	}
	if rows := importAudit(t, f, "import-aborted"); len(rows) != 1 || rows[0] != "aborted the import of "+importHost {
		t.Fatalf("import-aborted rows = %q", rows)
	}

	// Again: 200, and nothing written, audited or bumped.
	before = serial(t, f.fx)
	if code, out := postImport(t, f, importHost, "abort"); code != http.StatusOK || len(out) != 1 || out["state"] != "aborted" {
		t.Fatalf("abort again = %d %v, want 200", code, out)
	}
	if serial(t, f.fx) != before || len(importAudit(t, f, "import-aborted")) != 1 {
		t.Fatal("a repeated abort bumped the serial or wrote an audit row")
	}

	// After commit, abort is refused: delete is the verb then.
	f.seedRecord(t, "done.shop.example.com", "", newSeedID(t), seeds.KindImport, seeds.StateCommitted)
	if code, out := postImport(t, f, "done.shop.example.com", "abort"); code != http.StatusConflict ||
		str(t, out, "error") != importCommitted("done.shop.example.com") {
		t.Fatalf("abort after commit = %d %v", code, out)
	}
	if rec, _ := f.record(t, "done.shop.example.com"); rec.Seed.State != seeds.StateCommitted {
		t.Fatalf("a refused abort moved a committed import to %s", rec.Seed.State)
	}
}

// Every state before committed aborts. A refused import keeps the reason
// the box gave (preflight M14: the no-reason SeedAbort).
func TestImportAbortFromEveryStateBeforeCommit(t *testing.T) {
	f := newImportFixture(t)
	for _, state := range []string{seeds.StateAwaitingUpload, seeds.StateUploaded, seeds.StateReady,
		seeds.StateSwapped, seeds.StatePlaced, seeds.StateRefused} {
		host := state + ".shop.example.com"
		id := newSeedID(t)
		f.seedRecord(t, host, "", id, seeds.KindImport, state)
		reason := ""
		if state == seeds.StateRefused {
			reason = "the box found no addr row"
			rec, etag, err := provision.GetRecordETag(context.Background(), f.fx.rel, "bab", fromApp, host)
			if err != nil {
				t.Fatal(err)
			}
			rec.Seed.Reason = reason
			raw, _ := json.Marshal(rec)
			if _, err := f.fx.rel.Put(context.Background(), provision.RecordKey("bab", fromApp, host), raw, etag); err != nil {
				t.Fatal(err)
			}
		}
		if code, out := postImport(t, f, host, "abort"); code != http.StatusOK || out["state"] != "aborted" {
			t.Errorf("abort while %s = %d %v", state, code, out)
			continue
		}
		rec, _ := f.record(t, host)
		requireImportState(t, rec, id, seeds.StateAborted, reason)
	}
}

// A rolled-back import is finished by its abort: the record and the
// snapshot go, and the answer says rolled-back. A finish that fails still
// answers 200 (the abort has landed), and leaves the record rolled-back
// for the next abort or the sweep.
func TestImportAbortFinishesARolledBackImport(t *testing.T) {
	f := newImportFixture(t)
	body, _ := sqliteBytes(t)
	begun := f.begun(t, importHost, body)
	id := begun.Seed.ID
	requireImportState(t, f.uploaded(t, importHost, body), id, seeds.StateUploaded, "")
	f.setSeed(t, provision.SeedAbort(id))
	f.setSeed(t, provision.SeedRolledBack(id))

	data := f.fx.srv.cfg.SeedData
	f.fx.srv.cfg.SeedData = nil // the finish cannot delete the snapshot
	if code, out := postImport(t, f, importHost, "abort"); code != http.StatusOK || len(out) != 1 || out["state"] != "rolled-back" {
		t.Fatalf("abort with a failing finish = %d %v", code, out)
	}
	rec, _ := f.record(t, importHost)
	requireImportState(t, rec, id, seeds.StateRolledBack, "")

	f.fx.srv.cfg.SeedData = data
	if code, out := postImport(t, f, importHost, "abort"); code != http.StatusOK || len(out) != 1 || out["state"] != "rolled-back" {
		t.Fatalf("abort of a rolled-back import = %d %v", code, out)
	}
	f.requireRecordGone(t, importHost)
	f.requireNothingUploaded(t)
	if rows := importAudit(t, f, "import-aborted"); len(rows) != 0 {
		t.Fatalf("finishing a rolled-back import audited an abort: %q", rows)
	}
}

// An abort that loses its race re-reads once and answers by what it finds
// (preflight M14): another abort is a 200, a commit is SD-N56, a rollback
// is finished and a 200, a run that opened is SD-N58, and anything else is
// SD-N60. The loser writes no audit row and bumps nothing.
func TestImportAbortLosesARace(t *testing.T) {
	ctx := context.Background()
	for _, c := range []struct {
		name  string
		hook  func(f fromFixture, id string) error
		code  int
		state string // the 200's state, or "" for an error
		msg   string
		after string // the record's state after, "" for gone
	}{
		{"to another abort", func(f fromFixture, id string) error {
			_, err := provision.SetRecordSeed(ctx, f.fx.rel, "bab", fromApp, importHost, provision.SeedAbort(id))
			return err
		}, http.StatusOK, seeds.StateAborted, "", seeds.StateAborted},
		{"to a commit", func(f fromFixture, id string) error {
			_, err := provision.SetRecordSeed(ctx, f.fx.rel, "bab", fromApp, importHost, provision.SeedCommit(id))
			return err
		}, http.StatusConflict, "", importCommitted(importHost), seeds.StateCommitted},
		{"to a rollback", func(f fromFixture, id string) error {
			for _, step := range []func(string) func(*provision.Record) error{provision.SeedAbort, provision.SeedRolledBack} {
				if _, err := provision.SetRecordSeed(ctx, f.fx.rel, "bab", fromApp, importHost, step(id)); err != nil {
					return err
				}
			}
			return nil
		}, http.StatusOK, seeds.StateRolledBack, "", ""},
		{"to a run", func(f fromFixture, id string) error {
			rec, etag, err := provision.GetRecordETag(ctx, f.fx.rel, "bab", fromApp, importHost)
			if err != nil {
				return err
			}
			rec.ActiveRun = &provision.RunState{ID: "run_1", State: provision.RunQueued, Deadline: time.Now().Add(time.Hour)}
			return provision.UpdateRecord(ctx, f.fx.rel, *rec, etag)
		}, http.StatusConflict, "", importRunBusy, seeds.StatePlaced},
		{"to another import of the host", func(f fromFixture, id string) error {
			if err := f.fx.rel.Delete(ctx, provision.RecordKey("bab", fromApp, importHost)); err != nil {
				return err
			}
			other, err := seeds.NewID()
			if err != nil {
				return err
			}
			rec := provision.Record{Schema: 1, Host: importHost, Slug: "old", App: fromApp, Account: "bab", Channel: "stable", CreatedAt: time.Now().UTC(),
				Seed: &provision.SeedSource{ID: other, Kind: seeds.KindImport, State: seeds.StateAborted, At: time.Now().UTC()}}
			return provision.CreateRecord(ctx, f.fx.rel, rec)
		}, http.StatusConflict, "", importChanged(importHost), seeds.StateAborted},
	} {
		t.Run(c.name, func(t *testing.T) {
			f := newImportFixture(t)
			id := newSeedID(t)
			f.seedRecord(t, importHost, "", id, seeds.KindImport, seeds.StatePlaced)
			f.fx.srv.cfg.Releases = &movesOnWrite{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, importHost), hook: func() {
				if err := c.hook(f, id); err != nil {
					t.Error(err)
				}
			}}
			before := serial(t, f.fx)
			code, out := postImport(t, f, importHost, "abort")
			if code != c.code || (c.state != "" && (len(out) != 1 || out["state"] != c.state)) || (c.msg != "" && str(t, out, "error") != c.msg) {
				t.Fatalf("abort = %d %v, want %d %q %q", code, out, c.code, c.state, c.msg)
			}
			rec, err := f.record(t, importHost)
			switch {
			case c.after == "":
				f.requireRecordGone(t, importHost)
			case err != nil || rec.Seed == nil || rec.Seed.State != c.after:
				t.Fatalf("record after = %+v, %v; want %s", rec, err, c.after)
			}
			if c.after != "" && serial(t, f.fx) != before {
				t.Fatal("the losing abort bumped the serial")
			}
			if rows := importAudit(t, f, "import-aborted"); len(rows) != 0 {
				t.Fatalf("the losing abort audited: %q", rows)
			}
		})
	}
}

// Abort shares the import gate.
func TestImportAbortGate(t *testing.T) {
	f := newImportFixture(t)
	f.seedRecord(t, fromHost, "", newSeedID(t), seeds.KindReplica, seeds.StateReady)
	f.seedRecord(t, importHost, "", newSeedID(t), seeds.KindImport, seeds.StatePlaced)
	for _, c := range []struct {
		name, host string
		code       int
		msg        string
	}{
		{"no record", "ghost.shop.example.com", http.StatusNotFound, "ghost.shop.example.com is not an instance of shop"},
		{"not a host", "Not_A_Host", http.StatusNotFound, "Not_A_Host is not an instance of shop"},
		{"an ordinary instance", fromSource, http.StatusConflict, fromSource + " is not an import."},
		{"a --from seed", fromHost, http.StatusConflict, fromHost + " is not an import."},
	} {
		if code, out := postImport(t, f, c.host, "abort"); code != c.code || str(t, out, "error") != c.msg {
			t.Errorf("abort %s = %d %v, want %d %q", c.name, code, out, c.code, c.msg)
		}
	}
	f.fx.srv.cfg.HostClaim = true
	if code, out := postImport(t, f, importHost, "abort"); code != http.StatusConflict ||
		str(t, out, "error") != seedClaimConsoleRefusal().Msg {
		t.Errorf("abort with claims armed = %d %v", code, out)
	}
	f.fx.srv.cfg.HostClaim = false
	if rec, _ := f.record(t, importHost); rec.Seed.State != seeds.StatePlaced {
		t.Fatalf("a refused abort moved the import to %s", rec.Seed.State)
	}
}

// The delete door's import rule (spec §5.6, codex R2-7): an import no box
// has swapped is rolled back and finished by the delete, answered as the
// ordinary delete is; a rolled-back one is finished; every other
// uncommitted import is refused with SD-N57; a committed one is an
// ordinary instance.
func TestImportDeleteRefusedUntilCommittedOrUnswapped(t *testing.T) {
	f := newImportFixture(t)
	body, sha := sqliteBytes(t)
	code, out := beginImport(t, f, "early.shop.example.com", sha, int64(len(body)))
	f.requireBegun(t, code, out, "early.shop.example.com")
	before := serial(t, f.fx)
	if code, out := deleteInstance(t, f, "early.shop.example.com"); code != http.StatusOK || len(out) != 1 || out["result"] != "deleted" {
		t.Fatalf("delete before upload = %d %v, want 200 deleted", code, out)
	}
	f.requireRecordGone(t, "early.shop.example.com")
	if serial(t, f.fx) <= before {
		t.Fatal("the delete did not bump the serial")
	}
	if rows := importAudit(t, f, "instance-delete"); len(rows) != 1 || rows[0] != "deleted instance early.shop.example.com (cli)" {
		t.Fatalf("instance-delete rows = %q", rows)
	}
	// Again: the ordinary delete's idempotent 200.
	if code, out := deleteInstance(t, f, "early.shop.example.com"); code != http.StatusOK || out["result"] != "deleted" {
		t.Fatalf("second delete = %d %v", code, out)
	}

	// Uploaded: the snapshot goes too.
	begun := f.begun(t, importHost, body)
	requireImportState(t, f.uploaded(t, importHost, body), begun.Seed.ID, seeds.StateUploaded, "")
	if keys, _ := f.data.List(context.Background(), seeds.Prefix); len(keys) != 1 {
		t.Fatalf("seeds/ before the delete = %v, want the snapshot", keys)
	}
	if code, out := deleteInstance(t, f, importHost); code != http.StatusOK || out["result"] != "deleted" {
		t.Fatalf("delete after upload = %d %v", code, out)
	}
	f.requireRecordGone(t, importHost)
	f.requireNothingUploaded(t)

	// Refused (no box swapped) and rolled-back are finished; aborted,
	// ready, swapped and placed are the box's to roll back first.
	for _, c := range []struct {
		state string
		gone  bool
	}{
		{seeds.StateRefused, true},
		{seeds.StateRolledBack, true},
		{seeds.StateReady, false},
		{seeds.StateSwapped, false},
		{seeds.StatePlaced, false},
		{seeds.StateAborted, false},
	} {
		host := c.state + ".shop.example.com"
		f.seedRecord(t, host, "", newSeedID(t), seeds.KindImport, c.state)
		code, out := deleteInstance(t, f, host)
		if c.gone {
			if code != http.StatusOK || out["result"] != "deleted" {
				t.Errorf("delete while %s = %d %v, want 200", c.state, code, out)
			}
			f.requireRecordGone(t, host)
			continue
		}
		want := importAbortFirst(host)
		if c.state == seeds.StateAborted {
			want = importAlreadyAborted(host)
		}
		if code != http.StatusConflict || str(t, out, "error") != want {
			t.Errorf("delete while %s = %d %v, want 409 %q", c.state, code, out, want)
		}
		if rec, err := f.record(t, host); err != nil || rec.Seed.State != c.state {
			t.Errorf("a refused delete while %s left %+v, %v", c.state, rec, err)
		}
	}

	// Committed, a --from seed in any state, and an unseeded instance: the
	// ordinary delete.
	f.seedRecord(t, "done.shop.example.com", "", newSeedID(t), seeds.KindImport, seeds.StateCommitted)
	f.seedRecord(t, fromHost, "", newSeedID(t), seeds.KindReplica, seeds.StateReady)
	for _, host := range []string{"done.shop.example.com", fromHost, fromSource} {
		if code, out := deleteInstance(t, f, host); code != http.StatusOK || out["result"] != "deleted" {
			t.Fatalf("delete of %s = %d %v", host, code, out)
		}
		f.requireRecordGone(t, host)
	}
}

// A record the delete door can read but not decode, or one that disagrees
// with its own path, is no import: the ordinary delete removes it, as it
// did before the import rule (review I1).
func TestImportDeleteRemovesARecordItCannotDecode(t *testing.T) {
	f := newImportFixture(t)
	var logs bytes.Buffer
	log.SetOutput(&logs)
	defer log.SetOutput(os.Stderr)
	other, _ := json.Marshal(provision.Record{Schema: 1, Host: "other.shop.example.com", Slug: "other", App: fromApp, Account: "bab",
		Channel: "stable", CreatedAt: time.Now().UTC(),
		Seed: &provision.SeedSource{ID: newSeedID(t), Kind: seeds.KindImport, State: seeds.StateAwaitingUpload, At: time.Now().UTC()}})
	for host, body := range map[string][]byte{
		"junk.shop.example.com":     []byte("not json"),
		"mismatch.shop.example.com": other,
	} {
		key := provision.RecordKey("bab", fromApp, host)
		if _, err := f.fx.rel.Put(context.Background(), key, body, ""); err != nil {
			t.Fatal(err)
		}
		if code, out := deleteInstance(t, f, host); code != http.StatusOK || out["result"] != "deleted" {
			t.Fatalf("delete of %s = %d %v, want the ordinary 200", host, code, out)
		}
		if _, _, err := f.fx.rel.Get(context.Background(), key); !errors.Is(err, os.ErrNotExist) {
			t.Fatalf("%s survived its delete: %v", host, err)
		}
		if !strings.Contains(logs.String(), host+": the record could not be read as an import") {
			t.Fatalf("no log line names %s: %s", host, logs.String())
		}
	}
}

// deleteRace plays a second delete of the same import against this one:
// hook runs just before this delete's rollback CAS write (so it loses),
// and gone, when set, deletes the record just before the re-read that
// follows the loss, as the other delete's finish would.
type deleteRace struct {
	release.Store
	key   string
	hook  func()
	gone  bool
	puts  atomic.Int32
	reads atomic.Int32
}

func (s *deleteRace) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if key == s.key && ifMatch != "" && s.puts.Add(1) == 1 {
		s.hook()
	}
	return s.Store.Put(ctx, key, body, ifMatch)
}

func (s *deleteRace) Get(ctx context.Context, key string) ([]byte, string, error) {
	// Reads after the hook: the CAS's retry, then the delete door's re-read.
	if key == s.key && s.gone && s.puts.Load() > 0 && s.reads.Add(1) == 2 {
		if err := s.Store.Delete(ctx, key); err != nil {
			return nil, "", err
		}
	}
	return s.Store.Get(ctx, key)
}

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

// Two deletes of one pre-swap import overlap. The one whose rollback CAS
// loses finishes the import when it finds it rolled back, and answers the
// ordinary idempotent 200 when the other has already removed the record
// (review M3).
func TestImportDeletesOverlap(t *testing.T) {
	for _, gone := range []bool{false, true} {
		t.Run(map[bool]string{false: "rolled back", true: "finished"}[gone], func(t *testing.T) {
			f := newImportFixture(t)
			id := newSeedID(t)
			f.seedRecord(t, importHost, "", id, seeds.KindImport, seeds.StateUploaded)
			f.fx.srv.cfg.Releases = &deleteRace{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, importHost), gone: gone, hook: func() {
				if _, err := provision.SetRecordSeed(context.Background(), f.fx.rel, "bab", fromApp, importHost, provision.SeedRolledBackBeforeSwap(id)); err != nil {
					t.Error(err)
				}
			}}
			if code, out := deleteInstance(t, f, importHost); code != http.StatusOK || out["result"] != "deleted" {
				t.Fatalf("overlapping delete = %d %v, want 200", code, out)
			}
			f.requireRecordGone(t, importHost)
		})
	}
}

// The pre-swap delete of a private-storage import tears down the empty
// store its begin made (preflight m11).
func TestImportDeleteTearsDownTheStore(t *testing.T) {
	f := newImportFixture(t)
	enableInstances(t, f.fx, fromApp, fromManifest(true))
	code, out := beginImport(t, f, importHost, importSHA, 10)
	rec, _ := f.requireBegun(t, code, out, importHost)
	store, _, err := objstore.GetIn(context.Background(), f.fx.rel, "bab", fromApp, rec.Environment)
	if err != nil || store == nil || store.Declared == nil || store.Declared.Seed != rec.Seed.ID {
		t.Fatalf("store = %+v, %v; want one marked with seed %s", store, err, rec.Seed.ID)
	}
	bucket := storeBucketName(store)
	if _, ok := f.fake.Buckets[bucket]; !ok {
		t.Fatalf("bucket %s was never made", bucket)
	}
	if code, out := deleteInstance(t, f, importHost); code != http.StatusOK || out["result"] != "deleted" {
		t.Fatalf("delete = %d %v", code, out)
	}
	f.requireRecordGone(t, importHost)
	if store, _, err := objstore.GetIn(context.Background(), f.fx.rel, "bab", fromApp, rec.Environment); err != nil || store != nil {
		t.Fatalf("store after the delete = %+v, %v; want it gone", store, err)
	}
	if _, ok := f.fake.Buckets[bucket]; ok {
		t.Fatalf("bucket %s survived the delete", bucket)
	}
}

// failGetOn fails reads of key while armed.
type failGetOn struct {
	release.Store
	key   string
	armed atomic.Bool
}

func (s *failGetOn) Get(ctx context.Context, key string) ([]byte, string, error) {
	if key == s.key && s.armed.Load() {
		return nil, "", errors.New("the bucket could not be read")
	}
	return s.Store.Get(ctx, key)
}

// The delete door's import rule fails closed (preflight M15, M8): a record
// it cannot read, a finish that fails, and a seed id that cannot name a key
// each answer main's delete sentence and delete nothing. A finish that
// fails leaves the record rolled-back for a retry, which completes it.
func TestImportDeleteFailsClosed(t *testing.T) {
	f := newImportFixture(t)
	id := newSeedID(t)
	f.seedRecord(t, importHost, "", id, seeds.KindImport, seeds.StateUploaded)

	rel := &failGetOn{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, importHost)}
	rel.armed.Store(true)
	f.fx.srv.cfg.Releases = rel
	if code, out := deleteInstance(t, f, importHost); code != http.StatusBadGateway || str(t, out, "error") != importDeleteRetry {
		t.Fatalf("delete with an unreadable record = %d %v", code, out)
	}
	f.fx.srv.cfg.Releases = f.fx.rel
	rec, _ := f.record(t, importHost)
	requireImportState(t, rec, id, seeds.StateUploaded, "")

	data := f.fx.srv.cfg.SeedData
	f.fx.srv.cfg.SeedData = nil
	if code, out := deleteInstance(t, f, importHost); code != http.StatusBadGateway || str(t, out, "error") != importDeleteRetry {
		t.Fatalf("delete with a failing finish = %d %v", code, out)
	}
	rec, _ = f.record(t, importHost)
	requireImportState(t, rec, id, seeds.StateRolledBack, "")
	if rows := importAudit(t, f, "instance-delete"); len(rows) != 0 {
		t.Fatalf("a failed delete audited: %q", rows)
	}
	f.fx.srv.cfg.SeedData = data
	if code, out := deleteInstance(t, f, importHost); code != http.StatusOK || out["result"] != "deleted" {
		t.Fatalf("retried delete = %d %v", code, out)
	}
	f.requireRecordGone(t, importHost)

	// A seed id that cannot name a key: nothing is written or deleted, and
	// the shared sentinel key survives.
	sentinel := seeds.SnapshotKey("bad.shop.example.com", "not-a-seed-id")
	if _, err := f.data.Put(context.Background(), sentinel, []byte("another record's object"), ""); err != nil {
		t.Fatal(err)
	}
	f.seedRecord(t, "bad.shop.example.com", "", "not-a-seed-id", seeds.KindImport, seeds.StateUploaded)
	if code, out := deleteInstance(t, f, "bad.shop.example.com"); code != http.StatusBadGateway || str(t, out, "error") != importDeleteRetry {
		t.Fatalf("delete of a bad seed id = %d %v", code, out)
	}
	if rec, err := f.record(t, "bad.shop.example.com"); err != nil || rec.Seed.State != seeds.StateUploaded {
		t.Fatalf("record of a bad seed id = %+v, %v; want it untouched", rec, err)
	}
	if got := f.rawKey(t, f.data, sentinel); got != "another record's object" {
		t.Fatalf("sentinel = %q", got)
	}
}

// On a claim console the delete of an uncommitted import is refused as
// every import door is: the finish runs nowhere there, and the claim's
// ordinary retire would delete the record and leak what the seed made.
func TestImportDeleteOnAClaimConsole(t *testing.T) {
	f := newImportFixture(t)
	id := newSeedID(t)
	f.seedRecord(t, importHost, "", id, seeds.KindImport, seeds.StateAwaitingUpload)
	f.fx.srv.cfg.HostClaim = true
	defer func() { f.fx.srv.cfg.HostClaim = false }()
	if code, out := deleteInstance(t, f, importHost); code != http.StatusConflict || str(t, out, "error") != seedClaimConsoleRefusal().Msg {
		t.Fatalf("delete on a claim console = %d %v", code, out)
	}
	rec, _ := f.record(t, importHost)
	requireImportState(t, rec, id, seeds.StateAwaitingUpload, "")
}