package provision

import (
	"context"
	"errors"
	"reflect"
	"strings"
	"testing"
	"time"

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

// shopSetup is the import tests' neutral fixture: an account manifest
// (id acctID) and its app "shop" (domain shop.example.com, the shared
// pool, no seed command).
func shopSetup(t *testing.T) (*release.DirStore, string) {
	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"}); err != nil {
		t.Fatal(err)
	}
	return s, sqid
}

func shopRecord(sqid string) Record {
	return Record{Schema: 1, Host: "a.shop.example.com", Slug: "a", App: "shop",
		Account: sqid, Region: "ie", Channel: "stable", CreatedAt: time.Now()}
}

func newImport(sqid string) Record {
	rec := shopRecord(sqid)
	id, _ := seeds.NewID()
	rec.Seed = &SeedSource{ID: id, Kind: seeds.KindImport, State: seeds.StateReady, SHA256: strings.Repeat("a", 64), Bytes: 10,
		Replaces: "127.0.0.1:8401", VerifyHash: strings.Repeat("b", 64), By: "t", At: time.Now()}
	return rec
}

func importRecord(t *testing.T, s release.Store, sqid string) Record {
	t.Helper()
	rec := newImport(sqid)
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	return rec
}

// addrRow is the legacy row an import replaces: an addr route to the
// app's own process, not minted from any record.
func addrRow(rec Record, account string) registry.Route {
	return registry.Route{Host: rec.Host, Kind: "service", Addr: "127.0.0.1:8401", App: rec.App, Account: account,
		Region: "ie", Channel: "stable", Status: "ready", Unit: "shop@a.service"}
}

func importCfg(t *testing.T, owners map[string]seeds.LegacyOwner) ReconcileConfig {
	t.Helper()
	cfg := nestedSeedCfg(dirStore(t))
	cfg.Pools = nil
	cfg.Seeds.LegacyOwners = func(context.Context) (*seeds.LegacyOwners, error) {
		return &seeds.LegacyOwners{V: 1, Names: owners}, nil
	}
	// The host has no env file before the import, as a host the platform
	// never ran has none.
	cfg.Seeds.EnvBefore = func(string, string) (string, error) { return registry.EnvBeforeAbsent, nil }
	return cfg
}

func seedState(t *testing.T, s release.Store, rec Record) *SeedSource {
	t.Helper()
	r, err := GetRecord(ctx, s, rec.Account, rec.App, rec.Host)
	if err != nil {
		t.Fatal(err)
	}
	return r.Seed
}

// wantUntouched asserts a deferred swap: the record still ready with no
// reason, the stored row exactly row, and no journal.
func wantUntouched(t *testing.T, s release.Store, reg *registry.Registry, rec Record, row *registry.Route) {
	t.Helper()
	if got := seedState(t, s, rec); got.State != seeds.StateReady || got.Reason != "" {
		t.Fatalf("seed = %s %q, want ready (deferred, not refused)", got.State, got.Reason)
	}
	got, err := reg.Get(rec.Host)
	if err != nil {
		t.Fatal(err)
	}
	if row == nil {
		if got != nil {
			t.Fatalf("a row appeared: %+v", got)
		}
	} else if got == nil || !reflect.DeepEqual(*got, *row) {
		t.Fatalf("row = %+v, want it unchanged %+v", got, row)
	}
	if j, _ := reg.ImportJournal(rec.Host); j != nil {
		t.Fatalf("a deferred import left a journal row: %+v", j)
	}
}

func TestSwapReplacesTheAddrRowAndJournalsIt(t *testing.T) {
	s, sqid := shopSetup(t)
	rec := importRecord(t, s, sqid)
	reg := regWith(t, addrRow(rec, sqid))
	original := mustGet(t, reg, rec.Host)
	cfg := importCfg(t, nil)
	at := time.Unix(1_800_000_000, 0)
	cfg.Now = func() time.Time { return at }
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	got := mustGet(t, reg, rec.Host)
	if got.Kind != "instance" || got.Socket != NestedSocket(rec.Host) || got.Addr != "" || got.Addrs != "" ||
		got.ImportGate != rec.Seed.VerifyHash || got.DBPath != NestedDBPath("/data", rec.Host) || got.Account != sqid ||
		!got.Provisioned || got.Seed != "" || !got.Hibernate {
		t.Fatalf("swapped row = %+v", got)
	}
	if col, _ := seeds.ParseRowSeed(got.SeedSource); col == nil || col.ID != rec.Seed.ID || col.State != seeds.StateSwapped {
		t.Fatalf("swapped row's seed column = %+v, want this seed at swapped", col)
	}
	j, err := reg.ImportJournal(rec.Host)
	if err != nil || j == nil || j.SeedID != rec.Seed.ID || j.State != registry.JournalSwapped || !reflect.DeepEqual(j.Original, *original) ||
		j.RecordKey != RecordKey(rec.Account, rec.App, rec.Host) || j.At != at.Unix() {
		t.Fatalf("journal = %+v, %v", j, err)
	}
	if st := seedState(t, s, rec); st.State != seeds.StateSwapped {
		t.Fatalf("record = %s", st.State)
	}

	// The next pass converges the row as an ordinary owned row and leaves
	// the gate and the journal alone.
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	again := mustGet(t, reg, rec.Host)
	if again.ImportGate != rec.Seed.VerifyHash || again.Socket != got.Socket || again.DBPath != got.DBPath || again.Addr != "" {
		t.Fatalf("row after the next pass = %+v", again)
	}
	if j2, _ := reg.ImportJournal(rec.Host); j2 == nil || !reflect.DeepEqual(*j2, *j) {
		t.Fatalf("journal after the next pass = %+v", j2)
	}
}

// A row backed by an addrs set of exactly Replaces is the same address.
func TestSwapAcceptsAnAddrsRowOfExactlyReplaces(t *testing.T) {
	s, sqid := shopSetup(t)
	rec := importRecord(t, s, sqid)
	row := addrRow(rec, sqid)
	row.Addr, row.Addrs = "", "127.0.0.1:8401"
	reg := regWith(t, row)
	if err := Reconcile(ctx, s, reg, importCfg(t, nil)); err != nil {
		t.Fatal(err)
	}
	if got := mustGet(t, reg, rec.Host); got.ImportGate != rec.Seed.VerifyHash || got.Kind != "instance" {
		t.Fatalf("row = %+v", got)
	}
}

func TestSwapRefusesEachCondition(t *testing.T) {
	for _, c := range []struct {
		name   string
		row    func(Record, string) *registry.Route
		owners map[string]seeds.LegacyOwner
		cfg    func(*ReconcileConfig)
		want   string
	}{
		{"no row", func(Record, string) *registry.Route { return nil }, nil, nil, "there is no route at a.shop.example.com to replace"},
		{"addr changed", func(r Record, a string) *registry.Route { x := addrRow(r, a); x.Addr = "127.0.0.1:9"; return &x }, nil, nil,
			"the route's address is 127.0.0.1:9, not 127.0.0.1:8401"},
		{"addrs wider than replaces", func(r Record, a string) *registry.Route {
			x := addrRow(r, a)
			x.Addr, x.Addrs = "", "127.0.0.1:8401,127.0.0.1:8402"
			return &x
		}, nil, nil, "the route's address is 127.0.0.1:8401,127.0.0.1:8402, not 127.0.0.1:8401"},
		{"addrs behind a matching addr", func(r Record, a string) *registry.Route {
			x := addrRow(r, a)
			x.Addrs = "127.0.0.1:9"
			return &x
		}, nil, nil, "the route's address is 127.0.0.1:9, not 127.0.0.1:8401"},
		{"another app", func(r Record, a string) *registry.Route { x := addrRow(r, a); x.App = "other"; return &x }, nil, nil,
			"the route belongs to app other"},
		{"a third account", func(r Record, _ string) *registry.Route { x := addrRow(r, "zzz"); return &x }, nil, nil,
			"the route is filed under another account"},
		{"operator row, no assignment", func(r Record, _ string) *registry.Route { x := addrRow(r, release.OpsAccount); return &x }, nil, nil,
			"the route is filed under the operator account, and no assignment maps shop to this app"},
		{"legacy operator row, no assignment", func(r Record, _ string) *registry.Route { x := addrRow(r, ""); return &x }, nil, nil,
			"no assignment maps shop to this app"},
		{"operator row, assigned to another account", func(r Record, _ string) *registry.Route { x := addrRow(r, release.OpsAccount); return &x },
			map[string]seeds.LegacyOwner{"shop": {Account: "other", App: "shop"}}, nil, "no assignment maps shop to this app"},
		{"operator row, assigned to another app", func(r Record, _ string) *registry.Route { x := addrRow(r, release.OpsAccount); return &x },
			map[string]seeds.LegacyOwner{"shop": {Account: "SELF", App: "store"}}, nil, "no assignment maps shop to this app"},
		{"socket backed", func(r Record, a string) *registry.Route { x := addrRow(r, a); x.Socket = "/run/x.sock"; return &x }, nil, nil, "(socket)"},
		{"no address", func(r Record, a string) *registry.Route { x := addrRow(r, a); x.Addr = ""; return &x }, nil, nil, "(no address)"},
		{"provisioned", func(r Record, a string) *registry.Route { x := addrRow(r, a); x.Provisioned = true; return &x }, nil, nil, "(provisioned)"},
		{"static", func(r Record, a string) *registry.Route { x := addrRow(r, a); x.Kind = "static"; return &x }, nil, nil, "(kind static)"},
		{"backing", func(r Record, a string) *registry.Route {
			x := addrRow(r, a)
			x.Backing = registry.BackingUnit
			return &x
		}, nil, nil, "(backing)"},
		{"tunnel", func(r Record, a string) *registry.Route { x := addrRow(r, a); x.Tunnel = "t"; return &x }, nil, nil, "(tunnel)"},
		{"caps", func(r Record, a string) *registry.Route { x := addrRow(r, a); x.Caps = "host-role"; return &x }, nil, nil, "(caps)"},
		{"platform", func(r Record, a string) *registry.Route { x := addrRow(r, a); x.Platform = true; return &x }, nil, nil, "(platform)"},
		{"canary", func(r Record, a string) *registry.Route { x := addrRow(r, a); x.Canary = true; return &x }, nil, nil, "(canary)"},
		{"sidecar", func(r Record, a string) *registry.Route { x := addrRow(r, a); x.Kind = "sidecar"; return &x }, nil, nil, "(kind sidecar)"},
		// The edge serves these from somewhere other than the addr: a
		// fleet box over the reverse channel, or the activator's own child.
		{"fleet placed", func(r Record, a string) *registry.Route { x := addrRow(r, a); x.Region = "fleet/edge1"; return &x }, nil, nil, "(region)"},
		{"fleet box", func(r Record, a string) *registry.Route { x := addrRow(r, a); x.FleetBox = "box1"; return &x }, nil, nil, "(fleet_box)"},
		{"hibernating instance", func(r Record, a string) *registry.Route {
			x := addrRow(r, a)
			x.Kind, x.Hibernate = "instance", true
			return &x
		}, nil, nil, "(hibernate)"},
		{"claim owned", func(r Record, a string) *registry.Route {
			x := addrRow(r, a)
			x.ClaimOwnership = registry.ClaimOwnership{ClaimIncarnation: "inc", ClaimStamp: 1, ClaimOwner: "owner", ClaimRowID: "row"}
			return &x
		}, nil, nil, "(claim_owner)"},
		{"flat", func(r Record, a string) *registry.Route { x := addrRow(r, a); return &x }, nil, func(c *ReconcileConfig) { c.Layout = nil },
			"this box does not run instances nested"},
		{"flat intent", func(r Record, a string) *registry.Route { x := addrRow(r, a); return &x }, nil, func(c *ReconcileConfig) {
			c.Layout = func(context.Context, release.Store) (string, error) { return release.LayoutModeFlat, nil }
		}, "this box does not run instances nested"},
		{"claimed", func(r Record, a string) *registry.Route { x := addrRow(r, a); return &x }, nil, func(c *ReconcileConfig) { c.ClaimEnabled = true },
			"claim control is on for this box"},
	} {
		t.Run(c.name, func(t *testing.T) {
			s, sqid := shopSetup(t)
			rec := importRecord(t, s, sqid)
			var reg *registry.Registry
			var before *registry.Route
			if row := c.row(rec, sqid); row != nil {
				reg = regWith(t, *row)
				before = mustGet(t, reg, rec.Host)
			} else {
				reg = regWith(t)
			}
			owners := c.owners
			for k, v := range owners { // "SELF" is the record's own account
				if v.Account == "SELF" {
					owners[k] = seeds.LegacyOwner{Account: sqid, App: v.App}
				}
			}
			cfg := importCfg(t, owners)
			if c.cfg != nil {
				c.cfg(&cfg)
			}
			if err := Reconcile(ctx, s, reg, cfg); err != nil {
				t.Fatal(err)
			}
			if st := seedState(t, s, rec); st.State != seeds.StateRefused || !strings.Contains(st.Reason, c.want) {
				t.Fatalf("seed = %s %q, want refused with %q", st.State, st.Reason, c.want)
			}
			if j, _ := reg.ImportJournal(rec.Host); j != nil {
				t.Fatalf("a refused import left a journal row: %+v", j)
			}
			after, _ := reg.Get(rec.Host)
			if (before == nil) != (after == nil) || (before != nil && !reflect.DeepEqual(*before, *after)) {
				t.Fatalf("a refused import changed the row: %+v -> %+v", before, after)
			}
			// A refused import is skipped from then on.
			if err := Reconcile(ctx, s, reg, cfg); err != nil {
				t.Fatal(err)
			}
			if st := seedState(t, s, rec); st.State != seeds.StateRefused {
				t.Fatalf("seed after the next pass = %s", st.State)
			}
		})
	}
}

// M5: an operator-filed row ("bab", or a legacy row with no account) is
// accepted only through the assignment, and the new row is filed under
// the record's account.
func TestSwapAcceptsAnAssignedOperatorRow(t *testing.T) {
	for _, acct := range []string{release.OpsAccount, ""} {
		t.Run("account "+acct, func(t *testing.T) {
			s, sqid := shopSetup(t)
			rec := importRecord(t, s, sqid)
			reg := regWith(t, addrRow(rec, acct))
			if err := Reconcile(ctx, s, reg, importCfg(t, map[string]seeds.LegacyOwner{"shop": {Account: sqid, App: "shop"}})); err != nil {
				t.Fatal(err)
			}
			if got := mustGet(t, reg, rec.Host); got.Account != sqid || got.Kind != "instance" || got.ImportGate == "" {
				t.Fatalf("row = %+v; want it filed under the record's account, gated", got)
			}
			if st := seedState(t, s, rec); st.State != seeds.StateSwapped {
				t.Fatalf("record = %s", st.State)
			}
		})
	}
}

// The assignment is an authority check: an unreadable file or an unwired
// reader defers the swap, never refuses it and never accepts the row.
func TestSwapDefersWhenTheAssignmentCannotBeRead(t *testing.T) {
	for name, hook := range map[string]func(context.Context) (*seeds.LegacyOwners, error){
		"read error": func(context.Context) (*seeds.LegacyOwners, error) { return nil, errors.New("store down") },
		"unwired":    nil,
	} {
		t.Run(name, func(t *testing.T) {
			s, sqid := shopSetup(t)
			rec := importRecord(t, s, sqid)
			reg := regWith(t, addrRow(rec, release.OpsAccount))
			row := mustGet(t, reg, rec.Host)
			cfg := importCfg(t, nil)
			cfg.Seeds.LegacyOwners = hook
			if err := Reconcile(ctx, s, reg, cfg); err != nil {
				t.Fatal(err)
			}
			wantUntouched(t, s, reg, rec, row)
		})
	}
}

// M3: the swap is skipped, the record left as it is, unless the verify
// hash is a hash, the host and seed id pass seeds.Check, and the new row
// is nested and hibernating.
func TestSwapSkipsARecordItCannotGate(t *testing.T) {
	for _, c := range []struct {
		name string
		edit func(*Record)
	}{
		{"empty verify hash", func(r *Record) { r.Seed.VerifyHash = "" }},
		{"verify hash not a hash", func(r *Record) { r.Seed.VerifyHash = "closed" }},
		{"verify hash in capitals", func(r *Record) { r.Seed.VerifyHash = strings.Repeat("B", 64) }},
		{"bad seed id", func(r *Record) { r.Seed.ID = "sd_../../x" }},
		{"bad host", func(r *Record) { r.Host = "A.shop.example.com/.." }},
		{"does not hibernate", func(r *Record) { off := false; r.Hibernate = &off }},
		{"not an instance", func(r *Record) { r.Kind = "service" }},
		{"proxies to an address", func(r *Record) { r.Addr = "127.0.0.1:9000" }},
	} {
		t.Run(c.name, func(t *testing.T) {
			s, sqid := shopSetup(t)
			stored := importRecord(t, s, sqid)
			rec := stored
			seed := *stored.Seed
			rec.Seed = &seed
			c.edit(&rec)
			reg := regWith(t, addrRow(stored, sqid))
			row := mustGet(t, reg, stored.Host)
			cfg := importCfg(t, nil)
			got, _ := reg.Get(stored.Host)
			swapImport(ctx, s, reg, cfg, rec, got, mintLayoutReader(ctx, s, cfg), nil)
			wantUntouched(t, s, reg, stored, row)
		})
	}
}

// Each skip is importUngateable's own (not a later guard's), and a sound
// import record passes it.
func TestImportUngateable(t *testing.T) {
	good := newImport("abc")
	if why := importUngateable(good, importRow(good, "/data", nil), "/data"); why != "" {
		t.Fatalf("a sound record: %q", why)
	}
	for name, edit := range map[string]func(*Record){
		"empty verify hash": func(r *Record) { r.Seed.VerifyHash = "" },
		"closed gate":       func(r *Record) { r.Seed.VerifyHash = "closed" },
		"bad seed id":       func(r *Record) { r.Seed.ID = "sd_../../x" },
		"bad host":          func(r *Record) { r.Host = "A.shop.example.com/.." },
		"does not hibernate": func(r *Record) {
			off := false
			r.Hibernate = &off
		},
		"not an instance": func(r *Record) { r.Kind = "service" },
	} {
		rec := newImport("abc")
		seed := *rec.Seed
		rec.Seed = &seed
		edit(&rec)
		if why := importUngateable(rec, importRow(rec, "/data", nil), "/data"); why == "" {
			t.Errorf("%s: not skipped", name)
		}
	}
}

// The same rule through Reconcile, for a record shape CreateRecord stores.
func TestReconcileDoesNotSwapAnImportThatWouldNotHibernate(t *testing.T) {
	s, sqid := shopSetup(t)
	rec := newImport(sqid)
	off := false
	rec.Hibernate = &off
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t, addrRow(rec, sqid))
	row := mustGet(t, reg, rec.Host)
	if err := Reconcile(ctx, s, reg, importCfg(t, nil)); err != nil {
		t.Fatal(err)
	}
	wantUntouched(t, s, reg, rec, row)
}

// M4: a layout read error or a NestedOK error defers to the next pass and
// writes nothing; only a layout intent that says flat refuses (above).
func TestSwapDefersATransientLayoutAnswer(t *testing.T) {
	for _, c := range []struct {
		name string
		cfg  func(*ReconcileConfig)
	}{
		{"layout read error", func(c *ReconcileConfig) {
			c.Layout = func(context.Context, release.Store) (string, error) { return "", errors.New("store down") }
		}},
		{"nested probe error", func(c *ReconcileConfig) {
			c.Seeds.NestedOK = func() error { return errors.New("ready probe timed out") }
		}},
	} {
		t.Run(c.name, func(t *testing.T) {
			s, sqid := shopSetup(t)
			rec := importRecord(t, s, sqid)
			reg := regWith(t, addrRow(rec, sqid))
			row := mustGet(t, reg, rec.Host)
			cfg := importCfg(t, nil)
			c.cfg(&cfg)
			if err := Reconcile(ctx, s, reg, cfg); err != nil {
				t.Fatal(err)
			}
			wantUntouched(t, s, reg, rec, row)

			// The next pass, with the box answering, swaps.
			if err := Reconcile(ctx, s, reg, importCfg(t, nil)); err != nil {
				t.Fatal(err)
			}
			if st := seedState(t, s, rec); st.State != seeds.StateSwapped {
				t.Fatalf("record after the box answered = %s %q", st.State, st.Reason)
			}
		})
	}
}

func TestSwapRunsWhenTheNestedProbeAnswers(t *testing.T) {
	s, sqid := shopSetup(t)
	rec := importRecord(t, s, sqid)
	reg := regWith(t, addrRow(rec, sqid))
	cfg := importCfg(t, nil)
	asked := 0
	cfg.Seeds.NestedOK = func() error { asked++; return nil }
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if asked != 1 || mustGet(t, reg, rec.Host).ImportGate == "" {
		t.Fatalf("probe asked %d times, row %+v", asked, mustGet(t, reg, rec.Host))
	}
}

// Transient refusals of the write itself defer: the config gate, a row
// held by a running operation, and a dry run.
func TestSwapDefersWhatTheWriteCannotTakeYet(t *testing.T) {
	for _, c := range []struct {
		name string
		held bool
		cfg  func(*ReconcileConfig)
	}{
		{"config gate", false, func(c *ReconcileConfig) {
			c.ConfigGate = func(context.Context, registry.Route) error { return errors.New("no key yet") }
		}},
		{"held", true, nil},
		{"dry run", false, func(c *ReconcileConfig) { c.DryRun = true }},
	} {
		t.Run(c.name, func(t *testing.T) {
			s, sqid := shopSetup(t)
			rec := importRecord(t, s, sqid)
			r := addrRow(rec, sqid)
			if c.held {
				// A wrong address too: without the hold's deferral this row
				// would be refused SD-N92, not left for the next pass.
				r.Addr = "127.0.0.1:9"
			}
			reg := regWith(t, r)
			if c.held {
				if err := reg.SetHold(rec.Host, "moving", "op_1"); err != nil {
					t.Fatal(err)
				}
			}
			row := mustGet(t, reg, rec.Host)
			if c.held && row.HoldOp == "" {
				t.Fatalf("fixture: the row is not held: %+v", row)
			}
			cfg := importCfg(t, nil)
			if c.cfg != nil {
				c.cfg(&cfg)
			}
			if err := Reconcile(ctx, s, reg, cfg); err != nil {
				t.Fatal(err)
			}
			wantUntouched(t, s, reg, rec, row)
		})
	}
}

// A crash between the swap's transaction and the record's CAS: the
// journal names this seed and the record still says ready. The next pass
// sees the journal first, never judges the row it made as a legacy row,
// and only redoes the CAS.
func TestSwapResumesTheRecordCASAfterACrash(t *testing.T) {
	s, sqid := shopSetup(t)
	rec := importRecord(t, s, sqid)
	reg := regWith(t, addrRow(rec, sqid))
	original := mustGet(t, reg, rec.Host)
	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: *original,
		RecordKey: RecordKey(rec.Account, rec.App, rec.Host), At: 1}, next); err != nil {
		t.Fatal(err)
	}
	swapped := mustGet(t, reg, rec.Host)
	j, _ := reg.ImportJournal(rec.Host)

	if err := Reconcile(ctx, s, reg, importCfg(t, nil)); err != nil {
		t.Fatal(err)
	}
	if st := seedState(t, s, rec); st.State != seeds.StateSwapped || st.Reason != "" {
		t.Fatalf("record = %s %q, want swapped", st.State, st.Reason)
	}
	if got := mustGet(t, reg, rec.Host); !reflect.DeepEqual(*got, *swapped) {
		t.Fatalf("row changed on the resume: %+v -> %+v", swapped, got)
	}
	if j2, _ := reg.ImportJournal(rec.Host); j2 == nil || !reflect.DeepEqual(*j2, *j) {
		t.Fatalf("journal changed on the resume: %+v -> %+v", j, j2)
	}
}

// A journal row naming this seed ID under another record key is not this
// record's swap: no CAS for this record. The journal's own record (under
// the other key) is gone, so the journal rolls its import back: the
// original route returns exactly, and this record, which holds the host,
// keeps the host's replica and its own state.
func TestSwapResumeNeedsTheRecordKey(t *testing.T) {
	s, sqid := shopSetup(t)
	rec := importRecord(t, s, sqid)
	reg := regWith(t, addrRow(rec, sqid))
	original := mustGet(t, reg, rec.Host)
	next := importRow(rec, "/data", nil)
	if err := reg.SwapForImport(registry.ImportJournal{Host: rec.Host, SeedID: rec.Seed.ID, Original: *original,
		RecordKey: RecordKey(sqid, "other", rec.Host), At: 1}, next); 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 != seeds.StateReady || st.Reason != "" {
		t.Fatalf("record = %s %q, want ready", st.State, st.Reason)
	}
	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)
	}
}

// A journal row this record's seed does not own (another seed's, or one
// already rolling back) leaves the record alone: no refusal, no swap. This
// is the swap's own backstop (seedswap.go), called directly: in a pass the
// journal pass decides such a row first, and rolls it back.
func TestSwapLeavesAnotherSeedsJournalAlone(t *testing.T) {
	s, sqid := shopSetup(t)
	rec := importRecord(t, s, sqid)
	reg := regWith(t, addrRow(rec, sqid))
	original := mustGet(t, reg, rec.Host)
	other := rec
	seed := *rec.Seed
	seed.ID, _ = seeds.NewID()
	other.Seed = &seed
	if err := reg.SwapForImport(registry.ImportJournal{Host: rec.Host, SeedID: seed.ID, Original: *original,
		RecordKey: RecordKey(rec.Account, rec.App, rec.Host), At: 1}, importRow(other, "/data", nil)); err != nil {
		t.Fatal(err)
	}
	row := mustGet(t, reg, rec.Host)
	j, _ := reg.ImportJournal(rec.Host)
	swapImport(ctx, s, reg, importCfg(t, nil), rec, row, func() (bool, error) { return true, nil }, nil)
	if st := seedState(t, s, rec); st.State != seeds.StateReady || st.Reason != "" {
		t.Fatalf("record = %s %q, want ready", st.State, st.Reason)
	}
	if got := mustGet(t, reg, rec.Host); !reflect.DeepEqual(*got, *row) {
		t.Fatalf("row changed: %+v", got)
	}
	if j2, _ := reg.ImportJournal(rec.Host); !reflect.DeepEqual(j2, j) {
		t.Fatalf("journal changed: %+v", j2)
	}
}

// M2: before committed, an import's row is made only by the swap. A box
// with no row never mints one for a swapped or placed import, and neither
// does the swapping box after its gated row was removed.
func TestReconcileNeverMintsAnUncommittedImport(t *testing.T) {
	for _, state := range []string{seeds.StateSwapped, seeds.StatePlaced} {
		t.Run("no row, "+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)
			if err := Reconcile(ctx, s, reg, importCfg(t, nil)); err != nil {
				t.Fatal(err)
			}
			if rt, _ := reg.Get(rec.Host); rt != nil {
				t.Fatalf("minted %+v", rt)
			}
		})
	}
	t.Run("after the gated row was removed", func(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)
		}
		if st := seedState(t, s, rec); st.State != seeds.StateSwapped {
			t.Fatalf("record = %s", st.State)
		}
		if err := reg.Delete(rec.Host); err != nil { // carlos remove
			t.Fatal(err)
		}
		for _, mut := range []func(*Record) error{nil, SeedPlace(rec.Seed.ID)} {
			if mut != nil {
				if _, err := SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, mut); err != nil {
					t.Fatal(err)
				}
			}
			if err := Reconcile(ctx, s, reg, cfg); err != nil {
				t.Fatal(err)
			}
			if rt, _ := reg.Get(rec.Host); rt != nil {
				t.Fatalf("%s: minted %+v", seedState(t, s, rec).State, rt)
			}
		}
	})
}

// After commit the import is an ordinary instance: a box with no row
// mints it (ungated), as any other record.
func TestReconcileMintsACommittedImport(t *testing.T) {
	s, sqid := shopSetup(t)
	rec := newImport(sqid)
	rec.Seed.State = seeds.StateCommitted
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	if err := Reconcile(ctx, s, reg, importCfg(t, nil)); err != nil {
		t.Fatal(err)
	}
	if got := mustGet(t, reg, rec.Host); got.ImportGate != "" || got.DBPath != NestedDBPath("/data", rec.Host) {
		t.Fatalf("row = %+v", got)
	}
}

func TestImportRefusalOrder(t *testing.T) {
	rec := newImport("abc")
	row := addrRow(rec, "zzz")
	row.App = "other"
	cfg := ReconcileConfig{ClaimEnabled: true}
	if got := importRefusal(rec, &row, cfg, nil, false); got != "claim control is on for this box" {
		t.Fatalf("claims first: %q", got)
	}
	if got := importRefusal(rec, &row, ReconcileConfig{}, nil, false); got != "this box does not run instances nested" {
		t.Fatalf("then layout: %q", got)
	}
	if got := importRefusal(rec, nil, ReconcileConfig{}, nil, true); got != "there is no route at a.shop.example.com to replace" {
		t.Fatalf("then the row: %q", got)
	}
	if got := importRefusal(rec, &row, ReconcileConfig{}, nil, true); got != "the route belongs to app other" {
		t.Fatalf("app before account: %q", got)
	}
	ok := addrRow(rec, "abc")
	if got := importRefusal(rec, &ok, ReconcileConfig{}, nil, true); got != "" {
		t.Fatalf("the record's own row: %q", got)
	}
}