package host

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

	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/park"

	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/release"
)

const mvEvMoveID = "mv_abcdefghijklmnopqrstuvwxyz"

// evRec is a record whose move is in the given state, edited by mut.
func mvEvRec(state string, mut func(*provision.MoveState)) provision.Record {
	m := &provision.MoveState{ID: mvEvMoveID, Incarnation: strings.Repeat("ab", 32), From: "ie", To: "uk", Cut: "auto", State: state, RequestedAt: time.Unix(1_800_000_000, 0).UTC()}
	if mut != nil {
		mut(m)
	}
	return provision.Record{Account: "bab", App: "titogo", Host: "acme.go.tito.io", Region: "ie", Move: m}
}

func mvEvAt(sec int64) *time.Time {
	t := time.Unix(1_800_000_000+sec, 0).UTC()
	return &t
}

// The events a full move that ends done goes through, each record the one a
// writer leaves, in order. The console's request is seen by the source's
// pass (requested); every later record is a box writer's.
func doneFlow() []provision.Record {
	cut := func(phase string) provision.Record {
		return mvEvRec(provision.MoveCutting, func(m *provision.MoveState) {
			m.CutPhase, m.CutAt, m.CutBudgetS = phase, mvEvAt(10), 45
		})
	}
	land := func(phase string, attempts int) provision.Record {
		r := mvEvRec(provision.MoveLanding, func(m *provision.MoveState) {
			m.CutPhase, m.CutAt, m.CutBudgetS, m.FlippedAt, m.LandingPhase = provision.CutFlipped, mvEvAt(10), 45, mvEvAt(20), phase
			if attempts > 0 {
				m.Landing = &provision.MoveLandingJournal{Attempts: attempts}
			}
		})
		r.Region = "uk"
		return r
	}
	committed := func(mut func(*provision.MoveState)) provision.Record {
		r := mvEvRec(provision.MoveCommitted, func(m *provision.MoveState) {
			m.CutPhase, m.CutAt, m.FlippedAt, m.CommittedAt = provision.CutFlipped, mvEvAt(10), mvEvAt(20), mvEvAt(40)
			m.Landing = &provision.MoveLandingJournal{Attempts: 1}
			if mut != nil {
				mut(m)
			}
		})
		r.Region = "uk"
		return r
	}
	closed := provision.Record{Account: "bab", App: "titogo", Host: "acme.go.tito.io", Region: "uk",
		LastMove: &provision.MoveOutcome{ID: mvEvMoveID, Outcome: "done", At: *mvEvAt(60), OfflineS: 41.2}}
	return []provision.Record{
		mvEvRec(provision.MovePreparing, nil),
		mvEvRec(provision.MoveReady, nil),
		cut(provision.CutRequested),
		cut(provision.CutAdmitted), cut(provision.CutStopping), cut(provision.CutStopped), cut(provision.CutStaged), cut(provision.CutArchived),
		land(provision.LandChecking, 0),
		land(provision.LandParking, 0), land(provision.LandBuilding, 0), land(provision.LandPublished, 0), land(provision.LandLocal, 0),
		land(provision.LandConfigured, 0), land(provision.LandRestored, 0), land(provision.LandProbing, 0), land(provision.LandProbing, 1),
		land(provision.LandProbed, 1),
		committed(nil),
		committed(func(m *provision.MoveState) { m.OpenedAt = mvEvAt(51) }),
		committed(func(m *provision.MoveState) { m.OpenedAt, m.SourceParkedAt = mvEvAt(51), mvEvAt(55) }),
		closed,
	}
}

var doneFlowEvents = []string{
	"requested", "ready", "cut_requested", "admitted", "stopping", "stopped", "staged", "archived", "flipped",
	"parking", "building", "published", "local", "configured", "restored", "probing", "probe_attempt", "probed",
	"committed", "opened", "source_parked", "done",
}

// eventWorker is a worker whose LogEvent hook records every event it is
// handed.
func eventWorker(got *[]moves.LogEvent) *movesWorker {
	w := &movesWorker{mirrors: map[string]lease.MirrorStore{}}
	w.cfg.Box = "box-test"
	w.cfg.Now = func() time.Time { return time.Unix(1_800_000_100, 0).UTC() }
	w.cfg.LogEvent = func(_ context.Context, _ release.Store, _ provision.Record, ev moves.LogEvent) error {
		*got = append(*got, ev)
		return nil
	}
	return w
}

// Review Focus 3: across a full done flow, every transition gives one event,
// in order, with its identity; the record seen again (a status poll, a pass
// that changes nothing, a replay of the same write) gives none.
func TestEveryTransitionEmitsOneEvent(t *testing.T) {
	ctx := context.Background()
	var got []moves.LogEvent
	w := eventWorker(&got)
	flow := doneFlow()
	w.observeMove(ctx, nil, flow[0])
	for i := 1; i < len(flow); i++ {
		w.logTransition(ctx, nil, flow[i-1], flow[i])
		// A status poll, a pass that changes nothing, a replayed write. A
		// closed move's identities are forgotten (create-only storage refuses
		// a replay of the close), so the close is not replayed here.
		w.observeMove(ctx, nil, flow[i])
		w.logTransition(ctx, nil, flow[i], flow[i])
		if i < len(flow)-1 {
			w.logTransition(ctx, nil, flow[i-1], flow[i])
		}
	}
	var events []string
	for _, ev := range got {
		events = append(events, ev.Event)
		if ev.Op != mvEvMoveID || ev.Host != "acme.go.tito.io" || ev.Actor != "box-test" || ev.V != 1 || ev.At.IsZero() {
			t.Errorf("event %+v", ev)
		}
	}
	if strings.Join(events, " ") != strings.Join(doneFlowEvents, " ") {
		t.Fatalf("events\n got %v\nwant %v", events, doneFlowEvents)
	}
	ids := map[string]bool{}
	for _, ev := range got {
		if ids[ev.Identity] {
			t.Fatalf("identity %q twice", ev.Identity)
		}
		ids[ev.Identity] = true
	}
	for _, want := range []string{"requested", "ready", "cutting-requested", "cutting-stopped", "cutting-flipped", "landing-probing", "probe_attempt-1", "committed", "committed-opened", "done"} {
		if !ids[want] {
			t.Errorf("no event with identity %q in %v", want, ids)
		}
	}
	if f := got[len(got)-1].Fields; f["offline_s"] != "41.2" {
		t.Errorf("done carries %v, want offline_s", f)
	}
	if f := got[19].Fields; got[19].Event != "opened" || f["offline_s"] != "41" {
		t.Errorf("opened carries %v, want offline_s from cut_at to the open", f)
	}
}

// A rollback, a stuck and a retry: the rollback's phases are named for the
// half that moved; stuck and retry are keyed by the recovery retry epoch
// (claim Plan 10's RecoveryRetryEpoch, preflight L11).
func TestRollbackStuckAndRetryEvents(t *testing.T) {
	ctx := context.Background()
	var got []moves.LogEvent
	w := eventWorker(&got)
	rb := func(target, source string, mut func(*provision.MoveState)) provision.Record {
		return mvEvRec(provision.MoveRollingBack, func(m *provision.MoveState) {
			m.CutAt = mvEvAt(10)
			m.Rollback = &provision.MoveRollback{ID: "rb_x", Reason: provision.RollbackLandingDeadline, TargetPhase: target, SourcePhase: source}
			if mut != nil {
				mut(m)
			}
		})
	}
	stuck := func(epoch uint64) provision.Record {
		return mvEvRec(provision.MoveStuck, func(m *provision.MoveState) {
			m.CutAt = mvEvAt(10)
			m.Rollback = &provision.MoveRollback{ID: "rb_x", TargetPhase: provision.RbTargetCleaned, SourcePhase: provision.RbSourceVerified, StuckAt: provision.RbSourceVerified}
			m.StuckReason, m.RecoveryRetryEpoch = provision.StuckBaselineMismatch, epoch
		})
	}
	landing := mvEvRec(provision.MoveLanding, func(m *provision.MoveState) { m.LandingPhase = provision.LandProbing })
	seq := []provision.Record{
		landing,
		rb(provision.RbTargetReleasing, provision.RbSourceWaiting, nil),
		rb(provision.RbTargetReleased, provision.RbSourceWaiting, nil),
		rb(provision.RbTargetReleased, provision.RbSourceClaimed, nil),
		rb(provision.RbTargetCleaned, provision.RbSourceVerified, nil),
		stuck(0),
	}
	for i := 1; i < len(seq); i++ {
		w.logTransition(ctx, nil, seq[i-1], seq[i])
	}
	// The operator's retry (the console's write, seen by the box) advances
	// the epoch and leaves stuck; the second stuck is a new entry.
	retried := rb(provision.RbTargetCleaned, provision.RbSourceVerified, func(m *provision.MoveState) { m.RecoveryRetryEpoch = 1 })
	w.observeMove(ctx, nil, retried)
	w.observeMove(ctx, nil, retried)
	w.logTransition(ctx, nil, retried, stuck(1))
	w.logTransition(ctx, nil, stuck(1), provision.Record{Account: "bab", App: "titogo", Host: "acme.go.tito.io", Region: "ie",
		LastMove: &provision.MoveOutcome{ID: mvEvMoveID, Outcome: "rolled_back", At: *mvEvAt(90), OfflineS: 80}})
	var events, ids []string
	for _, ev := range got {
		events = append(events, ev.Event)
		ids = append(ids, ev.Identity)
	}
	want := "rolling_back released claimed cleaned stuck retry stuck rolled_back"
	if strings.Join(events, " ") != want {
		t.Fatalf("events %v, want %s", events, want)
	}
	wantIDs := "rolling_back rolling_back-target-released rolling_back-source-claimed rolling_back-target-cleaned stuck-0 retry-1 stuck-1 rolled_back"
	if strings.Join(ids, " ") != wantIDs {
		t.Fatalf("identities %v, want %s", ids, wantIDs)
	}
}

type memberSink struct{ events []string }

func (s *memberSink) App(acct, app, host, line string) {}
func (s *memberSink) Event(acct, app, host, level, msg string) {
	s.events = append(s.events, acct+"/"+app+"/"+host+" "+level+" "+msg)
}

// Preflight L9 and L27: the production LogEvent writes the object
// create-only, and only the writer that created it tells members; a second
// writer, or a replay, does neither. Steps members do not need (stopping)
// are logged and not shown to them.
func TestMoveLogWriterTellsMembersOnce(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	sink := &memberSink{}
	write := moveLogWriter(sink)
	flow := doneFlow()
	ev := func(event, identity string, fields map[string]string) moves.LogEvent {
		return moves.LogEvent{V: 1, Op: mvEvMoveID, Host: "acme.go.tito.io", Event: event, Identity: identity, Actor: "box-uk", At: time.Now().UTC(), Fields: fields}
	}
	done := flow[len(flow)-1]
	write(ctx, s, done, ev("done", "done", map[string]string{"offline_s": "41.2", "from": "ie", "to": "uk"}))
	write(ctx, s, done, ev("done", "done", map[string]string{"offline_s": "41.2", "from": "ie", "to": "uk"}))
	write(ctx, s, flow[4], ev("stopping", "cutting-stopping", map[string]string{"from": "ie", "to": "uk"}))
	if len(sink.events) != 1 || sink.events[0] != "bab/titogo/acme.go.tito.io info Moved. acme.go.tito.io is served from uk. It was offline for 41 seconds." {
		t.Fatalf("member events %q", sink.events)
	}
	prefix := moves.ReportPrefix("bab", "titogo", "acme.go.tito.io", mvEvMoveID)
	logged, err := moves.ReadLog(ctx, s, prefix)
	if err != nil || len(logged) != 2 || logged[0].Actor != "box-uk" {
		t.Fatalf("log %+v %v", logged, err)
	}
}

// The member lines (preflight L9, Paul's choice): M-C1, M-C5, M-C6, M-C7,
// M-C8, M-C9, and for stuck a line with no operator command in it.
func TestMemberLines(t *testing.T) {
	flow := doneFlow()
	rolling := mvEvRec(provision.MoveRollingBack, func(m *provision.MoveState) {
		m.Rollback = &provision.MoveRollback{ID: "rb_x", Reason: provision.RollbackLandingDeadline}
	})
	stuck := mvEvRec(provision.MoveStuck, nil)
	rolledBack := provision.Record{Account: "bab", App: "titogo", Host: "acme.go.tito.io", Region: "ie",
		LastMove: &provision.MoveOutcome{ID: mvEvMoveID, Outcome: "rolled_back"}}
	for _, c := range []struct {
		event string
		rec   provision.Record
		level string
		line  string
	}{
		{"requested", flow[0], "info", "Moving acme.go.tito.io from ie to uk."},
		{"admitted", flow[3], "info", "uk is ready. Starting the cut. acme.go.tito.io will be offline for about 45 seconds."},
		{"flipped", flow[8], "info", "Stopped in ie. Starting in uk."},
		{"rolling_back", rolling, "warn", "uk could not start acme.go.tito.io: " + provision.RollbackLandingDeadline + ". Moving it back to ie."},
		{"rolled_back", rolledBack, "warn", "acme.go.tito.io is back in ie and serving. Nothing was lost."},
		{"stuck", stuck, "error", "acme.go.tito.io could not start in uk or in ie. It is offline. An operator has been paged."},
		{"stopping", flow[4], "", ""},
	} {
		// The entry's own fields, as the writer names the transition (the
		// admitted cut's budget, the rollback's reason): the line needs
		// nothing else, so an entry the outbox writes later says the same.
		ev := moves.LogEvent{Event: c.event, Fields: map[string]string{"from": "ie", "to": "uk"}}
		if before := (provision.Record{Host: c.rec.Host}); c.rec.Move != nil {
			before.Move = &provision.MoveState{ID: c.rec.Move.ID, From: "ie", To: "uk", State: provision.MoveLanding}
			if c.event == "admitted" {
				before = flow[2]
			}
			if tr, ok := moves.TransitionOf(before, c.rec); ok && tr.Event == c.event {
				ev.Fields = tr.Fields
			}
		}
		line, ok := memberLine(ev, c.rec.Host)
		level := ""
		if ok {
			level = memberLevel(c.event)
		}
		if ok != (c.line != "") || level != c.level || line != c.line {
			t.Errorf("%s: %q %q %v, want %q %q", c.event, level, line, ok, c.level, c.line)
		}
	}
}

// ---- Task 3 fix round 1 ----

// Ruling 4: the box knows the home region only as "", so the member line
// names it with the reviewed phrase (MV-N381), never dropping the line.
func TestHomeRegionMemberLines(t *testing.T) {
	fromHome := mvEvRec(provision.MovePreparing, func(m *provision.MoveState) { m.From = "" })
	toHome := mvEvRec(provision.MovePreparing, func(m *provision.MoveState) { m.To = "" })
	for _, c := range []struct {
		rec   provision.Record
		event string
		want  string
	}{
		{fromHome, "requested", "Moving acme.go.tito.io from the home region to uk."},
		{toHome, "requested", "Moving acme.go.tito.io from ie to the home region."},
		{toHome, "flipped", "Stopped in ie. Starting in the home region."},
	} {
		ev := moves.LogEvent{Event: c.event, Fields: map[string]string{"from": c.rec.Move.From, "to": c.rec.Move.To}}
		if line, ok := memberLine(ev, c.rec.Host); !ok || line != c.want {
			t.Errorf("%s: %q %v, want %q", c.event, line, ok, c.want)
		}
	}
}

// Ruling 5: the events carry the transition's evidence. The cut's stopped
// carries the source's digests, the landing's probed the installed
// database, a stuck its role, and the source's recovered the recovery's
// timing.
func TestEventsCarryTheirEvidence(t *testing.T) {
	ctx := context.Background()
	var got []moves.LogEvent
	w := eventWorker(&got)
	cut := func(phase string, src *provision.MoveSource) provision.Record {
		return mvEvRec(provision.MoveCutting, func(m *provision.MoveState) { m.CutPhase, m.Source = phase, src })
	}
	src := &provision.MoveSource{FinalTXID: 42, DBSHA256: strings.Repeat("d", 64), TimersSHA256: strings.Repeat("e", 64),
		Files: &provision.MoveFilesDigest{Plain: strings.Repeat("f", 64)}, Evidence: "moves/x/stopped-source.json"}
	w.logTransition(ctx, nil, cut(provision.CutStopping, nil), cut(provision.CutStopped, src))
	land := func(phase, db string) provision.Record {
		return mvEvRec(provision.MoveLanding, func(m *provision.MoveState) {
			m.LandingPhase = phase
			m.Landing = &provision.MoveLandingJournal{InstalledDBSHA256: db, Attempts: 1}
		})
	}
	w.logTransition(ctx, nil, land(provision.LandProbing, strings.Repeat("a", 64)), land(provision.LandProbed, strings.Repeat("a", 64)))
	rb := func(target, source string, stuck bool) provision.Record {
		state := provision.MoveRollingBack
		if stuck {
			state = provision.MoveStuck
		}
		return mvEvRec(state, func(m *provision.MoveState) {
			m.CutAt = mvEvAt(10)
			m.Rollback = &provision.MoveRollback{ID: "rb_x", RequestedAt: mvEvAt(30), TargetPhase: target, SourcePhase: source}
			if source == provision.RbSourceRecovered {
				m.Rollback.RecoveredAt = mvEvAt(70)
			}
		})
	}
	w.logTransition(ctx, nil, rb(provision.RbTargetReturning, provision.RbSourceWaiting, false), rb(provision.RbTargetReturning, provision.RbSourceWaiting, true))
	w.logTransition(ctx, nil, rb(provision.RbTargetCleaned, provision.RbSourceProbed, false), rb(provision.RbTargetCleaned, provision.RbSourceRecovered, false))
	// The source's stuck, on another box's worker (the target's stuck above
	// is stuck-0 in this one).
	eventWorker(&got).logTransition(ctx, nil, rb(provision.RbTargetCleaned, provision.RbSourceVerified, false), rb(provision.RbTargetCleaned, provision.RbSourceVerified, true))
	by := map[string]map[string]string{}
	for _, ev := range got {
		by[ev.Identity] = ev.Fields
	}
	if f := by["cutting-stopped"]; f["db_sha256"] != src.DBSHA256 || f["final_txid"] != "42" || f["files_plain"] != src.Files.Plain || f["timers_sha256"] != src.TimersSHA256 {
		t.Errorf("stopped carries %v", f)
	}
	if f := by["landing-probed"]; f["installed_db_sha256"] != strings.Repeat("a", 64) {
		t.Errorf("probed carries %v", f)
	}
	if f := by["rolling_back-source-recovered"]; f["recovery_s"] != "40" || f["offline_s"] != "60" {
		t.Errorf("recovered carries %v, want recovery_s 40 (rollback request to recovered) and offline_s 60 (cut to recovered)", f)
	}
	var roles []string
	for _, ev := range got {
		if ev.Event == "stuck" {
			roles = append(roles, ev.Fields["role"])
		}
	}
	if strings.Join(roles, " ") != "target source" {
		t.Errorf("stuck roles %v, want the target's (at returning), then the source's", roles)
	}
}

// Ruling 7: the real writers log, through the move harness. The landing
// harness drives the source's readiness, the console's cut, the cut and the
// flip, then the landing to installed; the close harness the probe, the
// commit, the open, the source's park and the close. A writer whose log
// call went missing drops out of the list.
func TestRealWritersLogEveryTransition(t *testing.T) {
	t.Parallel()
	t.Run("readiness to installed", func(t *testing.T) {
		h := newLandingHarness(t)
		var got []moves.LogEvent
		h.cfg.LogEvent = func(_ context.Context, _ release.Store, _ provision.Record, ev moves.LogEvent) error {
			got = append(got, ev)
			return nil
		}
		h.flip(t)
		h.land(t)
		h.installed(t)
		wantSubsequence(t, got, "ready", "cut_requested", "admitted", "stopping", "stopped", "staged", "archived", "flipped", "parking", "building", "published", "local", "configured", "restored")
	})
	t.Run("probe to done", func(t *testing.T) {
		h := newCloseHarness(t)
		var got []moves.LogEvent
		h.cfg.LogEvent = func(_ context.Context, _ release.Store, _ provision.Record, ev moves.LogEvent) error {
			got = append(got, ev)
			return nil
		}
		h.land(t)
		h.committedAndOpen(t)
		h.sourcePass(t)
		h.land(t)
		lane := h.chore(t, "uk", true)
		h.clockPast(lane.Park.EvidenceAt.Add(park.SignatureAge + lane.Park.RequestLimit))
		h.land(t)
		src := h.chore(t, "ie", false)
		h.clockPast(src.Park.EvidenceAt.Add(park.SignatureAge + src.Park.RequestLimit))
		if err := h.regionalAck(t, src); err != nil {
			t.Fatal(err)
		}
		h.land(t)
		h.done(t)
		wantSubsequence(t, got, "probing", "probe_attempt", "probed", "committed", "opened", "source_parked", "done")
	})
}

// wantSubsequence requires want's events, in order, among got's, and no
// identity twice.
func wantSubsequence(t *testing.T, got []moves.LogEvent, want ...string) {
	t.Helper()
	var events []string
	seen := map[string]bool{}
	for _, ev := range got {
		events = append(events, ev.Event)
		if seen[ev.Identity] {
			t.Errorf("identity %q twice", ev.Identity)
		}
		seen[ev.Identity] = true
	}
	i := 0
	for _, e := range events {
		if i < len(want) && e == want[i] {
			i++
		}
	}
	if i != len(want) {
		t.Fatalf("events %v lack %v in order (matched %d)", events, want, i)
	}
}

// flakyLogStore fails every write under a move's log while fail is set.
type flakyLogStore struct {
	release.Store
	fail bool
}

func (f *flakyLogStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if f.fail && strings.Contains(key, "/log/") {
		return "", errors.New("the bucket is down")
	}
	return f.Store.Put(ctx, key, body, ifMatch)
}

// outboxWorker is a worker with the production log writer over s and an
// outbox under dataDir.
func outboxWorker(dataDir string) *movesWorker {
	w := &movesWorker{mirrors: map[string]lease.MirrorStore{}}
	w.cfg.Box = "box-test"
	w.cfg.DataDir = dataDir
	w.cfg.Now = func() time.Time { return time.Unix(1_800_000_100, 0).UTC() }
	w.cfg.LogEvent = moveLogWriter(nil)
	return w
}

func outboxFiles(t *testing.T, dataDir string) []string {
	t.Helper()
	var out []string
	root := moveLogOutboxDir(dataDir)
	err := filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error {
		if errors.Is(err, fs.ErrNotExist) {
			return fs.SkipAll
		}
		if err != nil {
			return err
		}
		if !d.IsDir() {
			rel, _ := filepath.Rel(root, p)
			out = append(out, filepath.ToSlash(rel))
		}
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
	return out
}

// Fix round 2, ruling 1: a log write that fails is kept, with the entry's
// bytes, in the worker's outbox on disk, and every pass retries the outbox,
// whatever the record now says. Here the write fails at the request and
// again at the close; the move is gone from the record by the time the
// bucket answers, and a fresh worker (an agent restart) writes both.
func TestLogOutboxOutlivesTheMove(t *testing.T) {
	ctx := context.Background()
	dataDir := t.TempDir()
	s := &flakyLogStore{Store: release.NewDirStore(t.TempDir()), fail: true}
	w := outboxWorker(dataDir)
	flow := doneFlow()
	w.observeMove(ctx, s, flow[0])
	last := flow[len(flow)-1]
	w.logTransition(ctx, s, flow[len(flow)-2], last)
	if last.Move != nil {
		t.Fatal("the fixture's last record still has its move")
	}
	files := outboxFiles(t, dataDir)
	if strings.Join(files, " ") != "acme.go.tito.io/"+mvEvMoveID+"/done.json acme.go.tito.io/"+mvEvMoveID+"/requested.json" {
		t.Fatalf("outbox %v", files)
	}
	w.flushLogOutbox(ctx, s) // still failing: kept
	if len(outboxFiles(t, dataDir)) != 2 {
		t.Fatal("a failed retry dropped an entry")
	}
	pending, err := os.ReadFile(filepath.Join(moveLogOutboxDir(dataDir), "acme.go.tito.io", mvEvMoveID, "requested.json"))
	if err != nil {
		t.Fatal(err)
	}
	var kept moveLogPending
	if err := json.Unmarshal(pending, &kept); err != nil {
		t.Fatal(err)
	}
	s.fail = false
	outboxWorker(dataDir).flushLogOutbox(ctx, s)
	if f := outboxFiles(t, dataDir); len(f) != 0 {
		t.Fatalf("written entries left in the outbox: %v", f)
	}
	prefix := moves.ReportPrefix("bab", "titogo", "acme.go.tito.io", mvEvMoveID)
	got, err := moves.ReadLog(ctx, s, prefix)
	if err != nil || len(got) != 2 {
		t.Fatalf("log %+v %v", got, err)
	}
	body, _, err := s.Get(ctx, moves.LogKeyFor(prefix, "requested"))
	if err != nil || string(body) != string(kept.Entry) {
		t.Fatalf("the object is not the kept bytes:\n%s\n%s", body, kept.Entry)
	}
	// Another sighting of the closed move writes nothing more.
	w.observeMove(ctx, s, last)
	if f := outboxFiles(t, dataDir); len(f) != 0 {
		t.Fatalf("outbox %v", f)
	}
}

// Fix round 2, ruling 2: recovery probe starts are logged through the
// journal as a box builds it (movesWorker.probeAttempts), and a start whose
// log write fails is kept in the outbox.
func TestRecoveryProbeAttemptsAreLogged(t *testing.T) {
	ctx := context.Background()
	f := recoveryPAFixture(t, hostclaim.KindRestoring)
	s := &flakyLogStore{Store: f.claims, fail: true}
	w := outboxWorker(f.dataDir)
	ids := 0
	ops := w.probeAttempts(s, func(context.Context, string, string, string) (*provision.Record, error) { return f.rec, nil },
		func() string { ids++; return "id-" + strings.Repeat("x", ids) })
	for n := 0; n < 3; n++ {
		if _, err := ops.OpenProbe(ctx, f.start, time.Now()); err != nil {
			t.Fatal(err)
		}
		if err := ops.Reserve(ctx, f.start, n); err != nil {
			t.Fatalf("start %d: %v", n, err)
		}
		if err := ops.Failed(ctx, f.start, n); err != nil {
			t.Fatal(err)
		}
		s.fail = false // only the first start's write fails
	}
	if files := outboxFiles(t, f.dataDir); len(files) != 1 || !strings.HasSuffix(files[0], "/probe_attempt-0-1.json") {
		t.Fatalf("outbox %v, want the first start's entry", files)
	}
	w.flushLogOutbox(ctx, s)
	got, err := moves.ReadLog(ctx, s, moves.ReportPrefix(f.rec.Account, f.rec.App, f.rec.Host, f.rec.Move.ID))
	if err != nil {
		t.Fatal(err)
	}
	var names []string
	for _, ev := range got {
		if ev.Event != "probe_attempt" {
			t.Fatalf("event %+v", ev)
		}
		names = append(names, ev.Identity)
	}
	sort.Strings(names)
	if strings.Join(names, " ") != "probe_attempt-0-1 probe_attempt-0-2 probe_attempt-0-3" {
		t.Fatalf("logged %v", names)
	}
}

// Fix round 2, ruling 3: the memo forgets a move as soon as a pass sees its
// record with no move or another move: here a source box whose move was
// closed on the target, then the same box when the next move replaced it.
// The memo is also bounded, dropping its oldest identities.
func TestLogMemoRetiresAMoveClosedElsewhere(t *testing.T) {
	ctx := context.Background()
	var got []moves.LogEvent
	w := eventWorker(&got)
	flow := doneFlow()
	w.observeMove(ctx, nil, flow[0])
	w.logTransition(ctx, nil, flow[0], flow[1])
	if w.logged.len() == 0 {
		t.Fatal("nothing remembered")
	}
	w.cfg.Enabled = true // the pass forgets only with the switch on (I1)
	s := release.NewDirStore(t.TempDir())
	closed := flow[len(flow)-1]
	if err := provision.CreateRecord(ctx, s, closed); err != nil {
		t.Fatal(err)
	}
	if err := movesPass(ctx, s, w); err != nil {
		t.Fatal(err)
	}
	if n := w.logged.len(); n != 0 {
		t.Fatalf("%d identities remembered after the pass saw the move closed elsewhere", n)
	}
	w.observeMove(ctx, nil, flow[0])
	next := mvEvRec(provision.MovePreparing, func(m *provision.MoveState) { m.ID = "mv_next" })
	s = release.NewDirStore(t.TempDir())
	if err := provision.CreateRecord(ctx, s, next); err != nil {
		t.Fatal(err)
	}
	if err := movesPass(ctx, s, w); err != nil {
		t.Fatal(err)
	}
	if w.logged.seen(next.Host + "/" + mvEvMoveID + "/requested") {
		t.Fatal("the replaced move is still remembered")
	}
}

func TestLogMemoIsBounded(t *testing.T) {
	var m moveLogMemo
	for i := 0; i < moveLogMemoCap+10; i++ {
		m.add("h", "op", "h/op/"+strconv.Itoa(i))
	}
	if m.len() != moveLogMemoCap {
		t.Fatalf("%d entries, want %d", m.len(), moveLogMemoCap)
	}
	if m.seen("h/op/0") || !m.seen("h/op/"+strconv.Itoa(moveLogMemoCap+9)) {
		t.Fatal("not the oldest dropped")
	}
	m.forget("h", "op")
	if m.len() != 0 {
		t.Fatalf("%d entries after forget", m.len())
	}
}

// Fix round 3, ruling 4: the outbox keeps the first entry queued for an
// identity. A restart during the outage forgets what was handed off, so the
// next pass sees the transition again and builds it again, later; that never
// replaces the kept bytes or their time.
func TestLogOutboxKeepsTheFirstEntry(t *testing.T) {
	ctx := context.Background()
	dataDir := t.TempDir()
	s := &flakyLogStore{Store: release.NewDirStore(t.TempDir()), fail: true}
	rec := doneFlow()[0]
	w := outboxWorker(dataDir)
	w.observeMove(ctx, s, rec)
	kept := filepath.Join(moveLogOutboxDir(dataDir), rec.Host, mvEvMoveID, "requested.json")
	first, err := os.ReadFile(kept)
	if err != nil {
		t.Fatal(err)
	}
	restarted := outboxWorker(dataDir)
	restarted.cfg.Now = func() time.Time { return time.Unix(1_800_000_900, 0).UTC() }
	restarted.flushLogOutbox(ctx, s) // still failing
	restarted.observeMove(ctx, s, rec)
	again, err := os.ReadFile(kept)
	if err != nil {
		t.Fatal(err)
	}
	if string(again) != string(first) {
		t.Fatalf("the kept entry was replaced:\n%s\n%s", first, again)
	}
	s.fail = false
	restarted.flushLogOutbox(ctx, s)
	got, err := moves.ReadLog(ctx, s, moves.ReportPrefix(rec.Account, rec.App, rec.Host, mvEvMoveID))
	if err != nil || len(got) != 1 || !got[0].At.Equal(time.Unix(1_800_000_100, 0)) {
		t.Fatalf("log %+v %v, want the first entry's time", got, err)
	}
}

// syncSpy stands in for outboxSync: it records every path synced, and
// fails the paths in fail.
type syncSpy struct {
	fail   map[string]bool
	synced map[string]int
}

func spyOutboxSync(t *testing.T) *syncSpy {
	t.Helper()
	spy := &syncSpy{fail: map[string]bool{}, synced: map[string]int{}}
	saved := outboxSync
	outboxSync = func(f *os.File) error {
		// Keyed by the absolute path, so a relative data directory's
		// syncs are named the same way.
		name := f.Name()
		if abs, err := filepath.Abs(name); err == nil {
			name = abs
		}
		if spy.fail[name] {
			return errors.New("injected sync failure")
		}
		spy.synced[name]++
		return f.Sync()
	}
	t.Cleanup(func() { outboxSync = saved })
	return spy
}

// reset forgets what was synced and fails nothing.
func (s *syncSpy) reset() {
	s.fail, s.synced = map[string]bool{}, map[string]int{}
}

// Final review I2: the log outbox acknowledges an entry only once it is
// durable. Every directory the write created is synced into its parent, a
// failed directory sync is an error, and an entry already there (an earlier
// try whose sync failed) is synced, file and directory, before it counts.
// The caller does not remember an entry the outbox could not make durable,
// so the next sighting tries again.
func TestLogOutboxIsDurableBeforeItAcknowledges(t *testing.T) {
	ctx := context.Background()
	spy := spyOutboxSync(t)
	rec := doneFlow()[0]
	ev := moves.LogEvent{V: 1, Op: mvEvMoveID, Host: rec.Host, Event: "requested", Identity: "requested"}

	t.Run("new directories are synced into their parents", func(t *testing.T) {
		spy.reset()
		dd := t.TempDir()
		if err := outboxWorker(dd).queueLog(rec, ev); err != nil {
			t.Fatal(err)
		}
		root := moveLogOutboxDir(dd)
		for _, d := range []string{dd, filepath.Join(dd, ".moves"), root, filepath.Join(root, rec.Host), filepath.Join(root, rec.Host, mvEvMoveID)} {
			if spy.synced[d] == 0 {
				t.Errorf("%s was not synced; synced %v", d, spy.synced)
			}
		}
	})

	t.Run("a failed directory sync is an error, and the retry syncs the existing entry", func(t *testing.T) {
		spy.reset()
		dd := t.TempDir()
		dir := filepath.Join(moveLogOutboxDir(dd), rec.Host, mvEvMoveID)
		spy.fail[dir] = true
		if err := outboxWorker(dd).queueLog(rec, ev); err == nil {
			t.Fatal("an entry whose directory sync failed was acknowledged")
		}
		file := filepath.Join(dir, "requested.json")
		if _, err := os.Lstat(file); err != nil {
			t.Fatalf("the fixture wants the entry linked before the failed sync: %v", err)
		}
		spy.reset()
		spy.fail[file] = true
		if err := outboxWorker(dd).queueLog(rec, ev); err == nil {
			t.Fatal("an existing entry whose file sync failed was acknowledged")
		}
		spy.reset()
		if err := outboxWorker(dd).queueLog(rec, ev); err != nil {
			t.Fatal(err)
		}
		if spy.synced[file] == 0 || spy.synced[dir] == 0 {
			t.Fatalf("an existing entry was acknowledged without syncing it and its directory: %v", spy.synced)
		}
	})

	t.Run("the caller retries an entry the outbox could not make durable", func(t *testing.T) {
		spy.reset()
		dd := t.TempDir()
		s := &flakyLogStore{Store: release.NewDirStore(t.TempDir()), fail: true}
		w := outboxWorker(dd)
		spy.fail[filepath.Join(moveLogOutboxDir(dd), rec.Host, mvEvMoveID)] = true
		w.observeMove(ctx, s, rec)
		key := rec.Host + "/" + mvEvMoveID + "/requested"
		if w.logged.seen(key) {
			t.Fatal("an entry the outbox could not make durable counts as handed off")
		}
		spy.reset()
		w.observeMove(ctx, s, rec)
		if !w.logged.seen(key) {
			t.Fatal("the next sighting did not hand the entry off")
		}
	})
}

// Final re-review 1, I2: the worker is built with its data directory made
// absolute (movesDataDir), so the outbox's ancestor walk always ends at that
// root. With CARLOS_DATA_DIR=".", an ancestor sync that fails leaves the new
// directories behind; the retry finds them there and still syncs every one
// of them, up to the data directory, before it acknowledges the entry.
func TestLogOutboxUnderARelativeDataDir(t *testing.T) {
	spy := spyOutboxSync(t)
	t.Chdir(t.TempDir())
	t.Setenv("CARLOS_DATA_DIR", ".")
	root, err := filepath.Abs(".")
	if err != nil {
		t.Fatal(err)
	}
	w := outboxWorker(movesDataDir(dataDir()))
	rec := doneFlow()[0]
	ev := moves.LogEvent{V: 1, Op: mvEvMoveID, Host: rec.Host, Event: "requested", Identity: "requested"}
	spy.fail[filepath.Join(root, ".moves")] = true
	if err := w.queueLog(rec, ev); err == nil {
		t.Fatal("an entry whose new directory's sync failed was acknowledged")
	}
	spy.reset()
	if err := w.queueLog(rec, ev); err != nil {
		t.Fatal(err)
	}
	out := moveLogOutboxDir(root)
	for _, d := range []string{root, filepath.Join(root, ".moves"), out, filepath.Join(out, rec.Host), filepath.Join(out, rec.Host, mvEvMoveID)} {
		if spy.synced[d] == 0 {
			t.Errorf("the retry acknowledged the entry without syncing %s; synced %v", d, spy.synced)
		}
	}
}