package activator

import (
	"context"
	"errors"
	"fmt"
	"log"
	"time"

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

var (
	// errHoldStopSlow: the old generation's stop outlasted wakeWaitBound, the
	// bound a hold holds requests for (spec §7.8). The hold is withdrawn
	// before its stop finishes, so nothing of the attempt's build runs.
	errHoldStopSlow = errors.New("swap: the old generation's stop outlasted the hold's bound; the hold is withdrawn")
	// errPinMismatch: the backend ran something other than the hold's
	// choice. A process ran, so the hold is down, but the text is the
	// attempt's diagnostic and never a member's event.
	errPinMismatch = errors.New("swap: the backend ran a different generation from the hold's choice")
)

// holdSpec is a hold swap waiting for its entry's stop (spec §7.8).
// finishStop answers on woke exactly once, unless swapHold withdraws the
// hold first (under a.mu, by clearing the entry's holdNext).
type holdSpec struct {
	gen    Gen // the attempt's build, staged path, digest and demand; N is drawn at spawn
	at     Attempt
	newest func(string) uint64
	woke   chan holdWoke // buffered 1
	// settleCancel ends a canary's settle observation. holdWakeLocked puts
	// it on the pinned entry, so an intentional stop of that entry cancels
	// the observation before anything is stopped (spec §6.4).
	settleCancel context.CancelFunc
}

// holdWoke is finishStop's answer: the pinned wake it installed, or why it
// installed none.
type holdWoke struct {
	next *actEntry
	err  error
}

// lifeDone is the agent lifetime's Done channel. It is nil (blocks forever
// in a select) for a Service built without New.
func (a *Service) lifeDone() <-chan struct{} {
	if a.lifeCtx == nil {
		return nil
	}
	return a.lifeCtx.Done()
}

// swapHold is §7.8: stop-then-start on the existing path, through the
// barrier. The old generation stops while begin() holds requests, up to
// wakeWaitBound. When the stop finishes, finishStop installs a waking
// entry pinned to the attempt's staged build before it releases anyone
// waiting on the stop. Held requests are served once that generation is
// healthy.
//
// A row that cannot run the staged build at all is refused before
// anything stops: a stop-then-start whose start cannot happen would only
// cost the instance its running process. So is a hold whose build cannot be
// published in the generations file first (zdd-readiness Task 13): the
// admission registers the build as a pending generation, the file is written
// off a.mu, and the admission runs again before the stop begins.
//
// The caller waits for the stop for at most wakeWaitBound, and no longer
// than its own context or the agent's life. Any of the three withdraws the
// hold under a.mu while the stop is still under way, and the attempt is
// abandoned: nothing of its build has run. Once finishStop has taken the
// hold, the pinned wake may exec the build against the database, so the
// result follows that wake to its end (bounded by the wake's own deadlines)
// and is never abandoned on the caller's account.
func (a *Service) swapHold(ctx context.Context, at Attempt, newest func(string) uint64) SwapResult {
	if err := verifyStaged(at.StagedPath, at.Digest); err != nil {
		return abandon(fmt.Errorf("staging: %w", err))
	}
	work, err := a.selectWork(ctx, at.Host)
	if err != nil {
		return abandon(fmt.Errorf("admission: %w", err))
	}
	if !hostwrite.Legacy(work) {
		// Claims on: refused before anything stops, whatever the runtime's
		// capabilities (claimsSwapRefusal). The pinned wake would meet the
		// closed claims launch, and a stop whose start cannot happen would
		// only cost the instance its running process.
		if err := a.claimsSwapRefusal(false); err != nil {
			return abandon(err)
		}
	}
	sctx, scancel := context.WithCancel(ctx)
	defer scancel()
	h := &holdSpec{
		gen:    Gen{Build: at.Build, StagedPath: at.StagedPath, Digest: at.Digest, Demand: at.Demand},
		at:     at,
		newest: newest,
		woke:   make(chan holdWoke, 1),
	}
	if at.Canary {
		h.settleCancel = scancel
	}
	var (
		held     *actEntry // the entry whose stop the hold waits on
		stopAt   time.Time // wall clock: the old generation's stop, for the drain time (spec §9.4)
		admitted *actEntry // the entry the first admission registered the build against
		pend     pendingGeneration
	)
	// The hold's build is published BEFORE the incumbent stops
	// (zdd-readiness Task 13, Astra R1 T12, R3 §5 item 9). From the stop
	// until the pinned wake's own publication nothing serves and, without
	// this, nothing names the build, so root could sweep or restage it under
	// the wake about to run it. The record is dropped when swapHold returns,
	// whatever the outcome: a withdrawn hold ran nothing, and a taken one is
	// named by its pinned entry (or a retained failed wake) from then on.
	defer func() {
		if pend == 0 {
			return
		}
		a.mu.Lock()
		a.dropPendingGenerationLocked(pend)
		a.mu.Unlock()
		_ = a.persistGenerations()
	}()
	// Admission, then the registration in the same critical section.
	err = a.admit(ctx, work, func() error {
		a.mu.Lock()
		defer a.mu.Unlock()
		e := a.entries[at.Host]
		if err := a.holdAdmitLocked(e, at, newest); err != nil {
			return err
		}
		admitted = e
		if id, ok := a.genIdentity(h.gen); ok {
			pend = a.addPendingGenerationLocked(at.Host, id)
		}
		return nil
	})
	if err != nil {
		return abandon(err)
	}
	// persistGenerations takes genMu and then a.mu, so it runs here, between
	// the two critical sections, never inside the admission. A write that
	// fails stops nothing: the incumbent keeps serving and the attempt is
	// abandoned with nothing of its build run (errStagedBuildNotPublished,
	// which is never a failure of the build).
	if err := a.publishBeforeExec(); err != nil {
		return abandon(err)
	}
	// The admission once more, with a.mu re-taken: anything the first one
	// checked may have changed while the file was written (plan review #10).
	// The same entry, still awake and owned, the attempt still the newest,
	// and every gate of the first admission. A refusal withdraws the pending
	// build (the deferred drop) and stops nothing.
	err = a.admit(ctx, work, func() error {
		a.mu.Lock()
		defer a.mu.Unlock()
		e := a.entries[at.Host]
		if err := a.holdAdmitLocked(e, at, newest); err != nil {
			return err
		}
		if e != admitted {
			// The entry the build was admitted against has gone (a sleep and
			// a fresh wake in the window). The checks pass on its successor,
			// but this attempt was never admitted there.
			return errSwapNotAwake
		}
		e.holdNext = h
		held, stopAt = e, time.Now()
		// A run the stop interrupts reads the attempt's cause: a restart or
		// a config change is a restart, not a new release (stopWhyFor).
		a.beginStopLocked(e, stopWhyFor(at.Cause))
		return nil
	})
	if err != nil {
		if held != nil {
			// The work refused after its body began the stop. The stop
			// stands; the hold does not.
			if !a.withdrawHold(held, h) {
				return a.holdResult(sctx, at, newest, <-h.woke, time.Since(stopAt))
			}
		}
		return abandon(err)
	}
	bound := time.NewTimer(wakeWaitBound)
	defer bound.Stop()
	var hw holdWoke
	select {
	case hw = <-h.woke:
	case <-bound.C:
		hw, err = a.withdrawOrTake(held, h, errHoldStopSlow)
	case <-ctx.Done():
		hw, err = a.withdrawOrTake(held, h, errSwapStopped)
	case <-a.lifeDone():
		hw, err = a.withdrawOrTake(held, h, errSwapShutdown)
	}
	if err != nil {
		return abandon(err)
	}
	return a.holdResult(sctx, at, newest, hw, time.Since(stopAt)) // finishStop answers once the old generation is proven gone
}

// holdAdmitLocked is a hold's admission (spec §7.4, §7.8): the attempt's
// checks on e, then the gates that refuse a row before anything stops (the
// kind, the layout, an operation already under way). swapHold asks it twice,
// before the build's publication and again after it, with a.mu re-taken.
// The caller holds a.mu.
func (a *Service) holdAdmitLocked(e *actEntry, at Attempt, newest func(string) uint64) error {
	if err := a.swapChecksLocked(e, at, newest); err != nil {
		return err
	}
	if e.worker || e.route.Kind != "instance" || e.route.Backing == registry.BackingUnit {
		return errSwapKind
	}
	// The layouts that can run a staged generation: flat, foreign on its
	// canonical socket, and nested (root mounts the staged build, Task 9).
	if l, ok := a.swapLayout(e.route); ok && l != LayoutFlat && l != LayoutForeign && l != LayoutNested {
		return errSwapLayout
	}
	if len(e.ops.members) > 0 {
		return errSwapBusy
	}
	return nil
}

// withdrawHold clears h from e while the stop has not taken it, and reports
// whether it did. False means finishStop already took the hold and answered
// on h.woke in the same critical section.
func (a *Service) withdrawHold(e *actEntry, h *holdSpec) bool {
	a.mu.Lock()
	defer a.mu.Unlock()
	if e.holdNext != h {
		return false
	}
	e.holdNext = nil
	return true
}

// withdrawOrTake withdraws the hold for why, or, when finishStop has taken
// it already, returns its answer.
func (a *Service) withdrawOrTake(e *actEntry, h *holdSpec, why error) (holdWoke, error) {
	if a.withdrawHold(e, h) {
		return holdWoke{}, why
	}
	return <-h.woke, nil
}

// holdResult follows a hold's pinned wake to its outcome. Its execution
// evidence is the pinned wake's (pinExec, set from the spawn's result), and
// a wake that came up healthy ran the build: SwapResult.Executed (final
// review F8). A pinned start handed to root whose outcome is uncertain
// (startUncertain) is executed-or-unknown too (final review wave 2, N4),
// and it is the build's failure: the hold is down (Astra R3 ruling 206).
func (a *Service) holdResult(sctx context.Context, at Attempt, newest func(string) uint64, hw holdWoke, drain time.Duration) (out SwapResult) {
	if hw.err != nil {
		return abandon(hw.err)
	}
	<-hw.next.ready
	a.mu.Lock()
	g, werr, proc, execed := hw.next.gen, hw.next.err, hw.next.proc, hw.next.pinExec
	a.mu.Unlock()
	defer func() { out.Executed = execed || werr == nil || startMayHaveRun(nil, werr) }()
	switch {
	case werr != nil && proc == nil && startMayHaveRun(nil, werr):
		// A nested start root accepted whose outcome is uncertain (it
		// exited before it was observed, root's start call failed, or its
		// kill is unproven) may have run the build against the database
		// (Astra R3 ruling 206, zdd-readiness Task 6). It is a failure of
		// the build whatever ended the wake: down, so done() holds the build
		// down and halts the cohort. An abandonment with Executed would be
		// deferred and could run the build again.
		return SwapResult{Outcome: SwapDown, Gen: g, Reason: werr.Error(), Drain: drain}
	case werr != nil && !execed:
		// The pinned wake failed before its build's exec: the lease, the
		// restore, the row, the staged bytes, the claims admission. Nothing
		// of the build ran, so this is no failure of it and nothing is held
		// down (spec §6.4, §8.3; index P5/P6). The attempt is abandoned.
		return abandon(werr)
	case werr == nil:
		if at.Canary {
			// A canary must survive the settle before it counts (spec §6.2,
			// §8.5). Plan 5 does not observe a pinned wake itself.
			r := observeSettle(sctx, proc, swapSettle)
			a.mu.Lock()
			hw.next.settleCancel = nil
			a.mu.Unlock()
			switch r {
			case settleExited:
				// Nothing falls back from a hold (spec §8.3): the instance is
				// down until the sweep notices the exit and a later wake runs.
				return SwapResult{Outcome: SwapDown, Gen: g, Reason: errSettleExit.Error(), Drain: drain}
			case settleCancelled:
				return abandon(errSwapStopped)
			}
		}
		// The checks once more before the result counts (spec §6.3, §8.1):
		// a newer attempt or a halted cohort since the wake began means this
		// build is not recorded good. It keeps serving until the newer
		// attempt replaces it. The same gate as every other exec point.
		a.mu.Lock()
		cerr := a.attemptCurrentLocked(hw.next, at, newest, false)
		a.mu.Unlock()
		if cerr != nil {
			return abandon(cerr)
		}
		return SwapResult{Outcome: SwapDone, Gen: g, Drain: drain}
	case classifyStep(nil, werr) == stepStale:
		// The pinned wake's own gate refused the exec (pinCheck): nothing
		// of the build ran, and this is no failure of it.
		return abandon(werr)
	case classifyStep(nil, werr) == stepCancelled:
		// An intentional stop ended the pinned wake (spec §6.4).
		return abandon(errSwapStopped)
	}
	return SwapResult{Outcome: SwapDown, Gen: g, Reason: werr.Error(), Drain: drain}
}

// holdWakeLocked is a hold's re-check before the spawn, made when its stop
// has finished and under the same lock. If the check passes, it installs
// the pinned waking entry. The caller holds a.mu and starts the wake.
func (a *Service) holdWakeLocked(e *actEntry, h *holdSpec) (*actEntry, error) {
	pin := h.gen
	next := &actEntry{host: e.host, route: e.route, state: actWaking, ready: make(chan struct{}), pin: &pin,
		settleCancel: h.settleCancel}
	// The same gate here, at the stop's end, and again at the pinned wake's
	// exec (spec §6.3, §7.4): the lease and the restore come between.
	next.pinCheck = func() error { return a.attemptCurrentLocked(next, h.at, h.newest, true) }
	a.entries[e.host] = next
	if err := next.pinCheck(); err != nil {
		delete(a.entries, e.host)
		return nil, err
	}
	return next, nil
}

// startPinnedChecked is a hold's pinned spawn behind its ownership and its
// attempt's checks. The gate is the attempt's checks (pinCheck, set by
// holdWakeLocked: the entry, the fences, the shutdown, the revision) plus
// the liveness of the lease this wake actually holds (leaseLost; the entry
// has no replica yet, so ownedLocked's e.rep check cannot see it). It runs
// under a.mu before the preparation, and again at the exec itself,
// immediately before StartStaged (final review F7): a renewal that failed
// or a fence set during Restore, the stage check or the tier read execs
// nothing. Its refusals are NoOwnership.
//
// The execution evidence comes from the spawn's result (final review F10):
// a returned process, or an unproven-start sentinel, marks the wake as one
// whose build may have run, before doWake cleans the failed wake up.
// Anything else ran nothing of the build.
func (a *Service) startPinnedChecked(ctx context.Context, e *actEntry, rt registry.Route, pin Gen, lease Lease) (Proc, string, Gen, error) {
	gate := func() error {
		a.mu.Lock()
		defer a.mu.Unlock()
		if e.pinCheck != nil {
			if err := e.pinCheck(); err != nil {
				return err
			}
		}
		return leaseLost(lease)
	}
	if err := gate(); err != nil {
		return nil, "", pin, err
	}
	proc, ver, g, err := a.startPinnedAt(ctx, rt, pin, gate, time.Time{})
	if proc != nil || genStartUnproven(err) {
		a.markPinExec(e)
	}
	return proc, ver, g, err
}

// markPinExec records e's pinned wake's execution evidence (pinExec).
// startPinnedChecked sets it from the spawn's result. A claims-on pinned
// spawn (Task 10) sets it from its exec hook; every claims-backed hold is
// refused before its stop on this base (claimsSwapRefusal), so that path
// is never a hold's here.
func (a *Service) markPinExec(e *actEntry) {
	a.mu.Lock()
	e.pinExec = true
	a.mu.Unlock()
}

// pinExecReached reports whether e's pinned wake has execution evidence.
func (a *Service) pinExecReached(e *actEntry) bool {
	a.mu.Lock()
	defer a.mu.Unlock()
	return e.pinExec
}

// wakeFailureEvent is the member's event for a failed wake, and whether
// there is one. An unpinned wake's is what it always was. A hold's pinned
// wake is the attempt's business (plan 5's swap events carry its member
// line): a refusal before the exec, a stale or cancelled one, or a spawn
// that ran the wrong build is logged only, so no swap text reaches the
// member's timeline. A pinned build that ran and failed gets the wording
// every failed wake gets.
func (a *Service) wakeFailureEvent(e *actEntry, err error) (string, bool) {
	if !a.allowFilesDiagnostic(e.host, err, "event") {
		return "", false
	}
	if e.pin == nil {
		return memberSafeWakeReason(err), true
	}
	if !a.pinExecReached(e) || classifyStep(nil, err) != stepFailed || errors.Is(err, errPinMismatch) {
		log.Printf("activator: %s: hold's pinned wake refused or abandoned: %v", e.host, err)
		return "", false
	}
	return memberSafeWakeReason(err), true
}

// startPinned is a hold's spawn: the attempt's staged build, verified and
// exec'd by path at its own tier, then the health check through the
// generation handle. StartStaged returns as soon as the process runs (Task
// 1), so this health check, and its deadline, is the only one. The proc
// comes back even on failure, so doWake's cleanupFailedWake stops it and
// records the outcome, and a caller that reads a nil proc as "nothing ran"
// is right to.
func (a *Service) startPinned(ctx context.Context, rt registry.Route, pin Gen) (Proc, string, Gen, error) {
	return a.startPinnedAt(ctx, rt, pin, nil, time.Time{})
}

// startPinnedAt is startPinned with gate, when set, asked immediately
// before StartStaged: a refusal there returns no process, with nothing of
// the build run. The same gate rides into StartStaged on its context
// (withExecGate; final review F1) and is asked again at the backend's final
// launch point, after the backend's own preparation (the config read, the
// sockets, a nested row's launch file). A refusal there is the same: no
// process, so no execution evidence (startPinnedChecked, spawnPlanned).
//
// healthBy, when set, caps the health check (zero-downtime spec §8.2): a
// first wake's B is checked until the earlier of healthBy (the waiters'
// deadline minus 25 s) and spawnHealthTimeout, which genHealthy applies
// itself. StartStaged returns the process at once, so this is the only
// health bound, and a B that misses it comes back with its process for the
// caller to record and kill. Zero leaves the check as it was.
func (a *Service) startPinnedAt(ctx context.Context, rt registry.Route, pin Gen, gate func() error, healthBy time.Time) (Proc, string, Gen, error) {
	gb, ok := a.backend.(SwapBackend)
	if !ok {
		return nil, "", pin, errSwapNoGens
	}
	if err := verifyStaged(pin.StagedPath, pin.Digest); err != nil {
		return nil, "", pin, fmt.Errorf("staging: %w", err)
	}
	pin.N = nextGen()
	if pin.Tier == (cgroup.Limit{}) {
		pin.Tier = gb.GenTier(rt, pin.Build) // the pinned artifact's own tier (spec §7.3)
	}
	if gate != nil {
		if err := gate(); err != nil {
			return nil, "", pin, err
		}
	}
	// Settings spec §4.2: the build state names pin.Build before it runs
	// against the data. Every staged instance exec funnels through here: a
	// hold's pinned wake, a planned boot, the good build booted untried and
	// a planned fallback (a different build touching the data, so it
	// records again). A refusal returns no process: nothing ran. If the
	// backend's own exec gate then refuses, the state names a build about to
	// run that did not, which the spec allows ("or one about to").
	// Timed for the journal (wake latency, pass 2).
	began := time.Now()
	if err := a.recordBuild(ctx, rt.Host, pin.Build); err != nil {
		return nil, "", pin, err
	}
	recorded := time.Now()
	g, proc, err := gb.StartStaged(withExecGate(ctx, gate), rt, pin)
	if err != nil {
		return proc, "", pin, err
	}
	started := time.Now()
	if g.N != pin.N || g.Build != pin.Build || g.StagedPath != pin.StagedPath {
		return proc, "", pin, fmt.Errorf("%w: generation %d of %s from %q; the hold chose %s from %q", errPinMismatch, g.N, g.Build, g.StagedPath, pin.Build, pin.StagedPath)
	}
	hctx := ctx
	if !healthBy.IsZero() {
		var cancel context.CancelFunc
		hctx, cancel = context.WithDeadline(ctx, healthBy)
		defer cancel()
	}
	if err := a.genHealthy(hctx, g); err != nil {
		return proc, "", g, err
	}
	log.Printf("activator: %s: generation %d of %s: record %s, start %s, healthy %s later", rt.Host, g.N, g.Build,
		recorded.Sub(began).Round(time.Millisecond), started.Sub(recorded).Round(time.Millisecond), time.Since(started).Round(time.Millisecond))
	gb.GenReady(rt, g, proc)
	return proc, g.Build, g, nil
}

// foreignRow reports whether host's row has the foreign layout (a
// hand-set socket path), which keeps its canonical socket. A flat row whose
// generation path would not fit a unix socket path reads foreign here too
// (swapLayout, rev 19 rule 4).
func (a *Service) foreignRow(host string) bool {
	a.mu.Lock()
	e := a.entries[host]
	var rt registry.Route
	if e != nil {
		rt = e.route
	}
	a.mu.Unlock()
	if e == nil {
		return false
	}
	l, ok := a.swapLayout(rt)
	return ok && l == LayoutForeign
}

// swapLayout is rt's layout for the swap's decisions: the backend's
// GenLayout where it has one (a flat row whose generation path would not
// fit reads foreign, rev 19 rule 4), else its Classifier layout. ok is false
// for a backend that can say neither, whose rows are never refused on
// layout here.
func (a *Service) swapLayout(rt registry.Route) (Layout, bool) {
	if g, ok := a.backend.(genLayouter); ok {
		return g.GenLayout(rt), true
	}
	if c, ok := a.backend.(Classifier); ok {
		l, _ := c.Layout(rt)
		return l, true
	}
	return "", false
}