package provision

import (
	"bytes"
	"context"
	"database/sql"
	"errors"
	"log"
	"os"
	"path/filepath"
	"reflect"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/seeds"
)

// captureLog collects the package's log lines for the rest of the test.
func captureLog(t *testing.T) *bytes.Buffer {
	t.Helper()
	var b bytes.Buffer
	log.SetOutput(&b)
	t.Cleanup(func() { log.SetOutput(os.Stderr) })
	return &b
}

// journaled is an import this box has swapped: the gated exec row in
// place of the addr row, and the journal at swapped under the record's
// key. The record is whatever state was asked for.
type journaled struct {
	s   *release.DirStore
	reg *registry.Registry
	rec Record
	row registry.Route
	j   registry.ImportJournal
	cfg ReconcileConfig
}

func journaledImport(t *testing.T, state string) journaled {
	t.Helper()
	s, sqid := shopSetup(t)
	rec := newImport(sqid)
	rec.Seed.State = state
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t, addrRow(rec, sqid))
	original := mustGet(t, reg, rec.Host)
	if err := reg.SwapForImport(registry.ImportJournal{Host: rec.Host, SeedID: rec.Seed.ID, Original: *original,
		RecordKey: RecordKey(rec.Account, rec.App, rec.Host), At: 1}, importRow(rec, "/data", nil)); err != nil {
		t.Fatal(err)
	}
	j, err := reg.ImportJournal(rec.Host)
	if err != nil || j == nil {
		t.Fatalf("journal = %+v, %v", j, err)
	}
	return journaled{s: s, reg: reg, rec: rec, row: *mustGet(t, reg, rec.Host), j: *j, cfg: importCfg(t, nil)}
}

// box wires the rollback's box hooks (seedrollback_test.go) into the
// scene, so a rollback shows by its effects and the lack of one by none.
func (x *journaled) box(t *testing.T) *rollbackBox {
	t.Helper()
	b := newRollbackBox(t, x.reg, x.j.Original)
	b.hooks(x.cfg.Seeds)
	x.cfg.Seeds.Data = faultData{x.cfg.Seeds.Data, b}
	return b
}

// rolledBack asserts the rollback ran to its end: the original route back
// exactly, the journal row gone, the host fenced before its local data
// went, and its replica deleted or kept as replicaGone says.
func (x journaled) rolledBack(t *testing.T, b *rollbackBox, replicaGone bool) {
	t.Helper()
	if got := mustGet(t, x.reg, x.rec.Host); !reflect.DeepEqual(*got, x.j.Original) {
		t.Fatalf("row = %+v, want the original %+v", *got, x.j.Original)
	}
	if j, _ := x.reg.ImportJournal(x.rec.Host); j != nil {
		t.Fatalf("journal = %+v", j)
	}
	if b.local || b.replica == replicaGone || count(b.steps, "fence") != 1 {
		t.Fatalf("local %v replica %v steps %q", b.local, b.replica, b.steps)
	}
}

// untouched asserts the journal owned the host this pass: the swapped row
// and the journal row exactly as they were.
func (x journaled) untouched(t *testing.T) {
	t.Helper()
	if got := mustGet(t, x.reg, x.rec.Host); !reflect.DeepEqual(*got, x.row) {
		t.Fatalf("row changed:\n got %+v\nwant %+v", *got, x.row)
	}
	if j, _ := x.reg.ImportJournal(x.rec.Host); j == nil || !reflect.DeepEqual(*j, x.j) {
		t.Fatalf("journal changed: %+v", j)
	}
}

func TestJournalResumesTheRecordCAS(t *testing.T) {
	s, sqid := shopSetup(t)
	rec := importRecord(t, s, sqid)
	reg := regWith(t, addrRow(rec, sqid))
	stored, _ := reg.Get(rec.Host)
	// The box crashed after the transaction and before the record CAS.
	next := RouteFor(rec, "/data", "", true, nil)
	next.ImportGate = rec.Seed.VerifyHash
	if err := reg.SwapForImport(registry.ImportJournal{Host: rec.Host, SeedID: rec.Seed.ID, Original: *stored,
		RecordKey: RecordKey(rec.Account, rec.App, rec.Host)}, next); err != nil {
		t.Fatal(err)
	}
	if err := Reconcile(ctx, s, reg, importCfg(t, nil)); err != nil {
		t.Fatal(err)
	}
	r, _ := GetRecord(ctx, s, rec.Account, rec.App, rec.Host)
	if r.Seed.State != seeds.StateSwapped {
		t.Fatalf("state = %s, want the CAS redone", r.Seed.State)
	}
	if got := mustGet(t, reg, rec.Host); got.Kind != "instance" {
		t.Fatalf("the swapped row was refused or changed: %+v", got)
	}
}

func TestCommitClearsTheGateAndTheJournal(t *testing.T) {
	s, sqid := shopSetup(t)
	rec := importRecord(t, s, sqid)
	reg := regWith(t, addrRow(rec, sqid))
	cfg := importCfg(t, nil)
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	for _, step := range []func(string) func(*Record) error{SeedPlace, SeedCommit} {
		if _, err := SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, step(rec.Seed.ID)); err != nil {
			t.Fatal(err)
		}
	}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	got := mustGet(t, reg, rec.Host)
	if got.ImportGate != "" {
		t.Fatalf("gate after commit = %q", got.ImportGate)
	}
	if j, _ := reg.ImportJournal(rec.Host); j != nil {
		t.Fatalf("journal after commit = %+v", j)
	}
	// The same pass converged the row: its seed column says committed.
	if col, _ := seeds.ParseRowSeed(got.SeedSource); col == nil || col.State != seeds.StateCommitted {
		t.Fatalf("seed column after commit = %+v", col)
	}
}

// Each record state a journal row at swapped can meet (the Task 4 table,
// spec §5.4 "Crashes" and §5.6). A rollback case must leave the host to
// the rollback alone: no ordinary handling, no removal, and none of the
// no-journal paths.
func TestJournalDecidesEachRecordState(t *testing.T) {
	type want int
	const (
		rollback want = iota
		nothing
		ordinary
	)
	for _, c := range []struct {
		name  string
		state string
		edit  func(t *testing.T, x *journaled)
		want  want
		// for a rollback: the record's seed state after it, and whether
		// the host's replica was this import's to delete
		after       string
		replicaGone bool
	}{
		{name: "record gone", state: seeds.StateReady, want: rollback, replicaGone: true, edit: func(t *testing.T, x *journaled) {
			if err := x.s.Delete(ctx, RecordKey(x.rec.Account, x.rec.App, x.rec.Host)); err != nil {
				t.Fatal(err)
			}
		}},
		{name: "record with no seed", state: seeds.StateReady, want: rollback, edit: func(t *testing.T, x *journaled) {
			plain := x.rec
			plain.Seed = nil
			putReplace(t, x.s, RecordKey(x.rec.Account, x.rec.App, x.rec.Host), mustJSON(t, plain))
		}},
		{name: "another seed", state: seeds.StateReady, want: rollback, edit: func(t *testing.T, x *journaled) {
			other := x.rec
			seed := *x.rec.Seed
			seed.ID, _ = seeds.NewID()
			other.Seed = &seed
			putReplace(t, x.s, RecordKey(x.rec.Account, x.rec.App, x.rec.Host), mustJSON(t, other))
		}},
		{name: "aborted", state: seeds.StateAborted, want: rollback, after: seeds.StateRolledBack, replicaGone: true},
		// m-c: a refusal that lost the race to this box's swap. The
		// journal wins.
		{name: "refused", state: seeds.StateRefused, want: rollback, after: seeds.StateRefused, replicaGone: true},
		{name: "rolled back", state: seeds.StateRolledBack, want: rollback, after: seeds.StateRolledBack, replicaGone: true},
		// m12: states that come before any swap. Unreachable, so the
		// host is held and logged.
		{name: "reserving", state: seeds.StateReserving, want: nothing},
		{name: "awaiting upload", state: seeds.StateAwaitingUpload, want: nothing},
		{name: "uploaded", state: seeds.StateUploaded, want: nothing},
		{name: "swapped", state: seeds.StateSwapped, want: ordinary},
		{name: "placed", state: seeds.StatePlaced, want: ordinary},
	} {
		t.Run(c.name, func(t *testing.T) {
			x := journaledImport(t, c.state)
			if c.edit != nil {
				c.edit(t, &x)
			}
			b := x.box(t)
			before, _ := GetRecord(ctx, x.s, x.rec.Account, x.rec.App, x.rec.Host)
			if c.want != rollback {
				// Converging the row as an ordinary owned row would rewrite
				// its seed column: make it stale so ordinary handling shows.
				// (A rollback acts only on the import's row as the swap
				// wrote it, so its cases keep the column.)
				if err := x.reg.SetSeedSource(x.rec.Host, "{}"); err != nil {
					t.Fatal(err)
				}
				x.row.SeedSource = "{}"
			}
			logged := captureLog(t)
			if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil {
				t.Fatal(err)
			}
			after, _ := GetRecord(ctx, x.s, x.rec.Account, x.rec.App, x.rec.Host)
			switch c.want {
			case rollback:
				x.rolledBack(t, b, c.replicaGone)
				if c.after != "" {
					if after == nil || after.Seed == nil || after.Seed.State != c.after {
						t.Fatalf("record after the rollback = %+v, want seed %s", after, c.after)
					}
				} else if !reflect.DeepEqual(after, before) {
					t.Fatalf("record changed: %+v -> %+v", before, after)
				}
			case nothing:
				x.untouched(t)
				if len(b.steps) != 0 {
					t.Fatalf("a rollback started: %q", b.steps)
				}
				if !strings.Contains(logged.String(), "is journaled but its record says "+c.state) {
					t.Fatalf("the host was not held with a log line:\n%s", logged)
				}
				if !reflect.DeepEqual(after, before) {
					t.Fatalf("record changed: %+v -> %+v", before, after)
				}
			case ordinary:
				if len(b.steps) != 0 {
					t.Fatalf("rolled back a live import: %q", b.steps)
				}
				got := mustGet(t, x.reg, x.rec.Host)
				if got.SeedSource != SeedColumn(x.rec) || got.ImportGate != x.rec.Seed.VerifyHash {
					t.Fatalf("row = %+v, want it converged and still gated", got)
				}
				if j, _ := x.reg.ImportJournal(x.rec.Host); j == nil || !reflect.DeepEqual(*j, x.j) {
					t.Fatalf("journal changed: %+v", j)
				}
			}
		})
	}
}

// The redo and the commit act only for the record the journal names. A
// record holding this seed ID under another key moves nothing forward;
// the journal's own record is the one that decides, and here it is gone,
// so the import is rolled back. The record under the other key holds the
// host, so the host's replica is left to it, and the record is not moved.
func TestJournalNeedsItsRecordKeyToMoveForward(t *testing.T) {
	for _, state := range []string{seeds.StateReady, seeds.StateCommitted} {
		t.Run(state, func(t *testing.T) {
			s, sqid := shopSetup(t)
			rec := newImport(sqid)
			rec.Seed.State = state
			if err := CreateRecord(ctx, s, rec); err != nil {
				t.Fatal(err)
			}
			reg := regWith(t, addrRow(rec, sqid))
			original := mustGet(t, reg, rec.Host)
			if err := reg.SwapForImport(registry.ImportJournal{Host: rec.Host, SeedID: rec.Seed.ID, Original: *original,
				RecordKey: RecordKey(sqid, "other", rec.Host), At: 1}, importRow(rec, "/data", nil)); err != nil {
				t.Fatal(err)
			}
			cfg := importCfg(t, nil)
			b := newRollbackBox(t, reg, *original)
			b.hooks(cfg.Seeds)
			if err := Reconcile(ctx, s, reg, cfg); err != nil {
				t.Fatal(err)
			}
			if st := seedState(t, s, rec); st.State != state {
				t.Fatalf("record = %s, want %s", st.State, state)
			}
			if got := mustGet(t, reg, rec.Host); !reflect.DeepEqual(*got, *original) {
				t.Fatalf("row = %+v, want the original", got)
			}
			if j, _ := reg.ImportJournal(rec.Host); j != nil {
				t.Fatalf("journal = %+v", j)
			}
			if b.local || !b.replica {
				t.Fatalf("local %v replica %v: steps %q", b.local, b.replica, b.steps)
			}
		})
	}
}

// m3: a dry run redoes no CAS and clears no gate.
func TestJournalDryRunWritesNothing(t *testing.T) {
	for _, state := range []string{seeds.StateReady, seeds.StateCommitted} {
		t.Run(state, func(t *testing.T) {
			x := journaledImport(t, state)
			x.cfg.DryRun = true
			if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil {
				t.Fatal(err)
			}
			if st := seedState(t, x.s, x.rec); st.State != state {
				t.Fatalf("record = %s, want %s", st.State, state)
			}
			x.untouched(t)
		})
	}
}

// m4: a dry run starts no rollback either.
func TestJournalDryRunStartsNoRollback(t *testing.T) {
	x := journaledImport(t, seeds.StateSwapped)
	b := x.box(t)
	x.cfg.DryRun = true
	if err := x.s.Delete(ctx, RecordKey(x.rec.Account, x.rec.App, x.rec.Host)); err != nil {
		t.Fatal(err)
	}
	logged := captureLog(t)
	if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil {
		t.Fatal(err)
	}
	if len(b.steps) != 0 || !strings.Contains(logged.String(), "dry-run: would roll back "+x.rec.Host) {
		t.Fatalf("a dry run started a rollback (%q), or did not say it would:\n%s", b.steps, logged)
	}
	x.untouched(t)
}

// I1 (Task 4 review): another record for the journal's host never decides
// the journal's import. Here an app named other, under the same account,
// holds the host in the pass, while the import's own record is committed.
// The import is committed, never rolled back, and the other record never
// touches the row.
func TestJournalIgnoresAnotherRecordForItsHost(t *testing.T) {
	for _, c := range []struct {
		name  string
		state string
		edit  func(t *testing.T, x *journaled)
		dup   bool // the pass logs the duplicate host, keeping the other record
	}{
		{name: "another record wins the duplicate pick", state: seeds.StateCommitted, dup: true},
		{name: "another record wins the duplicate pick, swapped", state: seeds.StateSwapped, dup: true},
		{name: "the import's record is out of scope", state: seeds.StateCommitted, edit: func(t *testing.T, x *journaled) {
			moved := x.rec
			moved.Region = "us"
			putReplace(t, x.s, RecordKey(x.rec.Account, x.rec.App, x.rec.Host), mustJSON(t, moved))
		}},
	} {
		t.Run(c.name, func(t *testing.T) {
			x := journaledImport(t, c.state)
			if c.edit != nil {
				c.edit(t, &x)
			}
			other := shopRecord(x.rec.Account)
			other.App = "other" // listed before shop, so it wins a duplicate host
			// A run on the other record makes the loop's handling of it
			// visible (RunSeen): the loop must take up no other record for
			// a host the journal owns.
			other.ActiveRun = &RunState{ID: "run_abcdefghijklmnopqrstuvwxyz", State: RunQueued, Deadline: time.Now().Add(time.Minute)}
			var seen []string
			x.cfg.RunSeen = func(r Record) { seen = append(seen, RecordKey(r.Account, r.App, r.Host)) }
			otherKey := RecordKey(other.Account, other.App, other.Host)
			if _, err := x.s.Put(ctx, otherKey, mustJSON(t, other), ""); err != nil {
				t.Fatal(err)
			}
			b := x.box(t)
			logged := captureLog(t)
			if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil {
				t.Fatal(err)
			}
			out := logged.String()
			if c.dup && !strings.Contains(out, "duplicate host "+x.rec.Host+": keeping "+otherKey) {
				t.Fatalf("the scene did not put the other record in the pass:\n%s", out)
			}
			if len(b.steps) != 0 || !b.local || !b.replica {
				t.Fatalf("another record sent the import to rollback: %q\n%s", b.steps, out)
			}
			if len(seen) != 0 {
				t.Fatalf("the loop took up %v for a host the journal owns", seen)
			}
			if c.state == seeds.StateSwapped {
				x.untouched(t)
				return
			}
			got := mustGet(t, x.reg, x.rec.Host)
			if got.ImportGate != "" || got.App != "shop" || got.Account != x.rec.Account || got.Kind != "instance" || !got.Provisioned {
				t.Fatalf("row = %+v, want the import's row, committed", got)
			}
			if got.SeedSource != x.row.SeedSource || got.Socket != x.row.Socket {
				t.Fatalf("the row was converged by another record: %+v", got)
			}
			if j, _ := x.reg.ImportJournal(x.rec.Host); j != nil {
				t.Fatalf("journal = %+v, want it dropped by the commit", j)
			}
		})
	}
}

// getFails is a store whose reads fail with something other than "not
// found".
type getFails struct {
	*release.DirStore
}

func (getFails) Get(context.Context, string) ([]byte, string, error) {
	return nil, "", errors.New("connection reset")
}

// m1, m2 (Task 4 review): every way journalRecord cannot tell decides
// nothing (known false), and only "not found" under the journal's own key
// means gone.
func TestJournalRecordFailsClosed(t *testing.T) {
	s, sqid := shopSetup(t)
	rec := newImport(sqid)
	key := RecordKey(rec.Account, rec.App, rec.Host)
	j := registry.ImportJournal{Host: rec.Host, SeedID: rec.Seed.ID, RecordKey: key, State: registry.JournalSwapped}
	other := shopRecord(sqid)
	other.App = "other"
	inPass := map[string]Record{rec.Host: rec}
	otherInPass := map[string]Record{rec.Host: other}
	none := map[string]bool{}

	for _, c := range []struct {
		name       string
		store      release.Store
		j          registry.ImportJournal
		recs       map[string]Record
		readFailed bool
		protect    map[string]bool
		body       []byte // written under the journal's key first
		known      bool
		gone       bool
		pass       bool
	}{
		{name: "the pass's own record", store: s, j: j, recs: inPass, protect: none, known: true, pass: true},
		{name: "the pass's own record, another read failed", store: s, j: j, recs: inPass, readFailed: true, protect: none, known: true, pass: true},
		{name: "protected", store: s, j: j, recs: inPass, protect: map[string]bool{rec.Host: true}},
		{name: "no record key", store: s, j: registry.ImportJournal{Host: rec.Host, SeedID: rec.Seed.ID, State: registry.JournalSwapped}, recs: map[string]Record{}, protect: none},
		{name: "missing from a pass that failed a read", store: s, j: j, recs: map[string]Record{}, readFailed: true, protect: none},
		{name: "another record in the pass, a read failed", store: s, j: j, recs: otherInPass, readFailed: true, protect: none},
		{name: "the read fails", store: getFails{s}, j: j, recs: otherInPass, protect: none},
		{name: "the body does not decode", store: s, j: j, recs: otherInPass, protect: none, body: []byte("{")},
		{name: "the body names another host", store: s, j: j, recs: otherInPass, protect: none, body: func() []byte {
			r := rec
			r.Host = "b.shop.example.com"
			return mustJSON(t, r)
		}()},
		{name: "gone", store: s, j: j, recs: otherInPass, protect: none, known: true, gone: true},
		{name: "out of the pass, read by its key", store: s, j: j, recs: otherInPass, protect: none, body: mustJSON(t, rec), known: true},
	} {
		t.Run(c.name, func(t *testing.T) {
			_ = s.Delete(ctx, key)
			if c.body != nil {
				if _, err := s.Put(ctx, key, c.body, ""); err != nil {
					t.Fatal(err)
				}
			}
			got, pass, known := journalRecord(ctx, c.store, c.j, c.recs, c.readFailed, c.protect)
			if known != c.known || pass != c.pass || (known && (got == nil) != c.gone) {
				t.Fatalf("record %v, in pass %v, known %v; want known %v, gone %v, in pass %v", got, pass, known, c.known, c.gone, c.pass)
			}
			if got != nil && RecordKey(got.Account, got.App, got.Host) != key {
				t.Fatalf("answered with %s, not the journal's record", RecordKey(got.Account, got.App, got.Host))
			}
		})
	}
}

// The removal phase never removes a journaled host's row, whatever the
// journal's state: the journal owns it until the row is gone (M1, M13).
// The gated row is Provisioned and its record is gone, which without the
// journal is exactly what the removal phase deletes.
func TestRemovalSkipsJournaledHosts(t *testing.T) {
	for _, state := range []string{registry.JournalSwapped, registry.JournalRollingBack} {
		t.Run(state, func(t *testing.T) {
			x := journaledImport(t, seeds.StateSwapped)
			if state == registry.JournalRollingBack {
				if err := x.reg.CloseImportGate(x.rec.Host, x.rec.Seed.ID, 7); err != nil {
					t.Fatal(err)
				}
				x.row = *mustGet(t, x.reg, x.rec.Host)
				j, _ := x.reg.ImportJournal(x.rec.Host)
				x.j = *j
			}
			if err := x.s.Delete(ctx, RecordKey(x.rec.Account, x.rec.App, x.rec.Host)); err != nil {
				t.Fatal(err)
			}
			if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil {
				t.Fatal(err)
			}
			x.untouched(t)
		})
	}
}

// M1: the journal is read even with seed hooks unarmed. Its hosts are
// protected from removal and from ordinary handling, and nothing is done
// for them, a commit included.
func TestJournalIsReadWithSeedHooksUnarmed(t *testing.T) {
	t.Run("record gone", func(t *testing.T) {
		x := journaledImport(t, seeds.StateSwapped)
		x.cfg.Seeds = nil
		if err := x.s.Delete(ctx, RecordKey(x.rec.Account, x.rec.App, x.rec.Host)); err != nil {
			t.Fatal(err)
		}
		if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil {
			t.Fatal(err)
		}
		x.untouched(t)
	})
	t.Run("record replaced by one with no seed", func(t *testing.T) {
		x := journaledImport(t, seeds.StateSwapped)
		x.cfg.Seeds = nil
		plain := x.rec
		plain.Seed = nil
		putReplace(t, x.s, RecordKey(x.rec.Account, x.rec.App, x.rec.Host), mustJSON(t, plain))
		if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil {
			t.Fatal(err)
		}
		x.untouched(t)
	})
	t.Run("committed", func(t *testing.T) {
		x := journaledImport(t, seeds.StateCommitted)
		x.cfg.Seeds = nil
		if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil {
			t.Fatal(err)
		}
		// Neither committed nor rolled back: the gated row and the
		// journal row exactly as they were.
		x.untouched(t)
	})
}

// corruptJournal makes ImportJournals fail: the row's original no longer
// parses.
func corruptJournal(t *testing.T, path, host string) {
	t.Helper()
	db, err := sql.Open("sqlite", path)
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	if _, err := db.Exec(`UPDATE import_journal SET original = '{' WHERE host = ?`, host); err != nil {
		t.Fatal(err)
	}
}

// regAt is regWith at a path the test can open beside the registry.
func regAt(t *testing.T, routes ...registry.Route) (*registry.Registry, string) {
	t.Helper()
	path := filepath.Join(t.TempDir(), "registry.db")
	reg, err := registry.Open(path)
	if err != nil {
		t.Fatal(err)
	}
	for _, rt := range routes {
		if err := reg.Upsert(rt); err != nil {
			t.Fatal(err)
		}
	}
	return reg, path
}

// M1: a journal that cannot be read is a failed read. Nothing is removed
// that pass, and no import is acted on: no swap, and no no-journal finish
// of an aborted import (it might have a journal row).
func TestJournalReadFailureRemovesNothingAndActsOnNoImport(t *testing.T) {
	s, sqid := shopSetup(t)
	// A swapped import, whose journal row will not parse.
	swapped := importRecord(t, s, sqid)
	// A ready import on another host, with its addr row.
	ready := newImport(sqid)
	ready.Host, ready.Slug = "b.shop.example.com", "b"
	if err := CreateRecord(ctx, s, ready); err != nil {
		t.Fatal(err)
	}
	// An aborted import on a third host.
	aborted := newImport(sqid)
	aborted.Host, aborted.Slug = "c.shop.example.com", "c"
	aborted.Seed.State = seeds.StateAborted
	if err := CreateRecord(ctx, s, aborted); err != nil {
		t.Fatal(err)
	}
	// And a provisioned row whose record is gone.
	orphan := RouteFor(shopRecord(sqid), "/data", "", true, nil)
	orphan.Host, orphan.Slug = "d.shop.example.com", "d"
	reg, path := regAt(t, addrRow(swapped, sqid), addrRow(ready, sqid), orphan)
	original := mustGet(t, reg, swapped.Host)
	if err := reg.SwapForImport(registry.ImportJournal{Host: swapped.Host, SeedID: swapped.Seed.ID, Original: *original,
		RecordKey: RecordKey(swapped.Account, swapped.App, swapped.Host), At: 1}, importRow(swapped, "/data", nil)); err != nil {
		t.Fatal(err)
	}
	if err := s.Delete(ctx, RecordKey(swapped.Account, swapped.App, swapped.Host)); err != nil {
		t.Fatal(err)
	}
	gated := mustGet(t, reg, swapped.Host)
	readyRow := mustGet(t, reg, ready.Host)
	corruptJournal(t, path, swapped.Host)

	if err := Reconcile(ctx, s, reg, importCfg(t, nil)); err != nil {
		t.Fatal(err)
	}
	if got := mustGet(t, reg, swapped.Host); !reflect.DeepEqual(*got, *gated) {
		t.Fatalf("the journaled row changed: %+v", got)
	}
	mustGet(t, reg, orphan.Host) // not removed: the pass could not read the journal
	if got := mustGet(t, reg, ready.Host); !reflect.DeepEqual(*got, *readyRow) {
		t.Fatalf("the ready import was swapped: %+v", got)
	}
	if st := seedState(t, s, ready); st.State != seeds.StateReady || st.Reason != "" {
		t.Fatalf("ready import = %s %q", st.State, st.Reason)
	}
	if st := seedState(t, s, aborted); st.State != seeds.StateAborted {
		t.Fatalf("aborted import = %s, want it left aborted", st.State)
	}
}

// A journal row's record that this box's scope excludes is still there:
// it is read by its key, never taken as gone, and a committed import is
// committed, never rolled back.
func TestJournalReadsAnOutOfScopeRecordByItsKey(t *testing.T) {
	x := journaledImport(t, seeds.StateCommitted)
	moved := x.rec
	moved.Region = "us"
	putReplace(t, x.s, RecordKey(x.rec.Account, x.rec.App, x.rec.Host), mustJSON(t, moved))
	b := x.box(t)
	if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil {
		t.Fatal(err)
	}
	if len(b.steps) != 0 || !b.local || !b.replica {
		t.Fatalf("rolled back a committed import: %q", b.steps)
	}
	if got := mustGet(t, x.reg, x.rec.Host); got.ImportGate != "" {
		t.Fatalf("gate = %q, want it cleared", got.ImportGate)
	}
	if j, _ := x.reg.ImportJournal(x.rec.Host); j != nil {
		t.Fatalf("journal = %+v", j)
	}
}

// m-d: before commit, only the swap's gated row is the import's row. An
// ungated row for a swapped or placed import, hand-added or with its gate
// cleared by hand, is neither adopted nor converged.
func TestReconcileLeavesAnUngatedRowOfAnUncommittedImport(t *testing.T) {
	for _, state := range []string{seeds.StateSwapped, seeds.StatePlaced} {
		for _, provisioned := range []bool{false, true} {
			name := state + "/hand-added"
			if provisioned {
				name = state + "/gate cleared"
			}
			t.Run(name, func(t *testing.T) {
				s, sqid := shopSetup(t)
				rec := newImport(sqid)
				rec.Seed.State = state
				if err := CreateRecord(ctx, s, rec); err != nil {
					t.Fatal(err)
				}
				row := importRow(rec, "/data", nil)
				row.ImportGate = ""
				row.Provisioned = provisioned
				row.SeedSource = "{}" // stale: a converge would rewrite it
				reg := regWith(t, row)
				stored := mustGet(t, reg, rec.Host)
				if err := Reconcile(ctx, s, reg, importCfg(t, nil)); err != nil {
					t.Fatal(err)
				}
				if got := mustGet(t, reg, rec.Host); !reflect.DeepEqual(*got, *stored) {
					t.Fatalf("row changed:\n got %+v\nwant %+v", *got, *stored)
				}
			})
		}
	}
}

// The journal check inside swapImport stays as a backstop (Task 3's
// departure 1): called with this seed's journal row at swapped, it only
// redoes the record CAS, even with no journal pass ahead of it.
func TestSwapImportBackstopRedoesOnlyTheCAS(t *testing.T) {
	x := journaledImport(t, seeds.StateReady)
	swapImport(ctx, x.s, x.reg, x.cfg, x.rec, &x.row, func() (bool, error) { return true, nil }, nil)
	if st := seedState(t, x.s, x.rec); st.State != seeds.StateSwapped || st.Reason != "" {
		t.Fatalf("record = %s %q, want swapped", st.State, st.Reason)
	}
	x.untouched(t)
}

// convergeScene is a --from seed whose record is placed and whose row on
// this box still holds the manifest's seed command and no seed column:
// the converge must set the column and clear the command.
func convergeScene(t *testing.T) (*release.DirStore, *registry.Registry, string, Record) {
	t.Helper()
	s := dirStore(t)
	sqid := writeAccountManifest(t, s, acctID, "Shop")
	if err := PutManifest(ctx, s, sqid, "shop", Manifest{Schema: 1, App: "shop", Domain: "shop.example.com", Seed: []string{"seed"}}); err != nil {
		t.Fatal(err)
	}
	rec := shopRecord(sqid)
	id, _ := seeds.NewID()
	rec.Seed = &SeedSource{ID: id, Kind: seeds.KindReplica, State: seeds.StatePlaced, SHA256: strings.Repeat("a", 64), Bytes: 10}
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	row := RouteFor(rec, "/data", `["seed"]`, true, nil)
	row.SeedSource = ""
	reg, path := regAt(t, row)
	return s, reg, path, rec
}

func addTrigger(t *testing.T, path, ddl string) {
	t.Helper()
	db, err := sql.Open("sqlite", path)
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	if _, err := db.Exec(ddl); err != nil {
		t.Fatal(err)
	}
}

// M10: the converge sets the seed column before it clears the seed
// command, so the row is never left holding neither (which could boot
// empty). The trigger refuses that state; in the other order the seed
// write is refused and the command stays.
func TestConvergeSetsTheSeedSourceBeforeClearingTheSeed(t *testing.T) {
	s, reg, path, rec := convergeScene(t)
	addTrigger(t, path, `CREATE TRIGGER neither BEFORE UPDATE ON routes
WHEN NEW.seed = '' AND NEW.seed_source = ''
BEGIN SELECT RAISE(ABORT, 'a row with neither a seed command nor a seed source'); END`)
	if err := Reconcile(ctx, s, reg, importCfg(t, nil)); err != nil {
		t.Fatal(err)
	}
	if got := mustGet(t, reg, rec.Host); got.Seed != "" || got.SeedSource != SeedColumn(rec) {
		t.Fatalf("row = seed %q source %q, want no seed command and this seed's column", got.Seed, got.SeedSource)
	}
}

// M10: a seed column that cannot be written leaves the seed command in
// place, so the row keeps the state it had.
func TestConvergeKeepsTheSeedWhenTheSeedSourceFails(t *testing.T) {
	s, reg, path, rec := convergeScene(t)
	addTrigger(t, path, `CREATE TRIGGER nosource BEFORE UPDATE OF seed_source ON routes
BEGIN SELECT RAISE(ABORT, 'seed_source is read-only here'); END`)
	if err := Reconcile(ctx, s, reg, importCfg(t, nil)); err != nil {
		t.Fatal(err)
	}
	if got := mustGet(t, reg, rec.Host); got.Seed != `["seed"]` || got.SeedSource != "" {
		t.Fatalf("row = seed %q source %q, want it as it was", got.Seed, got.SeedSource)
	}
}

// Merge review S2 (b): on a box with claim control on, the journal pass
// does nothing but protect. A journal row there is an import a claim
// caught mid-flight; the box arms no swap or rollback and the claim
// console refuses commit, abort and finish, so no single step is safe to
// take. In every state the row, the journal row and the record are exactly
// as they were: the swap's record CAS is not redone, a committed import's
// gate is not cleared, and no rollback step runs, even with the rollback's
// hooks wired. The removal phase leaves the row too.
func TestJournalPassOnlyProtectsWithClaimControlOn(t *testing.T) {
	for _, c := range []struct {
		name    string
		record  string // the record's seed state; "" deletes the record
		journal string // the journal row's state
	}{
		{"ready: the record CAS would be redone", seeds.StateReady, registry.JournalSwapped},
		{"swapped", seeds.StateSwapped, registry.JournalSwapped},
		{"placed", seeds.StatePlaced, registry.JournalSwapped},
		{"committed: the gate would be cleared", seeds.StateCommitted, registry.JournalSwapped},
		{"aborted: a rollback would start", seeds.StateAborted, registry.JournalSwapped},
		{"record gone: a rollback would start", "", registry.JournalSwapped},
		{"rolling-back", seeds.StateAborted, registry.JournalRollingBack},
		{"rolling-back, record gone", "", registry.JournalRollingBack},
		{"cleanup-pending", seeds.StateAborted, registry.JournalCleanupPending},
		{"cleanup-pending, record gone", "", registry.JournalCleanupPending},
	} {
		t.Run(c.name, func(t *testing.T) {
			state := c.record
			if state == "" {
				state = seeds.StateSwapped
			}
			x := journaledImport(t, state)
			if c.journal != registry.JournalSwapped {
				if err := x.reg.CloseImportGate(x.rec.Host, x.rec.Seed.ID, 7); err != nil {
					t.Fatal(err)
				}
			}
			if c.journal == registry.JournalCleanupPending {
				if err := x.reg.RestoreImportRoute(x.rec.Host, x.rec.Seed.ID); err != nil {
					t.Fatal(err)
				}
			}
			x.row = *mustGet(t, x.reg, x.rec.Host)
			j, _ := x.reg.ImportJournal(x.rec.Host)
			x.j = *j
			key := RecordKey(x.rec.Account, x.rec.App, x.rec.Host)
			if c.record == "" {
				if err := x.s.Delete(ctx, key); err != nil {
					t.Fatal(err)
				}
			}
			before, _, _ := x.s.Get(ctx, key)
			b := x.box(t)
			x.cfg.ClaimEnabled = true
			logged := captureLog(t)
			for pass := 0; pass < 2; pass++ {
				if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil {
					t.Fatalf("pass %d: %v", pass, err)
				}
				x.untouched(t)
				if after, _, _ := x.s.Get(ctx, key); !bytes.Equal(after, before) {
					t.Fatalf("pass %d: the record changed:\n got %s\nwant %s", pass, after, before)
				}
				if len(b.steps) != 0 || len(b.retired) != 0 || !b.local || !b.replica || !b.key {
					t.Fatalf("pass %d: the box was touched: steps %q, retired %q, local %v, replica %v, key %v",
						pass, b.steps, b.retired, b.local, b.replica, b.key)
				}
			}
			if !strings.Contains(logged.String(), "claim control is on for this box; its row is protected and nothing else is done") {
				t.Fatalf("the protection was not logged:\n%s", logged)
			}
		})
	}
}

// The same journal rows with claim control off move forward as before, so
// the test above fails for the reason it says when the guard is removed.
func TestJournalPassMovesForwardWithClaimControlOff(t *testing.T) {
	ready := journaledImport(t, seeds.StateReady)
	if err := Reconcile(ctx, ready.s, ready.reg, ready.cfg); err != nil {
		t.Fatal(err)
	}
	if st := seedState(t, ready.s, ready.rec); st.State != seeds.StateSwapped {
		t.Fatalf("record = %s, want the swap's CAS redone", st.State)
	}
	committed := journaledImport(t, seeds.StateCommitted)
	if err := Reconcile(ctx, committed.s, committed.reg, committed.cfg); err != nil {
		t.Fatal(err)
	}
	if got := mustGet(t, committed.reg, committed.rec.Host); got.ImportGate != "" {
		t.Fatalf("gate = %q, want it cleared", got.ImportGate)
	}
}