package console

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

	lsfile "github.com/benbjohnson/litestream/file"

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

const (
	fromApp    = "shop"
	fromDomain = "shop.example.com"
	fromSource = "main.shop.example.com"
	fromHost   = "canary.shop.example.com"
	fromBody   = `{"host":"canary.shop.example.com","from":{"host":"main.shop.example.com"}}`
	fromFailed = "could not copy main.shop.example.com. Nothing was created. Try again."
	// fromAbortedReason is what a copy that fails after the answer leaves on
	// its record: the host stays taken until the sweep finishes it.
	fromAbortedReason = "could not copy main.shop.example.com, so no instance was created. canary.shop.example.com is free again within about 5 minutes. Try again then."
	fromGrew          = "the copy of main.shop.example.com grew past its planned size, so it was stopped and no instance was created."
)

// fromFixture is a console ready for --from: the app shop (no pool, its own
// domain), a source instance main with a real litestream replica, a data
// bucket for seeds, a scratch pool, the store plane armed, and a ready
// home region.
type fromFixture struct {
	fx    *consoleFixture
	fake  *objstore.MemProvisioner
	token string
	data  release.Store
	url   string
	root  string // the replica bucket's root
	txns  int
}

func fromManifest(private bool) provision.Manifest {
	return provision.Manifest{Schema: 1, App: fromApp, Domain: fromDomain, Channel: "stable", PrivateStorage: private}
}

func newFromFixture(t *testing.T, txns int) fromFixture {
	t.Helper()
	fx, fake, token, _ := privateStorageFixture(t, "code-from")
	if _, err := fx.st.CreateAppIn(OpsAccountID, fromApp, "paul@keymail.dev", time.Now()); err != nil {
		t.Fatal(err)
	}
	enableInstances(t, fx, fromApp, fromManifest(false))
	fx.srv.cfg.HibernateBucket = "data-test"
	fx.srv.cfg.HibernatePrefix = "instances"
	fx.srv.cfg.Region = ""
	data := release.NewDirStore(t.TempDir())
	fx.srv.cfg.SeedData = data
	scratch := filepath.Join(t.TempDir(), "scratch", "seeds")
	if err := clearSeedScratch(scratch); err != nil {
		t.Fatal(err)
	}
	fx.srv.seedScratch = newSeedScratch(scratch, func(string) (int64, error) { return 64 << 30, nil })
	root := t.TempDir()
	fx.srv.dbReplica = func(_ context.Context, prefix string) (replicaReader, error) {
		return lsfile.NewReplicaClient(filepath.Join(root, prefix)), nil
	}
	f := fromFixture{fx: fx, fake: fake, token: token, data: data, url: instancesURL(fx, "bab", fromApp), root: root, txns: txns}
	f.addSource(t, provision.Record{Host: fromSource, Slug: "main"}, txns)
	publishSeedsFloorFx(t, fx, "v0.27.0")
	seedReadyBox(t, fx, "kass-1", "", 1, "v0.27.0", "v0.27.0", "nested", false)
	// A copy still running when the test ends would write into removed
	// directories: every test waits for its copies.
	t.Cleanup(fx.srv.seedCopies.Wait)
	return f
}

// addSource writes a source record of shop (rec's other fields as given)
// and, when txns > 0, a replica of that many batches under its prefix.
func (f fromFixture) addSource(t *testing.T, rec provision.Record, txns int) {
	t.Helper()
	if txns > 0 {
		dbdoortest.Replicate(t, filepath.Join(f.root, "instances", rec.Host), txns)
	}
	rec.Schema, rec.App, rec.Account, rec.Channel, rec.CreatedAt = 1, fromApp, "bab", "stable", time.Now().UTC()
	if err := provision.CreateRecord(context.Background(), f.fx.rel, rec); err != nil {
		t.Fatal(err)
	}
}

// create posts a create and returns what the door answered.
func (f fromFixture) create(t *testing.T, body string) (int, map[string]any) {
	t.Helper()
	return f.createAs(t, f.token, body)
}

func (f fromFixture) createAs(t *testing.T, token, body string) (int, map[string]any) {
	t.Helper()
	code, _, raw := rawReq(t, http.MethodPost, f.url, token, body)
	var out map[string]any
	_ = json.Unmarshal([]byte(raw), &out)
	return code, out
}

// copy posts a --from create, requires its 202 with a reserving record,
// waits for the detached copy to finish, and returns the record it left.
func (f fromFixture) copy(t *testing.T, body string) *provision.Record {
	t.Helper()
	return f.copyAs(t, f.token, body)
}

func (f fromFixture) copyAs(t *testing.T, token, body string) *provision.Record {
	t.Helper()
	code, out := f.createAs(t, token, body)
	seed, _ := out["seed"].(map[string]any)
	if code != http.StatusAccepted || seed == nil || seed["state"] != seeds.StateReserving || seed["kind"] != seeds.KindReplica {
		t.Fatalf("%s = %d %v, want 202 with a reserving record", body, code, out)
	}
	f.fx.srv.seedCopies.Wait()
	host, _ := out["host"].(string)
	rec, err := f.record(t, host)
	if err != nil || rec.Seed == nil || rec.Seed.ID != seed["id"] {
		t.Fatalf("record %s after its copy = %+v, %v", host, rec, err)
	}
	return rec
}

func (f fromFixture) record(t *testing.T, host string) (*provision.Record, error) {
	t.Helper()
	return provision.GetRecord(context.Background(), f.fx.rel, "bab", fromApp, host)
}

// requireAborted: the record's replica seed is aborted, saying why.
func requireAborted(t *testing.T, rec *provision.Record, reason string) {
	t.Helper()
	if rec.Seed == nil || rec.Seed.State != seeds.StateAborted || rec.Seed.Kind != seeds.KindReplica || rec.Seed.Reason != reason {
		t.Fatalf("record %s seed = %+v; want aborted with reason %q", rec.Host, rec.Seed, reason)
	}
}

func requireReady(t *testing.T, rec *provision.Record) {
	t.Helper()
	if rec.Seed == nil || rec.Seed.State != seeds.StateReady || rec.Seed.Reason != "" {
		t.Fatalf("record %s seed = %+v; want ready", rec.Host, rec.Seed)
	}
}

// requireNoRecord: the refusal came before the record was reserved.
func (f fromFixture) requireNoRecord(t *testing.T, host string) {
	t.Helper()
	if _, err := f.record(t, host); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("record %s: err = %v, want none written", host, err)
	}
}

// requireClean: the scratch is empty and the pool holds nothing for shop.
func (f fromFixture) requireClean(t *testing.T) {
	t.Helper()
	if ents, _ := os.ReadDir(f.fx.srv.seedScratch.dir()); len(ents) != 0 {
		t.Fatalf("scratch left behind: %v", ents)
	}
	rel, ref := f.fx.srv.seedScratch.reserve("bab/"+fromApp, 1)
	if ref != nil {
		t.Fatalf("the reservation was not released: %+v", ref)
	}
	rel()
}

func (f fromFixture) requireNothingUploaded(t *testing.T) {
	t.Helper()
	if keys, err := f.data.List(context.Background(), seeds.Prefix); err != nil || len(keys) != 0 {
		t.Fatalf("seeds/ = %v, %v; want nothing uploaded", keys, err)
	}
}

func snapshotRows(t *testing.T, raw []byte) int {
	t.Helper()
	path := filepath.Join(t.TempDir(), "snapshot.db")
	if err := os.WriteFile(path, raw, 0o600); err != nil {
		t.Fatal(err)
	}
	db, err := sql.Open("sqlite", "file:"+path+"?mode=ro")
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	var n int
	if err := db.QueryRow(`SELECT count(*) FROM t`).Scan(&n); err != nil {
		t.Fatal(err)
	}
	return n
}

func TestFromCreatesAReadySeededInstance(t *testing.T) {
	f := newFromFixture(t, 30)
	rec := f.copy(t, `{"host":"canary.shop.example.com","from":{"host":"main.shop.example.com"},"message":"a copy to try the upgrade on"}`)
	requireReady(t, rec)
	if rec.Seed.Kind != seeds.KindReplica || rec.Seed.From != fromSource || rec.Seed.TXID == "" || rec.Seed.Bytes == 0 ||
		len(rec.Seed.SHA256) != 64 || rec.Seed.By != "alice@keymail.dev" || !seeds.ValidID(rec.Seed.ID) {
		t.Fatalf("record seed = %+v", rec.Seed)
	}
	if rec.Environment != "" || rec.Slug != "canary" || rec.Channel != "stable" {
		t.Fatalf("record = %+v; want the ordinary create's fields", rec)
	}
	raw, _, err := f.data.Get(context.Background(), seeds.SnapshotKey(rec.Host, rec.Seed.ID))
	if err != nil {
		t.Fatalf("snapshot: %v", err)
	}
	sum := sha256.Sum256(raw)
	if hex.EncodeToString(sum[:]) != rec.Seed.SHA256 || int64(len(raw)) != rec.Seed.Bytes {
		t.Fatalf("snapshot is %d bytes, sha %x; record says %d, %s", len(raw), sum, rec.Seed.Bytes, rec.Seed.SHA256)
	}
	if n := snapshotRows(t, raw); n != f.txns*dbdoortest.Rows {
		t.Fatalf("snapshot holds %d rows, want %d", n, f.txns*dbdoortest.Rows)
	}
	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 == "instance-seed" {
			rows = append(rows, e.Detail+" | "+e.Address)
		}
		if e.Action == "instance-create" {
			t.Errorf("a seeded create also wrote an instance-create row: %+v", e)
		}
	}
	want := "seeded canary.shop.example.com from main.shop.example.com at txid " + rec.Seed.TXID + " | alice@keymail.dev"
	if len(rows) != 1 || rows[0] != want {
		t.Fatalf("instance-seed rows = %q, want [%q]", rows, want)
	}
}

func TestFromRefusals(t *testing.T) {
	f := newFromFixture(t, 30)
	for _, c := range []struct {
		body string
		code int
		want string
	}{
		{`{"host":"c1.shop.example.com","from":{"host":"nope.shop.example.com"}}`, http.StatusConflict, "no instance nope.shop.example.com in shop. A copy can only come from an instance of the same app in the same account."},
		{`{"host":"c2.shop.example.com","kind":"static","from":{"host":"main.shop.example.com"}}`, http.StatusBadRequest, "a seeded instance is an ordinary instance. Drop --kind, --addr and --role."},
		{`{"host":"c2.shop.example.com","addrs":["127.0.0.1:9000"],"from":{"host":"main.shop.example.com"}}`, http.StatusBadRequest, "a seeded instance is an ordinary instance"},
		{`{"host":"c2.shop.example.com","role":"replica","from":{"host":"main.shop.example.com"}}`, http.StatusBadRequest, "a seeded instance is an ordinary instance"},
		{`{"host":"c2.shop.example.com","replicas":["us"],"from":{"host":"main.shop.example.com"}}`, http.StatusBadRequest, "a seeded instance is an ordinary instance"},
		{`{"host":"main.shop.example.com","from":{"host":"main.shop.example.com"}}`, http.StatusConflict, "instance main.shop.example.com already exists"},
	} {
		code, out := f.create(t, c.body)
		if code != c.code || !strings.Contains(str(t, out, "error"), c.want) {
			t.Errorf("%s = %d %v, want %d %q", c.body, code, out, c.code, c.want)
		}
	}
	for _, h := range []string{"c1.shop.example.com", "c2.shop.example.com"} {
		f.requireNoRecord(t, h)
	}
	f.requireClean(t)
}

// --at: a target inside the history restores exactly there, and one at the
// end is the latest copy. One past the end is refused before anything is
// reserved, naming the latest transaction the source has replicated.
func TestFromAt(t *testing.T) {
	f := newFromFixture(t, 30)
	latest, err := dbdoor.PlanRestore(context.Background(), dbdoor.NewClient(replicaDoer(lsfile.NewReplicaClient(filepath.Join(f.root, "instances", fromSource)))), 0)
	if err != nil {
		t.Fatal(err)
	}
	for _, c := range []struct {
		host string
		txid uint64
		want uint64
	}{
		{"c3.shop.example.com", 1, 1},
		{"c4.shop.example.com", latest.TXID, latest.TXID},
	} {
		body, _ := json.Marshal(map[string]any{"host": c.host, "from": map[string]any{"host": fromSource, "txid": c.txid}})
		rec := f.copy(t, string(body))
		requireReady(t, rec)
		if rec.Seed.TXID != strconv.FormatUint(c.want, 10) {
			t.Fatalf("--at %d: restored at txid %s, want %d", c.txid, rec.Seed.TXID, c.want)
		}
	}
	code, out := f.create(t, `{"host":"c5.shop.example.com","from":{"host":"main.shop.example.com","txid":999999}}`)
	want := fmt.Sprintf("main.shop.example.com has replicated up to txid %d, so there is no txid 999999 to copy yet. Leave out --at to copy the latest.", latest.TXID)
	if code != http.StatusConflict || str(t, out, "error") != want {
		t.Fatalf("--at past the latest = %d %v, want 409 %q", code, out, want)
	}
	f.requireNoRecord(t, "c5.shop.example.com")
	f.requireClean(t)
}

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

// --from is refused while host claims are armed: replicaDoer has no
// pinned-version branch, and the seed floor's own claim rule refuses
// first, before any record is written (Task 4's carried item).
func TestFromRefusedWithClaimsArmed(t *testing.T) {
	f := newFromFixture(t, 10)
	f.fx.srv.cfg.HostClaim = true
	code, out := f.create(t, fromBody)
	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, fromHost)
}

func TestFromTooLarge(t *testing.T) {
	f := newFromFixture(t, 10)
	old := seedMaxBytes
	seedMaxBytes = 1024 // the test's database is larger than 1 KiB
	defer func() { seedMaxBytes = old }()
	code, out := f.create(t, fromBody)
	if code != http.StatusRequestEntityTooLarge || str(t, out, "error") != "a copy of main.shop.example.com would be 0.1 GiB, over the 4 GiB limit." {
		t.Fatalf("= %d %v", code, out)
	}
	f.requireNoRecord(t, fromHost)
	f.requireClean(t)
}

// SD-N26 rounds up, so a size over the limit never prints as the limit.
func TestSeedGiBRoundsUp(t *testing.T) {
	for n, want := range map[int64]string{1: "0.1 GiB", 4 << 30: "4.0 GiB", 4<<30 + 1: "4.1 GiB", 5 << 29: "2.5 GiB"} {
		if got := seedGiB(n); got != want {
			t.Errorf("seedGiB(%d) = %q, want %q", n, got, want)
		}
	}
}

func TestFromSourceRefusals(t *testing.T) {
	f := newFromFixture(t, 3)
	f.addSource(t, provision.Record{Host: "pooled.shop.example.com", Slug: "pooled", Pool: "shared"}, 0)
	f.addSource(t, provision.Record{Host: "fleet.shop.example.com", Slug: "fleet", Region: "fleet/racks"}, 0)
	f.addSource(t, provision.Record{Host: "site.shop.example.com", Slug: "site", Kind: "static"}, 0)
	f.addSource(t, provision.Record{Host: "far.shop.example.com", Slug: "far", Region: "us"}, 0)
	f.addSource(t, provision.Record{Host: "empty.shop.example.com", Slug: "empty"}, 0)
	for _, c := range []struct {
		source string
		code   int
		want   string
	}{
		{"pooled.shop.example.com", http.StatusConflict, "pooled.shop.example.com runs in the pool shared, so the platform keeps no replica of it to copy."},
		{"fleet.shop.example.com", http.StatusConflict, "fleet.shop.example.com runs on your fleet racks, so the platform keeps no replica of it to copy."},
		{"site.shop.example.com", http.StatusConflict, "site.shop.example.com is not an instance the platform keeps a replica of, so it cannot be copied."},
		{"far.shop.example.com", http.StatusConflict, "far.shop.example.com keeps its data in "},
		{"empty.shop.example.com", http.StatusConflict, "empty.shop.example.com has not replicated anything yet. Wake it once, then try again."},
	} {
		code, out := f.create(t, `{"host":"canary.shop.example.com","from":{"host":"`+c.source+`"}}`)
		if code != c.code || !strings.Contains(str(t, out, "error"), c.want) {
			t.Errorf("from %s = %d %v, want %d %q", c.source, code, out, c.code, c.want)
		}
	}
	f.requireNoRecord(t, fromHost)
	f.requireClean(t)
}

// A copy stays in its source's region: a target homed elsewhere is
// refused, before any record is written.
func TestFromRefusesAnotherRegion(t *testing.T) {
	f := newFromFixture(t, 3)
	seedReadyBox(t, f.fx, "far-1", "us", 1, "v0.27.0", "v0.27.0", "nested", false)
	code, out := f.create(t, `{"host":"canary.shop.example.com","region":"us","from":{"host":"main.shop.example.com"}}`)
	if code != http.StatusConflict || str(t, out, "error") != "a copy stays in its source's region. To run it elsewhere, create it here and then run carlos move." {
		t.Fatalf("= %d %v", code, out)
	}
	f.requireNoRecord(t, fromHost)
}

func TestFromPlanFailures(t *testing.T) {
	f := newFromFixture(t, 3)
	oldPlan := planRestore
	defer func() { planRestore = oldPlan }()
	for i, c := range []struct {
		err  error
		body string
		code int
		want string
	}{
		{dbdoor.ErrHistoryGone, `{"host":"canary.shop.example.com","from":{"host":"main.shop.example.com","txid":2}}`, http.StatusConflict, "history before 2 is no longer kept"},
		{dbdoor.ErrNothingReplicated, `{"host":"canary.shop.example.com","from":{"host":"main.shop.example.com","txid":2}}`, http.StatusConflict, "history before 2 is no longer kept"},
		{dbdoor.ErrNothingReplicated, fromBody, http.StatusConflict, "main.shop.example.com has not replicated anything yet. Wake it once, then try again."},
		{nil, `{"host":"canary.shop.example.com","from":{"host":"main.shop.example.com","txid":9}}`, http.StatusConflict, "main.shop.example.com has replicated up to txid 5, so there is no txid 9 to copy yet. Leave out --at to copy the latest."},
		{errors.New("dbdoor: the last planned file's header does not describe it"), fromBody, http.StatusBadGateway, fromFailed},
		{os.ErrNotExist, fromBody, http.StatusBadGateway, fromFailed},
	} {
		planRestore = func(context.Context, *dbdoor.Client, uint64) (dbdoor.RestorePlan, error) {
			if c.err == nil {
				return dbdoor.RestorePlan{TXID: 5}, nil // the plan clamps a later --at to the latest
			}
			return dbdoor.RestorePlan{}, c.err
		}
		code, out := f.create(t, c.body)
		if code != c.code || str(t, out, "error") != c.want {
			t.Errorf("%d: %v = %d %v, want %d %q", i, c.err, code, out, c.code, c.want)
		}
	}
	f.requireNoRecord(t, fromHost)
	f.requireClean(t)
}

// Every failure after the record is reserved aborts it with the sentence
// the door would have answered, and leaves nothing in scratch or the pool.
func TestFromRestoreFailuresAbort(t *testing.T) {
	oldRestore, oldDeadline := restoreToTarget, seedRestoreDeadline
	defer func() { restoreToTarget, seedRestoreDeadline = oldRestore, oldDeadline }()
	for i, c := range []struct {
		restore func(ctx context.Context) error
		body    string
		reason  string
	}{
		{func(context.Context) error { return dbdoor.ErrCheckFailed }, fromBody, "the copy of main.shop.example.com failed its integrity check, so no instance was created."},
		{func(context.Context) error { return dbdoor.ErrHistoryGone }, fromBody, fromAbortedReason},
		{func(context.Context) error { return dbdoor.ErrHistoryGone }, `{"host":"canary.shop.example.com","from":{"host":"main.shop.example.com","txid":2}}`, "history before 2 is no longer kept"},
		{func(context.Context) error { return errors.New("bucket read failed") }, fromBody, fromAbortedReason},
		{func(ctx context.Context) error { <-ctx.Done(); return ctx.Err() }, fromBody, "copying main.shop.example.com took longer than 30 minutes, so no instance was created."},
	} {
		f := newFromFixture(t, 3)
		seedRestoreDeadline = 200 * time.Millisecond
		restoreToTarget = func(ctx context.Context, _ *dbdoor.Client, parent string, _ uint64) (*dbdoor.Restored, error) {
			if _, err := os.MkdirTemp(parent, "carlos-db-"); err != nil { // scratch the door must clear
				return nil, err
			}
			return nil, c.restore(ctx)
		}
		rec := f.copy(t, c.body)
		if rec.Seed.State != seeds.StateAborted || rec.Seed.Reason != c.reason {
			t.Errorf("%d: seed = %+v, want aborted with %q", i, rec.Seed, c.reason)
			continue
		}
		f.requireClean(t)
		f.requireNothingUploaded(t)
	}
}

// The restore is pinned at the planned TXID, whatever the request said:
// only that restore produces the size that was admitted.
func TestFromRestoresAtThePlannedTXID(t *testing.T) {
	f := newFromFixture(t, 5)
	oldRestore := restoreToTarget
	defer func() { restoreToTarget = oldRestore }()
	var got []uint64
	restoreToTarget = func(ctx context.Context, c *dbdoor.Client, parent string, target uint64) (*dbdoor.Restored, error) {
		got = append(got, target)
		return oldRestore(ctx, c, parent, target)
	}
	plan, err := dbdoor.PlanRestore(context.Background(), dbdoor.NewClient(replicaDoer(lsfile.NewReplicaClient(filepath.Join(f.root, "instances", fromSource)))), 0)
	if err != nil {
		t.Fatal(err)
	}
	requireReady(t, f.copy(t, fromBody))
	requireReady(t, f.copy(t, fmt.Sprintf(`{"host":"c2.shop.example.com","from":{"host":"main.shop.example.com","txid":%d}}`, plan.TXID)))
	if len(got) != 2 || got[0] != plan.TXID || got[1] != plan.TXID {
		t.Fatalf("restored at %v, want [%d %d]", got, plan.TXID, plan.TXID)
	}
}

// A restore that finished larger than planned is refused after the fact
// (the watcher's tick can come too late for a fast restore).
func TestFromRefusesARestoreLargerThanPlanned(t *testing.T) {
	f := newFromFixture(t, 30)
	oldPlan := planRestore
	planRestore = func(ctx context.Context, c *dbdoor.Client, target uint64) (dbdoor.RestorePlan, error) {
		p, err := oldPlan(ctx, c, target)
		p.RestoredBytes = 1 // admit far less than the restore will write
		return p, err
	}
	defer func() { planRestore = oldPlan }()
	requireAborted(t, f.copy(t, fromBody), fromGrew)
	f.requireClean(t)
	f.requireNothingUploaded(t)
}

// The watcher cancels a restore that is still running: litestream writes
// replica.db.tmp and renames it only at the end, so the watcher must see
// the .tmp file grow.
func TestFromWatcherCancelsARestoreInFlight(t *testing.T) {
	f := newFromFixture(t, 3)
	oldRestore, oldEvery := restoreToTarget, restoreWatchEvery
	defer func() { restoreToTarget, restoreWatchEvery = oldRestore, oldEvery }()
	restoreWatchEvery = 20 * time.Millisecond
	cancelled := make(chan bool, 1)
	restoreToTarget = func(rctx context.Context, _ *dbdoor.Client, parent string, _ uint64) (*dbdoor.Restored, error) {
		dir, err := os.MkdirTemp(parent, "carlos-db-")
		if err != nil {
			return nil, err
		}
		defer os.RemoveAll(dir)
		out, err := os.Create(filepath.Join(dir, "replica.db.tmp"))
		if err != nil {
			return nil, err
		}
		defer out.Close()
		chunk := make([]byte, 64<<10)
		giveUp := time.After(10 * time.Second)
		for written := 0; ; {
			if written < 4<<20 { // well past the database's planned size, and bounded
				n, _ := out.Write(chunk)
				written += n
			}
			select {
			case <-rctx.Done():
				cancelled <- true
				return nil, rctx.Err()
			case <-giveUp:
				cancelled <- false
				return nil, errors.New("the watcher never cancelled the restore")
			case <-time.After(10 * time.Millisecond):
			}
		}
	}
	rec := f.copy(t, fromBody)
	if !<-cancelled {
		t.Fatal("the restore ran to its own timeout: the watcher did not cancel it in flight")
	}
	requireAborted(t, rec, fromGrew)
	f.requireClean(t)
}

// Once the door has answered 202 the copy no longer depends on the
// request: the request's context is cancelled when the handler returns,
// and the client then cancelling and closing its connection changes
// nothing. The copy finishes, ready or aborted, and cleans up after itself
// either way. (A disconnect before the 202 is a refusal like any other.)
func TestFromCopyOutlivesItsClient(t *testing.T) {
	for _, c := range []struct {
		name   string
		fail   error
		reason string
	}{
		{"ready", nil, ""},
		{"aborted", dbdoor.ErrCheckFailed, "the copy of main.shop.example.com failed its integrity check, so no instance was created."},
	} {
		t.Run(c.name, func(t *testing.T) {
			f := newFromFixture(t, 3)
			oldRestore := restoreToTarget
			defer func() { restoreToTarget = oldRestore }()
			started, proceed := make(chan struct{}), make(chan struct{})
			restoreToTarget = func(rctx context.Context, cl *dbdoor.Client, parent string, target uint64) (*dbdoor.Restored, error) {
				close(started)
				<-proceed
				if rctx.Err() != nil {
					return nil, rctx.Err()
				}
				if c.fail != nil {
					return nil, c.fail
				}
				return oldRestore(rctx, cl, parent, target)
			}
			tr := &http.Transport{}
			client := &http.Client{Transport: tr}
			ctx, cancel := context.WithCancel(context.Background())
			req, _ := http.NewRequestWithContext(ctx, http.MethodPost, f.url, strings.NewReader(fromBody))
			req.Header.Set("Authorization", "Bearer "+f.token)
			req.Header.Set("Content-Type", "application/json")
			res, err := client.Do(req)
			if err != nil {
				t.Fatal(err)
			}
			res.Body.Close()
			if res.StatusCode != http.StatusAccepted {
				t.Fatalf("= %d, want 202", res.StatusCode)
			}
			<-started
			cancel() // the client goes away, and its connection with it
			tr.CloseIdleConnections()
			close(proceed)
			f.fx.srv.seedCopies.Wait()
			rec, err := f.record(t, fromHost)
			if err != nil {
				t.Fatal(err)
			}
			if c.fail == nil {
				requireReady(t, rec)
			} else {
				requireAborted(t, rec, c.reason)
			}
			f.requireClean(t)
		})
	}
}

// A panic in the detached copy aborts its seed and leaves the console
// serving: nothing above the copy's goroutine would contain it.
func TestFromPanicAbortsTheSeedAndSparesTheConsole(t *testing.T) {
	f := newFromFixture(t, 3)
	oldRestore := restoreToTarget
	defer func() { restoreToTarget = oldRestore }()
	restoreToTarget = func(ctx context.Context, c *dbdoor.Client, parent string, target uint64) (*dbdoor.Restored, error) {
		if _, err := os.MkdirTemp(parent, "carlos-db-"); err != nil { // scratch the door must clear
			return nil, err
		}
		panic("a decoder fault")
	}
	requireAborted(t, f.copy(t, fromBody), fromAbortedReason)
	f.requireClean(t)
	f.requireNothingUploaded(t)
	// Still serving: the next copy runs to ready.
	restoreToTarget = oldRestore
	requireReady(t, f.copy(t, `{"host":"c2.shop.example.com","from":{"host":"main.shop.example.com"}}`))
	f.requireClean(t)
}

func TestFromUploadFailureAborts(t *testing.T) {
	f := newFromFixture(t, 3)
	f.fx.srv.cfg.SeedData = failingPuts{f.data}
	requireAborted(t, f.copy(t, fromBody), fromAbortedReason)
	f.requireClean(t)
}

type failingPuts struct{ release.Store }

func (failingPuts) Put(context.Context, string, []byte, string) (string, error) {
	return "", errors.New("access denied")
}

// A publish that fails aborts the seed, with the retry sentence.
func TestFromPublishFailureAborts(t *testing.T) {
	f := newFromFixture(t, 3)
	rel := &failOnceOn{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, fromHost)}
	f.fx.srv.cfg.Releases = rel
	oldRestore := restoreToTarget
	defer func() { restoreToTarget = oldRestore }()
	restoreToTarget = func(ctx context.Context, c *dbdoor.Client, parent string, target uint64) (*dbdoor.Restored, error) {
		rel.armed.Store(true) // the next write of the record is the publish
		return oldRestore(ctx, c, parent, target)
	}
	requireAborted(t, f.copy(t, fromBody), fromAbortedReason)
	f.requireClean(t)
}

// failOnceOn fails the first Put of key once armed.
type failOnceOn struct {
	release.Store
	key   string
	armed atomic.Bool
}

func (s *failOnceOn) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if key == s.key && s.armed.CompareAndSwap(true, false) {
		return "", errors.New("the bucket refused the write")
	}
	return s.Store.Put(ctx, key, body, ifMatch)
}

// A record that appears between the create door's checks and this door's
// reservation is answered as the ordinary create answers it, and leaves
// nothing behind.
func TestFromReservationLosesARace(t *testing.T) {
	f := newFromFixture(t, 3)
	oldPlan := planRestore
	defer func() { planRestore = oldPlan }()
	planRestore = func(ctx context.Context, c *dbdoor.Client, target uint64) (dbdoor.RestorePlan, error) {
		// Another writer's record, written from the handler's goroutine:
		// the test checks it below rather than failing here.
		_ = provision.CreateRecord(ctx, f.fx.rel, provision.Record{Schema: 1, Host: fromHost, Slug: "canary",
			App: fromApp, Account: "bab", Channel: "stable", CreatedAt: time.Now().UTC()})
		return oldPlan(ctx, c, target)
	}
	code, out := f.create(t, fromBody)
	if code != http.StatusConflict || str(t, out, "error") != "instance canary.shop.example.com already exists" {
		t.Fatalf("= %d %v", code, out)
	}
	if rec, err := f.record(t, fromHost); err != nil || rec.Seed != nil {
		t.Fatalf("record = %+v, %v; want the other writer's, unseeded", rec, err)
	}
	f.requireClean(t)
}

func TestFromNeedsADataBucket(t *testing.T) {
	f := newFromFixture(t, 3)
	f.fx.srv.cfg.SeedData = nil
	code, out := f.create(t, fromBody)
	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, fromHost)
}

func TestFromRefusesAPooledApp(t *testing.T) {
	f := newFromFixture(t, 3)
	m := fromManifest(false)
	m.Pool = "shared"
	enableInstances(t, f.fx, fromApp, m)
	code, out := f.create(t, fromBody)
	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, fromHost)
}

// The instance service role reaches --from through the create door's own
// route, as spec §6.1 says; the db door stays person-only.
func TestFromByAnInstanceServiceRole(t *testing.T) {
	f := newFromFixture(t, 3)
	svc := serviceToken(t, f.fx.st, OpsAccountID, "ci", fromApp, svcrole.Instance)
	rec := f.copyAs(t, svc, fromBody)
	requireReady(t, rec)
	if rec.Seed.By != "service:bab/ci" {
		t.Fatalf("record seed by %q, want the service principal", rec.Seed.By)
	}
}

func TestFromPrivateStorageRules(t *testing.T) {
	f := newFromFixture(t, 3)
	enableInstances(t, f.fx, fromApp, fromManifest(true))
	code, out := f.create(t, `{"host":"canary.shop.example.com","environment":"main","from":{"host":"main.shop.example.com"}}`)
	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)
	}
	// A store already provisioned for the derived environment, by no seed.
	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"},
		Grant:    &objstore.GrantBlock{Backend: "mem", Bucket: "carlos-bab-shop-canary", Source: objstore.GrantSourceAccountApproval, At: at, By: "operator@keymail.dev"},
	})
	code, out = f.create(t, fromBody)
	if code != http.StatusConflict || str(t, out, "error") != "environment canary already has a store (carlos-bab-shop-canary). A seeded instance needs a new, empty one." {
		t.Fatalf("an existing store = %d %v", code, out)
	}
	f.requireNoRecord(t, fromHost)
	f.requireClean(t)
}

// The store gates that write nothing refuse before the record is reserved
// (ruling R3), with the store plane's own answers.
func TestFromStoreGatesRefuseBeforeTheReservation(t *testing.T) {
	for name, c := range map[string]struct {
		unwire func(*Server)
		want   string
	}{
		"no backend": {func(s *Server) { s.cfg.StoreProvisioner = nil },
			"object storage is not configured for this deployment (CARLOS_STORE_BACKEND), and shop gives every instance its own storage — no instance can be created until one is wired or the policy is turned off (carlos instances enable --private-storage=false)"},
		"no sealing key": {func(s *Server) { s.cfg.SecretsPub = "" },
			"this console has no secrets key (CARLOS_SECRETS_PUBKEY) — it cannot deliver a store credential, so it will not mint one"},
	} {
		t.Run(name, func(t *testing.T) {
			f := newFromFixture(t, 3)
			enableInstances(t, f.fx, fromApp, fromManifest(true))
			c.unwire(f.fx.srv)
			code, out := f.create(t, fromBody)
			if code != http.StatusServiceUnavailable || str(t, out, "error") != c.want {
				t.Fatalf("= %d %v, want 503 %q", code, out, c.want)
			}
			f.requireNoRecord(t, fromHost)
			f.requireClean(t)
		})
	}
}

// A store walk refused after the reservation aborts the seed with the walk's
// own sentence as its reason.
func TestFromStoreWalkRefusalAborts(t *testing.T) {
	f := newFromFixture(t, 3)
	enableInstances(t, f.fx, fromApp, fromManifest(true))
	takeCanaryBucketName(t, f.fx)
	rec := f.copy(t, fromBody)
	requireAborted(t, rec, "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")
	f.requireClean(t)
	f.requireNothingUploaded(t)
}

func putCanaryStoreOf(t *testing.T, fx *consoleFixture, app string, rec *objstore.Record) {
	t.Helper()
	if _, err := objstore.PutIn(context.Background(), fx.rel, release.OpsAccount, app, "canary", rec, ""); err != nil {
		t.Fatal(err)
	}
}

// With private storage the copy gets a new store marked with its seed, and
// the source's own store is not read, bound or touched.
func TestFromPrivateStorageGetsANewStoreMarkedWithTheSeed(t *testing.T) {
	f := newFromFixture(t, 3)
	enableInstances(t, f.fx, fromApp, fromManifest(true))
	at := time.Now().UTC().Add(-time.Hour)
	if _, err := objstore.PutIn(context.Background(), f.fx.rel, release.OpsAccount, fromApp, "main", &objstore.Record{
		Declared: &objstore.Declared{EnvPrefix: "S3", At: at, By: "bob@keymail.dev"},
		Grant:    &objstore.GrantBlock{Backend: "mem", Bucket: "carlos-bab-shop-main", Source: objstore.GrantSourceAccountApproval, At: at, By: "operator@keymail.dev"},
	}, ""); err != nil {
		t.Fatal(err)
	}
	mainKey := objstore.KeyIn(release.OpsAccount, fromApp, "main")
	before, _, err := f.fx.rel.Get(context.Background(), mainKey)
	if err != nil {
		t.Fatal(err)
	}
	rec := f.copy(t, fromBody)
	requireReady(t, rec)
	if rec.Environment != "canary" {
		t.Fatalf("record environment = %q, want the host's own", rec.Environment)
	}
	store, _, err := objstore.GetIn(context.Background(), f.fx.rel, release.OpsAccount, fromApp, "canary")
	if err != nil || store == nil || store.Declared == nil || store.Declared.Seed != rec.Seed.ID || storeBucketName(store) == "" {
		t.Fatalf("canary store = %+v, %v; want one marked with seed %s", store, err, rec.Seed.ID)
	}
	after, _, err := f.fx.rel.Get(context.Background(), mainKey)
	if err != nil || string(after) != string(before) {
		t.Fatalf("the source's store record changed:\n before %s\n after  %s (%v)", before, after, err)
	}
	f.requireClean(t)
}

// The create door's sibling-apex re-check holds on the long --from path
// too: an app created under the host's label while the copy ran owns its
// apex, and the seed is aborted rather than published.
func TestFromAbortsWhenASiblingTakesTheApex(t *testing.T) {
	f := newFromFixture(t, 3)
	f.fx.srv.cfg.AppsDomain = "apps.example.net"
	m := fromManifest(false)
	m.Domain = "bab.apps.example.net"
	enableInstances(t, f.fx, fromApp, m)
	f.addSource(t, provision.Record{Host: "main.bab.apps.example.net", Slug: "main"}, 3)
	oldRestore := restoreToTarget
	defer func() { restoreToTarget = oldRestore }()
	restoreToTarget = func(ctx context.Context, c *dbdoor.Client, parent string, target uint64) (*dbdoor.Restored, error) {
		if _, err := f.fx.st.CreateAppIn(OpsAccountID, "canary", "paul@keymail.dev", time.Now()); err != nil {
			return nil, err
		}
		return oldRestore(ctx, c, parent, target)
	}
	rec := f.copy(t, `{"host":"canary.bab.apps.example.net","from":{"host":"main.bab.apps.example.net"}}`)
	requireAborted(t, rec, "canary.bab.apps.example.net is app canary's platform apex. That app was created while this instance was being written, so the instance was not kept.")
	f.requireClean(t)
}

// A door that held no answer never sends an empty 200 (ruling R1).
func TestHeldResponseNeverAnswersEmpty(t *testing.T) {
	w := httptest.NewRecorder()
	(&heldResponse{header: w.Header()}).flush(w, fromFailed)
	var out map[string]string
	if err := json.Unmarshal(w.Body.Bytes(), &out); err != nil || w.Code != http.StatusBadGateway || out["error"] != fromFailed {
		t.Fatalf("= %d %s, want 502 %q", w.Code, w.Body, fromFailed)
	}
}

func TestRestoreScratchUse(t *testing.T) {
	dir := t.TempDir()
	sub := filepath.Join(dir, "carlos-db-x")
	if err := os.Mkdir(sub, 0o700); err != nil {
		t.Fatal(err)
	}
	for name, n := range map[string]int{"replica.db.tmp": 100, "ltx-1": 40, "ltx-2": 60, "replica.db-shm": 32768} {
		if err := os.WriteFile(filepath.Join(sub, name), make([]byte, n), 0o600); err != nil {
			t.Fatal(err)
		}
	}
	db, total := restoreScratchUse(dir)
	if db != 100 || total != 200 {
		t.Fatalf("restoreScratchUse = %d, %d; want 100, 200 (SQLite's own -shm is not the restore's)", db, total)
	}
}