package host

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

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

// pagesWith counts the pages whose subject starts with prefix.
func pagesWith(pages pagesSeen, prefix string) (n int, level string) {
	for _, p := range pages {
		if strings.HasPrefix(p.subject, prefix) {
			n++
			level = p.level
		}
	}
	return n, level
}

// capturePagingLog records the paging code's agent log lines for the test.
func capturePagingLog(t *testing.T) *[]string {
	t.Helper()
	var lines []string
	saved := pagingLogf
	pagingLogf = func(format string, args ...any) { lines = append(lines, fmt.Sprintf(format, args...)) }
	t.Cleanup(func() { pagingLogf = saved })
	return &lines
}

// pageDelivered reports whether the page outbox under dataDir holds rule's
// entry for (host, op), delivered.
func pageDelivered(t *testing.T, dataDir, host, op, rule string) bool {
	t.Helper()
	e, err := readPageEntry(filepath.Join(pageEntryDir(dataDir, host, op), rule+".json"))
	if errors.Is(err, os.ErrNotExist) {
		return false
	}
	if err != nil {
		t.Fatal(err)
	}
	return e.Delivered != nil
}

// memoryPending counts the pages the index holds for dataDir that the
// outbox could not keep and that are not delivered yet.
func memoryPending(dataDir string) int {
	pageOutboxMu.Lock()
	defer pageOutboxMu.Unlock()
	n := 0
	for key, item := range pageIndex {
		if key.dataDir == dataDir && item.memory && !item.delivered {
			n++
		}
	}
	return n
}

// tFlush is one pass's page flush, with no record listing.
func tFlush(ctx context.Context, w *movesWorker) { w.flushPageOutbox(ctx) }

// readOnlyDir makes dir and everything under it unwritable until the test
// ends (or restore is called).
func readOnlyDir(t *testing.T, dir string) (restore func()) {
	t.Helper()
	var dirs []string
	filepath.WalkDir(dir, func(p string, d fs.DirEntry, err error) error {
		if err == nil && d.IsDir() {
			dirs = append(dirs, p)
		}
		return nil
	})
	for _, d := range dirs {
		if err := os.Chmod(d, 0o500); err != nil {
			t.Fatal(err)
		}
	}
	restore = func() {
		for _, d := range dirs {
			os.Chmod(d, 0o700)
		}
	}
	t.Cleanup(restore)
	return restore
}

// unreachableStore answers every read and listing with an error, as a bucket
// that cannot be reached.
type unreachableStore struct{ release.Store }

var errBucketDown = errors.New("the bucket cannot be reached")

func (unreachableStore) Get(context.Context, string) ([]byte, string, error) {
	return nil, "", errBucketDown
}
func (unreachableStore) List(context.Context, string) ([]string, error) { return nil, errBucketDown }
func (unreachableStore) ListDirs(context.Context, string) ([]string, error) {
	return nil, errBucketDown
}

// refusingPager is a pager that refuses while *refuse is set, recording
// what it takes into pages.
func refusingPager(pages *pagesSeen, refuse *bool) pageFunc {
	return func(ctx context.Context, level, subject, body string) error {
		if *refuse {
			return errors.New("the webhook is down")
		}
		return pages.hook()(ctx, level, subject, body)
	}
}

// TestPagingIsRateLimited (Review Focus 4; spec §13; preflight L8): a rule
// that fires every pass must not flood. On the box, stuck pages once per
// entry into stuck (a retry that sticks again pages again), rolled_back once
// per op, a stray process at the stop once per op, and a refused start check
// once per op; each rule's state is the op journal, so a restarted agent
// does not page it again. The console's own rules are tested where they
// live: rolling_back past release_deadline once an hour
// (TestConsolePagesThroughAlerting), preparing once
// (TestRecoveryPassAbortsAfterSixHours) from moves.PreparingSince
// (TestPreparingPageRunsFromPreparingSince).
func TestPagingIsRateLimited(t *testing.T) {
	ctx := context.Background()

	t.Run("stuck, once per entry", func(t *testing.T) {
		dd := t.TempDir()
		var pages pagesSeen
		worker := func() *movesWorker {
			return &movesWorker{cfg: moveCfg{DataDir: dd, Box: "uk-box", Page: pages.hook(), Now: func() time.Time { return time.Date(2026, 10, 11, 9, 0, 0, 0, time.UTC) }}}
		}
		rec := provision.Record{Host: p1Host, Move: &provision.MoveState{ID: p1ID, State: provision.MoveRollingBack,
			Rollback: &provision.MoveRollback{ID: "rb_abcdefghijklmnopqrstuvwxyz", TargetPhase: provision.RbTargetReturning}}}
		w := worker()
		for i := 0; i < 3; i++ {
			w.pageStuck(ctx, rec, "target", provision.StuckReturnTargetDatabase)
		}
		worker().pageStuck(ctx, rec, "target", provision.StuckReturnTargetDatabase) // a restarted agent
		if n, level := pagesWith(pages, "carlos move stuck: "); n != 1 || level != "incident" {
			t.Fatalf("stuck pages %d (%s), want one incident: %+v", n, level, pages)
		}
		if !pageDelivered(t, dd, p1Host, p1ID, "stuck-0") {
			t.Fatal("the outbox does not hold the stuck page delivered")
		}
		// An operator's retry advances the epoch; stuck again is a new entry.
		rec.Move.RecoveryRetryEpoch = 1
		w.pageStuck(ctx, rec, "target", provision.StuckReturnTargetDatabase)
		w.pageStuck(ctx, rec, "target", provision.StuckReturnTargetDatabase)
		if n, _ := pagesWith(pages, "carlos move stuck: "); n != 2 {
			t.Fatalf("stuck pages after a retry %d, want two: %+v", n, pages)
		}
		// One entry into stuck is one page, whichever half fires it.
		w.pageStuck(ctx, rec, "source", provision.StuckBaselineMismatch)
		if n, _ := pagesWith(pages, "carlos move stuck: "); n != 2 || !pageDelivered(t, dd, p1Host, p1ID, "stuck-1") {
			t.Fatalf("stuck pages %d, want still two, the second delivered: %+v", n, pages)
		}
	})

	t.Run("rolled_back, once per op, at the recovery's close", func(t *testing.T) {
		h := newRecoverHarness(t, false)
		h.released(t)
		h.srcPass(t)
		h.opened(t)
		h.closeOut(t)
		if rec := h.read(t, p1Host); rec.Move != nil || rec.LastMove == nil || rec.LastMove.Outcome != "rolled_back" {
			t.Fatalf("not closed: %+v %+v", rec.Move, rec.LastMove)
		}
		h.srcPass(t)
		h.sw = nil // a restarted agent
		h.srcPass(t)
		h.rollbackPass(t)
		n, level := pagesWith(h.pages, "carlos move rolled back: ")
		if n != 1 || level != "warning" {
			t.Fatalf("rolled_back pages %d (%s), want one warning: %+v", n, level, h.pages)
		}
		want := "The move " + p1ID + " of " + p1Host + " rolled back, and " + p1Host + " is serving on ie-box again."
		for _, p := range h.pages {
			if strings.HasPrefix(p.subject, "carlos move rolled back: ") && p.body != want {
				t.Fatalf("the page's body %q, want %q", p.body, want)
			}
		}
	})

	t.Run("rolled_back, once per op, at the revert's close", func(t *testing.T) {
		h := newRevertHarness(t)
		h.fromStopped(t)
		h.arm(t)
		var pages pagesSeen
		h.cfg.Page = pages.hook()
		h.drive(t)
		h.closed(t)
		h.pass(t)
		if n, level := pagesWith(pages, "carlos move rolled back: "); n != 1 || level != "warning" {
			t.Fatalf("rolled_back pages %d (%s), want one warning: %+v", n, level, pages)
		}
	})

	t.Run("a stray process at the stop, once per op, with M-C20 to operators only", func(t *testing.T) {
		logged := capturePagingLog(t)
		h := newCutHarness(t)
		h.ready(t)
		var pages pagesSeen
		sink := &memberSink{}
		h.cfg.Page, h.cfg.Events = pages.hook(), sink
		evidence := h.cfg.Evidence
		h.cfg.Evidence = func(ctx context.Context, host, location string, dbs []string) (moves.StoppedReport, error) {
			r, err := evidence(ctx, host, location, dbs)
			r.Steps[moves.StepStray] = moves.Check{Reason: "1 process(es) besides the agent in its leaf"}
			r.Stray = []int{4242}
			return r, err
		}
		h.cut(t)
		for i := 0; i < 3; i++ {
			h.pass(t)
		}
		h.freshWorker() // a restarted agent
		h.pass(t)
		if m := h.move(t); m.CutPhase != provision.CutStopping || h.ev.calls < 4 {
			t.Fatalf("the stray process did not hold the cut at stopping: %+v after %d evidence calls", m, h.ev.calls)
		}
		if n, level := pagesWith(pages, "carlos move stray process: "); n != 1 || level != "incident" {
			t.Fatalf("stray pages %d (%s), want one incident: %+v", n, level, pages)
		}
		// Fix round 1, ruling 1: M-C20 is operator copy. It is the page's
		// body and the agent's log line, and the app's members are told
		// nothing.
		mc20 := "ie-box is running a process for " + p1Host + " that this move cannot account for. An operator has been paged."
		for _, p := range pages {
			if strings.HasPrefix(p.subject, "carlos move stray process: ") && p.body != mc20 {
				t.Fatalf("the stray page's body %q, want M-C20 %q", p.body, mc20)
			}
		}
		if len(sink.events) != 0 {
			t.Fatalf("the app's members were told %q; M-C20 is the operators'", sink.events)
		}
		if n := strings.Count(strings.Join(*logged, "\n"), mc20); n != 1 {
			t.Fatalf("the agent's log says M-C20 %d times, want once: %q", n, *logged)
		}
		if !pageDelivered(t, h.cfg.DataDir, p1Host, p1ID, pageRuleStray) {
			t.Fatal("the cut's journal does not record the stray page")
		}
		// Evidence that fails for any other reason pages nothing.
		h2 := newCutHarness(t)
		h2.ready(t)
		var pages2 pagesSeen
		h2.cfg.Page = pages2.hook()
		h2.ev.stopped = false // the database is still held open: StepFDs
		h2.cut(t)
		h2.pass(t)
		h2.pass(t)
		if len(pages2) != 0 {
			t.Fatalf("an unproven stop with no stray process paged: %+v", pages2)
		}
	})

	t.Run("a refused start check, once per op", func(t *testing.T) {
		f := newPOFixture(t)
		var pages pagesSeen
		f.owners.Page = pages.hook()
		o := f.read()
		in, err := f.owners.Select(ctx, f.route, o, poProbe)
		if err != nil {
			t.Fatal(err)
		}
		admitted := f.admit(o, in)
		spawn := poWork{action: "spawn", f: f}
		f.writeLocal("mv_other", poOp, f.inc)
		for i := 0; i < 3; i++ {
			if err := f.owners.Materialize(ctx, admitted, in, spawn); !errors.Is(err, moves.ErrStartFiles) {
				t.Fatalf("files of another op: %v", err)
			}
		}
		// A restarted agent: the owners' memory is gone, the journal is not.
		f.owners.mu.Lock()
		f.owners.published, f.owners.fences = nil, nil
		f.owners.mu.Unlock()
		f.writeLocal(poOp, "src", f.inc)
		if err := f.owners.Materialize(ctx, admitted, in, spawn); err == nil {
			t.Fatal("a database of another generation started")
		}
		if n, level := pagesWith(pages, "carlos move start refused: "); n != 1 || level != "incident" {
			t.Fatalf("start check pages %d (%s), want one incident: %+v", n, level, pages)
		}
		if !pageDelivered(t, f.dataDir, poHost, poOp, pageRuleStartCheck) {
			t.Fatal("the landing's journal does not record the start check page")
		}
	})

	// Fix round 1, ruling 2: production stays unwired until Plan 5 9b, and
	// an unwired pager is never silent. Each rule says so in the agent's
	// log, loudly, once per move while the agent runs (fix round 2: the
	// worker's memory, not the journal); nothing is refused or kept. The probe owners' refused start check says what
	// became of its page, so the result is never dropped.
	t.Run("an unwired pager is loud, once per rule per move", func(t *testing.T) {
		logged := capturePagingLog(t)
		dd := t.TempDir()
		worker := func() *movesWorker { return &movesWorker{cfg: moveCfg{DataDir: dd, Box: "uk-box"}} }
		rec := provision.Record{Host: p1Host, Move: &provision.MoveState{ID: p1ID, State: provision.MoveStuck}}
		loud := func() int { return strings.Count(strings.Join(*logged, "\n"), "WARNING: no operator pager is wired") }
		w := worker()
		w.pageStuck(ctx, rec, "target", provision.StuckReturnTargetDatabase)
		w.pageStuck(ctx, rec, "target", provision.StuckReturnTargetDatabase)
		w.pageStuck(ctx, rec, "target", provision.StuckReturnTargetDatabase)
		if loud() != 1 {
			t.Fatalf("the unwired stuck page warned %d times, want once: %q", loud(), *logged)
		}
		w.pageRolledBack(ctx, rec)
		w.pageRolledBack(ctx, rec)
		if loud() != 2 {
			t.Fatalf("a second rule warned %d times in all, want 2: %q", loud(), *logged)
		}
		for _, l := range *logged {
			if strings.Contains(l, "WARNING") && (!strings.Contains(l, p1Host) || !strings.Contains(l, p1ID)) {
				t.Fatalf("the warning does not name the host and the move: %q", l)
			}
		}

		// The probe owners: the start check is refused as before, the
		// warning comes once, and every refusal's line says it was not paged.
		*logged = nil
		f := newPOFixture(t)
		o := f.read()
		in, err := f.owners.Select(ctx, f.route, o, poProbe)
		if err != nil {
			t.Fatal(err)
		}
		admitted := f.admit(o, in)
		f.writeLocal("mv_other", poOp, f.inc)
		for i := 0; i < 2; i++ {
			if err := f.owners.Materialize(ctx, admitted, in, poWork{action: "spawn", f: f}); !errors.Is(err, moves.ErrStartFiles) {
				t.Fatalf("files of another op: %v", err)
			}
		}
		if loud() != 1 {
			t.Fatalf("the unwired start check page warned %d times, want once: %q", loud(), *logged)
		}
		if n := strings.Count(strings.Join(*logged, "\n"), "refused to start "+poHost+" for "+poOp+" (not paged: no pager is wired)"); n != 2 {
			t.Fatalf("the refusals' lines %d, want 2 saying the page was not sent: %q", n, *logged)
		}
		if _, err := os.Stat(pageOutboxDir(dd)); !os.IsNotExist(err) {
			t.Fatalf("an unwired worker kept pages: %v", err)
		}
	})

	// Fix round 2, ruling 1: the page outbox. A refused page is sent again
	// by a later pass, whatever its move has done since; it is never lost
	// to the journal being parked; a double fault keeps it in memory; a
	// rule that fires again after delivery never pages again.
	t.Run("outbox: a refused stuck page is sent by a later pass", func(t *testing.T) {
		h := newRevertHarness(t)
		h.fromStopped(t)
		h.arm(t)
		var pages pagesSeen
		refuse := true
		h.cfg.Page = refusingPager(&pages, &refuse)
		if err := os.WriteFile(filepath.Join(h.cfg.DataDir, p1Host, "upload.txt"), []byte("changed while stopped"), 0o644); err != nil {
			t.Fatal(err)
		}
		h.pass(t)
		if m := h.move(t); m.State != provision.MoveStuck || m.StuckReason != provision.StuckBaselineMismatch {
			t.Fatalf("not stuck: %+v", m)
		}
		if len(pages) != 0 || pageDelivered(t, h.cfg.DataDir, p1Host, p1ID, "stuck-0") {
			t.Fatalf("a refused page counted as sent: %+v", pages)
		}
		refuse = false
		h.pass(t) // inside the backoff: not yet
		if len(pages) != 0 {
			t.Fatalf("retried inside its backoff: %+v", pages)
		}
		h.clock.Advance(2 * time.Minute)
		h.pass(t) // no new stuck event: the pass's flush sends it
		pagedOnce(t, pages, "ie-box", "source", provision.StuckBaselineMismatch)
		if !pageDelivered(t, h.cfg.DataDir, p1Host, p1ID, "stuck-0") {
			t.Fatal("the flushed page is not marked delivered")
		}
		h.clock.Advance(time.Hour)
		h.pass(t)
		h.freshWorker() // a restarted agent
		h.pass(t)
		pagedOnce(t, pages, "ie-box", "source", provision.StuckBaselineMismatch)
	})

	t.Run("outbox: a refused rolled_back page is delivered after the journal is parked", func(t *testing.T) {
		h := newRecoverHarness(t, false)
		refuse := true
		var pages pagesSeen
		h.srcPage = refusingPager(&pages, &refuse) // the source's pager
		h.released(t)
		h.srcPass(t)
		h.opened(t)
		h.closeOut(t)
		if rec := h.read(t, p1Host); rec.Move != nil || rec.LastMove == nil || rec.LastMove.Outcome != "rolled_back" {
			t.Fatalf("not closed: %+v %+v", rec.Move, rec.LastMove)
		}
		if live, parked := h.parkedJournal(t, p1ID); live || !parked {
			t.Fatalf("the source's journal of the move: live %v, parked %v; want parked", live, parked)
		}
		if n, _ := pagesWith(pages, "carlos move rolled back: "); n != 0 {
			t.Fatalf("a refused page was delivered: %+v", pages)
		}
		refuse = false
		h.clock.Advance(2 * time.Minute)
		h.srcPass(t)
		if n, level := pagesWith(pages, "carlos move rolled back: "); n != 1 || level != "warning" {
			t.Fatalf("rolled_back pages after the park %d (%s), want one warning: %+v", n, level, pages)
		}
		// Delivered, and the move retired on this box (no record carries it,
		// its journals are parked): the tombstone goes, at the next pass.
		h.srcPass(t)
		if _, err := os.Stat(pageEntryDir(h.sdir, p1Host, p1ID)); !os.IsNotExist(err) {
			t.Fatalf("the retired move's tombstone is still kept: %v", err)
		}
		h.clock.Advance(time.Hour)
		h.srcPass(t)
		if n, _ := pagesWith(pages, "carlos move rolled back: "); n != 1 {
			t.Fatalf("rolled_back paged again: %+v", pages)
		}
	})

	t.Run("outbox: the outbox write and the delivery fail, a later pass delivers from memory", func(t *testing.T) {
		logged := capturePagingLog(t)
		dd := t.TempDir()
		// The outbox cannot be made: a file stands where its directory goes.
		if err := os.MkdirAll(filepath.Join(dd, ".moves"), 0o700); err != nil {
			t.Fatal(err)
		}
		if err := os.WriteFile(pageOutboxDir(dd), []byte("in the way"), 0o600); err != nil {
			t.Fatal(err)
		}
		now := time.Date(2026, 10, 11, 9, 0, 0, 0, time.UTC)
		var pages pagesSeen
		refuse := true
		w := &movesWorker{cfg: moveCfg{DataDir: dd, Box: "uk-box", Now: func() time.Time { return now }, Page: refusingPager(&pages, &refuse)}}
		if o := w.firePage(ctx, p1Host, p1ID, "stuck-0", "incident", "carlos move stuck: "+p1Host, "b"); o != pageUndelivered {
			t.Fatalf("the double fault: %s", o)
		}
		if memoryPending(dd) != 1 || !strings.Contains(strings.Join(*logged, "\n"), "kept in memory only") {
			t.Fatalf("the page is not kept in memory, loudly: %d %q", memoryPending(dd), *logged)
		}
		refuse = false
		tFlush(ctx, w) // inside the backoff
		if len(pages) != 0 {
			t.Fatalf("retried inside its backoff: %+v", pages)
		}
		now = now.Add(2 * time.Minute)
		tFlush(ctx, w)
		w.retirePageTombstones(nil)
		if n, _ := pagesWith(pages, "carlos move stuck: "); n != 1 || memoryPending(dd) != 0 {
			t.Fatalf("the memory's page: %d delivered, %d kept; want 1 and 0", n, memoryPending(dd))
		}
		now = now.Add(time.Hour)
		tFlush(ctx, w)
		w.retirePageTombstones(nil)
		if len(pages) != 1 {
			t.Fatalf("a delivered memory page was sent again: %+v", pages)
		}
	})

	t.Run("outbox: a rule that fires again after delivery does not page again", func(t *testing.T) {
		dd := t.TempDir()
		now := time.Date(2026, 10, 11, 9, 0, 0, 0, time.UTC)
		var pages pagesSeen
		worker := func() *movesWorker {
			return &movesWorker{cfg: moveCfg{DataDir: dd, Box: "ie-box", Now: func() time.Time { return now }, Page: pages.hook()}}
		}
		rec := provision.Record{Host: p1Host, Move: &provision.MoveState{ID: p1ID, State: provision.MoveRollingBack}}
		w := worker()
		w.pageRolledBack(ctx, rec)
		if !pageDelivered(t, dd, p1Host, p1ID, pageRuleRolledBack) {
			t.Fatal("not delivered")
		}
		// The record still carries the move: the tombstone stays through
		// passes, and a firing again, here or after a restart, finds it.
		live := []provision.Record{rec}
		for i := 0; i < 3; i++ {
			now = now.Add(time.Hour)
			tFlush(ctx, w)
			w.retirePageTombstones(live)
			w.pageRolledBack(ctx, rec)
			worker().pageRolledBack(ctx, rec)
		}
		if n, _ := pagesWith(pages, "carlos move rolled back: "); n != 1 {
			t.Fatalf("rolled_back pages %d, want one: %+v", n, pages)
		}
		// A journal of the move on this box keeps it too.
		j, err := moves.ReadJournal(dd, p1Host, p1ID)
		if err != nil {
			t.Fatal(err)
		}
		j.Mark("x", "y")
		if err := j.Commit(dd); err != nil {
			t.Fatal(err)
		}
		tFlush(ctx, w)
		w.retirePageTombstones(nil)
		if !pageDelivered(t, dd, p1Host, p1ID, pageRuleRolledBack) {
			t.Fatal("the tombstone went while the move's journal is on the box")
		}
	})

	t.Run("outbox: a flush-delivered M-C20 still has its box log line", func(t *testing.T) {
		logged := capturePagingLog(t)
		h := newCutHarness(t)
		h.ready(t)
		var pages pagesSeen
		refuse := true
		h.cfg.Page = refusingPager(&pages, &refuse)
		evidence := h.cfg.Evidence
		h.cfg.Evidence = func(ctx context.Context, host, location string, dbs []string) (moves.StoppedReport, error) {
			r, err := evidence(ctx, host, location, dbs)
			r.Steps[moves.StepStray] = moves.Check{Reason: "1 process(es) besides the agent in its leaf"}
			r.Stray = []int{4242}
			return r, err
		}
		h.cut(t)
		h.pass(t)
		mc20 := "ie-box is running a process for " + p1Host + " that this move cannot account for. An operator has been paged."
		if n := strings.Count(strings.Join(*logged, "\n"), mc20); n != 1 {
			t.Fatalf("M-C20 in the agent's log %d times at the detection, want once: %q", n, *logged)
		}
		refuse = false
		h.clock.Advance(2 * time.Minute)
		h.pass(t)
		h.pass(t)
		if n, _ := pagesWith(pages, "carlos move stray process: "); n != 1 || !pageDelivered(t, h.cfg.DataDir, p1Host, p1ID, pageRuleStray) {
			t.Fatalf("the stray page after the flush: %+v", pages)
		}
		if n := strings.Count(strings.Join(*logged, "\n"), mc20); n != 1 {
			t.Fatalf("M-C20 in the agent's log %d times, want once: %q", n, *logged)
		}
	})

	// Fix round 3: the outbox's failure paths. An in-memory index over the
	// disk outbox keeps each (host, move, rule) once, through any storage
	// failure.
	t.Run("index: a delivered tombstone on read-only storage stops every re-fire", func(t *testing.T) {
		dd := t.TempDir()
		now := time.Date(2026, 10, 11, 9, 0, 0, 0, time.UTC)
		var pages pagesSeen
		w := &movesWorker{cfg: moveCfg{DataDir: dd, Box: "ie-box", Now: func() time.Time { return now }, Page: pages.hook()}}
		rec := provision.Record{Host: p1Host, Move: &provision.MoveState{ID: p1ID, State: provision.MoveStuck}}
		w.pageStuck(ctx, rec, "source", provision.StuckBaselineMismatch)
		if !pageDelivered(t, dd, p1Host, p1ID, "stuck-0") {
			t.Fatal("not delivered")
		}
		readOnlyDir(t, pageOutboxDir(dd))
		for i := 0; i < 3; i++ {
			w.pageStuck(ctx, rec, "source", provision.StuckBaselineMismatch)
			(&movesWorker{cfg: w.cfg}).pageStuck(ctx, rec, "source", provision.StuckBaselineMismatch)
		}
		if len(pages) != 1 {
			t.Fatalf("re-fires on read-only storage paged again: %d pages", len(pages))
		}
	})

	t.Run("index: a backoff the disk cannot record still holds", func(t *testing.T) {
		dd := t.TempDir()
		now := time.Date(2026, 10, 11, 9, 0, 0, 0, time.UTC)
		attempts := 0
		refuse := func(context.Context, string, string, string) error {
			attempts++
			return errors.New("the webhook is down")
		}
		w := &movesWorker{cfg: moveCfg{DataDir: dd, Box: "ie-box", Now: func() time.Time { return now }, Page: refuse}}
		rec := provision.Record{Host: p1Host, Move: &provision.MoveState{ID: p1ID, State: provision.MoveStuck}}
		w.pageStuck(ctx, rec, "source", provision.StuckBaselineMismatch) // attempt 1, next in 1 minute
		readOnlyDir(t, pageOutboxDir(dd))
		now = now.Add(2 * time.Minute)
		tFlush(ctx, w) // attempt 2: the disk cannot record its backoff (2 minutes)
		if attempts != 2 {
			t.Fatalf("attempts %d, want 2", attempts)
		}
		for i := 0; i < 3; i++ {
			now = now.Add(30 * time.Second)
			tFlush(ctx, w)
		}
		if attempts != 2 {
			t.Fatalf("retried inside a backoff the disk could not record: %d attempts", attempts)
		}
		now = now.Add(time.Minute) // past the 2-minute backoff
		tFlush(ctx, w)
		if attempts != 3 {
			t.Fatalf("attempts after the backoff %d, want 3", attempts)
		}
		// The count the disk could not record is kept too (re-review 3): the
		// third attempt's backoff is 4 minutes, not the 2 a stale disk count
		// of one would give again.
		now = now.Add(2*time.Minute + 30*time.Second)
		tFlush(ctx, w)
		if attempts != 3 {
			t.Fatalf("the third attempt's backoff restarted from the disk's stale count: %d attempts at 2m30s", attempts)
		}
		now = now.Add(2 * time.Minute) // 4m30s after the third attempt
		tFlush(ctx, w)
		if attempts != 4 {
			t.Fatalf("attempts after the 4-minute backoff %d, want 4", attempts)
		}
	})

	t.Run("index: stray fires in a double fault page once, and once the storage is back the tombstone is written", func(t *testing.T) {
		dd := t.TempDir()
		blocker := pageOutboxDir(dd)
		if err := os.MkdirAll(filepath.Dir(blocker), 0o700); err != nil {
			t.Fatal(err)
		}
		if err := os.WriteFile(blocker, []byte("in the way"), 0o600); err != nil {
			t.Fatal(err)
		}
		now := time.Date(2026, 10, 11, 9, 0, 0, 0, time.UTC)
		var pages pagesSeen
		refuse := true
		w := &movesWorker{cfg: moveCfg{DataDir: dd, Box: "ie-box", Now: func() time.Time { return now }, Page: refusingPager(&pages, &refuse)}}
		rec := provision.Record{Host: p1Host, Move: &provision.MoveState{ID: p1ID, State: provision.MoveCutting}}
		for i := 0; i < 3; i++ {
			w.pageStray(ctx, rec)
		}
		refuse = false
		for i := 0; i < 3; i++ {
			w.pageStray(ctx, rec) // re-fires while the page waits in memory
		}
		now = now.Add(2 * time.Minute)
		tFlush(ctx, w)
		tFlush(ctx, w)
		w.pageStray(ctx, rec)
		if n, _ := pagesWith(pages, "carlos move stray process: "); n != 1 {
			t.Fatalf("stray pages during the double fault %d, want one: %+v", n, pages)
		}
		// The storage is back: the next pass writes the tombstone, and a
		// re-fire, here or after a restart, does not page.
		if err := os.Remove(blocker); err != nil {
			t.Fatal(err)
		}
		tFlush(ctx, w)
		if !pageDelivered(t, dd, p1Host, p1ID, pageRuleStray) {
			t.Fatal("the delivered page's tombstone was not written once the storage came back")
		}
		w.pageStray(ctx, rec)
		(&movesWorker{cfg: w.cfg}).pageStray(ctx, rec)
		tFlush(ctx, w)
		if n, _ := pagesWith(pages, "carlos move stray process: "); n != 1 {
			t.Fatalf("stray pages after the storage came back %d, want still one: %+v", n, pages)
		}
	})

	t.Run("index: a pending page is delivered while records and the fleet cannot be read", func(t *testing.T) {
		dd := t.TempDir()
		now := time.Date(2026, 10, 11, 9, 0, 0, 0, time.UTC)
		var pages pagesSeen
		refuse := true
		w := &movesWorker{cfg: moveCfg{DataDir: dd, Box: "ie-box", Enabled: true, Now: func() time.Time { return now }, Page: refusingPager(&pages, &refuse)},
			mirrors: map[string]lease.MirrorStore{}}
		rec := provision.Record{Host: p1Host, Move: &provision.MoveState{ID: p1ID, State: provision.MoveStuck}}
		w.pageStuck(ctx, rec, "source", provision.StuckBaselineMismatch)
		refuse = false
		now = now.Add(2 * time.Minute)
		if err := movesPass(ctx, unreachableStore{}, w); err == nil {
			t.Fatal("a pass over an unreachable bucket succeeded")
		}
		if n, _ := pagesWith(pages, "carlos move stuck: "); n != 1 || !pageDelivered(t, dd, p1Host, p1ID, "stuck-0") {
			t.Fatalf("the pending page with the bucket down: %+v", pages)
		}
	})
}

// outboxTree lists every file under dataDir's .moves directory, with its
// bytes, so a test can tell whether a pass changed anything there.
func outboxTree(t *testing.T, dataDir string) map[string]string {
	t.Helper()
	out := map[string]string{}
	root := filepath.Join(dataDir, ".moves")
	err := filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error {
		if errors.Is(err, fs.ErrNotExist) && p == root {
			return fs.SkipAll
		}
		if err != nil || d.IsDir() {
			return err
		}
		b, err := os.ReadFile(p)
		if err != nil {
			return err
		}
		rel, _ := filepath.Rel(root, p)
		out[filepath.ToSlash(rel)] = string(b)
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
	return out
}

// Final review I1: with the switch off, a pass leaves both outboxes alone.
// The pending page, the delivered page's tombstone and the kept log entry
// stay on disk untouched, and nothing is sent or written, until the switch
// is on again; then the same pass delivers them.
func TestSwitchOffPassLeavesTheOutboxes(t *testing.T) {
	ctx := context.Background()
	dd := t.TempDir()
	now := time.Date(2026, 10, 11, 9, 0, 0, 0, time.UTC)
	var pages pagesSeen
	refuse := true
	var logged []moves.LogEvent
	logFails := true
	w := &movesWorker{cfg: moveCfg{DataDir: dd, Box: "ie-box", Enabled: true, Now: func() time.Time { return now }, Page: refusingPager(&pages, &refuse),
		LogEvent: func(_ context.Context, _ release.Store, _ provision.Record, ev moves.LogEvent) error {
			if logFails {
				return errors.New("the bucket is down")
			}
			logged = append(logged, ev)
			return nil
		}}, mirrors: map[string]lease.MirrorStore{}}
	rec := provision.Record{Account: "bab", App: "shop", Host: p1Host, Move: &provision.MoveState{ID: p1ID, State: provision.MoveStuck, From: "ie", To: "uk"}}
	// A page the pager refused: pending on disk.
	w.pageStuck(ctx, rec, "source", provision.StuckBaselineMismatch)
	// A delivered page of a move no record carries any more: a tombstone a
	// pass with the switch on would retire.
	refuse = false
	other := provision.Record{Host: p1Host, Move: &provision.MoveState{ID: "mv_zyxwvutsrqponmlkjihgfedcba"}}
	w.pageRolledBack(ctx, other)
	pages = nil
	// A log entry whose write failed: kept in the log outbox.
	w.observeMove(ctx, nil, provision.Record{Account: "bab", App: "shop", Host: p1Host, Region: "ie",
		Move: &provision.MoveState{ID: p1ID, State: provision.MovePreparing, From: "ie", To: "uk"}})
	before := outboxTree(t, dd)
	if len(before) != 3 {
		t.Fatalf("the fixture's outboxes %v, want a pending page, a tombstone and a kept log entry", before)
	}
	logFails = false
	now = now.Add(time.Hour) // every backoff is over

	s := release.NewDirStore(t.TempDir())
	w.cfg.Enabled = false
	if err := movesPass(ctx, s, w); err != nil {
		t.Fatal(err)
	}
	if len(pages) != 0 || len(logged) != 0 {
		t.Fatalf("a pass with the switch off sent pages %+v and wrote log entries %+v", pages, logged)
	}
	if after := outboxTree(t, dd); fmt.Sprint(after) != fmt.Sprint(before) {
		t.Fatalf("a pass with the switch off changed the outboxes:\nbefore %v\nafter  %v", before, after)
	}

	w.cfg.Enabled = true
	if err := movesPass(ctx, s, w); err != nil {
		t.Fatal(err)
	}
	if n, _ := pagesWith(pages, "carlos move stuck: "); n != 1 || len(logged) != 1 || logged[0].Identity != "requested" {
		t.Fatalf("with the switch on again: pages %+v, log entries %+v", pages, logged)
	}
}

// Final review I2: the page outbox's replace and its delivered mark are
// durable before they count. A failed directory sync after the replace is
// an error, and a delivered entry already on disk is synced before the mark
// is acknowledged.
func TestPageOutboxIsDurableBeforeItAcknowledges(t *testing.T) {
	spy := spyOutboxSync(t)
	dd := t.TempDir()
	now := time.Date(2026, 10, 11, 9, 0, 0, 0, time.UTC)
	e := pageEntry{V: 1, Host: p1Host, Op: p1ID, Rule: "stuck-0", Level: "incident", Subject: "s", Body: "b", At: now, Tries: 1, Next: now.Add(time.Minute)}
	if err := writePageEntryFixture(dd, e); err != nil {
		t.Fatal(err)
	}
	dir := pageEntryDir(dd, p1Host, p1ID)
	file := pageEntryPath(dd, p1Host, p1ID, "stuck-0")

	spy.fail[dir] = true
	e.Tries = 2
	if err := writePageEntry(dd, e); err == nil {
		t.Fatal("a replace whose directory sync failed was acknowledged")
	}

	spy.reset()
	if err := markPageDelivered(dd, e, now); err != nil {
		t.Fatal(err)
	}
	spy.reset()
	spy.fail[file] = true
	if err := markPageDelivered(dd, e, now); err == nil {
		t.Fatal("a delivered entry whose sync failed was acknowledged")
	}
	spy.reset()
	if err := markPageDelivered(dd, e, now); err != nil {
		t.Fatal(err)
	}
	if spy.synced[file] == 0 || spy.synced[dir] == 0 {
		t.Fatalf("a delivered entry was acknowledged without syncing it and its directory: %v", spy.synced)
	}
}

// writePageEntryFixture writes a page entry with no sync spy in the way.
func writePageEntryFixture(dataDir string, e pageEntry) error {
	b, err := json.Marshal(e)
	if err != nil {
		return err
	}
	dir := pageEntryDir(dataDir, e.Host, e.Op)
	if err := os.MkdirAll(dir, 0o700); err != nil {
		return err
	}
	return os.WriteFile(filepath.Join(dir, e.Rule+".json"), b, 0o600)
}

// Final re-review 1, I1: with the switch off a pass touches no outbox, index
// or memo state. Here the move log's memory remembers two moves; the store
// then holds one host's record closed and another's replaced by a new move,
// which a pass with the switch on would forget. Off, the memory is exactly
// what it was.
func TestSwitchOffPassLeavesTheLogMemo(t *testing.T) {
	ctx := context.Background()
	var got []moves.LogEvent
	w := eventWorker(&got)
	flow := doneFlow()
	w.observeMove(ctx, nil, flow[0]) // acme.go.tito.io, mvEvMoveID: requested
	other := mvEvRec(provision.MovePreparing, func(m *provision.MoveState) { m.ID = "mv_otherotherotherotherother" })
	other.Host = "b.example.com"
	w.observeMove(ctx, nil, other)
	keys := []string{flow[0].Host + "/" + mvEvMoveID + "/requested", other.Host + "/mv_otherotherotherotherother/requested"}
	before := w.logged.len()
	if before != 2 || !w.logged.seen(keys[0]) || !w.logged.seen(keys[1]) {
		t.Fatalf("the fixture's memory: %d", before)
	}
	s := release.NewDirStore(t.TempDir())
	if err := provision.CreateRecord(ctx, s, flow[len(flow)-1]); err != nil { // closed
		t.Fatal(err)
	}
	replaced := mvEvRec(provision.MovePreparing, func(m *provision.MoveState) { m.ID = "mv_nextnextnextnextnextnextne" })
	replaced.Host = other.Host
	if err := provision.CreateRecord(ctx, s, replaced); err != nil {
		t.Fatal(err)
	}
	w.cfg.Enabled = false
	if err := movesPass(ctx, s, w); err != nil {
		t.Fatal(err)
	}
	if n := w.logged.len(); n != before || !w.logged.seen(keys[0]) || !w.logged.seen(keys[1]) {
		t.Fatalf("a pass with the switch off changed the move log's memory: %d identities, want %d", n, before)
	}
	w.cfg.Enabled = true
	if err := movesPass(ctx, s, w); err != nil {
		t.Fatal(err)
	}
	if w.logged.seen(keys[0]) || w.logged.seen(keys[1]) {
		t.Fatal("with the switch on the closed and replaced moves are still remembered")
	}
}