package activator

import (
	"context"
	"errors"
	"log"
	"os"
	"slices"
	"strings"
	"sync"
	"testing"
	"time"
)

// Discussion #292, the production sequence with swaps off. `systemctl
// restart carlos-edge` stops every carlos-instance@ unit first
// (BindsTo= and After=carlos-edge.service), while the old agent is still
// running and has not had its SIGTERM. The agent sees the child gone,
// recycles the entry and gives its lease back, and a request queued on
// the stop wakes the host again, taking a fresh lease. That wake's start
// is a systemctl start job systemd queues behind the edge restart, so it
// is still in flight when the SIGTERM arrives. Shutdown has to abort it
// and give its lease back before the process exits; otherwise the next
// agent refuses the host for the lease's whole TTL.
//
// Aborting the client's start does not abort the start job systemd queued
// (Astra, review of 9a8e295c, P1): released over that job, the lease would
// let the job launch a writer nobody holds a lease for. So the abort stops
// the unit and proves it gone first, and only then gives the lease back.
func TestShutdownAbortsAWakeAndReleasesItsLease(t *testing.T) {
	t.Parallel()
	a, b, reqErr := arrangeQueuedRewake(t, "restart.example.com")

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	a.Shutdown(ctx)
	if ctx.Err() != nil {
		t.Fatal("Shutdown ran out of time instead of aborting the wake")
	}

	// Every lease this agent took is given back by the time Shutdown
	// returns: the first wake's through its replica's Close, the
	// re-wake's through Release, and never over a start job still queued.
	acq, _, _ := b.counts()
	if got := b.releaseCount(); got != acq {
		t.Fatalf("lease releases = %d after Shutdown, want %d (one per acquire)", got, acq)
	}
	b.mu.Lock()
	queuedAt, queuedNow, stops := slices.Clone(b.queuedAtRelease), b.queuedJobs, b.legacyStops
	b.mu.Unlock()
	if stops == 0 || queuedNow != 0 {
		t.Fatalf("the aborted start's queued job was never stopped (stops %d, queued %d)", stops, queuedNow)
	}
	for _, q := range queuedAt {
		if q != 0 {
			t.Fatalf("a lease was released with %d start job(s) still queued: %v", q, queuedAt)
		}
	}
	for _, err := range b.releaseCtxErrors() {
		if err != nil {
			t.Fatalf("a lease was released under a dead context (%v)", err)
		}
	}
	if n := a.entryCount(); n != 0 {
		t.Fatalf("Shutdown left %d entr(ies) behind", n)
	}
	select {
	case err := <-reqErr:
		if err == nil {
			t.Fatal("the request was served by a wake Shutdown aborted")
		}
	case <-time.After(5 * time.Second):
		t.Fatal("the queued request never returned")
	}
}

// arrangeQueuedRewake is #292's arrangement: host awake, its unit stopped
// by systemd, and a request's re-wake inside a start that systemd has
// queued behind the edge restart. It returns the request's outcome.
func arrangeQueuedRewake(t *testing.T, host string) (*Service, *fakeBackend, <-chan error) {
	t.Helper()
	a, b, _, _ := testActivator(t, time.Hour)
	arrangeAwake(t, a, testRoute(host))

	// systemd stops the bound unit, well before the agent's SIGTERM.
	b.lastProc().kill()
	queued := make(chan struct{})
	b.mu.Lock()
	b.startQueued = queued
	b.mu.Unlock()
	t.Cleanup(func() { close(queued) })

	reqErr := make(chan error, 1)
	go func() {
		release, err := a.beginRequest(context.Background(), testRoute(host))
		if err == nil {
			release()
		}
		reqErr <- err
	}()
	waitFor(t, "the re-wake to reach its queued start", func() bool { return b.startCount() == 2 })
	if acq, _, _ := b.counts(); acq != 2 {
		t.Fatalf("acquires = %d, want 2 (the first wake and the re-wake)", acq)
	}
	return a, b, reqErr
}

// P1's other half: an aborted legacy start whose unit cannot be proven
// stopped KEEPS its lease. A lease left to its TTL is a 15-minute stall; a
// lease given back over a queued start job is two writers.
func TestAnAbortedStartNotProvenStoppedKeepsItsLease(t *testing.T) {
	t.Parallel()
	a, b, _ := arrangeQueuedRewake(t, "unproven.example.com")
	b.mu.Lock()
	b.legacyStopErr = errors.New("stop carlos-instance@unproven: instance control socket: connection refused")
	b.mu.Unlock()

	// An unproven start keeps Shutdown polling until its budget ends, so
	// the budget is all this test waits: half a second is five of its polls.
	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
	defer cancel()
	a.Shutdown(ctx)

	acq, _, _ := b.counts()
	if got := b.releaseCount(); got != acq-1 {
		t.Fatalf("lease releases = %d, want %d: the first wake's, and NOT the aborted start's", got, acq-1)
	}
	b.mu.Lock()
	stops := b.legacyStops
	b.mu.Unlock()
	if stops == 0 {
		t.Fatal("the aborted start's unit was never asked to stop")
	}
	a.mu.Lock()
	_, retained := a.unprovenWakes["unproven.example.com"]
	a.mu.Unlock()
	if !retained {
		t.Fatal("the unproven start was not retained with its lease")
	}
}

// P2: a wake that ignores its cancellation holds Shutdown's join, and an
// entry whose stop was left pending must still be finished meanwhile:
// its process exits, and its lease goes back then, not once every wake
// has joined.
func TestShutdownFinishesPendingStopsWhileJoiningWakes(t *testing.T) {
	t.Parallel()
	a, b, _, _ := testActivator(t, time.Hour)
	stuckHost, wakingHost := "stuck.example.com", "waking.example.com"
	b.procStuck = true
	arrangeAwake(t, a, testRoute(stuckHost))
	stuck := b.lastProc()
	stuckRep := b.lastRep()

	b.mu.Lock()
	b.procStuck = false
	gate := make(chan struct{})
	b.startGate = gate
	b.mu.Unlock()
	go func() {
		if release, err := a.beginRequest(context.Background(), testRoute(wakingHost)); err == nil {
			release()
		}
	}()
	waitFor(t, "the wake to reach Start", func() bool { return b.startCount() == 2 })

	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
	defer cancel()
	done := make(chan struct{})
	go func() { a.Shutdown(ctx); close(done) }()
	waitFor(t, "the stuck stop to be left pending", func() bool {
		a.mu.Lock()
		defer a.mu.Unlock()
		e := a.entries[stuckHost]
		return e != nil && e.pendingStop
	})
	stuck.exit()
	waitFor(t, "the pending stop's lease to go back while a wake is still joining", stuckRep.isClosed)
	select {
	case <-done:
		t.Fatal("Shutdown returned while a wake was still spawning")
	default:
	}
	close(gate)
	<-done
}

// P3: a failed release is an obligation, not a forgotten handle. A wake
// that fails gives its lease back, the store refuses the DELETE, and
// Shutdown retries it (conditionally) inside its own budget.
func TestShutdownRetriesAFailedLeaseRelease(t *testing.T) {
	t.Parallel()
	a, b, _, _ := testActivator(t, time.Hour)
	b.startErr = errors.New("start failed")
	b.releaseErrs = []error{errors.New("delete lease: context deadline exceeded")}
	if _, err := a.beginRequest(context.Background(), testRoute("refused.example.com")); err == nil {
		t.Fatal("the wake succeeded")
	}
	if got := b.releaseCount(); got != 0 {
		t.Fatalf("lease releases = %d before Shutdown, want 0 (the DELETE failed)", got)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	a.Shutdown(ctx)
	if ctx.Err() != nil {
		t.Fatal("Shutdown ran out of time")
	}
	b.mu.Lock()
	retries := b.retries
	b.mu.Unlock()
	if got := b.releaseCount(); got != 1 || retries == 0 {
		t.Fatalf("lease releases = %d after %d retries, want the failed release retried and landed", got, retries)
	}
}

// P3's other half: a release still unresolved when Shutdown's budget ends
// is logged by host, not dropped silently.
func TestShutdownLogsALeaseReleaseItCouldNotResolve(t *testing.T) {
	logs := lockedLog(t)
	a, b, _, _ := testActivator(t, time.Hour)
	b.startErr = errors.New("start failed")
	b.releaseErrs = []error{errors.New("delete lease: timeout")}
	for i := 0; i < 1000; i++ {
		b.retryErrs = append(b.retryErrs, errors.New("delete lease: timeout"))
	}
	if _, err := a.beginRequest(context.Background(), testRoute("dark.example.com")); err == nil {
		t.Fatal("the wake succeeded")
	}
	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
	defer cancel()
	a.Shutdown(ctx)
	if got := b.releaseCount(); got != 0 {
		t.Fatalf("lease releases = %d, want 0", got)
	}
	if out := logs.String(); !strings.Contains(out, "unresolved") || !strings.Contains(out, "dark.example.com") {
		t.Fatalf("no unresolved-release line naming the host at exit; log:\n%s", out)
	}
}

// lockedBuf is a log sink safe against the activator's own goroutines.
type lockedBuf struct {
	mu  sync.Mutex
	buf strings.Builder
}

func (l *lockedBuf) Write(p []byte) (int, error) {
	l.mu.Lock()
	defer l.mu.Unlock()
	return l.buf.Write(p)
}

func (l *lockedBuf) String() string {
	l.mu.Lock()
	defer l.mu.Unlock()
	return l.buf.String()
}

func lockedLog(t *testing.T) *lockedBuf {
	t.Helper()
	l := &lockedBuf{}
	flags := log.Flags()
	log.SetOutput(l)
	log.SetFlags(0)
	t.Cleanup(func() {
		log.SetOutput(os.Stderr)
		log.SetFlags(flags)
	})
	return l
}

// Once Shutdown has begun, nothing wakes. A request queued on a stop must
// not wake the host afresh when that stop finishes after Shutdown began:
// the wake would take a lease once Shutdown had stopped looking. Startup
// is closed, as it is on a running agent, and the request is already past
// the startup gate (queued on a Bounce's stop) when the shutdown begins.
func TestNoWakeOnceShutdownHasBegun(t *testing.T) {
	t.Parallel()
	a, b, _, _ := testActivator(t, time.Hour)
	started := make(chan struct{})
	close(started)
	a.Startup = started
	host := "queued.example.com"
	arrangeAwake(t, a, testRoute(host))
	stopGate := make(chan struct{})
	b.lastProc().setStopGate(stopGate)
	if !a.Bounce(host) {
		t.Fatal("Bounce(awake host) = false")
	}
	waitFor(t, "the Bounce's stop", func() bool { return b.lastProc().stopEntered() == 1 })

	reqErr := make(chan error, 1)
	go func() {
		release, err := a.beginRequest(context.Background(), testRoute(host))
		if err == nil {
			release()
		}
		reqErr <- err
	}()
	// The request is queued on the stop before the shutdown begins.
	time.Sleep(50 * time.Millisecond)

	done := make(chan struct{})
	go func() {
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		a.Shutdown(ctx)
		close(done)
	}()
	waitFor(t, "Shutdown to begin", func() bool {
		a.mu.Lock()
		defer a.mu.Unlock()
		return a.shuttingDown
	})
	close(stopGate)
	<-done

	select {
	case err := <-reqErr:
		if err == nil {
			t.Fatal("a request woke the host after Shutdown had begun")
		}
	case <-time.After(5 * time.Second):
		t.Fatal("the queued request never returned")
	}
	if acq, _, starts := b.counts(); acq != 1 || starts != 1 {
		t.Fatalf("acquires, starts = %d, %d; want 1, 1 (nothing woken after Shutdown began)", acq, starts)
	}
	time.Sleep(50 * time.Millisecond)
	if n := a.entryCount(); n != 0 {
		t.Fatalf("%d entr(ies) live after Shutdown", n)
	}
}

// An exit seen once Shutdown has begun is the shutdown's, not the
// build's: neither the sweep nor a request reads it as unexpected or
// recycles it. Shutdown's own stop takes the entry down.
func TestAnExitAfterShutdownBeganIsNotUnexpected(t *testing.T) {
	a, b, _, _ := testActivator(t, time.Hour)
	host := "exit.example.com"
	arrangeAwake(t, a, testRoute(host))
	b.lastProc().kill()
	// The moment between Shutdown announcing itself and taking its
	// snapshot of the entries.
	a.mu.Lock()
	a.shuttingDown = true
	a.mu.Unlock()

	a.sweepOnce()
	if _, err := a.beginRequest(context.Background(), testRoute(host)); err == nil {
		t.Fatal("a request was served once Shutdown had begun")
	}
	a.mu.Lock()
	e := a.entries[host]
	state, unexpected := actAwake, false
	if e != nil {
		state, unexpected = e.state, e.unexpectedExit
	}
	a.mu.Unlock()
	if e == nil || state != actAwake {
		t.Fatalf("the exit was recycled after Shutdown began (entry %v, state %v)", e != nil, state)
	}
	if unexpected {
		t.Fatal("an exit after Shutdown began was read as unexpected")
	}
	if acq, _, _ := b.counts(); acq != 1 {
		t.Fatalf("acquires = %d, want 1", acq)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	a.Shutdown(ctx)
	if n := a.entryCount(); n != 0 {
		t.Fatalf("Shutdown left %d entr(ies) behind", n)
	}
	if got := b.releaseCount(); got != 1 {
		t.Fatalf("lease releases = %d, want 1", got)
	}
}

// Round 2 P2: release obligations expire. Under a persistent DELETE refusal
// with acquisition still allowed, every failed wake adds one; once their
// records pass their own expiry the retries retire them (logged) instead
// of running forever.
func TestOwedReleasesAreRetiredOnceTheirRecordsExpire(t *testing.T) {
	logs := lockedLog(t)
	a, b, _, _ := testActivator(t, time.Hour)
	b.startErr = errors.New("start failed")
	hosts := []string{"one.example.com", "two.example.com", "three.example.com"}
	for range hosts {
		b.releaseErrs = append(b.releaseErrs, errors.New("delete lease: AccessDenied"))
	}
	for i := 0; i < 1000; i++ {
		b.retryErrs = append(b.retryErrs, errors.New("delete lease: AccessDenied"))
	}
	for _, h := range hosts {
		if _, err := a.beginRequest(context.Background(), testRoute(h)); err == nil {
			t.Fatalf("%s: the wake succeeded", h)
		}
	}
	owed := func() int {
		a.mu.Lock()
		defer a.mu.Unlock()
		a.finishPendingStops()
		return len(a.owedReleases)
	}
	time.Sleep(50 * time.Millisecond)
	if n := owed(); n != len(hosts) {
		t.Fatalf("owed releases = %d before expiry, want %d", n, len(hosts))
	}
	b.mu.Lock()
	b.leaseLapsed = true
	b.mu.Unlock()
	waitFor(t, "the expired obligations to be retired", func() bool { return owed() == 0 })
	out := logs.String()
	if !strings.Contains(out, "retired") {
		t.Fatalf("no retirement line; log:\n%s", out)
	}
	if strings.Contains(out, "given back on retry") || b.releaseCount() != 0 {
		t.Fatalf("an expired, never-deleted lease was reported as given back; log:\n%s", out)
	}
}

// Astra, lower priority: the legacy proof is bounded by Shutdown's budget
// as well as its own 30 s, whichever ends first, so no proof outlives
// Shutdown. Unproven, the lease is still kept.
func TestTheLegacyProofEndsWithShutdownsBudget(t *testing.T) {
	t.Parallel()
	a, b, _ := arrangeQueuedRewake(t, "slowproof.example.com")
	b.mu.Lock()
	b.legacyBlock = true
	b.mu.Unlock()

	ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond)
	defer cancel()
	a.Shutdown(ctx)
	waitFor(t, "the proof to end with Shutdown's budget", func() bool {
		b.mu.Lock()
		defer b.mu.Unlock()
		return b.legacyEnded > 0
	})
	acq, _, _ := b.counts()
	if got := b.releaseCount(); got != acq-1 {
		t.Fatalf("lease releases = %d, want %d: an unproven start keeps its lease", got, acq-1)
	}
}

// Astra round 3 P2: a proof already running when Shutdown begins (here a
// failed start's, from before the shutdown, under its own 30 s budget)
// ends when Shutdown's budget does. Cut short, it is unproven, and the
// lease is kept.
func TestAProofRunningBeforeShutdownEndsWithItsBudget(t *testing.T) {
	t.Parallel()
	a, b, _, _ := testActivator(t, time.Hour)
	b.mu.Lock()
	b.startLegacyFail, b.legacyBlock = true, true
	b.mu.Unlock()
	go func() {
		if release, err := a.beginRequest(context.Background(), testRoute("early.example.com")); err == nil {
			release()
		}
	}()
	waitFor(t, "the failed start's proof to begin", func() bool {
		b.mu.Lock()
		defer b.mu.Unlock()
		return b.legacyStops == 1
	})
	ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond)
	defer cancel()
	a.Shutdown(ctx)
	waitFor(t, "the proof to end with Shutdown's budget", func() bool {
		b.mu.Lock()
		defer b.mu.Unlock()
		return b.legacyEnded > 0
	})
	if got := b.releaseCount(); got != 0 {
		t.Fatalf("lease releases = %d, want 0: a proof cut short keeps the lease", got)
	}
}