package activator

import (
	"errors"
	"slices"
	"sync"
	"testing"
	"time"

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

// Final review F1: a worker respawn asks its ownership again at the exec
// (composed into the before-exec hook, after the build's publication, or
// right before StartWorker on a backend without the staged capability),
// and once more before it installs the child. A lease lost while the
// respawn's preparation is under way (here, held open in the config
// admission) execs nothing; one lost after the exec installs nothing, and
// the child it made is shed. With generations on or off, on a backend with
// or without the staged capability.
func TestWorkerRespawnRechecksOwnershipAtTheExecAndTheInstall(t *testing.T) {
	t.Parallel()
	type rig struct {
		a      *Service
		b      *fakeBackend
		staged bool
	}
	setup := func(t *testing.T, on, staged bool) rig {
		a, b, _, _ := testActivator(t, time.Hour)
		a.gens = on
		if staged {
			a.backend = &stagedFake{fakeBackend: b}
		}
		return rig{a, b, staged}
	}
	for _, on := range []bool{true, false} {
		for _, staged := range []bool{true, false} {
			name := map[bool]string{true: "gens on", false: "gens off"}[on] + map[bool]string{true: ", staged", false: ", plain"}[staged]
			t.Run(name+": lost during the preparation", func(t *testing.T) {
				r := setup(t, on, staged)
				rt := testWorkerRoute("bot6.example.com")
				if err := r.a.WakeWorker(rt); err != nil {
					t.Fatal(err)
				}
				old := r.b.lastWorker()
				rep := r.b.lastRep()
				entered, release := make(chan struct{}), make(chan struct{})
				var once sync.Once
				r.b.mu.Lock()
				r.b.admitHook = func() {
					once.Do(func() { close(entered) })
					<-release
				}
				r.b.mu.Unlock()
				r.a.mu.Lock()
				r.a.respawnWorkerLocked(r.a.entries[rt.Host], nil, "test: the lease fails mid-preparation", nil)
				r.a.mu.Unlock()
				<-entered // past the first ownership check, inside the preparation
				rep.setErr(errors.New("renewal failed"))
				close(release)
				waitFor(t, "the respawn to finish", func() bool { return !r.a.entryRespawning(rt.Host) })
				if got := r.b.workerEnters(); got != 1 {
					t.Fatalf("worker spawns begun = %d: the respawn exec'd after its lease was lost", got)
				}
				r.a.mu.Lock()
				e := r.a.entries[rt.Host]
				proc := e.proc
				r.a.mu.Unlock()
				if proc != Proc(old) {
					t.Fatal("the respawn installed a child after its lease was lost")
				}
				if _, members := r.a.genStateForTest(rt.Host); len(members) != 0 {
					t.Fatalf("members after the refused respawn = %v", members)
				}
			})
			t.Run(name+": lost after the exec", func(t *testing.T) {
				r := setup(t, on, staged)
				rt := testWorkerRoute("bot7.example.com")
				if err := r.a.WakeWorker(rt); err != nil {
					t.Fatal(err)
				}
				old := r.b.lastWorker()
				rep := r.b.lastRep()
				r.b.mu.Lock()
				r.b.workerStartHook = func(_ registry.Route) { rep.setErr(errors.New("renewal failed")) }
				r.b.mu.Unlock()
				r.a.mu.Lock()
				r.a.respawnWorkerLocked(r.a.entries[rt.Host], nil, "test: the lease fails at the exec", nil)
				r.a.mu.Unlock()
				waitFor(t, "the respawn to finish", func() bool { return !r.a.entryRespawning(rt.Host) })
				if got := r.b.workerStarts(); got != 2 {
					t.Fatalf("worker starts = %d, want 2 (the wake and the respawn's exec)", got)
				}
				fresh := r.b.lastWorker()
				r.a.mu.Lock()
				proc := r.a.entries[rt.Host].proc
				r.a.mu.Unlock()
				if proc != Proc(old) || proc == Proc(fresh) {
					t.Fatal("the respawn installed its child after its lease was lost")
				}
				if !fresh.isStopped() {
					t.Fatal("the uninstalled child was left running")
				}
			})
		}
	}
}

// entryPendingStop reports whether host's entry is a stop held pending with
// its lease kept.
func (a *Service) entryPendingStop(host string) bool {
	a.mu.Lock()
	defer a.mu.Unlock()
	e := a.entries[host]
	return e != nil && e.pendingStop
}

func (a *Service) entryPresent(host string) bool {
	a.mu.Lock()
	defer a.mu.Unlock()
	return a.entries[host] != nil
}

// Final review R2-1: with generations off, a respawn child that exec'd and
// was then refused at its installation (here: the lease failed after the
// exec) stays the entry's to clean up until its exit is proven. Its build
// stays pinned, and the terminal stop that the lost lease begins keeps the
// lease and the entry until the child has exited, whether the shed Stop
// failed to prove it or is still blocked when the stop's bounded wait on
// the respawn runs out.
func TestAShedRespawnChildIsRetainedUntilItsExitIsProven(t *testing.T) {
	for _, blocked := range []bool{false, true} {
		name := map[bool]string{false: "the shed Stop fails", true: "the shed Stop blocks past the stop's respawn wait"}[blocked]
		t.Run(name, func(t *testing.T) {
			if blocked {
				old := respawnJoinBound
				respawnJoinBound = 100 * time.Millisecond
				t.Cleanup(func() { respawnJoinBound = old })
			}
			a, b, _, _, ver := stagedActivator(t)
			a.gens = false
			path := release.GenerationsPath(t.TempDir())
			a.GenerationsFile = path
			rt := alwaysOnWorkerRoute("bot10.example.com")
			if err := a.WakeWorker(rt); err != nil {
				t.Fatal(err)
			}
			old := b.lastWorker()
			rep := b.lastRep()
			ver.Store("v2") // the respawn (from the live link) execs v2
			gate := make(chan struct{})
			b.mu.Lock()
			if blocked {
				b.workerStopGate = gate
			} else {
				b.workerStuck = true
			}
			b.workerStartHook = func(registry.Route) { rep.setErr(errors.New("renewal failed")) }
			b.mu.Unlock()
			a.mu.Lock()
			e := a.entries[rt.Host]
			a.respawnWorkerLocked(e, nil, "test: shed after the exec", nil)
			done := e.respawnDone
			a.mu.Unlock()
			var fresh *fakeProc
			waitFor(t, "the respawn's exec", func() bool { fresh = b.lastWorker(); return fresh != old })
			if blocked {
				waitFor(t, "the shed Stop to block", func() bool { return fresh.stopEntered() > 0 })
			} else {
				<-done
			}
			if fresh.Exited() {
				t.Fatal("fixture: the shed child should still be running")
			}
			if got := generationsHost(t, a, path, rt.Host); !slices.Contains(got, "v2") {
				t.Errorf("generations file lists %v while the shed v2 child may still run; want v2 kept", got)
			}
			// The lost lease begins the terminal stop.
			a.sweepOnce()
			waitFor(t, "the stop to finish or be held", func() bool {
				return rep.isClosed() || !a.entryPresent(rt.Host) || a.entryPendingStop(rt.Host)
			})
			if rep.isClosed() || !a.entryPresent(rt.Host) {
				t.Fatalf("the stop released the lease (closed %v) and forgot the entry (present %v) while the shed child still runs", rep.isClosed(), a.entryPresent(rt.Host))
			}
			if !old.isStopped() {
				t.Fatal("the stop did not stop the installed child")
			}
			if got := generationsHost(t, a, path, rt.Host); !slices.Contains(got, "v2") {
				t.Fatalf("generations file lists %v during the held stop; want v2 kept", got)
			}
			// Proof: the shed child exits. The stop then finishes, and the
			// build is unpinned.
			if blocked {
				close(gate)
			} else {
				fresh.exit()
			}
			waitFor(t, "the stop to finish once the shed child is proven gone", func() bool {
				a.sweepOnce()
				return rep.isClosed() && !a.entryPresent(rt.Host)
			})
			if got := generationsHost(t, a, path, rt.Host); slices.Contains(got, "v2") {
				t.Fatalf("generations file lists %v after the shed child's exit; want v2 dropped", got)
			}
		})
	}
}