package activator

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

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

// cmdSwapBackend is the swap fixture's backend with the run capability, on
// the nested layout (merge ruling M2).
type cmdSwapBackend struct {
	*fakeSwapBackend
	cmu     sync.Mutex
	unit    *fakeCmdUnit
	specs   []CmdSpec
	gate    chan struct{} // when set, StartCmd blocks until it closes
	entered chan struct{} // when set, closed as StartCmd is entered
	onStart func()        // when set, runs inside StartCmd, before the unit is handed back
}

func (b *cmdSwapBackend) StartCmd(_ context.Context, _ registry.Route, spec CmdSpec) (CmdUnit, error) {
	b.cmu.Lock()
	gate, entered, onStart := b.gate, b.entered, b.onStart
	b.entered = nil
	b.cmu.Unlock()
	if entered != nil {
		close(entered)
	}
	if onStart != nil {
		onStart()
	}
	if gate != nil {
		<-gate
	}
	b.cmu.Lock()
	defer b.cmu.Unlock()
	b.specs = append(b.specs, spec)
	return b.unit, nil
}

func (b *cmdSwapBackend) started() []CmdSpec {
	b.cmu.Lock()
	defer b.cmu.Unlock()
	return slices.Clone(b.specs)
}

func newCmdSwapFixture(t *testing.T) (*swapFixture, *cmdSwapBackend) {
	t.Helper()
	var cb *cmdSwapBackend
	f := newSwapFixtureOn(t, func(b *fakeSwapBackend) Backend {
		cb = &cmdSwapBackend{fakeSwapBackend: b, unit: &fakeCmdUnit{}}
		return cb
	})
	f.b.layoutFn = func(registry.Route) (Layout, string) { return LayoutNested, "" }
	return f, cb
}

func (a *Service) runGenOf(h CmdHandle) Gen {
	a.mu.Lock()
	defer a.mu.Unlock()
	return h.(*cmdRun).h.gen
}

// A run whose unit is starting refuses an overlap swap's admission: the
// attempt is abandoned with nothing spawned, never admitted to retire the
// generation that run is starting against. Once the start has returned,
// the same attempt is admitted (merge ruling M2).
func TestARunStartInFlightRefusesAnOverlapSwap(t *testing.T) {
	f, cb := newCmdSwapFixture(t)
	cb.gate, cb.entered = make(chan struct{}), make(chan struct{})
	entered := cb.entered
	type ran struct {
		h   CmdHandle
		err error
	}
	done := make(chan ran, 1)
	go func() {
		h, err := f.a.RunCmd(context.Background(), testRoute(f.host), CmdSpec{RunID: "run_a"})
		done <- ran{h, err}
	}()
	<-entered
	res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest)
	if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errSwapBusy.Error()) || res.Executed {
		t.Fatalf("Swap during the run's start = %+v, want abandoned as busy with nothing run", res)
	}
	if gens, raises, _ := f.b.calls(); gens != 0 || raises != 0 {
		t.Fatalf("the refused attempt spawned %d generations and raised %d times", gens, raises)
	}
	close(cb.gate)
	r := <-done
	if r.err != nil {
		t.Fatal(r.err)
	}
	if g := f.a.runGenOf(r.h); g.N != f.incumbent.N {
		t.Fatalf("the run was admitted against generation %d, want the incumbent %d", g.N, f.incumbent.N)
	}
	f.rev.Add(1)
	if res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest); res.Outcome != SwapDone {
		t.Fatalf("Swap after the start = %+v", res)
	}
}

// A run that arrives while an overlap successor is spawning starts no unit
// against the incumbent, which is about to retire: it waits for the switch
// and then runs the successor's build, admitted against the successor's
// generation (merge ruling M2).
func TestARunWaitsForAPendingSwitch(t *testing.T) {
	t.Parallel()
	f, cb := newCmdSwapFixture(t)
	f.b.gmu.Lock()
	f.b.genGate = make(chan struct{})
	f.b.gmu.Unlock()
	swapped := make(chan SwapResult, 1)
	go func() { swapped <- f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest) }()
	waitFor(t, "the successor in the operation set", func() bool {
		_, members := f.a.genStateForTest(f.host)
		return len(members) == 1
	})
	type ran struct {
		h   CmdHandle
		err error
	}
	done := make(chan ran, 1)
	go func() {
		h, err := f.a.RunCmd(context.Background(), testRoute(f.host), CmdSpec{RunID: "run_a"})
		done <- ran{h, err}
	}()
	time.Sleep(3 * cmdSwitchPoll)
	if n := len(cb.started()); n != 0 {
		t.Fatalf("%d units started while a successor was pending", n)
	}
	f.b.gmu.Lock()
	close(f.b.genGate)
	f.b.gmu.Unlock()
	res := <-swapped
	if res.Outcome != SwapDone {
		t.Fatalf("Swap = %+v", res)
	}
	r := <-done
	if r.err != nil {
		t.Fatal(r.err)
	}
	defer r.h.Release()
	if g := f.a.runGenOf(r.h); g.N != res.Gen.N {
		t.Fatalf("the run was admitted against generation %d, want the successor %d", g.N, res.Gen.N)
	}
	specs := cb.started()
	if len(specs) != 1 || specs[0].serving == nil || specs[0].serving.Path != f.staged {
		t.Fatalf("the run started %+v, want one unit on the successor's build %s", specs, f.staged)
	}
	if n, _ := cb.unit.stopped(); n != 0 {
		t.Fatal("a run on the current generation was interrupted")
	}
}

// A running command on a generation a swap retires is interrupted and
// settled, and the run reads why: a new release for a release attempt, a
// restart for a scheduled restart or config change (exit 125, RUN-C75;
// merge ruling M2).
func TestARunOnARetiredGenerationIsInterrupted(t *testing.T) {
	t.Parallel()
	for _, c := range []struct {
		cause Cause
		why   string
	}{{CauseRelease, StopRelease}, {CauseRestart, StopRestart}, {CauseConfig, StopRestart}} {
		t.Run(string(c.cause), func(t *testing.T) {
			f, cb := newCmdSwapFixture(t)
			h, err := f.a.RunCmd(context.Background(), testRoute(f.host), CmdSpec{RunID: "run_a"})
			if err != nil {
				t.Fatal(err)
			}
			if g := f.a.runGenOf(h); g.N != f.incumbent.N || g.StagedPath != f.incumbent.StagedPath {
				t.Fatalf("the run was admitted against %+v, want the incumbent", g)
			}
			at := f.attempt(ModeOverlap)
			at.Cause = c.cause
			if res := f.a.Swap(context.Background(), at, f.newest); res.Outcome != SwapDone {
				t.Fatalf("Swap = %+v", res)
			}
			waitFor(t, "the run interrupted and settled", func() bool {
				n, _ := cb.unit.stopped()
				return n == 1 && h.(*cmdRun).h.provenSettled()
			})
			if why, at := h.Interrupted(); why != c.why || at.IsZero() {
				t.Fatalf("Interrupted() = %q at %v, want %q", why, at, c.why)
			}
			if f.a.entryCount() != 1 {
				t.Fatal("interrupting the run stopped the instance")
			}
			h.Release()
		})
	}
}

// A run on a retired generation keeps that generation's build in the
// generations file until the run is proven settled, so root neither
// restages nor sweeps the bytes the command runs; the settle that proves
// it releases the pin (merge ruling M2).
func TestARetiredRunKeepsItsBuildPinnedUntilItSettles(t *testing.T) {
	t.Parallel()
	f, cb := newCmdSwapFixture(t)
	path := release.GenerationsPath(t.TempDir())
	f.a.GenerationsFile = path
	cb.unit.setSettleErr(errors.New("processes remained"))
	h, err := f.a.RunCmd(context.Background(), testRoute(f.host), CmdSpec{RunID: "run_a"})
	if err != nil {
		t.Fatal(err)
	}
	res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest)
	if res.Outcome != SwapDone {
		t.Fatalf("Swap = %+v", res)
	}
	waitFor(t, "the interrupt's failed settle", func() bool { n, _ := cb.unit.stopped(); return n >= 1 })
	pinned := func() bool { return slices.Contains(readPinPaths(t, path, f.host), f.incumbent.StagedPath) }
	if err := f.a.persistGenerations(); err != nil {
		t.Fatal(err)
	}
	_, members := f.a.genStateForTest(f.host)
	if len(members) != 0 {
		t.Fatalf("the retiring generation is still a member: %v", members)
	}
	if !pinned() {
		t.Fatal("the retired build went from the generations file while its run was unsettled")
	}
	cb.unit.setSettleErr(nil)
	if err := h.Stop(context.Background()); err != nil {
		t.Fatalf("Stop: %v", err)
	}
	waitFor(t, "the pin released", func() bool { return !pinned() })
	h.Release()
}

// A run waiting for a switch carries its own deadline and cancellation:
// cancelled or expired while the successor is held, it starts nothing,
// before or after the switch, and returns its context's error. A Recheck
// that refuses after the wait starts nothing either (re-review 5).
func TestARunEndedDuringASwitchWaitNeverStarts(t *testing.T) {
	t.Parallel()
	for _, c := range []struct {
		name    string
		ctx     func() (context.Context, context.CancelFunc)
		cancel  bool
		recheck error
		want    error
	}{
		{name: "cancelled", ctx: func() (context.Context, context.CancelFunc) { return context.WithCancel(context.Background()) }, cancel: true, want: context.Canceled},
		{name: "expired", ctx: func() (context.Context, context.CancelFunc) {
			return context.WithTimeout(context.Background(), 3*cmdSwitchPoll)
		}, want: context.DeadlineExceeded},
		{name: "recheck refuses", ctx: func() (context.Context, context.CancelFunc) { return context.WithCancel(context.Background()) },
			recheck: errors.New("the record no longer names the run"), want: nil},
	} {
		t.Run(c.name, func(t *testing.T) {
			f, cb := newCmdSwapFixture(t)
			f.b.gmu.Lock()
			f.b.genGate = make(chan struct{})
			f.b.gmu.Unlock()
			swapped := make(chan SwapResult, 1)
			go func() { swapped <- f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest) }()
			waitFor(t, "the successor in the operation set", func() bool {
				_, members := f.a.genStateForTest(f.host)
				return len(members) == 1
			})
			ctx, cancel := c.ctx()
			defer cancel()
			var rechecks int
			spec := CmdSpec{RunID: "run_a"}
			if c.recheck != nil {
				spec.Recheck = func(context.Context) error { rechecks++; return c.recheck }
			}
			done := make(chan error, 1)
			go func() {
				_, err := f.a.RunCmd(ctx, testRoute(f.host), spec)
				done <- err
			}()
			waitFor(t, "the run waiting", func() bool {
				f.a.mu.Lock()
				defer f.a.mu.Unlock()
				e := f.a.entries[f.host]
				return e != nil && e.cmd != nil
			})
			time.Sleep(cmdSwitchPoll)
			if c.cancel {
				cancel()
			}
			if c.want != nil {
				if err := <-done; !errors.Is(err, c.want) {
					t.Fatalf("RunCmd = %v, want %v", err, c.want)
				}
			}
			f.b.gmu.Lock()
			close(f.b.genGate)
			f.b.gmu.Unlock()
			if res := <-swapped; res.Outcome != SwapDone {
				t.Fatalf("Swap = %+v", res)
			}
			if c.want == nil {
				if err := <-done; !errors.Is(err, c.recheck) || rechecks != 1 {
					t.Fatalf("RunCmd = %v after %d rechecks, want the recheck's refusal after one", err, rechecks)
				}
			}
			time.Sleep(2 * cmdSwitchPoll)
			if n := len(cb.started()); n != 0 {
				t.Fatalf("%d units started for a run that ended before its launch", n)
			}
			f.a.mu.Lock()
			attached := f.a.entries[f.host].cmd != nil
			f.a.mu.Unlock()
			if attached {
				t.Fatal("the ended run is still attached to the instance")
			}
		})
	}
}

// A run on a generation a swap retires, whose interruption's stop fails
// once while the instance stays awake: the sweeps alone retry the settle,
// actStopTimeout apart, prove it, drop the retired build's pin, and the
// next run is admitted. The run's handle and the pin stay until then
// (re-review 6).
func TestARetiredRunsFailedStopIsSettledBySweeps(t *testing.T) {
	t.Parallel()
	for _, released := range []bool{false, true} {
		t.Run(map[bool]string{false: "attached", true: "released"}[released], func(t *testing.T) {
			f, cb := newCmdSwapFixture(t)
			path := release.GenerationsPath(t.TempDir())
			f.a.GenerationsFile = path
			cb.unit.setSettleErr(errors.New("instance control: 502 mid-roll"))
			h, err := f.a.RunCmd(context.Background(), testRoute(f.host), CmdSpec{RunID: "run_a"})
			if err != nil {
				t.Fatal(err)
			}
			if res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest); res.Outcome != SwapDone {
				t.Fatalf("Swap = %+v", res)
			}
			waitFor(t, "the interruption's failed stop", func() bool {
				n, _ := cb.unit.stopped()
				return n == 1 && h.(*cmdRun).h.settlingDone()
			})
			if why, _ := h.Interrupted(); why != StopRelease {
				t.Fatalf("Interrupted() = %q, want %q", why, StopRelease)
			}
			if released {
				h.Release() // the worker wrote 125 and let go, its own Stop failed too
			}
			if _, err := f.a.RunCmd(context.Background(), testRoute(f.host), CmdSpec{RunID: "run_b"}); !errors.Is(err, ErrCmdBusy) {
				t.Fatalf("a run while the retired one is unproven: %v, want ErrCmdBusy", err)
			}
			pinned := func() bool {
				if err := f.a.persistGenerations(); err != nil {
					t.Fatal(err)
				}
				return slices.Contains(readPinPaths(t, path, f.host), f.incumbent.StagedPath)
			}
			if !pinned() {
				t.Fatal("the retired build lost its pin while its run was unproven")
			}
			cb.unit.setSettleErr(nil)
			f.a.sweepOnce() // within actStopTimeout of the failure: not yet
			time.Sleep(50 * time.Millisecond)
			if n, _ := cb.unit.stopped(); n != 1 {
				t.Fatalf("the sweep retried %d times inside actStopTimeout", n-1)
			}
			f.clk.advance(actStopTimeout)
			f.a.sweepOnce()
			waitFor(t, "the sweep's retry proving the run stopped", func() bool { return h.(*cmdRun).h.provenSettled() })
			if n, _ := cb.unit.stopped(); n != 2 {
				t.Fatalf("stops %d, want the failed one and the sweep's", n)
			}
			if pinned() {
				t.Fatal("the retired build is still pinned after its run was proven stopped")
			}
			if !released {
				h.Release()
			}
			h2, err := f.a.RunCmd(context.Background(), testRoute(f.host), CmdSpec{RunID: "run_b"})
			if err != nil {
				t.Fatalf("the next run: %v", err)
			}
			h2.Release()
			if f.a.entryCount() != 1 {
				t.Fatal("the instance stopped")
			}
		})
	}
}

// settlingDone reports whether no settle attempt is in flight.
func (h *cmdHold) settlingDone() bool {
	h.mu.Lock()
	defer h.mu.Unlock()
	return h.settling == nil
}

// readPinPaths is host's staged paths in the generations file. A read
// that meets a write in progress (another goroutine republishing) is
// tried again, as root would on its next pass.
func readPinPaths(t *testing.T, path, host string) []string {
	t.Helper()
	deadline := time.Now().Add(5 * time.Second)
	for {
		g, err := release.ReadGenerations(path)
		if err == nil && g != nil {
			return g.Paths[host]
		}
		if time.Now().After(deadline) {
			t.Fatalf("ReadGenerations: %v", err)
		}
		time.Sleep(10 * time.Millisecond)
	}
}

// buildRecorder is a Service.BeforeExec that keeps every record in order:
// its last entry for a host is that host's last_executed.
type buildRecorder struct {
	mu     sync.Mutex
	builds []string
	err    error
	hook   func() // runs inside the write, when set
}

func (r *buildRecorder) record(_ context.Context, _, build string) error {
	r.mu.Lock()
	hook, err := r.hook, r.err
	if err == nil {
		r.builds = append(r.builds, build)
	}
	r.mu.Unlock()
	if hook != nil {
		hook()
	}
	return err
}

func (r *buildRecorder) last() string {
	r.mu.Lock()
	defer r.mu.Unlock()
	if len(r.builds) == 0 {
		return ""
	}
	return r.builds[len(r.builds)-1]
}

// A successor B records its build and then fails its health check, so the
// incumbent A goes on serving under a build state that names B. A command
// then runs A: its build is recorded before its unit starts, so the build
// state names A again, as it does for the sidecar (settings spec §4.2;
// second merge review, finding 1).
func TestARunRecordsItsServingBuildBeforeItStarts(t *testing.T) {
	f, cb := newCmdSwapFixture(t)
	rec := &buildRecorder{}
	f.a.BeforeExec = rec.record
	f.a.genHealth = func(context.Context, Gen) error { return errors.New("502 from the successor") }
	if res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest); res.Outcome != SwapFailed || !res.Executed {
		t.Fatalf("Swap = %+v, want the successor to have run and failed", res)
	}
	if got := rec.last(); got != "v2" {
		t.Fatalf("after the failed successor the build state names %q, want its v2", got)
	}
	if cur, _ := f.a.genStateForTest(f.host); cur.N != f.incumbent.N {
		t.Fatalf("generation %d is serving, want the incumbent", cur.N)
	}
	// What the build state names at the moment the unit starts.
	var atStart string
	cb.cmu.Lock()
	cb.onStart = func() { atStart = rec.last() }
	cb.cmu.Unlock()
	h, err := f.a.RunCmd(context.Background(), testRoute(f.host), CmdSpec{RunID: "run_a"})
	if err != nil {
		t.Fatal(err)
	}
	defer h.Release()
	if got := rec.last(); got != f.incumbent.Build {
		t.Fatalf("after the command the build state names %q, want the serving %q", got, f.incumbent.Build)
	}
	if specs := cb.started(); len(specs) != 1 || specs[0].serving == nil || specs[0].serving.Path != f.incumbent.StagedPath {
		t.Fatalf("the command started %+v, want one unit on the incumbent's build", specs)
	}
	if atStart != f.incumbent.Build {
		t.Fatalf("the unit started with the build state naming %q: the record must come first", atStart)
	}
}

// A build state that cannot be written launches no command (settings spec
// §4.2): the run is refused, the hold is released, and the instance and its
// next run are unaffected. A stop, or the run's own cancel, that lands
// during the write starts nothing either.
func TestARunWhoseBuildRecordFailsOrIsOvertakenNeverStarts(t *testing.T) {
	t.Parallel()
	t.Run("the write fails", func(t *testing.T) {
		f, cb := newCmdSwapFixture(t)
		rec := &buildRecorder{err: errors.New("bucket: 503")}
		f.a.BeforeExec = rec.record
		if _, err := f.a.RunCmd(context.Background(), testRoute(f.host), CmdSpec{RunID: "run_a"}); !errors.Is(err, errBuildStateUnwritten) {
			t.Fatalf("RunCmd = %v, want errBuildStateUnwritten", err)
		}
		if n := len(cb.started()); n != 0 {
			t.Fatalf("%d units started over an unwritten build state", n)
		}
		rec.mu.Lock()
		rec.err = nil
		rec.mu.Unlock()
		h, err := f.a.RunCmd(context.Background(), testRoute(f.host), CmdSpec{RunID: "run_b"})
		if err != nil {
			t.Fatalf("the next run: %v (the refused one must leave no hold)", err)
		}
		h.Release()
	})
	t.Run("a stop lands during the write", func(t *testing.T) {
		f, cb := newCmdSwapFixture(t)
		rec := &buildRecorder{}
		rec.hook = func() { f.a.Bounce(f.host) }
		f.a.BeforeExec = rec.record
		_, err := f.a.RunCmd(context.Background(), testRoute(f.host), CmdSpec{RunID: "run_a"})
		var se *CmdStoppedError
		if !errors.As(err, &se) || se.Reason != StopRestart {
			t.Fatalf("RunCmd = %v, want CmdStoppedError{%q}", err, StopRestart)
		}
		if n := len(cb.started()); n != 0 {
			t.Fatalf("%d units started under a stopping instance", n)
		}
		rec.mu.Lock()
		rec.hook = nil
		rec.mu.Unlock()
		waitFor(t, "the instance stopped", func() bool { return f.a.entryCount() == 0 })
	})
	t.Run("the run is cancelled during the write", func(t *testing.T) {
		f, cb := newCmdSwapFixture(t)
		ctx, cancel := context.WithCancel(context.Background())
		defer cancel()
		rec := &buildRecorder{hook: cancel}
		f.a.BeforeExec = rec.record
		if _, err := f.a.RunCmd(ctx, testRoute(f.host), CmdSpec{RunID: "run_a"}); !errors.Is(err, context.Canceled) {
			t.Fatalf("RunCmd = %v, want context.Canceled", err)
		}
		if n := len(cb.started()); n != 0 {
			t.Fatalf("%d units started for a cancelled run", n)
		}
		f.a.mu.Lock()
		attached := f.a.entries[f.host].cmd != nil
		f.a.mu.Unlock()
		if attached {
			t.Fatal("the cancelled run is still attached")
		}
	})
}