package activator

import (
	"errors"
	"fmt"
	"log"
	"os"
	"path/filepath"
	"strings"
	"sync/atomic"
	"testing"
	"time"

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

// stagedActivator is shippingActivatorWithRegistry over stagedFake: ver is
// what live/<host> names, and every spawn reports the identity it exec'd.
func stagedActivator(t *testing.T) (*Service, *stagedFake, *fakeClock, *registry.Registry, *atomic.Value) {
	t.Helper()
	fb := &fakeBackend{}
	b := &stagedFake{fakeBackend: fb}
	clk := &fakeClock{t: time.Unix(1_700_000_000, 0)}
	reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatalf("open registry: %v", err)
	}
	a := New(Config{Bucket: "test-bucket", Idle: time.Hour}, b, reg)
	a.now = clk.now
	var ver atomic.Value
	ver.Store("v1")
	fb.execVerFn = func() string { return ver.Load().(string) }
	a.liveVer = func(string) (string, error) { return ver.Load().(string), nil }
	return a, b, clk, reg, &ver
}

func (a *Service) entryServing(host string) release.StagedIdentity {
	a.mu.Lock()
	defer a.mu.Unlock()
	if e := a.entries[host]; e != nil {
		return e.workerServing
	}
	return release.StagedIdentity{}
}

// Spec §7.9: a crash recycle respawns the serving build, not the installed one.
func TestWorkerCrashRecyclesTheCapturedServingBuild(t *testing.T) {
	t.Parallel()
	a, b, _, _, ver := stagedActivator(t)
	host := "bot.x.internal"
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	if got := a.entryServing(host); got != fakeIdentity("v1") {
		t.Fatalf("serving after wake = %+v, want the captured v1", got)
	}
	ver.Store("v2") // root installs v2 under the running child
	b.lastWorker().kill()
	a.sweepOnce()
	waitForRestart(t, a, b.fakeBackend, host, 2)
	calls := b.calls()
	if len(calls) != 2 || calls[0] != nil {
		t.Fatalf("spawns = %v; want a wake from the live link, then a recycle", calls)
	}
	if calls[1] == nil || *calls[1] != fakeIdentity("v1") {
		t.Fatalf("crash recycle asked for %v; want the captured v1 build", calls[1])
	}
	if got := a.entrySpawnVersion(host); got != "v1" {
		t.Fatalf("spawn version = %q, want v1", got)
	}
}

// Spec §7.9: so does an always-on worker's max_run recycle.
func TestWorkerMaxRunRecyclesTheCapturedServingBuild(t *testing.T) {
	t.Parallel()
	a, b, clk, _, ver := stagedActivator(t)
	host := "bot.x.internal"
	rt := alwaysOnWorkerRoute(host)
	rt.MaxRunSeconds = 30
	if err := a.WakeWorker(rt); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	ver.Store("v2")
	clk.advance(31 * time.Second)
	a.sweepOnce()
	waitForRestart(t, a, b.fakeBackend, host, 2)
	if calls := b.calls(); len(calls) != 2 || calls[1] == nil || *calls[1] != fakeIdentity("v1") {
		t.Fatalf("max_run recycle asked for %v; want the captured v1 build", calls)
	}
}

// Until plan 5 queues it, the drift check is the one path that moves a
// worker to a different installed build, and it reads the live link.
func TestWorkerConvergeStillMovesOntoTheInstalledBuild(t *testing.T) {
	t.Parallel()
	a, b, clk, _, ver := stagedActivator(t)
	host := "bot.x.internal"
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	ver.Store("v2")
	a.sweepOnce()
	clk.advance(ConvergeGrace + time.Minute)
	a.sweepOnce()
	waitFor(t, "the converge onto v2", func() bool { return a.entrySpawnVersion(host) == "v2" && !a.entryRespawning(host) })
	if calls := b.calls(); len(calls) != 2 || calls[1] != nil {
		t.Fatalf("converge asked for %v; want the live link (nil)", calls)
	}
	if got := a.entryServing(host); got != fakeIdentity("v2") {
		t.Fatalf("serving after converge = %+v, want v2", got)
	}
}

// Plan 5's hook: an admissible queued attempt takes the spawn.
func TestWorkerRecycleDefersToCoalesceRecycle(t *testing.T) {
	t.Parallel()
	a, b, _, _, _ := stagedActivator(t)
	host := "bot.x.internal"
	var asked []string
	a.CoalesceRecycle = func(h string) bool { asked = append(asked, h); return true }
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	b.lastWorker().kill()
	a.sweepOnce()
	if a.entryRespawning(host) || b.workerStarts() != 1 {
		t.Fatalf("workerStarts = %d; the hook said a queued attempt makes the spawn", b.workerStarts())
	}
	if len(asked) != 1 || asked[0] != host {
		t.Fatalf("hook asked about %v, want [%s]", asked, host)
	}
}

// Review Focus 4, and spec §7.9: the report is display metadata; registry
// Version is root's install stamp.
func TestNoteWorkerReadyIsDisplayOnly(t *testing.T) {
	t.Parallel()
	a, b, _, reg := testActivator(t, time.Hour)
	host := "bot.x.internal"
	if err := reg.Upsert(testWorkerRoute(host)); err != nil {
		t.Fatal(err)
	}
	if err := reg.SetVersions(host, "v1", ""); err != nil {
		t.Fatal(err)
	}
	if err := a.WakeWorker(testWorkerRoute(host)); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	noteReadyAsChild(t, a, b, host, "1.4.2-app-chosen")
	if rt, _ := reg.Get(host); rt == nil || rt.Version != "v1" {
		t.Fatalf("row = %+v; a worker's report must not overwrite root's stamp", rt)
	}
	if got := a.WorkerReportedVersion(host); got != "1.4.2-app-chosen" {
		t.Fatalf("WorkerReportedVersion = %q", got)
	}
	a.NoteWorkerReady(host, "")
	if rt, _ := reg.Get(host); rt == nil || rt.Version != "v1" {
		t.Fatalf("row = %+v after an empty report", rt)
	}
	if got := a.WorkerReportedVersion("nobody.x.internal"); got != "" {
		t.Fatalf("WorkerReportedVersion for no worker = %q", got)
	}
}

// Spec §7.9: no spawn path writes registry Version: not a wake, a crash
// recycle or a converge.
func TestWorkerSpawnsLeaveTheInstalledStampAlone(t *testing.T) {
	t.Parallel()
	a, b, clk, reg, ver := shippingActivatorWithRegistry(t, time.Hour)
	host := "bot.x.internal"
	rt := alwaysOnWorkerRoute(host)
	if err := reg.Upsert(rt); err != nil {
		t.Fatal(err)
	}
	if err := reg.SetVersions(host, "v1", "v0"); err != nil {
		t.Fatal(err)
	}
	stamp := func(when string) {
		t.Helper()
		if got, _ := reg.Get(host); got == nil || got.Version != "v1" || got.PrevVersion != "v0" {
			t.Fatalf("%s: row = %+v; want root's stamp v1 (prev v0) untouched", when, got)
		}
	}
	if err := a.WakeWorker(rt); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	stamp("after a wake")
	noteReadyAsChild(t, a, b, host, "whatever-the-app-says")
	stamp("after a ready report")
	b.lastWorker().kill()
	a.sweepOnce()
	waitForRestart(t, a, b, host, 2)
	stamp("after a crash recycle")
	ver.Store("v2")
	a.sweepOnce()
	clk.advance(ConvergeGrace + time.Minute)
	a.sweepOnce()
	waitFor(t, "the converge onto v2", func() bool { return a.entrySpawnVersion(host) == "v2" && !a.entryRespawning(host) })
	stamp("after a converge")
}

// Review Focus 4: an app-chosen report changes nothing about what recycles,
// and a new child's report starts out unknown.
func TestWorkerReportingAnAppChosenVersionStillRecyclesItsCapturedBuild(t *testing.T) {
	t.Parallel()
	a, b, _, _, _ := stagedActivator(t)
	host := "bot.x.internal"
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	noteReadyAsChild(t, a, b.fakeBackend, host, "latest")
	b.lastWorker().kill()
	a.sweepOnce()
	waitForRestart(t, a, b.fakeBackend, host, 2)
	if calls := b.calls(); len(calls) != 2 || calls[1] == nil || *calls[1] != fakeIdentity("v1") {
		t.Fatalf("recycle asked for %v; want the captured v1, whatever the child said", calls)
	}
	if got := a.WorkerReportedVersion(host); got != "" {
		t.Fatalf("WorkerReportedVersion = %q after a recycle; the new child has not reported", got)
	}
}

func (b *stagedFake) failNextStagedSpawn(err error) {
	b.smu.Lock()
	defer b.smu.Unlock()
	b.failNextStaged = err
}

// waitForSpawnCalls waits until the backend has been asked for n spawns and
// the respawn that asked has finished, failed or not.
func waitForSpawnCalls(t *testing.T, a *Service, b *stagedFake, host string, n int) {
	t.Helper()
	waitFor(t, fmt.Sprintf("spawn call #%d of %s", n, host), func() bool {
		return len(b.calls()) == n && !a.entryRespawning(host)
	})
}

// Ruling R17 (spec §4.2): a recycle the backend refuses because the captured
// copy changed stays refused. The worker stays down, it keeps its captured
// build, and the next crash restart asks for that same build again, never
// for what live/<host> names. Here live/<host> still names v1, so there is
// no drift that could admit a converge either.
func TestWorkerRecycleRefusedForAChangedCopyStaysRefused(t *testing.T) {
	t.Parallel()
	a, b, clk, _, _ := stagedActivator(t)
	host := "bot.x.internal"
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	changed := fmt.Errorf("worker %s: %w", host, release.ErrStagedCopyChanged)
	b.failNextStagedSpawn(changed)
	b.lastWorker().kill()
	a.sweepOnce()
	waitForSpawnCalls(t, a, b, host, 2)
	if calls := b.calls(); calls[1] == nil || *calls[1] != fakeIdentity("v1") {
		t.Fatalf("crash recycle asked for %v; want the captured v1 build", calls)
	}
	if got := a.entryServing(host); got != fakeIdentity("v1") {
		t.Fatalf("serving after a refused recycle = %+v; want the captured v1 kept", got)
	}
	if b.workerStarts() != 1 || !b.lastWorker().Exited() {
		t.Fatalf("workerStarts = %d; a refused recycle must leave the worker down", b.workerStarts())
	}
	// The bytes still differ: the next attempt asks for the captured build
	// again and is refused again.
	b.failNextStagedSpawn(changed)
	clk.advance(sidecarRestartMax + time.Second)
	a.sweepOnce()
	waitForSpawnCalls(t, a, b, host, 3)
	if calls := b.calls(); calls[2] == nil || *calls[2] != fakeIdentity("v1") {
		t.Fatalf("restart after the refusal asked for %v; want the captured v1 again, never the live link", calls[2])
	}
	if b.workerStarts() != 1 || !b.lastWorker().Exited() {
		t.Fatalf("workerStarts = %d; the refusal must stand while the bytes differ", b.workerStarts())
	}
	if got := a.entryServing(host); got != fakeIdentity("v1") {
		t.Fatalf("serving after a second refusal = %+v; want the captured v1 kept", got)
	}
}

// Any other spawn failure keeps the captured build too.
func TestWorkerRecycleFailingForAnotherReasonKeepsTheCapturedBuild(t *testing.T) {
	t.Parallel()
	a, b, clk, _, _ := stagedActivator(t)
	host := "bot.x.internal"
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	b.failNextStagedSpawn(errors.New("fork/exec: resource temporarily unavailable"))
	b.lastWorker().kill()
	a.sweepOnce()
	waitForSpawnCalls(t, a, b, host, 2)
	if got := a.entryServing(host); got != fakeIdentity("v1") {
		t.Fatalf("serving after a failed recycle = %+v; want the captured v1 kept", got)
	}
	clk.advance(sidecarRestartMax + time.Second)
	a.sweepOnce()
	waitForSpawnCalls(t, a, b, host, 3)
	if calls := b.calls(); calls[2] == nil || *calls[2] != fakeIdentity("v1") {
		t.Fatalf("restart after the failure asked for %v; want the captured v1 again", calls[2])
	}
	if b.workerStarts() != 2 {
		t.Fatalf("workerStarts = %d; want the worker running again", b.workerStarts())
	}
}

// Ruling R13: a max_run shorter than ConvergeGrace must not starve the
// converge. Recycles onto the captured build keep the drift clock, and the
// recycle that falls due past ConvergeGrace converges instead.
func TestWorkerMaxRunRecyclesConvergeOnceTheGraceHasPassed(t *testing.T) {
	t.Parallel()
	a, b, clk, _, ver := stagedActivator(t)
	host := "bot.x.internal"
	rt := alwaysOnWorkerRoute(host)
	rt.MaxRunSeconds = 30
	if err := a.WakeWorker(rt); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	ver.Store("v2")
	converged := -1
	for i := 1; i <= 6 && converged < 0; i++ {
		clk.advance(31 * time.Second)
		a.sweepOnce()
		waitForRestart(t, a, b.fakeBackend, host, i+1)
		if b.calls()[i] == nil {
			converged = i
		}
	}
	calls := b.calls()
	if converged < 0 {
		t.Fatalf("spawns = %v; six max_run recycles never moved the worker onto v2", calls)
	}
	for i := 1; i < converged; i++ {
		if calls[i] == nil || *calls[i] != fakeIdentity("v1") {
			t.Fatalf("recycle %d asked for %v; inside the grace a recycle keeps the captured v1", i, calls[i])
		}
	}
	if elapsed := time.Duration(converged) * 31 * time.Second; elapsed < ConvergeGrace {
		t.Fatalf("converged after %s; want the drift to have lasted ConvergeGrace first", elapsed)
	}
	if got := a.entryServing(host); got != fakeIdentity("v2") {
		t.Fatalf("serving after the converge = %+v, want v2", got)
	}
	// No drift now: the next max_run recycles the captured v2.
	clk.advance(31 * time.Second)
	a.sweepOnce()
	waitForRestart(t, a, b.fakeBackend, host, converged+2)
	if c := b.calls()[converged+1]; c == nil || *c != fakeIdentity("v2") {
		t.Fatalf("recycle after the converge asked for %v; want the captured v2", c)
	}
}

func (a *Service) entryStaleSince(host string) time.Time {
	a.mu.Lock()
	defer a.mu.Unlock()
	if e := a.entries[host]; e != nil {
		return e.staleSince
	}
	return time.Time{}
}

// Ruling R17 (N2): a converge by recycle draws on the sweep's converge
// budget. Two workers whose drift is overdue crash together; one sweep
// converges at most maxConvergeCyclesPerSweep of them, the other recycles its
// captured build and keeps its drift age, and converges on a later sweep.
func TestWorkerRecycleConvergesShareTheSweepsConvergeBudget(t *testing.T) {
	t.Parallel()
	a, b, clk, _, ver := stagedActivator(t)
	hosts := []string{"bot1.x.internal", "bot2.x.internal"}
	for _, h := range hosts {
		if err := a.WakeWorker(alwaysOnWorkerRoute(h)); err != nil {
			t.Fatalf("WakeWorker %s: %v", h, err)
		}
	}
	ver.Store("v2")
	a.sweepOnce() // the drift is first seen here, by the sweep's own check
	seen := a.entryStaleSince(hosts[0])
	if seen.IsZero() || !a.entryStaleSince(hosts[1]).Equal(seen) {
		t.Fatalf("drift clocks = %v, %v; want both started by the sweep", seen, a.entryStaleSince(hosts[1]))
	}
	clk.advance(ConvergeGrace + time.Minute)
	b.mu.Lock()
	workers := append([]*fakeProc(nil), b.workers...)
	b.mu.Unlock()
	for _, w := range workers {
		w.kill()
	}
	a.sweepOnce()
	waitFor(t, "both crash restarts", func() bool {
		return b.workerStarts() == 4 && !a.entryRespawning(hosts[0]) && !a.entryRespawning(hosts[1])
	})
	var converged, recycled []string
	for _, h := range hosts {
		switch a.entrySpawnVersion(h) {
		case "v2":
			converged = append(converged, h)
		case "v1":
			recycled = append(recycled, h)
		}
	}
	if len(converged) != maxConvergeCyclesPerSweep || len(recycled) != len(hosts)-maxConvergeCyclesPerSweep {
		t.Fatalf("in one sweep converged %v and recycled %v; want at most %d converge", converged, recycled, maxConvergeCyclesPerSweep)
	}
	if got := a.entryStaleSince(recycled[0]); !got.Equal(seen) {
		t.Fatalf("drift clock of the worker that recycled = %v; want its age kept (%v)", got, seen)
	}
	a.sweepOnce()
	waitFor(t, "the later converge", func() bool {
		return a.entrySpawnVersion(recycled[0]) == "v2" && !a.entryRespawning(recycled[0])
	})
}

// Ruling R17 (N2): with plan 5's scheduler present (CoalesceRecycle set), a
// recycle never converges, even past ConvergeGrace: plan 5's admission owns
// build changes. A false answer recycles the captured build, and the drift
// keeps its age.
func TestWorkerRecycleNeverConvergesWhenTheSchedulerIsPresent(t *testing.T) {
	t.Parallel()
	a, b, clk, _, ver := stagedActivator(t)
	host := "bot.x.internal"
	a.CoalesceRecycle = func(string) bool { return false }
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	ver.Store("v2")
	b.lastWorker().kill()
	a.sweepOnce() // the crash restart sees the drift first
	waitForRestart(t, a, b.fakeBackend, host, 2)
	seen := a.entryStaleSince(host)
	if seen.IsZero() {
		t.Fatal("the crash restart did not start the drift clock")
	}
	b.lastWorker().kill()
	clk.advance(ConvergeGrace + time.Minute)
	a.sweepOnce()
	waitForRestart(t, a, b.fakeBackend, host, 3)
	calls := b.calls()
	for i := 1; i < len(calls); i++ {
		if calls[i] == nil || *calls[i] != fakeIdentity("v1") {
			t.Fatalf("recycle %d asked for %v; with the scheduler present a recycle keeps the captured v1", i, calls[i])
		}
	}
	if got := a.entryStaleSince(host); !got.Equal(seen) {
		t.Fatalf("drift clock = %v after the recycle; want its age kept (%v)", got, seen)
	}
}

// Astra R3 §5 item 4: on a box whose scheduler (or its refusal) owns
// recycles (CoalesceRecycle set), a worker with no captured build is never
// respawned from the live link, which names whatever root installed last:
// a build no admission chose. A live child whose max_run falls due keeps
// running; a crashed one stays down, and the agent says so once for the
// host, not at every backoff. A release attempt (ReplaceWorker) or a wake
// is what starts it. Before the fix both respawned from live/<host>.
func TestAnArmedBoxNeverRecyclesAnUncapturedWorkerFromTheLiveLink(t *testing.T) {
	const said = "has no captured build, and the release scheduler owns its recycles"
	capture := func(t *testing.T) *syncLog {
		logs := &syncLog{}
		flags := log.Flags()
		log.SetOutput(logs)
		t.Cleanup(func() { log.SetOutput(os.Stderr); log.SetFlags(flags) })
		return logs
	}
	t.Run("crashed", func(t *testing.T) {
		a, b, clk, _ := testActivator(t, time.Hour) // no staged capability: nothing is captured
		host := "bot.x.internal"
		a.CoalesceRecycle = func(string) bool { return false }
		if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
			t.Fatalf("WakeWorker: %v", err)
		}
		if got := a.entryServing(host); got.StagedPath != "" {
			t.Fatalf("serving %+v, want no captured build", got)
		}
		logs := capture(t)
		b.lastWorker().kill()
		for range 3 {
			a.sweepOnce()
			clk.advance(sidecarRestartMax + time.Second)
		}
		time.Sleep(100 * time.Millisecond)
		if n := b.workerStarts(); n != 1 || a.entryRespawning(host) {
			t.Fatalf("workerStarts = %d (respawning %v): an uncaptured worker was respawned from the live link", n, a.entryRespawning(host))
		}
		if !a.Owns(host) {
			t.Fatal("the entry went: it stays, down, for a release attempt or a wake to start")
		}
		if n := strings.Count(logs.String(), said); n != 1 {
			t.Fatalf("said so %d times in %q, want once for the host", n, logs.String())
		}
	})
	t.Run("max_run", func(t *testing.T) {
		a, b, clk, _ := testActivator(t, time.Hour)
		host := "bot.x.internal"
		a.CoalesceRecycle = func(string) bool { return false }
		rt := alwaysOnWorkerRoute(host)
		rt.MaxRunSeconds = 30
		if err := a.WakeWorker(rt); err != nil {
			t.Fatalf("WakeWorker: %v", err)
		}
		logs := capture(t)
		for range 3 {
			clk.advance(31 * time.Second)
			a.sweepOnce()
		}
		time.Sleep(100 * time.Millisecond)
		if n := b.workerStarts(); n != 1 || a.entryRespawning(host) {
			t.Fatalf("workerStarts = %d (respawning %v): an uncaptured worker was recycled from the live link", n, a.entryRespawning(host))
		}
		if b.lastWorker().Exited() {
			t.Fatal("the running child was stopped by a recycle that could not respawn it")
		}
		if n := strings.Count(logs.String(), said); n != 1 {
			t.Fatalf("said so %d times in %q, want once for the host", n, logs.String())
		}
	})
}

// Ruling R13: an exited worker skips the sweep's drift check, so the crash
// restart measures the drift itself, and a crash loop converges once the
// grace has passed.
func TestWorkerCrashLoopConvergesOnceTheGraceHasPassed(t *testing.T) {
	t.Parallel()
	a, b, clk, _, ver := stagedActivator(t)
	host := "bot.x.internal"
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	ver.Store("v2")
	converged := -1
	for i := 1; i <= 6 && converged < 0; i++ {
		b.lastWorker().kill()
		clk.advance(sidecarRestartMax + time.Second)
		a.sweepOnce()
		waitForRestart(t, a, b.fakeBackend, host, i+1)
		if b.calls()[i] == nil {
			converged = i
		}
	}
	calls := b.calls()
	if converged < 0 {
		t.Fatalf("spawns = %v; a crash loop never moved the worker onto v2", calls)
	}
	if converged < 2 || calls[1] == nil || *calls[1] != fakeIdentity("v1") {
		t.Fatalf("spawns = %v; the first crash after the install must recycle the captured v1 (the drift has only just been seen)", calls)
	}
	if a.entrySpawnVersion(host) != "v2" {
		t.Fatalf("spawn version = %q after the converge, want v2", a.entrySpawnVersion(host))
	}
}