// The housekeeper: the ONE election for this deployment's global duties
// (spec 2026-08-20-multi-region-ingress-design.md, Part 1). A deployment now
// has N platform boxes converging one control bucket, and the rule the writer
// map states is "data that is per-box gets per-box keys; duties that are
// deployment-global get a lease". The releases retention prune is the first
// such duty to move behind it — five concurrent pruners racing a promote is a
// bug class the spec refuses to have — and Part 2's cert issuance is the next.
//
// The mechanism is deliberately NOT new. It is internal/lease.Acquire, the
// same conditional-write lock (litestream's lss3.Leaser writing
// <prefix>/lock.json) the status publisher has elected with since the fleet
// status page shipped — see statusLoop's lease.Acquire call in agent.go and
// the bucket-writer audit's §1. The spec's own words: "The status publisher's
// CAS-lease is the one election mechanism, reused; no second election pattern
// may be introduced." So there is no store-CAS lease here, no
// control/housekeeper.json record, and no bespoke TTL.
//
// Two things DO differ from the status election, both on purpose:
//
//   - It is long-lived, not per-tick. The status loop acquires and releases
//     inside one 5-minute tick because publishing is a single act; the
//     housekeeper's duties are continuous and Task 4's edge cert gate has to
//     be answerable at any instant, so this box acquires once and RENEWS
//     (lease.Acquire starts the renewer itself) for as long as it can, and
//     housekeeperHeld() is a non-blocking read of that state.
//   - It leases in the DEPLOYMENT bucket at a fixed key —
//     control/housekeeper-lease/lock.json — not under CARLOS_STATUS_PREFIX.
//     The duties act on the deployment bucket's own control plane
//     (releases/**, later control/certs/**), and that bucket is the one
//     linearizable home the spec pins the control plane to.
//
// The accepted price is lease.LeaseTTL, 15 minutes: a HARD-crashed
// housekeeper strands the lease until expiry, so handover can stall up to
// ~15m. Nothing here cares. The prune is idempotent and not urgent (a policy
// written now is promised "within ~10 minutes", and the loop is a 5-minute
// tick anyway), and cert renewals run ~30 days early.
//
// A GRACEFUL stand-down releases immediately, and that path has to be
// REACHABLE, not merely written: a `systemctl restart carlos-agent` is not a
// crash, and if it left the lock behind, every ordinary platform roll would
// cost the deployment up to ~15m (+ a retry) with no housekeeper — including
// a single-box deployment, which would be inert against its own lock, and
// from Part 2 that gap is "no cert issuer" after every deploy. So
// startHousekeeper returns a stop func and RunAgent calls it in the same
// post-drain block that shuts the activator's children down, beside the
// comment about not orphaning leases until TTL. That is the same principle.
//
// The hold owns a context, not just a flag. Every duty runs under a context
// derived from the holding, cancelled the instant the hold ends — a prune is
// a full list+delete sweep and Part 2's ACME order takes seconds to minutes,
// so a duty that kept running after a lost renewal would stretch the window
// where two boxes act on deployment-global state from housekeeperPoll to
// housekeeperPoll plus however long the duty runs. Duties must therefore
// honour their context; the ones here already do (every store call takes it).
package host

import (
	"context"
	"errors"
	"log"
	"math/rand/v2"
	"os"
	"sync/atomic"
	"time"

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

// housekeeperLeasePrefix is the deployment-bucket prefix the election's lock
// lands under: the leaser writes <prefix>/lock.json, so the object is
// control/housekeeper-lease/lock.json. Under control/ because it is control
// plane, and a fixed key because there is exactly one housekeeper per
// deployment — the same "the bucket IS the scope" property the status-lease
// relies on (one control bucket per deployment, forever).
const housekeeperLeasePrefix = "control/housekeeper-lease"

// housekeeperRetry spaces a losing contender's attempts. Each attempt is one
// conditional PUT against the lock (a GET first when a lock exists), so this
// is the same request-cost trade internal/lease's LeaseTTL comment weighs:
// 5 minutes is 288 attempts/day/box, in step with the status election's own
// per-tick contention, and bounds takeover after a hard crash at roughly the
// 15m TTL plus one retry.
const housekeeperRetry = 5 * time.Minute

// housekeeperStopWait bounds how long a graceful stand-down may take before
// the agent stops waiting for it. The release itself gets 5s
// (releaseHousekeeperLease) and a holder's poll tick is instant, so this is
// generous; it exists only so a wedged S3 call cannot hold a restart open.
const housekeeperStopWait = 10 * time.Second

// housekeeperPoll is how often a HOLDER checks that renewal is still winning.
// Renewal itself runs inside internal/lease (TTL/3 = 5m); this only decides
// how quickly a box that has LOST the lease stops believing it holds it —
// which is exactly how long a stale holder could keep running duties after a
// takeover. 30 seconds is cheap (a mutexed field read, no I/O at all).
const housekeeperPoll = 30 * time.Second

// heldLease is the slice of *lease.Held this file uses: renewal health and
// release. An interface only so the election transport is injectable — the
// lock's own semantics are internal/lease's to test, against a real
// S3-compatible store.
type heldLease interface {
	Err() error
	Release(ctx context.Context) error
}

// leaseAcquirer is lease.Acquire's shape, as a seam.
type leaseAcquirer func(ctx context.Context, bucket, prefix, region string) (heldLease, error)

// acquireDeploymentLease is the production acquirer. The region is
// deliberately empty: internal/lease's resolveRegion then answers from the
// AWS default chain (AWS_REGION / profile), which is precisely what
// release.NewS3Store signs for against this same bucket — a housekeeper lease
// signing for a different region than the store it guards would be a split
// waiting to happen. No new env var: the deployment bucket has never had a
// region variable of its own, and inventing one here would be a second answer
// to a question the chain already answers.
//
// The nil-into-interface hazard is why this is a func and not a method value:
// lease.Acquire returns a typed nil *lease.Held alongside its error, which
// would arrive as a NON-nil heldLease if handed straight back.
func acquireDeploymentLease(ctx context.Context, bucket, prefix, region string) (heldLease, error) {
	h, err := lease.Acquire(ctx, bucket, prefix, region)
	if err != nil {
		return nil, err
	}
	return h, nil
}

// housekeeperMode is this box's relationship to the election, decided once
// from the environment.
type housekeeperMode int

const (
	// housekeeperNever: this box must never run deployment-global duties —
	// a fleet box (a customer's machine; the platform's releases are not
	// its business, and the store door would refuse the deletes anyway), or
	// a box with no deployment store at all.
	housekeeperNever housekeeperMode = iota
	// housekeeperSole: a CARLOS_DEPLOYMENT_DIR deployment. There is no
	// bucket to hold a lock in, and a directory store is one box by
	// construction, so this box IS the housekeeper — anything else would
	// silently stop dev/test deployments pruning at all. Such a deployment
	// publishes no status, so retention reads this box's own build-state
	// report in its place (settings spec rev 11 §4.4, ruling R-T9-SOLE):
	// one box, so its own report is every box's.
	housekeeperSole
	// housekeeperElect: a bucket-backed platform box. Contend.
	housekeeperElect
)

// holding is one continuous stretch of this box being the housekeeper. Its
// context is what every duty runs under, so ending the holding ends the
// duties: there is no state where this box believes it does not hold the
// lease while work it started as holder is still running.
type holding struct {
	ctx    context.Context
	cancel context.CancelFunc
}

// housekeeper is the long-lived acquire/renew/stand-down lifecycle plus the
// state every gated caller reads.
type housekeeper struct {
	mode    housekeeperMode
	reason  string // the one line run() logs when it arms
	bucket  string
	acquire leaseAcquirer
	retry   time.Duration
	poll    time.Duration

	// cur is the current holding, or nil when this box is not the
	// housekeeper. ONE atomic, deliberately: a separate "held" flag beside
	// it would have an instant where the two disagree, and both the duty
	// gate and Task 4's cert gate read this on live paths.
	cur atomic.Pointer[holding]
}

// newHousekeeper decides the mode from the environment. Nothing is acquired
// and no holding is begun here — start() does both.
func newHousekeeper(getenv func(string) string, acquire leaseAcquirer) *housekeeper {
	hk := &housekeeper{acquire: acquire, retry: housekeeperRetry, poll: housekeeperPoll}
	switch {
	// Deliberately the bare CARLOS_FLEET test that retentionLoop and
	// instancesLoop already use, for the same reason retentionLoop states:
	// a fleet box whose LABEL is misconfigured is still a fleet box.
	case shouldSkipPublisherElection(getenv):
		hk.mode = housekeeperNever
		hk.reason = "fleet box — deployment-global duties are a platform-box job; never contending"
	case getenv("CARLOS_DEPLOYMENT_BUCKET") != "":
		// The bucket openDeployStore already adopts from — the duties act on
		// its contents, so the lock guarding them belongs in it.
		hk.mode = housekeeperElect
		hk.bucket = getenv("CARLOS_DEPLOYMENT_BUCKET")
	case getenv("CARLOS_DEPLOYMENT_DIR") != "":
		hk.mode = housekeeperSole
		hk.reason = "directory-backed deployment — this box is its own housekeeper (no bucket to elect in)"
	default:
		hk.mode = housekeeperNever
		hk.reason = "no deployment store configured — never contending"
	}
	return hk
}

// held reports whether this box currently holds the lease. Cheap and
// non-blocking by contract: it is read on a duty tick and, from Part 2, on
// the edge's certificate path.
func (hk *housekeeper) held() bool {
	_, ok := hk.gate()
	return ok
}

// gate is held() plus the context the caller's work must run under: valid
// only while the holding lasts, cancelled the moment it ends. A duty that
// takes this context cannot outlive the authority it started under.
func (hk *housekeeper) gate() (context.Context, bool) {
	if hk == nil {
		return nil, false
	}
	h := hk.cur.Load()
	if h == nil {
		return nil, false
	}
	return h.ctx, true
}

// beginHolding marks this box the housekeeper for as long as parent lives.
func (hk *housekeeper) beginHolding(parent context.Context) {
	ctx, cancel := context.WithCancel(parent)
	hk.cur.Store(&holding{ctx: ctx, cancel: cancel})
}

// endHolding stops this box being the housekeeper: the gate closes FIRST (no
// new duty can start) and only then is the context cancelled (whatever is
// already running unwinds). Idempotent — the stop func calls it after the
// loop has usually done it already.
func (hk *housekeeper) endHolding() {
	if h := hk.cur.Swap(nil); h != nil {
		h.cancel()
	}
}

// start arms this housekeeper and returns the stand-down: cancel the loop,
// wait for it to release and exit, and close the gate. Calling it is what
// makes the graceful release reachable — see the file header.
func (hk *housekeeper) start(parent context.Context) (stop func()) {
	ctx, cancel := context.WithCancel(parent)
	// A lone box is the housekeeper from the instant it is configured, not
	// from whenever a goroutine is scheduled: no acquisition is involved, so
	// there is nothing to wait for.
	if hk.mode == housekeeperSole {
		hk.beginHolding(ctx)
	}
	done := make(chan struct{})
	go func() {
		defer close(done)
		hk.run(ctx)
	}()
	return func() {
		// Close the gate BEFORE cancelling: cancel() only reaches the
		// holding's context by propagation (it is a child of this ctx via
		// beginHolding), so between cancel() and the run loop noticing and
		// calling endHolding() itself there is a window where hk.cur is
		// still non-nil over an already-cancelled context — held() reads
		// true against dead work. Calling endHolding() first closes hk.cur
		// synchronously, before ctx is even touched, so that window never
		// opens. The run path's own endHolding() call (in hold()) stays and
		// is idempotent — this is belt and braces for a path that doesn't
		// go through hold() (housekeeperSole) or that races it.
		hk.endHolding()
		cancel()
		select {
		case <-done:
		case <-time.After(housekeeperStopWait):
			// Never wedge a shutdown on this: the lock expires on its own
			// TTL, which is exactly the crash behaviour we already accept.
			log.Printf("agent: housekeeper: stand-down did not complete within %s — the lease will expire on its TTL instead", housekeeperStopWait)
		}
		hk.endHolding()
	}
}

// run is the election loop; it returns only when ctx ends (or immediately, on
// a box that never contends).
//
// NOT holding the lease is a normal, quiet state — it is what N-1 boxes are
// in, permanently. Contention (*lease.HeldError) is therefore never logged,
// exactly as statusLoop swallows it: a line per retry per box, forever, would
// bury the log. A real acquire failure (credentials, a broken bucket) IS
// logged, same as any other tick failure.
func (hk *housekeeper) run(ctx context.Context) {
	if hk.mode != housekeeperElect {
		log.Printf("agent: housekeeper: %s", hk.reason)
		return
	}
	log.Printf("agent: housekeeper: contending for s3://%s/%s/lock.json (lease TTL %s)",
		hk.bucket, housekeeperLeasePrefix, lease.LeaseTTL)
	for ctx.Err() == nil {
		held, err := hk.acquire(ctx, hk.bucket, housekeeperLeasePrefix, "")
		if err != nil {
			var heldErr *lease.HeldError
			if !errors.As(err, &heldErr) && ctx.Err() == nil {
				log.Printf("agent: housekeeper: lease acquire: %v", err)
			}
			if !sleepCtx(ctx, hk.retry) {
				return
			}
			continue
		}
		log.Printf("agent: housekeeper: lease acquired — this box runs the deployment's global duties")
		hk.beginHolding(ctx)
		hk.hold(ctx, held)
	}
}

// hold blocks while this box is the housekeeper, and ends the holding BEFORE
// it gives the lease up either way — closing the gate so no duty starts
// against a lease this box has decided to let go of, and cancelling any duty
// already in flight so none outlives the authority it began under.
//
// A failed renewal means leadership is lost (internal/lease stops renewing on
// the first failure, precisely so a re-try can't fence a claimant who took
// over after expiry). The lease is then NOT released: our etag is stale, so
// the release would just fail a precondition, and the TTL is what hands the
// lock over. A graceful stand-down (ctx ended: process shutting down) DOES
// release, so the next box takes over in seconds instead of ~15 minutes.
func (hk *housekeeper) hold(ctx context.Context, held heldLease) {
	for {
		if !sleepCtx(ctx, hk.poll) {
			hk.endHolding()
			releaseHousekeeperLease(held)
			return
		}
		if err := held.Err(); err != nil {
			hk.endHolding()
			log.Printf("agent: housekeeper: lease renewal failed — standing down: %v", err)
			return
		}
	}
}

// releaseHousekeeperLease hands the lock back on a graceful stand-down. The
// release gets its OWN budget, never the caller's: by the time this runs the
// agent's context is already dead, and a release under a dead context leaves
// the lock stranded until TTL — the exact trap internal/lease's
// Replicator.Close names, where it cost a wake up to a full TTL of 503s.
func releaseHousekeeperLease(held heldLease) {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if err := held.Release(ctx); err != nil {
		log.Printf("agent: housekeeper: lease release: %v", err)
	}
}

// sleepCtx waits for d, reporting false if ctx ended first.
func sleepCtx(ctx context.Context, d time.Duration) bool {
	t := time.NewTimer(d)
	defer t.Stop()
	select {
	case <-ctx.Done():
		return false
	case <-t.C:
		return true
	}
}

// theHousekeeper is the process's one election, published by
// startHousekeeper. A pointer rather than a bare flag so the loop owns its own
// state and nothing else can set "held" from the side; nil (never armed)
// reads as not held.
var theHousekeeper atomic.Pointer[housekeeper]

// housekeeperHeld reports whether THIS box currently holds the deployment's
// housekeeper lease. Cheap, non-blocking, and safe before (or without)
// startHousekeeper: false. This is the read for callers that only need the
// answer — Part 2's edge cert gate. Anything that then goes and DOES work
// wants housekeeperGate instead, so the work is bound to the holding.
func housekeeperHeld() bool { return theHousekeeper.Load().held() }

// housekeeperGate is housekeeperHeld plus the context the duty must run
// under — the gate every deployment-global duty consults.
func housekeeperGate() (context.Context, bool) { return theHousekeeper.Load().gate() }

// housekeeperContends reports whether this box takes part in the election at
// all — mode sole or elect, decided once from the environment. The edge's
// cert-backfill eligibility (edge.WithIssuerContender, #318): a fleet box
// grandfathered with the bucket env is agent-run and gated, but mode
// "never" — its disk holds a customer's certs, never this deployment's to
// publish.
func housekeeperContends() bool {
	// Nil-safe like held()/gate(): several agent loops start before
	// startHousekeeper, and a caller from one of them should read "does not
	// contend", not panic.
	hk := theHousekeeper.Load()
	return hk != nil && hk.mode != housekeeperNever
}

// startHousekeeper arms the election for this process and returns its
// stand-down. Called once from RunAgent, before the loops whose duties it
// gates; the returned func MUST be called on the way out (RunAgent's
// post-drain block) or a restart strands the lock until its TTL.
func startHousekeeper() (stop func()) {
	hk := newHousekeeper(os.Getenv, acquireDeploymentLease)
	theHousekeeper.Store(hk)
	return hk.start(context.Background())
}

// housekeeperDuty is one deployment-global job: a name for the log and the
// work. Today's set is the releases retention prune; Part 2 adds certificate
// issuance.
type housekeeperDuty struct {
	name string
	run  func(context.Context) error
}

// runHousekeeperDuties runs one tick of the given duties — but only on the
// box holding the lease, and only under the context of THAT holding. A
// non-holder does nothing and says nothing: that is the normal state of every
// box but one, on every tick, forever.
//
// The gate is consulted once per tick, not once per duty: within a tick the
// context is what carries a mid-flight stand-down, and re-reading the gate
// between duties would just replace one cancellation mechanism with two.
//
// Per-duty isolation, the shape retentionPass already uses for per-app
// errors: one duty's failure is logged and the next still runs.
func runHousekeeperDuties(gate func() (context.Context, bool), duties ...housekeeperDuty) {
	ctx, ok := gate()
	if !ok {
		return
	}
	for _, d := range duties {
		// The holding can end mid-tick. What is left is not this box's to
		// run, and the duty that was interrupted stood down — it did not
		// fail, so it gets no error line either.
		if ctx.Err() != nil {
			return
		}
		if err := d.run(ctx); err != nil && ctx.Err() == nil {
			log.Printf("agent: housekeeper: %s: %v", d.name, err)
		}
	}
}

// mintMeshKeyLoop mints the deployment's mesh key (multi-region backhaul
// spec, Part 3 Task 2) — a housekeeper duty, but deliberately its OWN loop
// rather than a duty folded into retentionLoop's runHousekeeperDuties call
// (agent.go's retentionLoop, gated on prune's cadence).
//
// The plan first drafted the mint as riding retentionLoop's tick; review
// caught the trap: retentionLoop returns early — never arming its loop at
// all, so runHousekeeperDuties is never even called — on a box missing
// CARLOS_RELEASE_PUBKEY (release-signature verification, something wholly
// unrelated to the mesh key). Critically, a box in that state still WINS
// the housekeeper election (newHousekeeper/hk.run don't consult
// CARLOS_RELEASE_PUBKEY at all), so the one box holding the lease would
// simply never mint, and no other box would ever try either — the mesh key
// would silently never exist, and Tasks 3/4's hop auth would never arm,
// fleet-wide, with no error anywhere. This loop is gated on nothing but
// deployment-store presence (shouldSkipPublisherElection, openDeployStore),
// mirroring edgesfile.go's edgesFileLoop shape exactly, so minting is
// available on every housekeeper independent of every other loop's
// preconditions.
//
// mintMeshKey itself is create-only and idempotent (mesh.go's own comment):
// every tick after the deployment's first successful mint finds the key
// already there and does nothing, silently.
func mintMeshKeyLoop() {
	ctx := context.Background()
	if shouldSkipPublisherElection(os.Getenv) {
		log.Printf("agent: mesh key mint loop is a platform-box job — skipped on a fleet box")
		return
	}
	store, err := openDeployStore(ctx)
	if err != nil {
		log.Printf("agent: mesh key mint loop disabled: %v", err)
		return
	}
	log.Printf("agent: mesh key mint loop armed (5min + jitter) — mints only while this box holds the housekeeper lease")
	mint := housekeeperDuty{"mesh key mint", func(ctx context.Context) error {
		return edge.MintMeshKey(ctx, store)
	}}
	delay := time.Minute
	for {
		time.Sleep(delay)
		delay = 5*time.Minute + rand.N(time.Minute)
		runHousekeeperDuties(housekeeperGate, mint)
	}
}