package host

// What is testable here without a live S3 is the GATING — who contends, who
// runs duties, and how the box behaves when it loses or never wins. The lock
// semantics themselves (a second claimant gets a *lease.HeldError; a dark
// holder cannot reclaim after failover) belong to internal/lease and are
// covered there against a real S3-compatible store — internal/lease's own
// fake (endpoint_test.go's newFakeS3) answers every GET with NoSuchKey, so it
// proves where the traffic goes and nothing about contention. This file
// deliberately invents no second fake S3: the election transport is injected
// as a seam instead.

import (
	"context"
	"errors"
	"fmt"
	"io"
	"log"
	"os"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/lease"
)

// ---- helpers ---------------------------------------------------------------

// lockedBuffer is an io.Writer safe for the election goroutine and the test
// body to touch at once — log output is the assertion in the "stays quiet"
// tests, so it has to be read while the loop is still running.
type lockedBuffer struct {
	mu sync.Mutex
	b  strings.Builder
}

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

func (l *lockedBuffer) String() string {
	l.mu.Lock()
	defer l.mu.Unlock()
	return l.b.String()
}

// captureRacyLog redirects the standard logger for one test, through a
// mutexed buffer: unlike this package's other captureLog, the writer here is
// shared with a live election goroutine.
func captureRacyLog(t *testing.T) *lockedBuffer {
	t.Helper()
	buf := &lockedBuffer{}
	log.SetOutput(buf)
	t.Cleanup(func() { log.SetOutput(os.Stderr) })
	return buf
}

// quietLog silences the standard logger for a test that doesn't assert on it.
func quietLog(t *testing.T) {
	t.Helper()
	log.SetOutput(io.Discard)
	t.Cleanup(func() { log.SetOutput(os.Stderr) })
}

// fakeLease stands in for *lease.Held: renewal health is a func the test
// flips, and releases are counted.
type fakeLease struct {
	mu       sync.Mutex
	err      error
	released int
}

func (f *fakeLease) fail(err error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.err = err
}

func (f *fakeLease) Err() error {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.err
}

func (f *fakeLease) Release(context.Context) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.released++
	return nil
}

func (f *fakeLease) releases() int {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.released
}

type acquireCall struct{ bucket, prefix, region string }

// acquirer records every acquisition attempt and answers from a per-attempt
// func, so a test can script "contended, contended, won".
type acquirer struct {
	mu     sync.Mutex
	calls  []acquireCall
	answer func(n int) (heldLease, error)
}

func (a *acquirer) acquire(_ context.Context, bucket, prefix, region string) (heldLease, error) {
	a.mu.Lock()
	a.calls = append(a.calls, acquireCall{bucket, prefix, region})
	n := len(a.calls)
	answer := a.answer
	a.mu.Unlock()
	return answer(n)
}

func (a *acquirer) count() int {
	a.mu.Lock()
	defer a.mu.Unlock()
	return len(a.calls)
}

func (a *acquirer) first() acquireCall {
	a.mu.Lock()
	defer a.mu.Unlock()
	return a.calls[0]
}

// contended is what a losing contender sees, wrapped exactly as
// lease.Acquire wraps it.
func contended() error {
	return fmt.Errorf("acquire lease: %w", &lease.HeldError{Owner: "other-box", ExpiresAt: time.Now().Add(time.Minute)})
}

func waitFor(t *testing.T, what string, cond func() bool) {
	t.Helper()
	deadline := time.Now().Add(3 * time.Second)
	for time.Now().Before(deadline) {
		if cond() {
			return
		}
		time.Sleep(time.Millisecond)
	}
	t.Fatalf("timed out waiting for %s", what)
}

// runElection arms hk the way RunAgent does and registers its stand-down, so
// every test exercises the same start/stop pair production uses. The returned
// func is safe to call twice (the cleanup calls it again).
func runElection(t *testing.T, hk *housekeeper) func() {
	t.Helper()
	stop := hk.start(context.Background())
	stopped := false
	once := func() {
		if stopped {
			return
		}
		stopped = true
		stop()
	}
	t.Cleanup(once)
	return once
}

// heldGate is the gate a duty sees on a holder, for the tests that are about
// duty dispatch rather than the election.
func heldGate() (context.Context, bool) { return context.Background(), true }

// noGate is the gate every non-holding box sees.
func noGate() (context.Context, bool) { return nil, false }

func fastHousekeeper(getenv func(string) string, a *acquirer) *housekeeper {
	hk := newHousekeeper(getenv, a.acquire)
	hk.retry = time.Millisecond
	hk.poll = time.Millisecond
	return hk
}

// ---- duty gating -----------------------------------------------------------

func TestRunHousekeeperDutiesRunsEachDutyOncePerTickWhenHeld(t *testing.T) {
	quietLog(t)
	a, b := 0, 0
	runHousekeeperDuties(heldGate,
		housekeeperDuty{"a", func(context.Context) error { a++; return nil }},
		housekeeperDuty{"b", func(context.Context) error { b++; return nil }},
	)
	if a != 1 || b != 1 {
		t.Fatalf("duty runs = (%d, %d), want (1, 1)", a, b)
	}
}

func TestRunHousekeeperDutiesSkipsQuietlyWhenNotHeld(t *testing.T) {
	buf := captureRacyLog(t)
	ran := 0
	runHousekeeperDuties(noGate,
		housekeeperDuty{"retention pass", func(context.Context) error { ran++; return nil }},
	)
	if ran != 0 {
		t.Fatalf("duty ran %d times on a non-holder, want 0", ran)
	}
	// Not holding the lease is the NORMAL state on every box but one, every
	// tick, forever: a log line here would bury the log.
	if out := buf.String(); out != "" {
		t.Fatalf("non-holder logged %q, want silence", out)
	}
}

func TestRunHousekeeperDutiesLogsOneDutysErrorAndRunsTheRest(t *testing.T) {
	buf := captureRacyLog(t)
	later := 0
	runHousekeeperDuties(heldGate,
		housekeeperDuty{"retention pass", func(context.Context) error { return errors.New("boom") }},
		housekeeperDuty{"later duty", func(context.Context) error { later++; return nil }},
	)
	if later != 1 {
		t.Fatalf("later duty ran %d times, want 1 — one duty's error must not stop the next", later)
	}
	out := buf.String()
	if !strings.Contains(out, "retention pass") || !strings.Contains(out, "boom") {
		t.Fatalf("log = %q, want the failing duty's name and error", out)
	}
}

// ---- who contends ----------------------------------------------------------

func TestHousekeeperFleetBoxNeverAttemptsAcquisition(t *testing.T) {
	quietLog(t)
	a := &acquirer{answer: func(int) (heldLease, error) { return &fakeLease{}, nil }}
	hk := fastHousekeeper(envFunc(map[string]string{
		"CARLOS_FLEET":             "acme",
		"CARLOS_DEPLOYMENT_BUCKET": "dep-bucket",
	}), a)

	// run returns immediately on a fleet box — nothing to loop over.
	hk.run(context.Background())

	if a.count() != 0 {
		t.Fatalf("fleet box attempted acquisition %d times, want 0", a.count())
	}
	if hk.held() {
		t.Fatal("fleet box reports holding the housekeeper lease")
	}
}

func TestHousekeeperWithNoDeploymentStoreNeverHolds(t *testing.T) {
	quietLog(t)
	a := &acquirer{answer: func(int) (heldLease, error) { return &fakeLease{}, nil }}
	hk := fastHousekeeper(envFunc(map[string]string{}), a)

	hk.run(context.Background())

	if a.count() != 0 {
		t.Fatalf("storeless box attempted acquisition %d times, want 0", a.count())
	}
	if hk.held() {
		t.Fatal("storeless box reports holding the housekeeper lease")
	}
}

func TestHousekeeperDirDeploymentIsItsOwnHousekeeper(t *testing.T) {
	quietLog(t)
	a := &acquirer{answer: func(int) (heldLease, error) { return &fakeLease{}, nil }}
	hk := fastHousekeeper(envFunc(map[string]string{"CARLOS_DEPLOYMENT_DIR": t.TempDir()}), a)
	stop := runElection(t, hk)

	// No goroutine to wait on: a lone box holds from the instant it is armed,
	// because there is nothing to acquire.
	if !hk.held() {
		t.Fatal("dir deployment does not hold: a directory-backed deployment is one box by construction and must keep pruning")
	}
	if a.count() != 0 {
		t.Fatalf("dir deployment attempted acquisition %d times, want 0 — there is no bucket to elect in", a.count())
	}

	// And it stands down with the process, so nothing believes it is the
	// housekeeper after the agent has gone.
	stop()
	if hk.held() {
		t.Fatal("dir deployment still reports holding after stand-down")
	}
}

// ---- the election ----------------------------------------------------------

func TestHousekeeperAcquiresTheDeploymentLeaseAndHolds(t *testing.T) {
	quietLog(t)
	a := &acquirer{answer: func(int) (heldLease, error) { return &fakeLease{}, nil }}
	hk := fastHousekeeper(envFunc(map[string]string{"CARLOS_DEPLOYMENT_BUCKET": "dep-bucket"}), a)
	runElection(t, hk)

	waitFor(t, "the lease to be held", hk.held)

	got := a.first()
	want := acquireCall{bucket: "dep-bucket", prefix: housekeeperLeasePrefix, region: ""}
	if got != want {
		t.Fatalf("acquire call = %+v, want %+v", got, want)
	}
	if housekeeperLeasePrefix != "control/housekeeper-lease" {
		t.Fatalf("lease prefix = %q, want control/housekeeper-lease (lock.json lands under it)", housekeeperLeasePrefix)
	}
}

func TestHousekeeperLosingTheElectionIsQuietAndKeepsTrying(t *testing.T) {
	buf := captureRacyLog(t)
	a := &acquirer{answer: func(int) (heldLease, error) { return nil, contended() }}
	hk := fastHousekeeper(envFunc(map[string]string{"CARLOS_DEPLOYMENT_BUCKET": "dep-bucket"}), a)
	// The arming line is not what this test is about; drop everything logged
	// before the loop starts contending.
	runElection(t, hk)

	waitFor(t, "several acquisition attempts", func() bool { return a.count() >= 3 })

	if hk.held() {
		t.Fatal("a losing contender reports holding the lease")
	}
	// Contention is normal on every tick a box doesn't win: an error loop
	// here would be a line every retry, forever, on N-1 boxes.
	for _, line := range strings.Split(strings.TrimSpace(buf.String()), "\n") {
		if strings.Contains(line, "lease acquire") {
			t.Fatalf("contention logged as a failure: %q", line)
		}
	}
}

func TestHousekeeperLogsARealAcquireFailure(t *testing.T) {
	buf := captureRacyLog(t)
	a := &acquirer{answer: func(int) (heldLease, error) { return nil, errors.New("no credentials") }}
	hk := fastHousekeeper(envFunc(map[string]string{"CARLOS_DEPLOYMENT_BUCKET": "dep-bucket"}), a)
	runElection(t, hk)

	waitFor(t, "the failure to be logged", func() bool { return strings.Contains(buf.String(), "no credentials") })

	if hk.held() {
		t.Fatal("a box that could not acquire reports holding the lease")
	}
}

func TestHousekeeperStandsDownWhenRenewalFails(t *testing.T) {
	quietLog(t)
	first := &fakeLease{}
	third := &fakeLease{}
	// The takeover, scripted: this box holds, loses renewal, and finds the
	// lock already taken by whoever won it after expiry — for as long as the
	// test says so. That middle state is what makes "stood down" observable
	// at all: if a lease were waiting on the very next attempt, the box would
	// be holding again within a retry and a housekeeper that never cleared
	// its flag would look identical to one that did.
	var freed atomic.Bool
	a := &acquirer{answer: func(n int) (heldLease, error) {
		switch {
		case n == 1:
			return first, nil
		case !freed.Load():
			return nil, contended()
		default:
			return third, nil
		}
	}}
	hk := fastHousekeeper(envFunc(map[string]string{"CARLOS_DEPLOYMENT_BUCKET": "dep-bucket"}), a)
	runElection(t, hk)

	waitFor(t, "the first lease to be held", hk.held)

	// Renewal fails: leadership is lost, so this box must stop reporting
	// itself as the housekeeper — a stale holder still running duties (or,
	// from Part 2, still issuing certs) against a lease another box now owns
	// is the whole reason this check exists.
	first.fail(errors.New("renew: precondition failed"))
	waitFor(t, "the stand-down", func() bool { return !hk.held() })
	waitFor(t, "a fresh acquisition attempt", func() bool { return a.count() > 1 })

	// A lost lease is NOT released: our etag is stale, and the TTL is what
	// hands it over.
	if first.releases() != 0 {
		t.Fatalf("released a lost lease %d times, want 0", first.releases())
	}

	// And it recovers on its own once the lock frees up again.
	freed.Store(true)
	waitFor(t, "the lease to be won back", hk.held)
	if third.releases() != 0 {
		t.Fatalf("released the recovered lease %d times, want 0 while still holding it", third.releases())
	}
}

// The stand-down is the path an ordinary `systemctl restart` takes, so it has
// to release the lock AND come back — a stop that hung would wedge the
// restart, and one that returned before the loop had released would strand
// control/housekeeper-lease/lock.json for its 15-minute TTL.
func TestHousekeeperStopReleasesTheLeaseAndReturns(t *testing.T) {
	quietLog(t)
	held := &fakeLease{}
	a := &acquirer{answer: func(int) (heldLease, error) { return held, nil }}
	hk := fastHousekeeper(envFunc(map[string]string{"CARLOS_DEPLOYMENT_BUCKET": "dep-bucket"}), a)
	stop := runElection(t, hk)

	waitFor(t, "the lease to be held", hk.held)

	returned := make(chan struct{})
	go func() {
		stop()
		close(returned)
	}()
	select {
	case <-returned:
	case <-time.After(3 * time.Second):
		t.Fatal("stop() did not return — a shutdown would hang here")
	}

	if hk.held() {
		t.Fatal("still reports holding the lease after shutdown")
	}
	if held.releases() != 1 {
		t.Fatalf("releases = %d, want 1 — a graceful stand-down hands the lease over immediately", held.releases())
	}
	// The election goroutine is gone, not merely told to go: stop() waits on
	// it, so a leaked loop would still be acquiring after this point.
	before := a.count()
	time.Sleep(20 * time.Millisecond) // many retries at the 1ms test cadence
	if after := a.count(); after != before {
		t.Fatalf("acquisitions after stop: %d → %d, want no further attempts (the loop leaked)", before, after)
	}
}

// ---- duties are bound to the holding ---------------------------------------

// A duty must not outlive the authority it started under: when renewal fails
// mid-prune, the sweep's context is cancelled rather than left to finish its
// deletes against a lock another box now owns.
func TestHousekeeperCancelsDutiesWhenTheHoldEnds(t *testing.T) {
	quietLog(t)
	held := &fakeLease{}
	a := &acquirer{answer: func(n int) (heldLease, error) {
		if n == 1 {
			return held, nil
		}
		return nil, contended()
	}}
	hk := fastHousekeeper(envFunc(map[string]string{"CARLOS_DEPLOYMENT_BUCKET": "dep-bucket"}), a)
	runElection(t, hk)
	waitFor(t, "the lease to be held", hk.held)

	started := make(chan struct{})
	observed := make(chan error, 1)
	go func() {
		runHousekeeperDuties(hk.gate, housekeeperDuty{"slow duty", func(ctx context.Context) error {
			close(started)
			select {
			case <-ctx.Done():
				observed <- ctx.Err()
			case <-time.After(3 * time.Second):
				observed <- nil
			}
			return ctx.Err()
		}})
	}()

	<-started
	held.fail(errors.New("renew: precondition failed"))

	select {
	case err := <-observed:
		if err == nil {
			t.Fatal("the duty ran to completion after the hold ended — it must be cancelled")
		}
	case <-time.After(3 * time.Second):
		t.Fatal("timed out waiting for the duty to observe the stand-down")
	}

	// And the gate is shut for the next tick, so nothing new starts either.
	if _, ok := hk.gate(); ok {
		t.Fatal("the gate is still open after the hold ended")
	}
}

// A duty that starts on a holder and is cancelled mid-flight stood down; it
// did not fail, so it must not be logged as a failure — and the duties behind
// it in the same tick are no longer this box's to run.
func TestRunHousekeeperDutiesStopQuietlyWhenTheHoldEndsMidTick(t *testing.T) {
	buf := captureRacyLog(t)
	ctx, cancel := context.WithCancel(context.Background())
	later := 0
	runHousekeeperDuties(
		func() (context.Context, bool) { return ctx, true },
		housekeeperDuty{"retention pass", func(ctx context.Context) error {
			cancel() // the hold ends while this duty is running
			return ctx.Err()
		}},
		housekeeperDuty{"later duty", func(context.Context) error { later++; return nil }},
	)
	if later != 0 {
		t.Fatalf("later duty ran %d times after the hold ended, want 0", later)
	}
	if out := buf.String(); out != "" {
		t.Fatalf("stand-down logged as a failure: %q", out)
	}
}

// ---- the process-wide gate -------------------------------------------------

func TestHousekeeperHeldIsFalseBeforeAnyElectionIsArmed(t *testing.T) {
	if housekeeperHeld() {
		t.Fatal("housekeeperHeld() is true with no housekeeper armed")
	}
	if _, ok := housekeeperGate(); ok {
		t.Fatal("housekeeperGate() reports held with no housekeeper armed")
	}
}

// ---- mintMeshKeyLoop (multi-region backhaul spec, Part 3 Task 2's review
// fix): the mesh key's own arming loop, independent of retentionLoop -------

// Mirrors TestRetentionLoopIsAPlatformBoxJob's shape (agent_test.go): a
// fleet box must skip before ever opening a store.
func TestMintMeshKeyLoopIsAPlatformBoxJob(t *testing.T) {
	const skipLine = "agent: mesh key mint loop is a platform-box job — skipped on a fleet box"

	logs := captureLog(t)
	door := newDoorFake(t)
	for k, v := range fleetDoorEnv(door.url) {
		t.Setenv(k, v)
	}
	t.Setenv("CARLOS_DEPLOYMENT_BUCKET", "")
	t.Setenv("CARLOS_DEPLOYMENT_DIR", "")

	mintMeshKeyLoop() // must return; a fleet box never reaches the loop

	if got := logs(); !strings.Contains(got, skipLine) {
		t.Fatalf("log = %q, want the skip line verbatim", got)
	}
	if hits := door.seen(); len(hits) != 0 {
		t.Fatalf("door saw %+v, want zero — the skip must land BEFORE the store opens", hits)
	}
}

// The regression guard for the Critical review finding: the mint must NOT be
// gated on CARLOS_RELEASE_PUBKEY (retentionLoop's own gate, unrelated to the
// mesh key) — a box missing that var still wins the housekeeper election, so
// a mint duty tied to retentionLoop's arming would silently never run
// fleet-wide. Proven by explicitly leaving CARLOS_RELEASE_PUBKEY unset and
// confirming the loop still arms.
func TestMintMeshKeyLoopArmsWithoutReleasePubkey(t *testing.T) {
	t.Setenv("CARLOS_FLEET", "")
	t.Setenv("CARLOS_DEPLOYMENT_BUCKET", "")
	t.Setenv("CARLOS_DEPLOYMENT_DIR", t.TempDir())
	t.Setenv("CARLOS_RELEASE_PUBKEY", "") // deliberately unset — the whole point

	buf := captureRacyLog(t)
	// mintMeshKeyLoop never returns once armed: run it in its own goroutine
	// (it parks in a time.Sleep for a minute, long past this test's end) and
	// wait for the arming log line against a synchronized sink.
	go mintMeshKeyLoop()

	waitFor(t, "the mint loop to arm despite no CARLOS_RELEASE_PUBKEY", func() bool {
		return strings.Contains(buf.String(), "agent: mesh key mint loop armed")
	})
}