package rollout

import (
	"context"
	"slices"
	"strings"
	"testing"
	"time"

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

// restartOn is an agent restart under the harness: a fresh Manager over
// book, its counters (attempt numbers, wake tokens) starting again from
// zero, as a new process's do. The old manager's schedulers stop with ctx.
func (h *harness) restartOn(book *FailureBook) {
	h.t.Helper()
	book.now = h.clock
	h.book = book
	ctx, cancel := context.WithCancel(context.Background())
	h.t.Cleanup(cancel)
	h.m = NewManager(ctx, Deps{
		Act: h.probe, Routes: h.rows, Book: book, OptDir: h.opt, DataDir: h.data, Box: "box-1",
		Demand: h.demand, Policy: h.policy, Pointer: h.pointerFor, Events: h.events,
		Touch: func() { h.touched.Add(1) }, Now: h.clock, Logf: func(string, ...any) {},
	})
}

// cancelHoldWake plans host's first wake (a hold build) and ends its
// observation cancelled after the build ran, which leaves the host's
// hold-ran marker and no failure record.
func (h *harness) cancelHoldWake(host, build string) activator.WakePlan {
	h.t.Helper()
	h.wakeHost(host)
	p := h.m.Plan(context.Background(), h.rows.get(host))
	if !p.FirstWake || p.Boot.Version != build || p.Mode != activator.ModeHold {
		h.t.Fatalf("plan %+v: want a hold first wake of %s", p, build)
	}
	g := activator.Gen{N: 7, Build: build, StagedPath: p.Boot.StagedPath, Digest: p.Boot.Digest}
	h.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: h.rows.get(host), Gen: g, Mode: p.Mode, FirstWake: true, Token: p.Token})
	h.m.OnGen(activator.GenEvent{Kind: activator.GenCancelled, Route: h.rows.get(host), Gen: g, Mode: p.Mode, FirstWake: true,
		Token: p.Token, Executed: true})
	return p
}

// Final review round 1, finding 1 (Astra): execution identities are unique
// across agent restarts. B runs on h and leaves a marker with no failure
// record; the agent restarts (a new Manager, its counters at zero again,
// over the same book on disk) and its first wake runs B again. The second
// execution is a new one: the marker gets a fresh Seq and At, so the retry
// grant the console signed for the first execution, and a rollback pointer
// signed between the two executions, clear nothing of it.
func TestAnExecutionAfterARestartIsANewExecution(t *testing.T) {
	dir := t.TempDir()
	h := newHarness(t, "s")
	book, err := OpenFailureBook(dir)
	if err != nil {
		t.Fatal(err)
	}
	h.restartOn(book)
	h.swap = appconfig.SwapHold
	h.stage("v2")
	h.install("v2", "s")
	h.sleepHost("s")
	h.pass()
	h.cancelHoldWake("s", "v2")
	first, ok := h.book.HoldRanOn("s")
	if !ok || first.Build != "v2" || !first.At.Equal(t0) {
		t.Fatalf("first marker %+v, %v: want v2's at %v", first, ok, t0)
	}
	staleGrant := h.book.RestrictionEpoch("bab", "blog", "v2", "s")

	// The agent restarts: the book reopens from disk, the manager's
	// counters start again.
	reopened, err := OpenFailureBook(dir)
	if err != nil {
		t.Fatal(err)
	}
	h.restartOn(reopened)
	h.advance(time.Minute)
	second := h.clock()
	h.pass()
	h.cancelHoldWake("s", "v2")
	got, ok := h.book.HoldRanOn("s")
	if !ok || got.Build != "v2" {
		t.Fatalf("marker after the second execution %+v, %v", got, ok)
	}
	if got.Exec == first.Exec {
		t.Fatalf("the second execution reused the first's identity %q", got.Exec)
	}
	if got.Seq <= first.Seq || !got.At.Equal(second) {
		t.Fatalf("marker %+v after the second execution: want a fresh Seq past %d and At %v", got, first.Seq, second)
	}

	// The first execution's grant and a rollback signed between the two runs
	// clear nothing of the second's marker.
	if rec, mark, _, _ := h.book.RetryHostAt("bab", "blog", "v2", "s", h.book.ID(), staleGrant); rec || mark {
		t.Fatal("a retry grant signed for the first execution cleared the second's marker")
	}
	if ok, _ := h.book.ClearHoldRanOnPointer("s", "bab", "blog", t0.Add(30*time.Second)); ok {
		t.Fatal("a rollback signed between the two executions cleared the second's marker")
	}
	if _, ok := h.book.HoldRanOn("s"); !ok {
		t.Fatal("the second execution's marker is gone")
	}
}

// Finding 1's identities: an attempt's and a wake's carry the manager's
// incarnation, so two managers (two agent processes) never hand out the same
// one, and the a/w prefix Begin's wake rule reads is kept.
func TestExecutionIdentitiesCarryTheManagersIncarnation(t *testing.T) {
	m1 := NewManager(context.Background(), Deps{})
	m2 := NewManager(context.Background(), Deps{})
	a1, a2 := m1.nextExecID(), m2.nextExecID()
	if a1 == a2 || !strings.HasPrefix(a1, "a") || !strings.HasPrefix(a2, "a") || wakeIntent(a1) {
		t.Fatalf("attempt identities %q and %q: want distinct, both attempts", a1, a2)
	}
	w1, w2 := m1.wakeExecID(1), m2.wakeExecID(1)
	if w1 == w2 || !wakeIntent(w1) || !wakeIntent(w2) {
		t.Fatalf("wake identities %q and %q: want distinct, both wakes", w1, w2)
	}
	if m1.wakeExecID(0) != "" {
		t.Fatal("token zero has an identity")
	}
	if again := m1.nextExecID(); again == a1 {
		t.Fatalf("the same manager handed out %q twice", a1)
	}
}

// grant is a console's host retry grant as the box's status lets it sign
// one: the restriction's book and epoch.
type grant struct {
	book  string
	epoch uint64
}

func grantFor(b *FailureBook, build, host string) grant {
	return grant{book: b.ID(), epoch: b.RestrictionEpoch("bab", "blog", build, host)}
}

func applyGrant(b *FailureBook, build, host string, g grant) (rec, mark bool) {
	rec, mark, _, _ = b.RetryHostAt("bab", "blog", build, host, g.book, g.epoch)
	return rec, mark
}

// Final review round 1, finding 3 (Astra): two independent books (a host
// relocated, or a box rebuilt) count from the same place, so the same
// numeric epoch names different restrictions in each. A grant signed for
// one book's restriction clears nothing in the other's.
func TestARetryGrantForOneBookClearsNothingInAnother(t *testing.T) {
	one, other := &FailureBook{}, &FailureBook{}
	for _, b := range []*FailureBook{one, other} {
		b.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "h", Mode: string(activator.ModeHold),
			HeldDown: []string{"h"}, MarkHoldRan: true, HoldRanExec: "a1", HoldRanAt: t0})
	}
	g := grantFor(one, "v2", "h")
	if g.epoch == 0 || g.epoch != other.RestrictionEpoch("bab", "blog", "v2", "h") {
		t.Fatalf("the two books' epochs differ (%d, %d): the case needs equal counters", g.epoch, other.RestrictionEpoch("bab", "blog", "v2", "h"))
	}
	if rec, mark := applyGrant(other, "v2", "h", g); rec || mark {
		t.Fatalf("a grant for one book cleared record %v, marker %v in another", rec, mark)
	}
	if !other.HeldDown("bab", "blog", "v2", "h") {
		t.Fatal("the other book's hold-down was lifted")
	}
	if rec, mark := applyGrant(one, "v2", "h", g); !rec || !mark {
		t.Fatalf("the grant cleared record %v, marker %v in its own book; want both", rec, mark)
	}
}

// Finding 3's failed-save half, under the crash-epoch hard limit (readiness
// round 2, Task 1, which replaced the reopen's bounded advance): a
// generation handed out after a save failed lives only in memory, so status
// never publishes it and no grant at it clears anything, before a crash or
// after. A reopen may hand the same number out again, which no grant can
// name; every value it hands out is past every one status published.
func TestAnUnsavedGenerationIsNeverPublished(t *testing.T) {
	dd := t.TempDir()
	b, err := OpenFailureBook(dd)
	if err != nil {
		t.Fatal(err)
	}
	if err := b.Flush(); err != nil {
		t.Fatal(err)
	}
	if _, err := b.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "h"}); err != nil {
		t.Fatal(err)
	}
	saved := b.PublishedEpoch("bab", "blog", "v2", "h")
	if saved == 0 {
		t.Fatal("a saved restriction published nothing")
	}
	readOnly(t, dd)
	if _, err := b.Record(FailureRecord{Account: "bab", App: "blog", Build: "v3", Host: "h", HeldDown: []string{"h"}}); err == nil {
		t.Skip("the save did not fail")
	}
	if e := b.PublishedEpoch("bab", "blog", "v3", "h"); e != 0 {
		t.Fatalf("status published %d for a restriction not on disk, want 0", e)
	}
	unsaved := grantFor(b, "v3", "h") // a console that signed the in-memory value anyway
	if rec, mark := applyGrant(b, "v3", "h", unsaved); rec || mark {
		t.Fatal("a grant at the unsaved generation cleared before the save landed")
	}
	writable(t, dd)

	b2, err := OpenFailureBook(dd) // the agent crashed: v3's record never reached the disk
	if err != nil {
		t.Fatal(err)
	}
	if err := b2.Flush(); err != nil {
		t.Fatal(err)
	}
	b2.Record(FailureRecord{Account: "bab", App: "blog", Build: "v3", Host: "h", HeldDown: []string{"h"}})
	if e := b2.PublishedEpoch("bab", "blog", "v3", "h"); e <= saved {
		t.Fatalf("the reopened book published %d, not past the published %d", e, saved)
	}
	if rec, mark := applyGrant(b2, "v2", "h", grant{book: b.ID(), epoch: saved}); rec || mark {
		t.Fatal("the grant for the published restriction cleared after the app's restrictions changed")
	}
}

// Final review round 1, finding 4 (Astra): a retry that removes only a
// hold-ran marker (no failure record holds the host) is still a retry. An
// execution of the marked build that began before it, a scheduled attempt or
// a planned wake, and reports failure after it, is stale: no record is
// written and the cohort is not halted. A no-op retry changes nothing
// (TestANoOpRetryNotesNoClear).
func TestAMarkerOnlyRetryOrdersALateFailureBeforeIt(t *testing.T) {
	for _, kind := range []string{"attempt", "wake"} {
		t.Run(kind, func(t *testing.T) {
			ctx := context.Background()
			k := appKey{"bab", "blog"}
			h := newHarness(t, "h")
			h.swap = appconfig.SwapHold
			h.stage("v2")
			h.book.MarkHoldRan("h", HoldRan{Account: "bab", App: "blog", Build: "v2", Exec: "a-earlier", At: t0})
			var plan activator.WakePlan
			if kind == "attempt" {
				gate := h.act.hold("h")
				h.install("v2", "h")
				h.pass()
				waitFor(t, "h's attempt onto v2", func() bool {
					_, started, _ := h.act.snapshot()
					return len(started) == 1 && started[0] == "h:v2"
				})
				if !h.retry("h", "v2") {
					close(gate)
					t.Fatal("the marker-only retry cleared nothing")
				}
				h.act.mu.Lock()
				h.act.outcome = func(activator.Attempt) activator.SwapOutcome { return activator.SwapFailed }
				h.act.mu.Unlock()
				next := h.act.hold("h") // a next attempt, if any, parks in Swap
				defer close(next)
				close(gate)
				waitFor(t, "h's attempt to finish", func() bool { calls, _, _ := h.act.snapshot(); return len(calls) == 1 })
			} else {
				h.install("v2", "h")
				h.sleepHost("h")
				h.pass()
				h.wakeHost("h")
				plan = h.m.Plan(ctx, h.rows.get("h"))
				if !plan.FirstWake || plan.Boot.Version != "v2" {
					t.Fatalf("plan %+v: want a first wake of the marked v2", plan)
				}
				if !h.retry("h", "v2") {
					t.Fatal("the marker-only retry cleared nothing")
				}
				h.m.OnGen(activator.GenEvent{Kind: activator.GenFailed, Route: h.rows.get("h"), Failed: plan.Boot, Mode: plan.Mode,
					FirstWake: true, Reason: "health check failed", Token: plan.Token, ConfigRev: plan.ConfigRev})
			}
			if _, ok := h.book.Lookup("bab", "blog", "v2"); ok {
				t.Fatal("a failure of an execution that began before the retry wrote a record")
			}
			s := h.m.scheduler(k)
			s.mu.Lock()
			halted := s.q.cohortFor(cohortKey("v2", classAdopt)).halted
			s.mu.Unlock()
			if halted {
				t.Fatal("a failure of an execution that began before the retry halted the cohort")
			}
		})
	}
}

// Finding 4's other half: a retry that clears nothing (no record, no marker
// of the build) notes no clear, so a failure of an execution that began
// before it is still recorded.
func TestANoOpRetryNotesNoClear(t *testing.T) {
	b := &FailureBook{}
	snap := b.Snapshot()
	if rec, mark, _, _ := b.RetryHostAt("bab", "blog", "v2", "h", b.ID(), 1); rec || mark {
		t.Fatal("a retry with nothing to clear cleared something")
	}
	if res, _ := b.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "h", Seq: snap}); res != Recorded {
		t.Fatalf("a failure after a no-op retry = %v, want recorded", res)
	}
}

// Final review round 1, finding 5 (Astra): a host retry for a build B the
// slot does not want now (it wants A) still entitles the host to a canary
// onto B. The entitlement survives an attempt onto another build in
// between, and is spent only by the adoption onto B: when the slot comes
// to want B while B's cohort is halted, that adoption is admitted as the
// host's canary.
func TestARetryForABuildNotWantedYetKeepsItsCanary(t *testing.T) {
	k := appKey{"bab", "blog"}
	h := newHarness(t, "h", "s")
	h.stage("v2", "v3")
	h.act.mu.Lock()
	h.act.outcome = func(a activator.Attempt) activator.SwapOutcome {
		if a.Host == "s" && a.Build == "v3" {
			return activator.SwapFailed
		}
		return activator.SwapDone
	}
	h.act.mu.Unlock()
	h.pass()

	// s fails v3: a record holds v3 for h too, and v3's cohort halts.
	h.install("v3", "s")
	h.pass()
	waitFor(t, "s's attempt onto v3 to fail", func() bool { calls, _, _ := h.act.snapshot(); return len(calls) == 1 })
	s := h.m.scheduler(k)
	halted := func() bool {
		s.mu.Lock()
		defer s.mu.Unlock()
		return s.q.cohortFor(cohortKey("v3", classAdopt)).halted
	}
	waitFor(t, "v3's cohort to halt", halted)

	// The console's retry of h names v3; h's slot wants v1, which it serves.
	if !h.retry("h", "v3") {
		t.Fatal("the retry for v3 cleared nothing")
	}
	// An adoption onto another build (A = v2) runs in between: it does not
	// spend the entitlement for v3.
	h.install("v2", "h")
	h.pass()
	waitFor(t, "h's attempt onto v2", func() bool { return h.servingAll("v2", "h") && !h.m.attemptRunning("h") })

	// Now h's slot wants v3, its cohort still halted: the adoption onto v3 is
	// admitted, as h's canary.
	if !halted() {
		t.Fatal("v3's cohort is no longer halted: the case needs the halt")
	}
	gate := h.act.hold("h")
	defer close(gate)
	h.install("v3", "h")
	h.pass()
	waitFor(t, "h's adoption onto v3 through the halt", func() bool {
		_, started, _ := h.act.snapshot()
		return slices.Contains(started, "h:v3")
	})
	s.mu.Lock()
	d := s.q.hosts["h"].running
	s.mu.Unlock()
	if d == nil || d.ID.Build != "v3" || !d.Canary || !d.Forced {
		t.Fatalf("h's running attempt %+v: want the forced canary onto v3", d)
	}
}

// The entitlement belongs to its build: it never lets an adoption onto
// another build through that build's halt.
func TestARetryEntitlementIsNeverSpentOnAnotherBuild(t *testing.T) {
	q := newAppQueue("bab", "blog", func(string, string, bool) bool { return false })
	q.concurrency = 1
	now := t0
	q.observe("h", "v1", "v1", now)
	q.wantBuild("h", activator.CauseRelease, "v1", now)
	q.retryHost("h", "v3")
	q.cohortFor(cohortKey("v2", classAdopt)).halted = true
	q.wantBuild("h", activator.CauseRelease, "v2", now)
	if _, ok := q.gate(q.hosts["h"], now); ok {
		t.Fatal("an entitlement for v3 let an adoption onto v2 through v2's halt")
	}
	q.cohortFor(cohortKey("v3", classAdopt)).halted = true
	q.wantBuild("h", activator.CauseRelease, "v3", now)
	d, ok := q.gate(q.hosts["h"], now)
	if !ok || !d.Forced || !d.Canary {
		t.Fatalf("gate onto v3 = %+v, %v: want the forced canary", d, ok)
	}
}