package activator

import (
	"context"
	"errors"
	"path/filepath"
	"sync"
	"testing"
	"time"

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

func TestWorkerGenerations(t *testing.T) {
	for _, on := range []bool{true, false} {
		a, b, _, _ := testActivator(t, time.Hour)
		a.gens = on
		old := sidecarSettle
		sidecarSettle = 20 * time.Millisecond
		t.Cleanup(func() { sidecarSettle = old })
		rt := testWorkerRoute("bot.example.com")
		if err := a.WakeWorker(rt); err != nil {
			t.Fatal(err)
		}
		first, members := a.genStateForTest(rt.Host)
		if on != (first.N != 0) || first.Socket != "" || len(members) != 0 {
			t.Fatalf("gens %v: worker generation %+v, members %v", on, first, members)
		}
		b.mu.Lock()
		b.workerStartDelay = 200 * time.Millisecond // hold the respawn open
		b.mu.Unlock()
		b.lastWorker().crash()
		a.sweepOnce() // a crashed hibernating worker respawns in place
		waitFor(t, "the respawn to begin", func() bool { return b.workerEnters() == 2 })
		_, members = a.genStateForTest(rt.Host)
		if on != (len(members) == 1) {
			t.Fatalf("gens %v: members during the respawn = %v", on, members)
		}
		waitFor(t, "the respawn to finish", func() bool { return b.workerStarts() == 2 })
		waitFor(t, "the member to leave", func() bool { _, m := a.genStateForTest(rt.Host); return len(m) == 0 })
		if on {
			if second, _ := a.genStateForTest(rt.Host); second.N <= first.N || !second.Worker {
				t.Fatalf("respawned worker generation %+v, want a Worker above %d", second, first.N)
			}
		}
	}
}

// StartWorker spawns every worker into the flat host cgroup, whatever the
// row's layout, so KillHost proves that cgroup for a nested-looking row too.
// Here the cgroup is a fixture, not cgroup v2, so the proof is refused.
// Answering nil for a nested layout skipped the proof for a worker that
// runs in carlos-exec@<host> all the same.
func TestKillHostProvesANestedLayoutWorkersHostCgroup(t *testing.T) {
	host := "nw.example.com"
	b, _ := delegatedBackend(t, host)
	rd := t.TempDir()
	b.RunDirOverride = rd
	rt := registry.Route{Host: host, Kind: "sidecar", DBPath: provision.NestedDBPath(b.DataDir, host),
		Socket: filepath.Join(rd, "instances", provision.InstanceID(host), "app.sock")}
	if err := b.KillHost(context.Background(), rt); err == nil {
		t.Fatal("KillHost answered a nested-layout worker's host cgroup as proven without proving it")
	}
}

// hostKillBackend is fakeBackend plus KillHost, failing while bad.
type hostKillBackend struct {
	*fakeBackend
	hmu   sync.Mutex
	bad   bool
	kills int
}

func (h *hostKillBackend) KillHost(context.Context, registry.Route) error {
	h.hmu.Lock()
	defer h.hmu.Unlock()
	h.kills++
	if h.bad {
		return errors.New("the host cgroup still holds a process")
	}
	return nil
}

func (h *hostKillBackend) killCount() int {
	h.hmu.Lock()
	defer h.hmu.Unlock()
	return h.kills
}

func (h *hostKillBackend) heal() {
	h.hmu.Lock()
	h.bad = false
	h.hmu.Unlock()
}

// Spec §7.4, §7.9: a worker runs in its host cgroup (plan 2 D4), so that is
// what proves it terminated, not a g<n> it never ran in (whose absence
// KillGen counts as empty). While the host cgroup still holds a process
// (a descendant the worker forked), a terminal stop keeps the lease and a
// respawn execs nothing beside it.
func TestWorkerTerminationProvesItsHostCgroup(t *testing.T) {
	t.Run("a terminal stop keeps the lease", func(t *testing.T) {
		a, b, _, _ := testActivator(t, time.Hour)
		a.gens = true
		hk := &hostKillBackend{fakeBackend: b, bad: true}
		a.backend = hk
		rt := testWorkerRoute("bot.example.com")
		if err := a.WakeWorker(rt); err != nil {
			t.Fatal(err)
		}
		rep := b.lastRep()
		a.Bounce(rt.Host)
		// The stop pends only once it has decided the lease stays: its proof
		// (KillHost) ran and failed, and the release was not made.
		waitFor(t, "the stop to pend", func() bool {
			a.mu.Lock()
			defer a.mu.Unlock()
			e := a.entries[rt.Host]
			return e != nil && e.pendingStop
		})
		if n := hk.killCount(); n == 0 {
			t.Fatal("the stop pended without asking the host cgroup's proof")
		}
		if rep.isClosed() {
			t.Fatal("the lease was released while the worker's host cgroup still held a process")
		}
		hk.heal()
		a.sweepOnce()
		waitFor(t, "the lease release", rep.isClosed)
	})
	t.Run("a respawn with no old handle still proves the host cgroup", func(t *testing.T) {
		a, b, _, _ := testActivator(t, time.Hour)
		a.gens = true
		hk := &hostKillBackend{fakeBackend: b}
		a.backend = hk
		rt := testWorkerRoute("bot3.example.com")
		if err := a.WakeWorker(rt); err != nil {
			t.Fatal(err)
		}
		hk.hmu.Lock()
		hk.bad = true
		hk.hmu.Unlock()
		a.mu.Lock()
		// main's signature; once plan 4 Task 9 has merged, pass nil as its
		// fourth (onto) argument.
		a.respawnWorkerLocked(a.entries[rt.Host], nil, "test: a restart with no old handle", nil)
		a.mu.Unlock()
		waitFor(t, "the respawn to finish", func() bool { return !a.entryRespawning(rt.Host) })
		if got := b.workerStarts(); got != 1 {
			t.Fatalf("worker starts = %d: a nil-stopOld respawn skipped the host cgroup proof", got)
		}
	})
	t.Run("a failed worker wake keeps its lease", func(t *testing.T) {
		old := sidecarSettle
		sidecarSettle = 20 * time.Millisecond
		t.Cleanup(func() { sidecarSettle = old })
		a, b, _, _ := testActivator(t, time.Hour)
		a.gens = true
		hk := &hostKillBackend{fakeBackend: b, bad: true}
		a.backend = hk
		b.mu.Lock()
		b.workerDeadOnStart = true // the wake fails its settle
		b.mu.Unlock()
		if err := a.WakeWorker(testWorkerRoute("bot4.example.com")); err == nil {
			t.Fatal("a worker dead on start woke")
		}
		// The lease goes back by Release, or by closing the replica when the
		// wake had one (cleanupFailedWake).
		given := func() bool {
			b.mu.Lock()
			released, reps := b.releases, append([]*fakeRep(nil), b.reps...)
			b.mu.Unlock()
			if released > 0 {
				return true
			}
			for _, r := range reps {
				if r.isClosed() {
					return true
				}
			}
			return false
		}
		if given() {
			t.Fatal("a failed worker wake gave its lease back over an unproven host cgroup")
		}
		hk.heal()
		a.sweepOnce()
		waitFor(t, "the lease given back", given)
	})
	t.Run("a respawn execs nothing beside the old child", func(t *testing.T) {
		old := sidecarSettle
		sidecarSettle = 20 * time.Millisecond
		t.Cleanup(func() { sidecarSettle = old })
		a, b, clk, _ := testActivator(t, time.Hour)
		a.gens = true
		hk := &hostKillBackend{fakeBackend: b}
		a.backend = hk
		rt := testWorkerRoute("bot2.example.com")
		if err := a.WakeWorker(rt); err != nil {
			t.Fatal(err)
		}
		hk.hmu.Lock()
		hk.bad = true
		hk.hmu.Unlock()
		b.lastWorker().crash()
		kills := hk.killCount()
		a.sweepOnce() // the respawn retires the crashed child, and its proof fails
		waitFor(t, "the respawn to finish", func() bool { return !a.entryRespawning(rt.Host) })
		if got := hk.killCount(); got <= kills {
			t.Fatalf("host cgroup kills = %d (was %d): the respawn never asked the old child's proof", got, kills)
		}
		if got := b.workerStarts(); got != 1 {
			t.Fatalf("worker starts = %d: a respawn exec'd beside an unproven old child", got)
		}
		// Positive control: once the proof succeeds, the next sweep's
		// crash restart respawns the worker.
		hk.heal()
		clk.advance(sidecarRestartMax + time.Second) // past the crash restart's backoff
		a.sweepOnce()
		waitFor(t, "the respawn after the proof healed", func() bool { return b.workerStarts() == 2 && !a.entryRespawning(rt.Host) })
	})
}

// Index P5/P6: every exec asks the gate's ownership check, whatever the
// switch says. A respawn whose lease has been lost execs nothing, with
// generations on or off.
func TestWorkerRespawnAsksOwnershipAtItsExec(t *testing.T) {
	t.Parallel()
	for _, on := range []bool{true, false} {
		a, b, _, _ := testActivator(t, time.Hour)
		a.gens = on
		rt := testWorkerRoute("bot5.example.com")
		if err := a.WakeWorker(rt); err != nil {
			t.Fatal(err)
		}
		b.lastRep().setErr(errors.New("renewal failed"))
		a.mu.Lock()
		a.respawnWorkerLocked(a.entries[rt.Host], nil, "test: the lease is lost", nil)
		a.mu.Unlock()
		waitFor(t, "the respawn to finish", func() bool { return !a.entryRespawning(rt.Host) })
		if got := b.workerStarts(); got != 1 {
			t.Fatalf("gens %v: worker starts = %d: a respawn exec'd without the lease", on, got)
		}
		if _, members := a.genStateForTest(rt.Host); len(members) != 0 {
			t.Fatalf("gens %v: members after the refused respawn = %v", on, members)
		}
	}
}