package rollout

import (
	"context"
	"testing"
	"time"

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

// workerHarness turns the harness's hosts into always-on sidecars.
func workerHarness(t *testing.T, hosts ...string) *harness {
	h := newHarness(t, hosts...)
	for _, host := range hosts {
		r := h.rows.get(host)
		r.Kind, r.Hibernate = "sidecar", false
		h.rows.put(r)
	}
	return h
}

// Spec §7.9: a socketless attempt is ReplaceWorker on the same entry, never
// Swap and never a stop and a wake. It carries the host's good build as its
// fallback.
func TestAWorkerReleaseReplacesOnTheSameEntry(t *testing.T) {
	h := workerHarness(t, "bot")
	h.stage("v2")
	h.install("v2", "bot")
	h.pass()
	waitFor(t, "the good build to move to v2", func() bool { return h.rows.get("bot").GoodVersion == "v2" })
	if calls, _, _ := h.act.snapshot(); len(calls) != 0 {
		t.Fatal("a socketless attempt went through Swap")
	}
	r := h.act.replacements()
	if len(r) != 1 || r[0].a.Build != "v2" || r[0].fallback == nil || r[0].fallback.Version != "v1" {
		t.Fatalf("replacements %+v, want one onto v2 with the good v1 behind it", r)
	}
	if fb := r[0].fallback; fb.Digest != "sha-v1" || fb.StagedPath != release.VersionPath(h.opt, "bab", "v1", "blog") {
		t.Fatalf("fallback %+v, want the good build by its retained identity", fb)
	}
}

// Spec §6.5, §8.5 (Astra R3 flag triage T9): a replacement whose build exits
// cleanly as a one-shot leaves nothing for the activator to serve, so the
// acknowledged check would drop its good write. done() records the build
// good from the result, and clears a hold-ran marker older than the attempt.
// A replacement that is not a clean exit still needs the activator to serve
// it.
func TestACleanOneShotReplacementRecordsItsBuildGood(t *testing.T) {
	h := workerHarness(t, "bot")
	h.stage("v2")
	h.book.MarkHoldRan("bot", HoldRan{Account: "bab", App: "blog", Build: "v9"}) // set before the attempt starts
	h.act.mu.Lock()
	h.act.worker = func(activator.Attempt, *release.StagedIdentity) activator.SwapResult {
		return activator.SwapResult{Outcome: activator.SwapDone, CleanExit: true}
	}
	h.act.mu.Unlock()
	h.install("v2", "bot")
	h.pass()
	waitFor(t, "the good build to move to v2", func() bool { return h.rows.get("bot").GoodVersion == "v2" })
	if r := h.rows.get("bot"); r.GoodDigest != "sha-v2" {
		t.Fatalf("row = %+v, want v2's digest as the good build's", r)
	}
	if _, ok := h.act.Current("bot"); ok {
		t.Fatal("the fake still serves the one-shot: the test does not model a parked worker")
	}
	if _, held := h.book.HoldRanOn("bot"); held {
		t.Fatal("the clean one-shot passed, but the older hold-ran marker stands")
	}
}

// Astra R3 N5: a hold build that ran on a worker without passing (a
// superseded hold, its marker set by done()) may have migrated its
// database, so the next scheduled replacement carries no fallback onto the
// good build. Another app's marker under the same host name changes nothing.
func TestAWorkerReplacementCarriesNoFallbackWhileAMarkerStands(t *testing.T) {
	h := workerHarness(t, "bot")
	h.stage("v2", "v3")
	h.book.MarkHoldRan("bot", markV2)
	h.install("v3", "bot")
	h.pass()
	waitFor(t, "the replacement onto v3", func() bool { return len(h.act.replacements()) == 1 })
	if r := h.act.replacements(); r[0].a.Build != "v3" || r[0].fallback != nil {
		t.Fatalf("replacement %+v (fallback %+v): want v3 with no fallback while v2's marker stands", r[0].a, r[0].fallback)
	}

	other := workerHarness(t, "bot")
	other.stage("v3")
	other.book.MarkHoldRan("bot", HoldRan{Account: "bab", App: "other", Build: "v2"})
	other.install("v3", "bot")
	other.pass()
	waitFor(t, "the replacement onto v3", func() bool { return len(other.act.replacements()) == 1 })
	if r := other.act.replacements(); r[0].fallback == nil || r[0].fallback.Version != "v1" {
		t.Fatalf("replacement %+v: another app's marker took away the good v1 fallback", r[0])
	}
}

func TestAFailedWorkerReleaseFallsBackAndHalts(t *testing.T) {
	h := workerHarness(t, "bot")
	h.stage("v2")
	h.act.mu.Lock()
	h.act.worker = func(a activator.Attempt, fb *release.StagedIdentity) activator.SwapResult {
		return activator.SwapResult{Outcome: activator.SwapFailed, Reason: "exited within 10s of starting"}
	}
	h.act.mu.Unlock()
	h.install("v2", "bot")
	h.pass()
	waitFor(t, "a failure record", func() bool { _, ok := h.book.Lookup("bab", "blog", "v2"); return ok })
	if r := h.rows.get("bot"); r.GoodVersion != "v1" {
		t.Fatalf("good build = %q after a failed worker release", r.GoodVersion)
	}
}

// Spec §8.1 (P5 review finding 7): the fallback's held publication is
// released only after done() has recorded the failure it falls back from.
func TestAWorkerFallbackIsPublishedAfterItsFailureIsRecorded(t *testing.T) {
	h := workerHarness(t, "bot")
	h.stage("v2")
	recorded := make(chan bool, 1)
	h.act.mu.Lock()
	h.act.worker = func(a activator.Attempt, fb *release.StagedIdentity) activator.SwapResult {
		return activator.SwapResult{Outcome: activator.SwapFailed, Reason: "exited within 10s of starting"}
	}
	h.act.onHeldPublish = func(string) {
		_, ok := h.book.Lookup("bab", "blog", "v2")
		select {
		case recorded <- ok:
		default:
		}
	}
	h.act.mu.Unlock()
	h.install("v2", "bot")
	h.pass()
	select {
	case ok := <-recorded:
		if !ok {
			t.Fatal("the fallback's publication was released before v2's failure was recorded")
		}
	case <-time.After(5 * time.Second):
		t.Fatal("the held publication was never released")
	}
}

// Spec §7.9: a recycle that falls due coalesces with a queued release attempt
// only when the scheduler would admit that attempt now.
func TestARecycleCoalescesOnlyWithAnAdmissibleRelease(t *testing.T) {
	q := awakeQueue(noBlocks, "a", "b")
	releaseAll(q, "v2", "a", "b")
	canary, ok := q.admitOne("a", t0)
	if !ok || !canary.Canary {
		t.Fatalf("admitOne(a) = %+v %v, want a's canary", canary, ok)
	}
	if _, ok := q.admitOne("b", t0); ok {
		t.Fatal("coalesced past swap_concurrency 1 while another worker's canary runs")
	}
	q.finish(canary, activator.SwapDone)
	if d, ok := q.admitOne("b", t0); !ok || d.ID.Build != "v2" {
		t.Fatalf("admitOne(b) = %+v %v after the canary settled", d, ok)
	}
	r := awakeQueue(noBlocks, "c")
	r.wantRestart("c", "r1", t0)
	if _, ok := r.admitOne("c", t0); ok {
		t.Fatal("a restart is not a release attempt; the recycle runs on the serving build")
	}
	h := newHarness(t, "a")
	if h.m.CoalesceRecycle("nobody") {
		t.Fatal("an unknown host coalesced")
	}
}

// Spec §7.9: while an attempt runs for the host, the recycle is skipped:
// that attempt's ReplaceWorker makes the spawn. A crash during its
// observation is the build's to report, never a recycle's to paper over.
func TestARecycleIsSkippedWhileTheHostsAttemptRuns(t *testing.T) {
	h := workerHarness(t, "bot")
	h.stage("v2")
	running := make(chan struct{})
	proceed := make(chan struct{})
	h.act.mu.Lock()
	h.act.worker = func(activator.Attempt, *release.StagedIdentity) activator.SwapResult {
		close(running)
		<-proceed
		return activator.SwapResult{Outcome: activator.SwapDone}
	}
	h.act.mu.Unlock()
	h.install("v2", "bot")
	h.pass()
	select {
	case <-running:
	case <-time.After(5 * time.Second):
		t.Fatal("the attempt never started")
	}
	if !h.m.CoalesceRecycle("bot") {
		t.Fatal("a recycle went ahead while the host's attempt was running")
	}
	close(proceed)
	waitFor(t, "the attempt to finish", func() bool { return h.rows.get("bot").GoodVersion == "v2" })
	if h.m.CoalesceRecycle("bot") {
		t.Fatal("coalesced with nothing queued or running")
	}
}

// Plan 4's contract: CoalesceRecycle is called under the activator's lock,
// so it must never call back into the activator. Here the fake's own lock
// stands in for activator.Service.mu, held across the call: a callback into
// Current, Owns or ReplaceWorker would never return.
func TestCoalesceRecycleNeverCallsTheActivator(t *testing.T) {
	h := workerHarness(t, "bot")
	h.pass() // the host is known to its scheduler
	waitFor(t, "the host's scheduler", func() bool { return h.m.schedulerOf("bot") != nil })
	answered := make(chan bool, 1)
	h.act.mu.Lock()
	go func() { answered <- h.m.CoalesceRecycle("bot") }()
	select {
	case <-answered:
	case <-time.After(2 * time.Second):
		h.act.mu.Unlock()
		t.Fatal("CoalesceRecycle waited on the activator")
	}
	h.act.mu.Unlock()
}

// Task 11 fix round 1, review finding 2 (spec §8.6): an adoption a recycle
// coalesces with runs on its own goroutine, and Close joins it as it joins
// one the scheduler loop admits: its outcome and its End are on disk when
// Close returns. Once Close has begun a recycle admits nothing; it is
// skipped, which is all a stopping agent needs.
func TestCloseJoinsACoalescedWorkerAttempt(t *testing.T) {
	h := workerHarness(t, "bot")
	dir := h.onDisk()
	h.stage("v2", "v3")
	h.pass()
	waitFor(t, "the host's scheduler to see it serve v1", func() bool {
		s := h.m.schedulerOf("bot")
		if s == nil {
			return false
		}
		s.mu.Lock()
		defer s.mu.Unlock()
		hs := s.q.hosts["bot"]
		return hs != nil && hs.serving == "v1"
	})
	running, proceed := make(chan struct{}), make(chan struct{})
	h.act.mu.Lock()
	h.act.worker = func(activator.Attempt, *release.StagedIdentity) activator.SwapResult {
		close(running)
		<-proceed
		return activator.SwapResult{Outcome: activator.SwapDone}
	}
	h.act.mu.Unlock()
	queue := func(build string) {
		h.install(build, "bot")
		s := h.m.schedulerOf("bot")
		s.mu.Lock()
		s.q.wantBuild("bot", activator.CauseRelease, build, h.clock()) // queued, not yet admitted by the loop
		s.mu.Unlock()
	}
	queue("v2")
	if !h.m.CoalesceRecycle("bot") {
		t.Fatal("the recycle did not coalesce with the admissible release")
	}
	select {
	case <-running:
	case <-time.After(5 * time.Second):
		t.Fatal("the coalesced attempt never started")
	}
	if n := len(fileOf(t, dir).Intents); n != 1 {
		t.Fatalf("%d intents on disk while the coalesced attempt runs, want 1", n)
	}
	go func() { time.Sleep(200 * time.Millisecond); close(proceed) }()
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	h.m.Close(ctx)
	if f := fileOf(t, dir); len(f.Intents) != 0 {
		t.Fatalf("intents %+v on disk after Close: a restart would reconcile the coalesced attempt as a failure", f.Intents)
	}
	if g := h.rows.get("bot").GoodVersion; g != "v2" {
		t.Fatalf("good build %q after Close, want v2: the coalesced attempt's outcome", g)
	}
	queue("v3")
	if !h.m.CoalesceRecycle("bot") {
		t.Fatal("a recycle after Close went ahead: want it skipped")
	}
	time.Sleep(100 * time.Millisecond)
	if r := h.act.replacements(); len(r) != 1 {
		t.Fatalf("replacements %+v: an attempt was admitted by a recycle after Close", r)
	}
}