package host

import (
	"context"
	"encoding/json"
	"errors"
	"io/fs"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"testing"

	"github.com/carlosframework/platform/internal/appconfig"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/moves/archive"
	"github.com/carlosframework/platform/internal/park"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// The shared installer on the source (Plan 4 Task 3, rulings J13, J14, J20,
// J24): a dirty return installs the generation the target sent back, from
// the return lane in the source's own bucket, through the same six phases
// the landing runs. The fixture is the landing harness driven to the flip,
// then taken by provision's and hostclaim's own transitions to the point
// Task 5's source recovery hands the install over: the record rolling_back,
// dirty, the source at parking; the claim restoring/claimed with the source
// executing at its pre-move stamp S; the source's installation record
// active at S; and the source's own pre-move files still at the flat paths.
// The return lane holds the staged seal's objects "as given" (J22), copied
// by version into moves/<host>/<rb id>/ of the source's bucket.

const p1RB = "rb_abcdefghijklmnopqrstuvwxyz"

type returnHarness struct {
	*landingHarness
	sdir      string // the source's data directory
	rcfg      moveCfg
	rw        *movesWorker
	seal      moves.Seal // the return seal
	baseline  provision.MoveSource
	srcRoute  registry.Route
	asked     []string // the releases ReleaseReady was asked about
	render    appconfig.HeldRender
	sealOf    func(*hostclaim.Claim) (moves.Seal, error)
	installed bool // stop the drive at restored
}

// newReturnHarness builds the fixture. registered lists the source's four
// flat paths in the claim's inventory at S and queues its park-local, as
// Task 5's claimed does when the inventory lists them (J13).
func newReturnHarness(t *testing.T, registered bool) *returnHarness {
	t.Helper()
	return newReturnHarnessWith(t, registered, true)
}

// returnJournal is the return's own op journal on the source.
func (h *returnHarness) returnJournal(t *testing.T) *moves.Journal {
	t.Helper()
	j, err := moves.ReadJournal(h.sdir, p1Host, p1RB)
	if err != nil {
		t.Fatal(err)
	}
	return j
}

// newReturnHarnessWith is newReturnHarness with the target's admission
// recorded or not (fix round 3): admitted, the claim holds the target's
// probe admission of the move's prepared generation, the evidence a dirty
// return needs before it parks the pre-move generations.
func newReturnHarnessWith(t *testing.T, registered, admitted bool) *returnHarness {
	t.Helper()
	ctx := context.Background()
	h := &returnHarness{landingHarness: newLandingHarness(t)}
	h.flip(t)
	if admitted {
		admitTarget(t, h.landingHarness)
	}
	h.sdir = h.cfg.DataDir
	rec := h.read(t, p1Host)
	h.baseline = *rec.Move.Source
	g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}

	// The return lane, as given: every sealed object copied by version into
	// moves/<host>/<rb id>/ of the source's bucket.
	staged := h.landingHarness.seal(t)
	lane, from := moves.StagingPrefix(p1Host, p1RB), moves.StagingPrefix(p1Host, p1ID)
	copyClass := func(objs []moves.SealedObject) []moves.SealedObject {
		var out []moves.SealedObject
		for _, o := range objs {
			key := lane + strings.TrimPrefix(o.Key, from)
			ver, _, err := h.vs.CopyVersion(ctx, "data-uk", o.Key, o.VersionID, "data-ie", key, true)
			if err != nil {
				t.Fatalf("copy %s into the return lane: %v", o.Key, err)
			}
			out = append(out, moves.SealedObject{Key: key, VersionID: ver, Size: o.Size})
		}
		return out
	}
	h.seal = staged
	h.seal.Replica, h.seal.Files, h.seal.Timers = copyClass(staged.Replica), copyClass(staged.Files), copyClass(staged.Timers)
	if err := h.seal.Validate(); err != nil {
		t.Fatal(err)
	}
	h.sealOf = func(*hostclaim.Claim) (moves.Seal, error) { return h.seal, nil }

	// The record: the rollback, the target released dirty and returned, the
	// source claimed and at parking.
	now := h.clock.Now()
	for _, step := range []func(*provision.Record) error{
		provision.MoveRollBack(g, p1RB, provision.RollbackLandingDeadline, now),
		provision.MoveRollbackTargetPhase(g, provision.RbTargetReleasing, provision.RbTargetReleased, func(rb *provision.MoveRollback) {
			rb.Dirty, rb.Evidence = true, "moves/bab/app/"+p1Host+"/"+p1ID+"/stopped-target.json"
		}),
		provision.MoveRollbackTargetPhase(g, provision.RbTargetReleased, provision.RbTargetReturning, nil),
		provision.MoveRollbackTargetPhase(g, provision.RbTargetReturning, provision.RbTargetReturned, nil),
		provision.MoveRollbackSourcePhase(g, provision.RbSourceWaiting, provision.RbSourceClaimed, nil),
		provision.MoveRollbackSourcePhase(g, provision.RbSourceClaimed, provision.RbSourceRestoring, nil),
		provision.MoveRollbackSourcePhase(g, provision.RbSourceRestoring, provision.RbSourceParking, nil),
	} {
		if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, step); err != nil {
			t.Fatal(err)
		}
	}
	// The claim: the rollback, the target's release, the source's return.
	for _, step := range []func(*hostclaim.Claim) error{
		hostclaim.RollBack(p1ID, p1RB),
		hostclaim.TargetReleased(p1ID, "uk-box", "uk", "stopped-target", hostclaim.ParkScope{Location: cutRoot, Incarnation: h.inc, EvidenceAt: now, RequestLimit: landingRequestLimit}, obligationID),
		hostclaim.ReturnToSource(p1ID, "ie-box", false, hostclaim.ReturnFence{}),
	} {
		obs, err := hostclaim.Read(ctx, h.s, p1Host)
		if err != nil {
			t.Fatal(err)
		}
		if _, err := hostclaim.Update(ctx, h.s, obs, step); err != nil {
			t.Fatal(err)
		}
	}
	if c := h.claim(t); c.Op.Kind != hostclaim.KindRestoring || c.Op.Phase != hostclaim.PhaseClaimed || c.Op.Exec != "ie-box" || c.Holder.Stamp != 1 {
		t.Fatalf("fixture claim: %+v holder %+v", c.Op, c.Holder)
	}
	// The source's installation record, active at S (Task 5's claimed).
	for _, d := range []string{"hostguard", ".parked"} {
		if err := os.MkdirAll(filepath.Join(h.sdir, d), 0o770); err != nil {
			t.Fatal(err)
		}
	}
	guard := hostguard.Guard{DataDir: h.sdir, LockDir: t.TempDir(), Claims: hostguard.StoreReader{Store: h.s}}
	if err := guard.Activate(ctx, p1Host, "ie-box", nil); err != nil {
		t.Fatal(err)
	}
	// The source's pre-move installation beside the files and env the cut
	// left: the database, its marker naming the source's generation, a WAL
	// and the litestream shadow.
	db := provision.DefaultDBPath(h.sdir, p1Host)
	for path, body := range map[string]string{db + ".gen": "cur\n", db + "-wal": "a stale wal", filepath.Join(h.sdir, ".activator-"+p1Host+".db-litestream", "generation"): "shadow"} {
		if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
			t.Fatal(err)
		}
		if err := os.WriteFile(path, []byte(body), 0o600); err != nil {
			t.Fatal(err)
		}
	}
	if registered {
		rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
			c.Inventory = map[string]hostclaim.InventoryEntry{"ie-box": {Layout: "flat", Paths: map[string]uint64{p1Host + ".db": 1, p1Host + ".db.gen": 1, p1Host: 1, p1Host + ".env": 1}}}
		})
		obs, err := hostclaim.Read(ctx, h.s, p1Host)
		if err != nil {
			t.Fatal(err)
		}
		if _, err := hostclaim.Update(ctx, h.s, obs, hostclaim.QueueLocalPark(p1ID, "ie-box", obligationID)); err != nil {
			t.Fatal(err)
		}
	}
	h.srcRoute = registry.Route{Host: p1Host, Kind: "instance", App: "app", Account: "bab", Hibernate: true, Status: "restoring", HoldOp: p1RB,
		Socket: provision.DefaultSocket(p1Host), DBPath: db, Provisioned: true}
	// The render answers the current bundle, which is not the pinned one: a
	// recovery records what it rendered and never compares it.
	h.render = appconfig.HeldRender{Path: filepath.Join(h.sdir, p1Host+".json"), SHA256: "y", BundleETags: map[string]string{"default": "e9", "environment": ""}, OverlayETag: "o9"}
	h.rcfg = h.sourceCfg(t)
	return h
}

// sourceCfg is the source box's configuration with the installer armed:
// the landing's hooks over the source's data directory, bucket and region.
func (h *returnHarness) sourceCfg(t *testing.T) moveCfg {
	t.Helper()
	c := h.landingCfg(t)
	c.Region, c.Box, c.OwnBucket, c.OwnBucketRegion = "ie", "ie-box", "data-ie", "eu-west-1"
	c.DataDir = h.sdir
	c.OptDir = h.optDir
	c.OwnsRoute = func(host string) bool { return host == p1Host }
	c.Route = func(host string) (*registry.Route, error) {
		if host != p1Host {
			return nil, nil
		}
		rt := h.srcRoute
		return &rt, nil
	}
	c.FilesDir = func(host string) (string, string) {
		return filepath.Join(h.sdir, host), filepath.Join(h.sdir, host+".env")
	}
	c.Guard = hostguard.Guard{DataDir: h.sdir, LockDir: t.TempDir()}
	c.Local = park.Local{DataDir: h.sdir, Box: "ie-box"}
	c.ParkView = recordView(h.s, "ie", roleSource)
	c.RenderHeld = func(_ context.Context, _ release.Store, rt registry.Route, op string, _ *provision.MoveInputs, before func(string) error) (appconfig.HeldRender, error) {
		h.renders++
		if h.renderHook != nil {
			return h.renderHook(before)
		}
		if err := before(h.render.SHA256); err != nil {
			return appconfig.HeldRender{}, err
		}
		if op != p1RB || rt.HoldOp != p1RB {
			t.Errorf("rendered for %q over a row held by %q, want the rollback %s", op, rt.HoldOp, p1RB)
		}
		return h.render, nil
	}
	c.ReleaseReady = func(_ context.Context, _ release.Store, _ provision.Record, v string) moves.Check {
		h.asked = append(h.asked, v)
		return h.release
	}
	c.RemoveVerifyEntry = func(context.Context, release.Store, string) error { return nil }
	c.WriteGen = hostbox.WriteGenerationMarker
	return c
}

// install drives the installer the way Task 5's pass will: one phase per
// call, the record re-read between them, until it stops advancing. It
// answers the last call's recorded refusal and error.
func (h *returnHarness) install(t *testing.T) (string, error) {
	t.Helper()
	ctx := context.Background()
	if h.rw == nil {
		h.rw = &movesWorker{mirrors: map[string]lease.MirrorStore{}}
	}
	h.rw.cfg = h.rcfg
	defer releaseOwnLease(h.rw, p1Host)
	for i := 0; i < 8; i++ {
		rec := h.read(t, p1Host)
		src, err := recoveryInstallSource(h.rw, h.s, rec, h.sealOf, &h.baseline)
		if err != nil {
			return "", err
		}
		advanced, refused, err := installGeneration(ctx, h.s, h.rw, rec, src)
		if err != nil || refused != "" || !advanced {
			return refused, err
		}
	}
	return "", nil
}

func (h *returnHarness) rollback(t *testing.T) (*provision.MoveState, *provision.MoveRollback) {
	t.Helper()
	m := h.read(t, p1Host).Move
	if m == nil || m.Rollback == nil {
		t.Fatalf("record: %+v", m)
	}
	return m, m.Rollback
}

func (h *returnHarness) journal(t *testing.T) *moves.Journal {
	t.Helper()
	j, err := moves.ReadJournal(h.sdir, p1Host, p1RB)
	if err != nil {
		t.Fatal(err)
	}
	return j
}

// parkedFiles maps every regular file under the source's .parked/ to its body.
func (h *returnHarness) parkedFiles(t *testing.T) map[string]string {
	t.Helper()
	out := map[string]string{}
	root := filepath.Join(h.sdir, ".parked")
	err := filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error {
		if err != nil || d.IsDir() {
			return err
		}
		body, err := os.ReadFile(p)
		if err != nil {
			return err
		}
		rel, _ := filepath.Rel(root, p)
		out[rel] = string(body)
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
	return out
}

func parkedHas(files map[string]string, suffix, body string) bool {
	for p, b := range files {
		if strings.HasSuffix(p, suffix) && b == body {
			return true
		}
	}
	return false
}

// installedReturn asserts the end state of a dirty return's install through
// restored, on the source.
func (h *returnHarness) installedReturn(t *testing.T, pinned *provision.MoveInputs) moves.InstallManifest {
	t.Helper()
	m, rb := h.rollback(t)
	if m.State != provision.MoveRollingBack || rb.SourcePhase != provision.RbSourceRestored || m.StuckReason != "" || !rb.Dirty {
		t.Fatalf("record: %+v %+v", m, rb)
	}
	// configured rendered the current inputs and recorded them, never
	// comparing them with the pin, which is untouched.
	want := provision.MoveInputs{BundleETags: map[string]string{"default": "e9", "environment": ""}, OverlayETag: "o9", Release: "v7"}
	if in := rb.RecoveryInputs; in == nil || in.Release != want.Release || in.OverlayETag != want.OverlayETag || !sameBundle(in.BundleETags, want.BundleETags) {
		t.Fatalf("recovery_inputs: %+v, want %+v", rb.RecoveryInputs, want)
	}
	if got := m.Inputs; got == nil || pinned == nil || got.Release != pinned.Release || !sameBundle(got.BundleETags, pinned.BundleETags) || got.OverlayETag != pinned.OverlayETag {
		t.Fatalf("move.inputs changed: %+v, was %+v", got, pinned)
	}
	if len(h.asked) == 0 {
		t.Fatal("the release gate was never asked")
	}
	for _, v := range h.asked {
		if v != "v7" {
			t.Fatalf("release gate asked about %v, want the live v7", h.asked)
		}
	}
	// The claim does not follow the source's install phases: it stays at
	// restoring/claimed with the source executing at S.
	c := h.claim(t)
	if c.Op.Kind != hostclaim.KindRestoring || c.Op.Phase != hostclaim.PhaseClaimed || c.Op.Exec != "ie-box" || c.Holder.Stamp != 1 {
		t.Fatalf("claim: %+v holder %+v", c.Op, c.Holder)
	}
	// The return generation, g-<rb id>, in the source's own bucket: the
	// build's four objects.
	prefix := moves.GenerationPrefix(cutRoot, p1RB)
	var keys []string
	for k := range h.f.buckets["data-ie"] {
		if strings.HasPrefix(k, prefix) {
			keys = append(keys, strings.TrimPrefix(k, prefix))
		}
	}
	if len(keys) != 4 {
		t.Fatalf("return generation objects: %v", keys)
	}
	var im moves.InstallManifest
	if err := json.Unmarshal([]byte(h.journal(t).Steps[landStepInstall]), &im); err != nil || im.Op != p1RB || im.N != h.seal.N {
		t.Fatalf("install manifest in the return's journal: %v %+v", err, im)
	}
	dbPath := provision.DefaultDBPath(h.sdir, p1Host)
	// The pre-move WAL and litestream shadow are gone from the live paths
	// (checked before anything here opens the database, which makes its own).
	for _, p := range []string{dbPath + "-wal", filepath.Join(h.sdir, ".activator-"+p1Host+".db-litestream")} {
		if _, err := os.Lstat(p); !errors.Is(err, os.ErrNotExist) {
			t.Fatalf("%s is still at the live path: %v", p, err)
		}
	}
	if got, err := fileSHA256(dbPath); err != nil || got != "sha256:"+im.DBSHA256 {
		t.Fatalf("the installed database: %v %s want %s", err, got, im.DBSHA256)
	}
	if got := countRowsForTest(t, dbPath); got != h.rows {
		t.Fatalf("the installed database holds %d rows, want %d", got, h.rows)
	}
	if gen, err := hostbox.ReadGenerationMarker(dbPath + ".gen"); err != nil || gen != p1RB {
		t.Fatalf("%s.gen: %q %v, want the return generation %s", dbPath, gen, err, p1RB)
	}
	body, err := os.ReadFile(filepath.Join(h.sdir, p1Host, archive.MarkerName))
	var fg filesGenMarker
	if err != nil || json.Unmarshal(body, &fg) != nil || fg.Op != p1RB || fg.Incarnation != h.inc {
		t.Fatalf(".files-gen: %s %v", body, err)
	}
	if body, err := os.ReadFile(filepath.Join(h.sdir, p1Host, "upload.txt")); err != nil || string(body) != "hello" {
		t.Fatalf("the extracted files: %q %v", body, err)
	}
	// The new installation is registered at S, the source's own stamp.
	inv := c.Inventory["ie-box"]
	if inv.Layout != "flat" || len(inv.Paths) != 4 {
		t.Fatalf("inventory: %+v", inv)
	}
	for _, p := range []string{p1Host + ".db", p1Host + ".db.gen", p1Host, p1Host + ".env"} {
		if inv.Paths[p] != 1 {
			t.Fatalf("inventory path %s at %d, want S 1", p, inv.Paths[p])
		}
	}
	// The pre-move generations at the location are residue parked in the
	// source's region, under the move's op.
	var gens []string
	for _, o := range c.Obligations {
		if o.Op == p1ID && o.Kind == "park-generation" && o.Region == "ie" && o.Park != nil && o.Park.Location == cutRoot {
			gens = append(gens, o.Generation)
		}
		if o.Kind == "park-local" {
			t.Fatalf("a park-local is still owed: %+v", o)
		}
	}
	sort.Strings(gens)
	if strings.Join(gens, " ") != "cur old" {
		t.Fatalf("residue chores: %v (%+v)", gens, c.Obligations)
	}
	for k := range h.f.buckets["data-ie"] {
		if strings.HasPrefix(k, cutRoot+"/g-cur/") || strings.HasPrefix(k, cutRoot+"/g-old/") {
			t.Fatalf("%s is still at the source's location", k)
		}
	}
	// Timers and the columns come from the return's baseline.
	if len(h.replaced) != 1 || moves.TimersDigest(h.replaced[0]) != h.baseline.TimersSHA256 {
		t.Fatalf("ReplaceTimers: %+v", h.replaced)
	}
	if len(h.repl) != 1 || h.repl[0] != h.baseline.Replication || len(h.nextDue) != 1 || h.nextDue[0] != h.baseline.NextDue {
		t.Fatalf("columns: %v %v", h.repl, h.nextDue)
	}
	if held := h.leases.holding(); len(held) != 0 {
		t.Fatalf("the install still holds %v", held)
	}
	return im
}

// A registered source (J13): RetireArtifacts(S) parks the inventoried
// pre-move paths, the moves-owned rename parks what the inventory does not
// list (the WAL, the litestream shadow), and only then is the return's
// installation extracted and registered at S.
func TestDirtyReturnInstallsUnderTheSourceStamp(t *testing.T) {
	t.Parallel()
	h := newReturnHarness(t, true)
	pinned := h.read(t, p1Host).Move.Inputs
	if refused, err := h.install(t); err != nil || refused != "" {
		t.Fatalf("install: %q %v", refused, err)
	}
	h.installedReturn(t, pinned)
	parked := h.parkedFiles(t)
	for suffix, body := range map[string]string{
		"/artifacts/" + p1Host + ".db":                                  "the stopped database",
		"/artifacts/" + p1Host + ".db.gen":                              "cur\n",
		"/artifacts/" + p1Host + "/upload.txt":                          "hello",
		"/artifacts/" + p1Host + ".db-wal":                              "a stale wal",
		"/artifacts/.activator-" + p1Host + ".db-litestream/generation": "shadow",
	} {
		if !parkedHas(parked, suffix, body) {
			t.Fatalf("%s was not parked (%v)", suffix, parked)
		}
	}
	// The registered half was park.Local's: its journal is in the
	// retirement. The rest was the moves-owned rename, recorded by the op
	// that ran it, the return.
	var receipts int
	for p, b := range parked {
		if strings.HasSuffix(p, "/"+localParkReceiptName) {
			receipts++
			var r localParkReceipt
			if err := json.Unmarshal([]byte(b), &r); err != nil || r.Op != p1RB {
				t.Fatalf("receipt %s: %s %v", p, b, err)
			}
		}
	}
	if receipts != 1 {
		t.Fatalf("%d moves-owned receipts", receipts)
	}
	if !h.journal(t).Done(installStepPreMove) || !h.journal(t).Done(landStepRetired) {
		t.Fatalf("journal: %v", h.journal(t).Steps)
	}
	// restored is the probe's (Task 9): another call changes nothing.
	before := h.read(t, p1Host).Move.UpdatedAt
	if refused, err := h.install(t); err != nil || refused != "" {
		t.Fatalf("install at restored: %q %v", refused, err)
	}
	if after := h.read(t, p1Host).Move; !after.UpdatedAt.Equal(before) {
		t.Fatalf("a call at restored wrote the record: %+v", after)
	}
}

// An unregistered source (Paul's H5(b)): no park-local is owed, and the
// moves-owned rename parks every pre-move artifact before extraction.
func TestDirtyReturnParksAnUnregisteredSource(t *testing.T) {
	t.Parallel()
	h := newReturnHarness(t, false)
	pinned := h.read(t, p1Host).Move.Inputs
	if refused, err := h.install(t); err != nil || refused != "" {
		t.Fatalf("install: %q %v", refused, err)
	}
	h.installedReturn(t, pinned)
	parked := h.parkedFiles(t)
	for suffix, body := range map[string]string{
		"/artifacts/" + p1Host + ".db":         "the stopped database",
		"/artifacts/" + p1Host + ".db.gen":     "cur\n",
		"/artifacts/" + p1Host + ".env":        "APP_MODE=prod\n",
		"/artifacts/" + p1Host + "/upload.txt": "hello",
		"/artifacts/" + p1Host + ".db-wal":     "a stale wal",
	} {
		if !parkedHas(parked, suffix, body) {
			t.Fatalf("%s was not parked (%v)", suffix, parked)
		}
	}
}

// A crash at each journalled step of the pre-move park and of the install
// is finished by the next call, from the journal, with nothing parked twice.
func TestDirtyReturnReplaysFromTheJournal(t *testing.T) {
	t.Parallel()
	for _, point := range []string{"after-retired", "after-" + installStepPreMove + "-plan", "after-" + installStepPreMove, "after-staged", "after-files-rename", "after-record-local", "after-db-rename", "after-record-restored"} {
		t.Run(point, func(t *testing.T) {
			h := newReturnHarness(t, false)
			pinned := h.read(t, p1Host).Move.Inputs
			h.landAt = point
			if _, err := h.install(t); !errors.Is(err, errCrash) {
				t.Fatalf("the crash at %s: %v (points %v)", point, err, h.landHits)
			}
			h.rw = nil
			h.leases.expire()
			if refused, err := h.install(t); err != nil || refused != "" {
				t.Fatalf("the replay: %q %v", refused, err)
			}
			h.installedReturn(t, pinned)
			if n := len(h.parkedFiles(t)); n == 0 {
				t.Fatal("nothing parked")
			}
		})
	}
}

// The source's own pre-move files are installed over only once parked: a
// files path that holds anything after the park is not the return's, and
// the install stops for an operator (stuck, never a landing's rollback)
// with nothing moved; a marker that names another generation is never
// replaced.
func TestDirtyReturnNeverInstallsOverWhatItDidNotPark(t *testing.T) {
	t.Parallel()
	t.Run("files", func(t *testing.T) {
		h := newReturnHarness(t, false)
		sentinel := filepath.Join(h.sdir, p1Host, "sentinel.txt")
		h.landFault = func(p string) error {
			if p == "after-"+landStepRetired {
				if err := os.MkdirAll(filepath.Dir(sentinel), 0o755); err != nil {
					t.Fatal(err)
				}
				return os.WriteFile(sentinel, []byte("not the return's"), 0o644)
			}
			return nil
		}
		refused, err := h.install(t)
		if err != nil || refused != provision.StuckReturnFilesOccupied {
			t.Fatalf("install: %q %v", refused, err)
		}
		m, rb := h.rollback(t)
		if m.State != provision.MoveStuck || m.StuckReason != provision.StuckReturnFilesOccupied || rb.StuckAt != provision.RbSourceLocal || m.Reason != "" {
			t.Fatalf("record: %+v %+v", m, rb)
		}
		if body, err := os.ReadFile(sentinel); err != nil || string(body) != "not the return's" {
			t.Fatalf("the sentinel: %q %v", body, err)
		}
		if _, err := os.Stat(filepath.Join(h.sdir, p1Host, "upload.txt")); !errors.Is(err, os.ErrNotExist) {
			t.Fatalf("the return's files were placed beside the sentinel: %v", err)
		}
		// A stuck install is not driven on.
		if _, err := h.install(t); err == nil {
			t.Fatal("the installer ran a stuck record")
		}
	})
	t.Run("marker", func(t *testing.T) {
		h := newReturnHarness(t, true)
		gen := provision.DefaultDBPath(h.sdir, p1Host) + ".gen"
		h.landFault = func(p string) error {
			if p == "after-"+landStepRetired {
				return os.WriteFile(gen, []byte("cur\n"), 0o600)
			}
			return nil
		}
		if _, err := h.install(t); !errors.Is(err, errLandingChanged) {
			t.Fatalf("install over the pre-move marker: %v", err)
		}
		if m, rb := h.rollback(t); m.State != provision.MoveRollingBack || rb.SourcePhase != provision.RbSourceConfigured {
			t.Fatalf("record: %+v %+v", m, rb)
		}
		if body, err := os.ReadFile(gen); err != nil || string(body) != "cur\n" {
			t.Fatalf("the marker was replaced: %q %v", body, err)
		}
		if _, err := os.Stat(provision.DefaultDBPath(h.sdir, p1Host)); !errors.Is(err, os.ErrNotExist) {
			t.Fatalf("a database was installed beside the marker: %v", err)
		}
	})
}

// A return build that does not verify stops the install at configured,
// stuck, with nothing installed.
func TestDirtyReturnCorruptBuildIsStuck(t *testing.T) {
	t.Parallel()
	h := newReturnHarness(t, true)
	h.landFault = func(point string) error {
		if point == "before-db-verify" {
			f, err := os.OpenFile(filepath.Join(h.sdir, ".moves", p1Host, p1RB, "build", p1Host+".db"), os.O_APPEND|os.O_WRONLY, 0o600)
			if err != nil {
				t.Fatal(err)
			}
			defer f.Close()
			if _, err := f.Write([]byte("corrupt")); err != nil {
				t.Fatal(err)
			}
		}
		return nil
	}
	refused, err := h.install(t)
	if err != nil || refused != provision.StuckReturnInstallMismatch {
		t.Fatalf("install: %q %v", refused, err)
	}
	m, rb := h.rollback(t)
	if m.State != provision.MoveStuck || m.StuckReason != provision.StuckReturnInstallMismatch || rb.StuckAt != provision.RbSourceConfigured || rb.RecoveryInputs != nil {
		t.Fatalf("record: %+v %+v", m, rb)
	}
	dbPath := provision.DefaultDBPath(h.sdir, p1Host)
	for _, p := range []string{dbPath, dbPath + ".gen"} {
		if _, err := os.Stat(p); !errors.Is(err, os.ErrNotExist) {
			t.Fatalf("%s exists after a refused install: %v", p, err)
		}
	}
}

// The claim check and the record check are the recovery's own (J13): a
// claim with another executor, or at another kind or phase, is refused
// before anything is written, and a clean rollback has nothing to install.
func TestDirtyReturnRequiresItsClaimAndRecord(t *testing.T) {
	t.Parallel()
	for name, edit := range map[string]func(*hostclaim.Claim){
		"another executor":  func(c *hostclaim.Claim) { c.Op.Exec = "uk-box" },
		"another phase":     func(c *hostclaim.Claim) { c.Op.Phase = hostclaim.PhaseProbing },
		"another rollback":  func(c *hostclaim.Claim) { c.Op.Intent["rollback"] = "rb_bcdefghijklmnopqrstuvwxya" },
		"another holder":    func(c *hostclaim.Claim) { c.Holder.Stamp = 7 },
		"a landing op kind": func(c *hostclaim.Claim) { c.Op.Kind = hostclaim.KindLanding },
	} {
		t.Run(name, func(t *testing.T) {
			h := newReturnHarness(t, true)
			rewriteP1Claim(t, h.s, edit)
			before := h.claim(t)
			if _, err := h.install(t); err == nil {
				t.Fatal("installed under the wrong claim")
			}
			if _, rb := h.rollback(t); rb.SourcePhase != provision.RbSourceParking {
				t.Fatalf("the record advanced: %+v", rb)
			}
			if after := h.claim(t); len(after.Obligations) != len(before.Obligations) {
				t.Fatalf("chores queued: %+v", after.Obligations)
			}
		})
	}
	t.Run("a clean rollback", func(t *testing.T) {
		h := newReturnHarness(t, true)
		rec := h.read(t, p1Host)
		rec.Move.Rollback.Dirty = false
		if _, err := recoveryInstallSource(&movesWorker{cfg: h.rcfg}, h.s, rec, h.sealOf, &h.baseline); err == nil {
			t.Fatal("a clean rollback has an install source")
		}
	})
	t.Run("a seal outside the lane", func(t *testing.T) {
		// A usable seal over the move's own staging lane in the source's
		// bucket: every object readable at its version, but not the return's.
		h := newReturnHarness(t, true)
		ctx := context.Background()
		other := h.landingHarness.seal(t)
		for _, class := range [][]moves.SealedObject{other.Replica, other.Files, other.Timers} {
			for i, o := range class {
				ver, _, err := h.vs.CopyVersion(ctx, "data-uk", o.Key, o.VersionID, "data-ie", o.Key, true)
				if err != nil {
					t.Fatal(err)
				}
				class[i].VersionID = ver
			}
		}
		h.sealOf = func(*hostclaim.Claim) (moves.Seal, error) { return other, nil }
		if _, err := h.install(t); !errors.Is(err, moves.ErrBadSeal) {
			t.Fatalf("built from the move's staging lane, not the return lane: %v", err)
		}
		if n := len(h.vs.callsOn(moves.GenerationPrefix(cutRoot, p1RB))); n != 0 {
			t.Fatalf("the build ran: %d calls", n)
		}
	})
}

// J24: the stand-in view's authoritative region is the other side's. A
// landing works before the target's probe is admitted, so the source's
// region (From) is authoritative; a dirty return works before the
// recovery's admission, so the target's (To) is.
func TestStandInViewTakesTheRole(t *testing.T) {
	m := &provision.MoveState{From: "ie", To: "uk"}
	obs := hostclaim.Observed{ETag: "e", Claim: &hostclaim.Claim{Holder: &hostclaim.Holder{Location: cutRoot + "/"}}}
	for role, want := range map[string][2]string{roleTarget: {"uk", "ie"}, roleSource: {"ie", "uk"}} {
		v, err := standInView(want[0], role, m)(context.Background(), obs)
		if err != nil || v.Region != want[0] || v.AuthoritativeRegion != want[1] {
			t.Fatalf("%s: %+v %v", role, v, err)
		}
		if g, ok := v.GenerationOf(park.Ref{Key: cutRoot + "/g-" + p1RB + "/0000/x.ltx"}); !ok || g != p1RB {
			t.Fatalf("%s: generation %q %v", role, g, ok)
		}
	}
	if _, err := standInView("ie", "", m)(context.Background(), obs); !errors.Is(err, park.ErrUnsafe) {
		t.Fatalf("no role: %v", err)
	}
}

// Fix round 1, ruling 1: the moves-owned rename of the source's pre-move
// artifacts, its plan and its renames, runs inside the host guard, with the
// claim and the record read fresh there. Authority that changes before it
// (another executor, or the record no longer this return at local) leaves
// every live path where it is and journals no plan.
func TestDirtyReturnRetirementIsGuarded(t *testing.T) {
	t.Parallel()
	g := provision.MoveGuard{ID: p1ID}
	for name, change := range map[string]func(t *testing.T, h *returnHarness){
		"the claim's executor": func(t *testing.T, h *returnHarness) {
			rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) { c.Op.Exec = "uk-box" })
		},
		"the record": func(t *testing.T, h *returnHarness) {
			g.Incarnation = h.inc
			if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, provision.MoveMarkStuck(g, provision.StuckReturnFilesOccupied)); err != nil {
				t.Fatal(err)
			}
		},
	} {
		t.Run(name, func(t *testing.T) {
			h := newReturnHarness(t, false)
			changed := false
			h.landFault = func(p string) error {
				if p == "before-"+installStepPreMove && !changed {
					changed = true
					change(t, h)
				}
				return nil
			}
			if _, err := h.install(t); err == nil {
				t.Fatal("the rename ran after its authority changed")
			}
			if !changed {
				t.Fatalf("the rename's guard point was never reached: %v", h.landHits)
			}
			db := provision.DefaultDBPath(h.sdir, p1Host)
			for path, body := range map[string]string{db: "the stopped database", db + ".gen": "cur\n", db + "-wal": "a stale wal",
				filepath.Join(h.sdir, p1Host, "upload.txt"): "hello", filepath.Join(h.sdir, p1Host+".env"): "APP_MODE=prod\n"} {
				if got, err := os.ReadFile(path); err != nil || string(got) != body {
					t.Fatalf("%s moved: %q %v", path, got, err)
				}
			}
			if j := h.journal(t); j.Done(installStepPreMove+"-plan") || j.Done(installStepPreMove) {
				t.Fatalf("a plan was journalled: %v", j.Steps)
			}
			if parked := h.parkedFiles(t); len(parked) != 0 {
				t.Fatalf("parked: %v", parked)
			}
		})
	}
}

// Fix round 1, ruling 2 (§7.8, rev 7.11): a leftover tagged with a stamp the
// claim shows as superseded on this box (a cold admission here that issued
// it, now neither the holder's stamp nor its live session) is retired
// through park.Local, so its inventory entries clear before the return's
// installation is registered; the moves-owned rename then finds nothing of
// it to move.
func TestDirtyReturnRetiresRetiredStampLeftovers(t *testing.T) {
	t.Parallel()
	h := newReturnHarness(t, true)
	wal, shadow := p1Host+".db-wal", ".activator-"+p1Host+".db-litestream"
	rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
		inv := c.Inventory["ie-box"]
		inv.Paths[wal], inv.Paths[shadow] = 5, 5
		c.Inventory["ie-box"] = inv
		c.Holder.Admissions = append(c.Holder.Admissions, hostclaim.AdmissionRecord{Region: "ie", Session: 5, Box: "ie-box", Generation: "g5", Mode: "cold", Covered: true})
	})
	pinned := h.read(t, p1Host).Move.Inputs
	if refused, err := h.install(t); err != nil || refused != "" {
		t.Fatalf("install: %q %v", refused, err)
	}
	h.installedReturn(t, pinned)
	parked := h.parkedFiles(t)
	for suffix, body := range map[string]string{"/artifacts/" + wal: "a stale wal", "/artifacts/" + shadow + "/generation": "shadow"} {
		if !parkedHas(parked, suffix, body) {
			t.Fatalf("%s was not parked (%v)", suffix, parked)
		}
	}
	// Retired by park.Local, not by the rename: the rename's plan found
	// them gone.
	plan, err := journalledLocalPlan(h.journal(t), installStepPreMove)
	if err != nil {
		t.Fatal(err)
	}
	for _, it := range plan.Items {
		if (it.Path == wal || it.Path == shadow || it.Path == p1Host+".db") && it.ID != "" {
			t.Fatalf("the rename planned %s, which the inventory claimed", it.Path)
		}
	}
}

// Fix round 1, ruling 2: a leftover tagged with any other stamp (live,
// unknown, or another holder's) stops the return for an operator before
// anything moves, with a recorded reason; once the operator has dealt with
// it, move --retry (MoveRetry) goes back to local and the install proceeds.
func TestDirtyReturnUnknownStampIsStuckUntilRetried(t *testing.T) {
	t.Parallel()
	for name, stamp := range map[string]uint64{"unknown": 9, "the live session": 0} {
		t.Run(name, func(t *testing.T) {
			h := newReturnHarness(t, true)
			wal := p1Host + ".db-wal"
			rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
				s := stamp
				if s == 0 {
					// The holder's live writer session on this box: never a
					// leftover, even with a cold admission record here.
					s = c.Holder.Session
					if s == 0 || s == c.Holder.Stamp {
						s = 11
						c.Holder.Session = s
					}
					c.Holder.Admissions = append(c.Holder.Admissions, hostclaim.AdmissionRecord{Region: "ie", Session: s, Box: "ie-box", Generation: "g-live", Mode: "cold", Covered: true})
				}
				inv := c.Inventory["ie-box"]
				inv.Paths[wal] = s
				c.Inventory["ie-box"] = inv
			})
			pinned := h.read(t, p1Host).Move.Inputs
			refused, err := h.install(t)
			if err != nil || refused != provision.StuckReturnUnknownStamp {
				t.Fatalf("install: %q %v", refused, err)
			}
			m, rb := h.rollback(t)
			if m.State != provision.MoveStuck || m.StuckReason != provision.StuckReturnUnknownStamp || rb.StuckAt != provision.RbSourceLocal {
				t.Fatalf("record: %+v %+v", m, rb)
			}
			db := provision.DefaultDBPath(h.sdir, p1Host)
			for path, body := range map[string]string{db: "the stopped database", db + ".gen": "cur\n", db + "-wal": "a stale wal"} {
				if got, err := os.ReadFile(path); err != nil || string(got) != body {
					t.Fatalf("%s moved: %q %v", path, got, err)
				}
			}
			if parked := h.parkedFiles(t); len(parked) != 0 {
				t.Fatalf("parked before the refusal: %v", parked)
			}
			if !hostclaim.InventoryHasStamp(h.claim(t), "ie-box", 1) {
				t.Fatal("S was retired before the refusal")
			}
			// The operator accounts for the file (here: its stray entry is
			// dropped), then retries.
			rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
				inv := c.Inventory["ie-box"]
				delete(inv.Paths, wal)
				c.Inventory["ie-box"] = inv
			})
			g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
			if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, provision.MoveRetry(g)); err != nil {
				t.Fatal(err)
			}
			if refused, err := h.install(t); err != nil || refused != "" {
				t.Fatalf("install after the retry: %q %v", refused, err)
			}
			h.installedReturn(t, pinned)
			if !parkedHas(h.parkedFiles(t), "/artifacts/"+wal, "a stale wal") {
				t.Fatal("the WAL was not parked after the retry")
			}
		})
	}
}