package rollout

import (
	"context"
	"slices"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/appconfig"
)

// N6, Astra's sequence: A's good write was decided before hold B ran on the
// host; B's marker is newer, and A's write must not clear it.
func TestAnOlderGoodWriteNeverClearsANewerMarker(t *testing.T) {
	b := &FailureBook{}
	snapA := b.Snapshot()
	if err := b.MarkHoldRan("h", HoldRan{Account: "bab", App: "blog", Build: "v2"}); err != nil {
		t.Fatal(err)
	}
	if cleared, _ := b.ClearHoldRanBefore("h", snapA); cleared {
		t.Fatal("a write decided before the marker cleared it")
	}
	if cleared, _ := b.ClearHoldRanBefore("h", b.Snapshot()); !cleared {
		t.Fatal("a write decided after the marker must clear it")
	}
}

func TestAZeroSnapshotClearsNothing(t *testing.T) {
	b := &FailureBook{}
	_ = b.MarkHoldRan("h", HoldRan{Account: "bab", App: "blog", Build: "v2"})
	if cleared, _ := b.ClearHoldRanBefore("h", 0); cleared {
		t.Fatal("no snapshot must clear nothing")
	}
}

// Plan review finding 4: a marker belongs to one execution, not to its
// build. B's marker (execution a1) stands before C's snapshot; B then runs
// again (a2) while C's good write is delayed. The second execution renews
// the marker's sequence, so C's write clears nothing. A repeated report of
// one execution changes nothing.
func TestAnotherExecutionOfTheMarkedBuildRenewsTheMarker(t *testing.T) {
	now := t0
	b := &FailureBook{now: func() time.Time { return now }}
	if err := b.MarkHoldRan("h", HoldRan{Account: "bab", App: "blog", Build: "v2", Exec: "a1", At: t0}); err != nil {
		t.Fatal(err)
	}
	snapC := b.Snapshot()
	second := t0.Add(time.Minute)
	if err := b.MarkHoldRan("h", HoldRan{Account: "bab", App: "blog", Build: "v2", Exec: "a2", At: second}); err != nil {
		t.Fatal(err)
	}
	got, _ := b.HoldRanOn("h")
	if got.Exec != "a2" || got.Seq < snapC || !got.At.Equal(second) {
		t.Fatalf("marker %+v after a second execution of v2: want a2's, sequence at least %d, stamped %v", got, snapC, second)
	}
	if cleared, _ := b.ClearHoldRanBefore("h", snapC); cleared {
		t.Fatal("C's good write cleared the marker of an execution that began after C's snapshot")
	}
	// The same execution reported again keeps its sequence and its stamp.
	_ = b.MarkHoldRan("h", HoldRan{Account: "bab", App: "blog", Build: "v2", Exec: "a2", At: second.Add(time.Minute)})
	if again, _ := b.HoldRanOn("h"); again != got {
		t.Fatalf("marker %+v after a repeated report of a2, want it unchanged at %+v", again, got)
	}
	// A late report of an earlier execution of the same build never moves
	// the stamp back: a rollback signed between the two runs predates a2.
	_ = b.MarkHoldRan("h", HoldRan{Account: "bab", App: "blog", Build: "v2", Exec: "a0", At: t0.Add(-time.Minute)})
	late, _ := b.HoldRanOn("h")
	if !late.At.Equal(second) || late.Seq <= got.Seq {
		t.Fatalf("marker %+v after a late report of an earlier run: want the stamp kept at %v and a newer sequence", late, second)
	}
	// The same for another build's older execution (Task 4 fix round 1):
	// the marker names the newest report's build, never an older stamp.
	_ = b.MarkHoldRan("h", HoldRan{Account: "bab", App: "blog", Build: "v3", Exec: "a9", At: t0})
	if other, _ := b.HoldRanOn("h"); other.Build != "v3" || !other.At.Equal(second) || other.Seq <= late.Seq {
		t.Fatalf("marker %+v after a late report of another build's earlier run: want v3's, stamped %v, newer sequence", other, second)
	}
}

// Retry ordering: a failure of an attempt that began before the host's retry
// never takes the retry back.
func TestAPreRetryFailureLeavesTheRetryStanding(t *testing.T) {
	b := &FailureBook{}
	before := b.Snapshot()
	b.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "s", Seq: before})
	if ok, _ := b.RetryHost("bab", "blog", "v2", "h"); !ok {
		t.Fatal("retry")
	}
	if res, _ := b.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "h", Seq: before, HeldDown: []string{"h"}}); res != PreRetry {
		t.Fatalf("Record = %v, want PreRetry", res)
	}
	if b.Blocks("bab", "blog", "v2", "h", false) || b.HeldDown("bab", "blog", "v2", "h") {
		t.Fatal("a pre-retry failure undid the retry")
	}
	b.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "h", Seq: b.Snapshot()})
	if !b.Blocks("bab", "blog", "v2", "h", false) {
		t.Fatal("a failure after the retry must stand")
	}
}

func TestAPreRetryFailureNeverRecreatesAClearedRecord(t *testing.T) {
	b := &FailureBook{}
	before := b.Snapshot()
	b.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "s", Seq: before})
	if ok, _ := b.ClearOnPointer("bab", "blog", "v2", time.Now().Add(time.Hour)); !ok {
		t.Fatal("clear")
	}
	b.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "t", Seq: before, MarkHoldRan: true, Mode: "hold"})
	if _, held := b.Lookup("bab", "blog", "v2"); held {
		t.Fatal("a pre-retry failure recreated the record")
	}
	if _, marked := b.HoldRanOn("t"); !marked {
		t.Fatal("the marker is a fact about t's database and must still be set")
	}
}

// A pre-retry hold-down is a hold build that ran against the instance's
// database, like a requested marker: the retry keeps the hold-down off, and
// the instance keeps the marker (no automatic fallback onto it, spec §6.4;
// Task 4 fix round 1), with the execution's identity and start.
func TestAPreRetryHoldDownLeavesTheMarker(t *testing.T) {
	b := &FailureBook{}
	before := b.Snapshot()
	b.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "s", Seq: before})
	b.RetryHost("bab", "blog", "v2", "h")
	started := t0.Add(-time.Minute)
	b.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "h", Seq: before, Mode: "hold", HeldDown: []string{"h"},
		HoldRanAt: started, HoldRanExec: "a7"})
	if b.HeldDown("bab", "blog", "v2", "h") {
		t.Fatal("a pre-retry failure held the retried host down")
	}
	got, ok := b.HoldRanOn("h")
	if !ok || got.Build != "v2" || got.Exec != "a7" || !got.At.Equal(started) {
		t.Fatalf("marker %+v, %v: want v2's from execution a7, stamped %v", got, ok, started)
	}
}

// Task 4 fix round 1, the reviewer's stranding sequence: a retry lets b's
// canary through the halt; a second retry arrives while that canary runs;
// the canary fails. Its failure began after the first retry, which is the
// one it spends: it is recorded and b is held again, and a later retry runs
// another attempt. A second retry of a host still retried notes no clear,
// or the canary's failure would be stale, b would stay retried behind the
// halt with its allowance spent, and no retry could free it.
func TestASecondRetryDuringTheCanaryNeverStrandsTheHost(t *testing.T) {
	h := newHarness(t, "a", "b")
	h.stage("v2")
	h.failing.Store(true)
	h.act.outcome = func(a activator.Attempt) activator.SwapOutcome {
		if a.Build == "v2" && h.failing.Load() {
			return activator.SwapFailed
		}
		return activator.SwapDone
	}
	h.install("v2", "a", "b")
	h.pass()
	waitFor(t, "a failure record", func() bool { _, ok := h.book.Lookup("bab", "blog", "v2"); return ok })
	gate := h.act.hold("b")
	if !h.retry("b", "v2") {
		t.Fatal("the first retry cleared nothing")
	}
	waitFor(t, "b's retry canary", func() bool {
		_, started, _ := h.act.snapshot()
		return slices.Contains(started, "b:v2")
	})
	h.retry("b", "v2") // the second retry, while the canary runs
	close(gate)
	waitFor(t, "b's canary to fail", func() bool { return !h.m.attemptRunning("b") && h.book.Blocks("bab", "blog", "v2", "b", false) })

	h.failing.Store(false)
	if !h.retry("b", "v2") {
		t.Fatal("a retry after the canary's failure cleared nothing: b is stranded")
	}
	waitFor(t, "b on v2", func() bool { return h.servingAll("v2", "b") })
	_, started, _ := h.act.snapshot()
	n := 0
	for _, st := range started {
		if st == "b:v2" {
			n++
		}
	}
	if n != 2 {
		t.Fatalf("attempts %v: want b's canary and the attempt after the third retry", started)
	}
}

// Task 4 fix round 1: a planned wake's snapshot is taken at the plan's end.
// A retry that lands during the planner's slow reads (here the policy load)
// is what let the plan boot the retried build, so the build's failure in
// that wake is a real one: recorded, and in hold mode a hold-down.
func TestARetryDuringThePlanLeavesTheWakesFailureStanding(t *testing.T) {
	ctx := context.Background()
	h := newHarness(t, "w")
	h.swap = appconfig.SwapHold
	h.stage("v2")
	h.install("v2", "w")
	h.sleepHost("w")
	h.pass()
	h.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "s", PromotedAt: t0, At: t0})
	h.wakeHost("w")
	h.advance(time.Minute) // the policy cache is stale: the plan loads it
	var once sync.Once
	h.mu.Lock()
	h.policyHook = func() {
		once.Do(func() {
			if !h.retry("w", "v2") {
				t.Error("the retry during the plan cleared nothing")
			}
		})
	}
	h.mu.Unlock()
	p := h.m.Plan(ctx, h.rows.get("w"))
	if !p.FirstWake || p.Boot.Version != "v2" || p.Mode != activator.ModeHold {
		t.Fatalf("plan %+v: want a hold first wake of the retried v2", p)
	}
	h.m.OnGen(activator.GenEvent{Kind: activator.GenFailed, Route: h.rows.get("w"), Failed: p.Boot, Mode: p.Mode,
		FirstWake: true, Reason: "health check failed", Token: p.Token, ConfigRev: p.ConfigRev})
	if !h.book.Blocks("bab", "blog", "v2", "w", false) || !h.book.HeldDown("bab", "blog", "v2", "w") {
		t.Fatal("the failure of a wake the retry let through was taken as pre-retry")
	}
}

// What Record reports, which failCohort acts on (plan review finding 3).
func TestRecordReportsWhatItDid(t *testing.T) {
	b := &FailureBook{}
	if res, _ := b.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "a"}); res != Recorded {
		t.Fatalf("first failure: %v, want Recorded", res)
	}
	if res, _ := b.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "b", HeldDown: []string{"b"}}); res != Merged {
		t.Fatalf("a new hold-down: %v, want Merged", res)
	}
	if res, _ := b.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "b", HeldDown: []string{"b"}}); res != Unchanged {
		t.Fatalf("the same failure again: %v, want Unchanged", res)
	}
	if res, _ := b.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "c", Seq: b.Snapshot()}); res == PreRetry {
		t.Fatal("a failure with no retry before it was taken as pre-retry")
	}
}

// A retry clear goes with its build's record when no route installs the
// build: Prune drops both.
func TestPruneDropsTheRetryClearsOfBuildsNoRouteInstalls(t *testing.T) {
	b := &FailureBook{}
	before := b.Snapshot()
	b.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "s"})
	b.RetryHost("bab", "blog", "v2", "h")
	if err := b.Prune(func(string, string, string) bool { return false }); err != nil {
		t.Fatal(err)
	}
	if len(b.cleared) != 0 {
		t.Fatalf("clears %v survived the prune of their build", b.cleared)
	}
	if res, _ := b.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "h", Seq: before}); res != Recorded {
		t.Fatalf("Record = %v after the prune, want Recorded", res)
	}
}

func TestTheSequenceSurvivesAReopen(t *testing.T) {
	dd := t.TempDir()
	b, _ := OpenFailureBook(dd)
	_ = b.MarkHoldRan("h", HoldRan{Account: "bab", App: "blog", Build: "v2"})
	b2, err := OpenFailureBook(dd)
	if err != nil {
		t.Fatal(err)
	}
	if cleared, _ := b2.ClearHoldRanBefore("h", b2.Snapshot()); !cleared {
		t.Fatal("after a reopen a fresh snapshot must exceed the old marker")
	}
}

// N6 through the Manager: h's first wake of v1 is planned (its snapshot
// taken) and is observed; meanwhile a scheduled hold attempt onto v2 runs
// on h and is superseded, which marks h. v1's settle then arrives and
// writes v1 as h's good build: the write was decided before v2 ran, so v2's
// marker stands.
func TestASettleDecidedBeforeAHoldRanKeepsItsMarker(t *testing.T) {
	ctx := context.Background()
	h := newHarness(t, "h")
	r := h.rows.get("h")
	r.GoodVersion, r.GoodDigest = "", "" // v1 has not passed here yet
	h.rows.put(r)
	h.swap = appconfig.SwapHold
	h.stage("v2")
	h.act.mode = activator.ModeHold
	h.act.executed = func(a activator.Attempt, _ activator.SwapOutcome) bool { return a.Build == "v2" }

	p := h.m.Plan(ctx, h.rows.get("h"))
	if !p.FirstWake || p.Boot.Version != "v1" {
		t.Fatalf("plan %+v: want a first wake of v1", p)
	}
	g, _ := h.act.Current("h")
	h.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: h.rows.get("h"), Gen: g, Mode: p.Mode, FirstWake: true, Token: p.Token})
	// Since Task 5 the wake's intent, which stands until its settle, refuses
	// an adoption of h (ErrIntentStands), so this ordering needs the intent
	// gone. It is removed by hand: the marker rule must hold whatever lets a
	// hold run between a plan and its settle.
	if err := h.book.End("h", h.m.wakeExecID(p.Token)); err != nil {
		t.Fatal(err)
	}

	gate := h.act.hold("h")
	h.install("v2", "h")
	h.pass()
	waitFor(t, "h's hold attempt onto v2", func() bool {
		_, started, _ := h.act.snapshot()
		return slices.Contains(started, "h:v2")
	})
	h.install("v3", "h") // v3 is not staged: its attempt is refused, unrun
	h.pass()
	close(gate)
	waitFor(t, "v2's marker", func() bool { m, ok := h.book.HoldRanOn("h"); return ok && m.Build == "v2" })

	h.m.OnGen(activator.GenEvent{Kind: activator.GenSettled, Route: h.rows.get("h"), Gen: g, Mode: p.Mode, FirstWake: true, Token: p.Token})
	if r := h.rows.get("h"); r.GoodVersion != "v1" {
		t.Fatalf("good build %q after v1's settle, want v1", r.GoodVersion)
	}
	if m, ok := h.book.HoldRanOn("h"); !ok || m.Build != "v2" {
		t.Fatalf("marker %+v, %v after v1's settle: want v2's, set after v1's wake was planned", m, ok)
	}
}

// Plan review finding 3: an outcome that began before a retry is stale, and
// none of the failure transition runs for it. After `carlos instances retry`
// of the failing host, or a `carlos promote --retry` pointer that cleared
// the record, a delayed failure of an attempt or of a planned wake that
// began before it leaves the cohort unhalted, supersedes no running attempt
// (y's), and leaves the retried host free: its queued adoption is admitted,
// or its next wake plans the retried build.
func TestAPreRetryOutcomeRunsNoFailureTransition(t *testing.T) {
	for _, retry := range []string{"host", "pointer"} {
		for _, kind := range []string{"attempt", "wake"} {
			t.Run(retry+" retry, "+kind, func(t *testing.T) {
				preRetryOutcome(t, retry, kind)
			})
		}
	}
}

func preRetryOutcome(t *testing.T, retry, kind string) {
	ctx := context.Background()
	k := appKey{"bab", "blog"}
	h := newHarness(t, "h", "y", "w")
	h.concurrency = 2
	h.swap = appconfig.SwapHold
	h.stage("v2")
	h.install("v2", "h", "y", "w")
	s := h.m.scheduler(k)
	s.mu.Lock()
	// The cohort's canary passed on another host already, so h and y run
	// side by side.
	*s.q.cohortFor(cohortKey("v2", classAdopt)) = cohort{canary: "x", settled: true}
	s.mu.Unlock()
	gateH, gateY := h.act.hold("h"), h.act.hold("y")
	h.sleepHost("w")
	h.pass()
	waitFor(t, "h's and y's attempts onto v2", func() bool {
		_, started, _ := h.act.snapshot()
		return slices.Contains(started, "h:v2") && slices.Contains(started, "y:v2")
	})
	host := "h"
	var plan activator.WakePlan
	if kind == "wake" {
		host = "w"
		h.wakeHost("w")
		plan = h.m.Plan(ctx, h.rows.get("w"))
		if !plan.FirstWake || plan.Boot.Version != "v2" {
			t.Fatalf("w's plan %+v: want a first wake of v2", plan)
		}
	}

	// A sibling's failure of v2, written straight into the book: no halt
	// stands, so a halt after this point is the stale outcome's.
	h.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "s", PromotedAt: t0, At: t0})
	switch retry {
	case "host":
		if !h.retry(host, "v2") {
			t.Fatalf("RetryHost(%s) cleared nothing", host)
		}
	case "pointer":
		h.advance(time.Minute)
		h.pointer("stable", "v2", h.clock())
		h.pass()
		if _, held := h.book.Lookup("bab", "blog", "v2"); held {
			t.Fatal("the retry pointer left v2's record standing")
		}
	}
	s.mu.Lock()
	revY := s.q.newest("y")
	s.mu.Unlock()

	// The outcome that began before the retry arrives now.
	if kind == "attempt" {
		h.act.mu.Lock()
		h.act.outcome = func(a activator.Attempt) activator.SwapOutcome {
			if a.Host == "h" && a.Cause != activator.CauseRetry {
				return activator.SwapFailed
			}
			return activator.SwapDone
		}
		h.act.mu.Unlock()
		gateH2 := h.act.hold("h") // h's next attempt parks in Swap
		defer close(gateH2)
		close(gateH)
		waitFor(t, "h's attempt after the stale failure", func() bool {
			calls, started, _ := h.act.snapshot()
			n := 0
			for _, st := range started {
				if st == "h:v2" {
					n++
				}
			}
			return len(calls) >= 1 && n == 2
		})
	} else {
		defer close(gateH)
		h.m.OnGen(activator.GenEvent{Kind: activator.GenFailed, Route: h.rows.get("w"), Failed: plan.Boot, Mode: plan.Mode,
			FirstWake: true, Reason: "health check failed", Token: plan.Token, ConfigRev: plan.ConfigRev})
	}
	defer close(gateY)

	s.mu.Lock()
	halted := s.q.cohortFor(cohortKey("v2", classAdopt)).halted
	y := s.q.hosts["y"]
	yRunning, yRev, yNewest := y.running != nil, uint64(0), y.buildRev
	if y.running != nil {
		yRev = y.running.SchedRev
	}
	s.mu.Unlock()
	if halted {
		t.Fatal("a stale outcome halted the cohort")
	}
	if !yRunning || yRev != revY || yNewest != revY {
		t.Fatalf("y's attempt: running %v, rev %d, newest %d; want it running unsuperseded at %d", yRunning, yRev, yNewest, revY)
	}
	if h.book.Blocks("bab", "blog", "v2", host, false) || h.book.HeldDown("bab", "blog", "v2", host) {
		t.Fatalf("the stale outcome took %s's retry back", host)
	}
	if retry == "pointer" {
		if _, held := h.book.Lookup("bab", "blog", "v2"); held {
			t.Fatal("the stale outcome recreated the cleared record")
		}
	}
	if kind == "attempt" {
		calls, _, _ := h.act.snapshot()
		if c := calls[0].a; c.Host != "h" || calls[0].outcome != activator.SwapFailed {
			t.Fatalf("first finished call %+v: want h's stale failure", calls[0])
		}
		return
	}
	h.wakeHost("w")
	if p := h.m.Plan(ctx, h.rows.get("w")); p.Refuse != "" || p.Boot.Version != "v2" || !p.FirstWake {
		t.Fatalf("w's next plan %+v: want the retried v2", p)
	}
}