package activator

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

	"github.com/carlosframework/platform/internal/cgroup"
	"github.com/carlosframework/platform/internal/instancectl"
	"github.com/carlosframework/platform/internal/registry"
)

// Final review F8: SwapResult.Executed says whether a process of the
// attempt's build started. The scheduler re-arms a `carlos instances retry`
// allowance only for an attempt that ended before that (an abandonment with
// Executed false); an attempt whose build ran spends it, whatever its
// outcome. These cases cover each mode's refusals before the exec, its
// abandonments after the exec, and its executed outcomes.
func TestSwapResultReportsWhetherTheAttemptsBuildExecuted(t *testing.T) {
	t.Parallel()
	for _, tc := range []struct {
		name     string
		mode     SwapMode
		arrange  func(f *swapFixture)
		outcome  SwapOutcome
		executed bool
	}{
		{"overlap, refused at the exec", ModeOverlap, func(f *swapFixture) {
			f.b.onRaise = func() { f.rev.Add(1) }
		}, SwapAbandoned, false},
		{"overlap, the spawn fails", ModeOverlap, func(f *swapFixture) {
			f.b.genErr = errors.New("exec format error")
		}, SwapFailed, false},
		// Amadan #272: a parent that holds a legacy process directly refuses
		// the generation start before any exec. That is the box's condition,
		// not a failure of the build, so the attempt is abandoned unspent.
		{"overlap, the host's parent holds a process", ModeOverlap, func(f *swapFixture) {
			f.b.genErr = fmt.Errorf("h.example.com: %w: holds 1 process(es)", cgroup.ErrLegacyParentPopulated)
		}, SwapAbandoned, false},
		{"hold, the host's parent holds a process", ModeHold, func(f *swapFixture) {
			f.b.genErr = fmt.Errorf("h.example.com: %w: holds 1 process(es)", cgroup.ErrLegacyParentPopulated)
		}, SwapAbandoned, false},
		// Amadan #272, Task 19: the cgroup the successor would spawn into
		// took cgroup.kill and could not be made fresh. The manager refuses
		// before it builds the command: the box's condition, nothing ran.
		{"overlap, a killed cgroup would not go", ModeOverlap, func(f *swapFixture) {
			f.b.genErr = fmt.Errorf("%w: carlos-exec@h.example.com/g2: device or resource busy", cgroup.ErrKilledCgroupKept)
		}, SwapAbandoned, false},
		{"hold, a killed cgroup would not go", ModeHold, func(f *swapFixture) {
			f.b.genErr = fmt.Errorf("%w: carlos-exec@h.example.com: device or resource busy", cgroup.ErrKilledCgroupKept)
		}, SwapAbandoned, false},
		// Task 7 (readiness round 2): the successor's limits could not be
		// applied, so the manager refused before it built the command: the
		// box's condition, nothing ran, the allowance is not spent.
		{"overlap, the generation's limits were not applied", ModeOverlap, func(f *swapFixture) {
			f.b.genErr = fmt.Errorf("%w: g2: memory.max: is a directory", cgroup.ErrLimitNotApplied)
		}, SwapAbandoned, false},
		{"hold, the generation's limits were not applied", ModeHold, func(f *swapFixture) {
			f.b.genErr = fmt.Errorf("%w: h.example.com: the placement shim is not available", cgroup.ErrLimitNotApplied)
		}, SwapAbandoned, false},
		{"overlap, superseded at the switch", ModeOverlap, func(f *swapFixture) {
			f.a.genHealth = func(context.Context, Gen) error { f.rev.Add(1); return nil }
		}, SwapAbandoned, true},
		{"overlap, done", ModeOverlap, func(*swapFixture) {}, SwapDone, true},
		{"hold, superseded before the exec", ModeHold, func(f *swapFixture) {
			var once sync.Once
			f.a.backend = &acquireHookBackend{fakeSwapBackend: f.b, hook: func() { once.Do(func() { f.rev.Add(1) }) }}
		}, SwapAbandoned, false},
		{"hold, superseded before the result", ModeHold, func(f *swapFixture) {
			f.a.genHealth = func(context.Context, Gen) error { f.rev.Add(1); return nil }
		}, SwapAbandoned, true},
		{"hold, unhealthy", ModeHold, func(f *swapFixture) {
			f.a.genHealth = func(context.Context, Gen) error { return errors.New("unhealthy: HTTP 500") }
		}, SwapDown, true},
		{"hold, done", ModeHold, func(*swapFixture) {}, SwapDone, true},
	} {
		t.Run(tc.name, func(t *testing.T) {
			f := newSwapFixture(t)
			tc.arrange(f)
			res := f.a.Swap(context.Background(), f.attempt(tc.mode), f.newest)
			if res.Outcome != tc.outcome {
				t.Fatalf("Swap = %+v, want %s", res, tc.outcome)
			}
			if res.Executed != tc.executed {
				t.Fatalf("Swap = %+v: Executed %v, want %v", res, res.Executed, tc.executed)
			}
		})
	}
}

// The same for a worker's same-entry replacement, from the respawn's
// execution evidence (final review F2's attemptExec).
func TestReplaceWorkerReportsWhetherTheAttemptsBuildExecuted(t *testing.T) {
	for _, tc := range []struct {
		name     string
		mode     SwapMode
		arrange  func(t *testing.T, a *Service, b *stagedFake, rev *atomic.Uint64)
		outcome  SwapOutcome
		executed bool
	}{
		{"hold, superseded at its exec gate", ModeHold, func(_ *testing.T, _ *Service, b *stagedFake, rev *atomic.Uint64) {
			var hooked atomic.Bool
			b.mu.Lock()
			b.admitHook = func() {
				if hooked.CompareAndSwap(false, true) {
					rev.Store(2)
				}
			}
			b.mu.Unlock()
		}, SwapAbandoned, false},
		{"hold, superseded after its exec", ModeHold, func(_ *testing.T, _ *Service, b *stagedFake, rev *atomic.Uint64) {
			var hooked atomic.Bool
			b.mu.Lock()
			b.workerStartHook = func(registry.Route) {
				if hooked.CompareAndSwap(false, true) {
					rev.Store(2)
				}
			}
			b.mu.Unlock()
		}, SwapAbandoned, true},
		{"refused for its publication", ModeOverlap, func(t *testing.T, a *Service, _ *stagedFake, _ *atomic.Uint64) {
			blocker := filepath.Join(t.TempDir(), "not-a-dir")
			if err := os.WriteFile(blocker, nil, 0o644); err != nil {
				t.Fatal(err)
			}
			a.mu.Lock()
			a.GenerationsFile = filepath.Join(blocker, "generations.json")
			a.mu.Unlock()
		}, SwapAbandoned, false},
		{"done", ModeOverlap, func(*testing.T, *Service, *stagedFake, *atomic.Uint64) {}, SwapDone, true},
	} {
		t.Run(tc.name, func(t *testing.T) {
			shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond)
			a, b, _, _, _ := stagedActivator(t)
			host := "bot.x.internal"
			if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
				t.Fatal(err)
			}
			var rev atomic.Uint64
			rev.Store(1)
			tc.arrange(t, a, b, &rev)
			res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, tc.mode, fakeIdentity("v2")), nil,
				func(string) uint64 { return rev.Load() })
			if res.Outcome != tc.outcome {
				t.Fatalf("ReplaceWorker = %+v, want %s", res, tc.outcome)
			}
			if res.Executed != tc.executed {
				t.Fatalf("ReplaceWorker = %+v: Executed %v, want %v", res, res.Executed, tc.executed)
			}
			waitFor(t, "the respawns to finish", func() bool { return !a.entryRespawning(host) })
		})
	}
}

// Final review wave 2 (N4): Executed is false only when it is PROVEN that
// nothing of the attempt's build ran. A nested start that root was asked
// for and that failed with no process (the unit exited before it was
// observed, or root's start answered an error), or whose kill is
// unproven, may have run the build: executed or unknown. Only root's
// refusal of the staged build before it started anything (ErrStagedExe)
// is proven unexecuted. Before the fix both modes read the nil handle of
// a kill-proven failure as "never ran", and an overlap read an unproven
// kill the same way, so an abandoned forced attempt regained an allowance
// it may have spent.
func TestASwapWhoseNestedStartIsUncertainReportsExecution(t *testing.T) {
	t.Parallel()
	unitFailed := errors.New("the unit failed to start")
	for _, tc := range []struct {
		name     string
		mode     SwapMode
		exeErr   error
		stopErr  error
		executed bool
	}{
		{"overlap, root's start failed, kill proven", ModeOverlap, unitFailed, nil, true},
		{"overlap, root's start failed, kill unproven", ModeOverlap, unitFailed, errors.New("stop timed out"), true},
		{"overlap, root refused the staged build", ModeOverlap, fmt.Errorf("%w: digest mismatch", instancectl.ErrStagedExe), nil, false},
		{"hold, root's start failed, kill proven", ModeHold, unitFailed, nil, true},
		{"hold, root refused the staged build", ModeHold, fmt.Errorf("%w: digest mismatch", instancectl.ErrStagedExe), nil, false},
	} {
		t.Run(tc.name, func(t *testing.T) {
			f := newSwapFixture(t)
			b := newLaunchBackend(t, "nested", f.b)
			b.ctl.exeErr, b.ctl.stopGenErr = tc.exeErr, tc.stopErr
			f.a.backend = b
			id := markerScript(t, t.TempDir(), "v2", filepath.Join(t.TempDir(), "v2.ran"))
			at := f.attempt(tc.mode)
			at.StagedPath, at.Digest = id.StagedPath, id.Digest
			res := f.a.Swap(context.Background(), at, f.newest)
			if res.Outcome == SwapDone {
				t.Fatalf("Swap = %+v, want it not done", res)
			}
			b.ctl.mu.Lock()
			asked := len(b.ctl.got)
			b.ctl.mu.Unlock()
			if asked != 1 {
				t.Fatalf("root was asked to start %d units, want 1", asked)
			}
			if res.Executed != tc.executed {
				t.Fatalf("Swap = %+v: Executed %v, want %v", res, res.Executed, tc.executed)
			}
		})
	}
}

// Astra R3 ruling 206 (zdd-readiness Task 6): a hold's nested start that
// root accepted and whose outcome is uncertain may have run its build, so
// the hold is down (SwapDown: done() holds the build down and halts the
// cohort), never an abandonment with Executed, whose deferral would let the
// scheduler run the build again. That holds for a unit that exited before
// it was observed (errStartedThenExited) and for any other uncertain start,
// whether the backend's kill proved the generation gone or not. Only root's
// refusal of the staged build before it started anything is still an
// abandonment, with nothing executed. Before the fix the kill-proven cases
// were abandoned with Executed true.
func TestAHoldWhoseNestedStartIsUncertainIsDown(t *testing.T) {
	t.Parallel()
	unitFailed := errors.New("the unit failed to start")
	exitedUnobserved := func(instancectl.StagedExe) (string, error) { return "", nil }
	rootFailed := func(instancectl.StagedExe) (string, error) { return "", unitFailed }
	rootRefused := func(instancectl.StagedExe) (string, error) {
		return "", fmt.Errorf("%w: digest mismatch", instancectl.ErrStagedExe)
	}
	for _, tc := range []struct {
		name     string
		answer   func(instancectl.StagedExe) (string, error)
		stopErr  error
		outcome  SwapOutcome
		executed bool
	}{
		{"the unit exited before it was observed, kill proven", exitedUnobserved, nil, SwapDown, true},
		{"the unit exited before it was observed, kill unproven", exitedUnobserved, errors.New("stop timed out"), SwapDown, true},
		{"root's start failed, kill proven", rootFailed, nil, SwapDown, true},
		{"root's start failed, kill unproven", rootFailed, errors.New("stop timed out"), SwapDown, true},
		{"root refused the staged build", rootRefused, nil, SwapAbandoned, false},
	} {
		t.Run(tc.name, func(t *testing.T) {
			f := newSwapFixture(t)
			b := newLaunchBackend(t, "nested", f.b)
			b.ctl.exeAnswer, b.ctl.stopGenErr = tc.answer, tc.stopErr
			f.a.backend = b
			id := markerScript(t, t.TempDir(), "v2", filepath.Join(t.TempDir(), "v2.ran"))
			at := f.attempt(ModeHold)
			at.StagedPath, at.Digest = id.StagedPath, id.Digest
			res := f.a.Swap(context.Background(), at, f.newest)
			if res.Outcome != tc.outcome || res.Executed != tc.executed {
				t.Fatalf("Swap = %+v, want %s with Executed %v", res, tc.outcome, tc.executed)
			}
			b.ctl.mu.Lock()
			asked := len(b.ctl.got)
			b.ctl.mu.Unlock()
			if asked != 1 {
				t.Fatalf("root was asked to start %d units, want 1", asked)
			}
		})
	}
}

// Final review wave 2 (N4): ReplaceWorker learns what its attempt's
// respawn did before it answers, even when its caller stops waiting. The
// respawn runs on (it is the entry's, on the service lifetime), so it is
// joined, bounded by respawnJoinBound, and one still running past that is
// reported executed-or-unknown. Before the fix a cancelled ReplaceWorker
// returned at once with Executed false, while the respawn went on to exec
// the attempt's build.
func TestReplaceWorkerLearnsItsRespawnsExecutionWhenItsCallerLeaves(t *testing.T) {
	for _, tc := range []struct {
		name string
		// supersede: the attempt is superseded while the respawn is held, so
		// its exec gate refuses: proven unexecuted.
		supersede bool
		// bound shrinks respawnJoinBound below the hold: the join gives up.
		bound    time.Duration
		executed bool
	}{
		{"the respawn execs after the caller left", false, 0, true},
		{"the respawn is refused at its exec after the caller left", true, 0, false},
		{"the respawn outlasts the join bound", false, 150 * time.Millisecond, true},
	} {
		t.Run(tc.name, func(t *testing.T) {
			if tc.bound > 0 {
				defer func(d time.Duration) { respawnJoinBound = d }(respawnJoinBound)
				respawnJoinBound = tc.bound
			}
			shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond)
			a, b, _, _, _ := stagedActivator(t)
			host := "bot.x.internal"
			if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
				t.Fatal(err)
			}
			var rev atomic.Uint64
			rev.Store(1)
			entered, release := make(chan struct{}), make(chan struct{})
			var releaseOnce sync.Once
			releaseAll := func() { releaseOnce.Do(func() { close(release) }) }
			t.Cleanup(releaseAll)
			var hooked atomic.Bool
			b.mu.Lock()
			b.admitHook = func() {
				if hooked.CompareAndSwap(false, true) {
					close(entered)
					<-release
				}
			}
			b.mu.Unlock()
			ctx, cancel := context.WithCancel(context.Background())
			defer cancel()
			out := make(chan SwapResult, 1)
			go func() {
				out <- a.ReplaceWorker(ctx, workerAttempt(CauseRelease, ModeHold, fakeIdentity("v2")), nil,
					func(string) uint64 { return rev.Load() })
			}()
			select {
			case <-entered:
			case <-time.After(5 * time.Second):
				t.Fatal("the attempt's respawn never reached its preparation")
			}
			cancel()
			if tc.bound > 0 {
				// The respawn stays held past the bound: the join gives up.
				select {
				case res := <-out:
					if res.Outcome != SwapAbandoned || res.Executed != tc.executed {
						t.Fatalf("ReplaceWorker = %+v: want abandoned, Executed %v", res, tc.executed)
					}
				case <-time.After(5 * time.Second):
					t.Fatal("ReplaceWorker never answered: the join is not bounded")
				}
				releaseAll()
				waitFor(t, "the respawn to finish", func() bool { return !a.entryRespawning(host) })
				return
			}
			select {
			case res := <-out:
				t.Fatalf("ReplaceWorker = %+v answered while its respawn was still preparing", res)
			case <-time.After(200 * time.Millisecond):
			}
			if tc.supersede {
				rev.Store(2)
			}
			releaseAll()
			select {
			case res := <-out:
				if res.Outcome != SwapAbandoned || res.Executed != tc.executed {
					t.Fatalf("ReplaceWorker = %+v: want abandoned, Executed %v", res, tc.executed)
				}
			case <-time.After(5 * time.Second):
				t.Fatal("ReplaceWorker never answered after its respawn finished")
			}
			waitFor(t, "the respawns to finish", func() bool { return !a.entryRespawning(host) })
		})
	}
}