package activator

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

	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// stagedWorkerStarter is the optional Backend capability behind spec §7.9:
// spawn a worker from a NAMED staged build (onto != nil, a recycle) or from
// whatever live/<host> names now (onto == nil, a wake or a converge), and say
// exactly which bytes were exec'd. ExecBackend and SplitBackend have it. A
// backend without it keeps today's StartWorker, and its workers have no
// captured identity to recycle onto. An implementation runs ctx's
// before-exec hook (runBeforeExec) once it has resolved the identity and
// before it execs, and does not exec when the hook fails (ruling R19 item 3).
// It asks ctx's exec gate (withExecGate) as StartStaged does: after all of
// its preparation, the cgroup's included, immediately before each start
// attempt, and when the gate refuses starts nothing and returns its error
// (Astra R3 §5 item 4).
type stagedWorkerStarter interface {
	StartWorkerStaged(ctx context.Context, rt registry.Route, env map[string]string, onto *release.StagedIdentity) (Proc, release.StagedIdentity, error)
}

var (
	_ stagedWorkerStarter = (*ExecBackend)(nil)
	_ stagedWorkerStarter = (*SplitBackend)(nil)
)

// ErrBuildNotPublished is a worker spawn refused because the build it was
// about to exec could not be written to the generations file first (ruling
// R19 item 3): root could not see that build as referenced, so it is not
// run. The failed-spawn handling applies, and the sweep retries the write.
// It is never a failure of the build (classifyStep). Its text is main's
// (final review F9): it reaches a member through a worker wake's failure
// event with the flag off, which must read as it always did. Plan 5's
// staged publication has its own sentinel, errStagedBuildNotPublished.
var ErrBuildNotPublished = errors.New("worker build not published before its exec")

// errStagedBuildNotPublished is ErrBuildNotPublished for plan 5's staged
// publication (Task 12, publishBeforeExec): the instance spawns (a planned
// or pinned wake, a replanned build, an overlap successor) and a planned
// wake's step 0.8, which a planned worker wake passes too before its lease.
// A worker's own exec-time publication (startWorker's hook) keeps
// ErrBuildNotPublished. Abandonment as well (classifyStep), and only
// reachable with the flag on. A member's wake event never shows its text:
// memberSafeWakeReason maps it to the maintenance fence's wording (final
// review F9).
var errStagedBuildNotPublished = errors.New("staged build not published before its exec")

// beforeExecKey carries the activator's before-exec hook on a spawn's
// context (withBeforeExec).
type beforeExecKey struct{}

// withBeforeExec returns ctx carrying f, the hook a staged worker spawn runs
// between resolving the identity it will exec and the exec itself (ruling R19
// item 3). It rides on the context so StartWorkerStaged's signature, and the
// stagedWorkerStarter interface, stay as they are.
//
// The hook is the build's publication, then the build-state record (settings
// spec §4.2) behind one ask of the caller's gate, as an instance's record
// sits behind its gate in startPinnedAt. That ask is not the exec-time
// check: the gate also rides the context as an instance spawn's does
// (withExecGate), because the hook runs before the cgroup's preparation and
// once, where ownership has to be asked after that preparation and before
// every start attempt (Astra R3 §5 item 4, ruling 206).
func withBeforeExec(ctx context.Context, f func(release.StagedIdentity) error) context.Context {
	return context.WithValue(ctx, beforeExecKey{}, f)
}

// runBeforeExec runs ctx's before-exec hook, if any, with the identity about
// to be exec'd. A backend that implements stagedWorkerStarter calls it after
// resolving that identity and before its spawn's preparation, and does not
// exec when it returns an error.
func runBeforeExec(ctx context.Context, id release.StagedIdentity) error {
	if f, ok := ctx.Value(beforeExecKey{}).(func(release.StagedIdentity) error); ok && f != nil {
		return f(id)
	}
	return nil
}

// startWorker spawns through the staged capability when the backend has it,
// and through StartWorker otherwise. Never call it with a.mu held.
//
// Through the staged capability, the build about to be exec'd is published
// BEFORE the exec (ruling R19 item 3): the backend's before-exec hook records
// it as a pending generation and writes the generations file. A write that
// fails refuses the spawn with ErrBuildNotPublished. The returned
// pendingGeneration names that record. On an error startWorker has already
// dropped it; on success the caller drops it (dropPendingGenerationLocked)
// under a.mu in the same critical section that installs the captured
// identity, or after shedding the child, and then publishes again.
//
// gate, when set, is the caller's ownership check at the exec (final review
// F1). Through the staged capability it rides the context as the exec gate
// (withExecGate), and the backend asks it as an instance spawn's: after the
// materialization, the publication, the cgroup's preparation and the
// process's placement, and before EACH exec, the PID-move fallback after a
// failed flat placement included (Astra R3 §5 item 4, ruling 206). In the
// before-exec hook, as it
// was, it was asked once and before the cgroup was prepared, so ownership
// lost during that preparation, or between a failed start and its retry,
// still started the worker. On a backend without the staged capability it
// is asked immediately before StartWorker. Its refusal execs nothing and is
// returned as is.
func (a *Service) startWorker(ctx context.Context, rt registry.Route, env map[string]string, onto *release.StagedIdentity, gate func() error) (Proc, release.StagedIdentity, pendingGeneration, error) {
	if a.cfg.HostWrites.Enabled {
		return nil, release.StagedIdentity{}, 0, ErrClaimIsolationUnavailable
	}
	s, ok := a.backend.(stagedWorkerStarter)
	if !ok {
		// No staged capability: no identity is captured, so no build state
		// is recorded. Production's SplitBackend always has the capability;
		// it sends a unit-backed row (which UnitBackend refuses as a worker,
		// and which holds no real lease) to StartWorker without the hook.
		// So this arm, and that one, record nothing: settings spec §4.2
		// covers the rows that run under a lease in the replica bucket.
		if gate != nil {
			if err := gate(); err != nil {
				return nil, release.StagedIdentity{}, 0, err
			}
		}
		p, ver, err := a.backend.StartWorker(ctx, rt, env)
		return p, release.StagedIdentity{Version: ver}, 0, err
	}
	var pend pendingGeneration
	hook := func(id release.StagedIdentity) error {
		if pend != 0 {
			a.mu.Lock()
			a.dropPendingGenerationLocked(pend)
			a.mu.Unlock()
		}
		pend = a.addPendingGeneration(rt.Host, id)
		if err := a.persistGenerations(); err != nil {
			log.Printf("activator: worker %s not spawned: the build it would exec (%s %s) could not be published first: %v", rt.Host, id.Version, id.StagedPath, err)
			return fmt.Errorf("%w: %v", ErrBuildNotPublished, err)
		}
		// Settings spec §4.2: the build state names the exact identity
		// about to be exec'd, under the worker's lease, after its
		// publication. Every worker spawn passes here: a wake (planned or
		// not, boot and fallback), a crash restart, a max_run recycle, a
		// converge and ReplaceWorker. The caller's gate is asked first, as
		// startPinnedAt and spawnSuccessor ask theirs before their record,
		// so a worker whose ownership is already gone records nothing. The
		// backend asks it again on the context (withExecGate) after the
		// cgroup's preparation and before each start attempt; a refusal
		// there leaves the state naming a build about to run that did not,
		// which the spec allows ("or one about to").
		if gate != nil {
			if err := gate(); err != nil {
				return err
			}
		}
		return a.recordBuild(ctx, rt.Host, id.Version)
	}
	p, id, err := s.StartWorkerStaged(withBeforeExec(withExecGate(ctx, gate), hook), rt, env, onto)
	if err != nil && pend != 0 {
		a.mu.Lock()
		a.dropPendingGenerationLocked(pend)
		a.mu.Unlock()
		_ = a.persistGenerations()
		pend = 0
	}
	return p, id, pend, err
}

// recycleWorkerLocked replaces a worker's child for a reason that is not a
// build change: a crash, or an always-on worker's max_run. Automatic
// recycling never changes a worker's build (spec §7.9): the child is
// respawned from its SERVING build, the staged path and digest captured at
// its exec, whatever live/<host> names by now. The caller holds a.mu.
//
// CoalesceRecycle is asked first. When it says a queued release attempt is
// admissible now, that attempt makes the one spawn and the recycle is
// skipped. A worker spawned by a backend without the staged capability has
// no captured build. With no scheduler (CoalesceRecycle nil) it keeps
// today's respawn from the live link. With one, armed or refusing, it never
// does (Astra R3 §5 item 4): the live link names whatever root installed
// last, a build no admission chose, and on a refusing box possibly one the
// unreadable failure book holds down. So a live child (stopOld, a max_run
// recycle) is left running, and a dead one stays down until a release
// attempt or a wake starts it. The agent says so once per host.
//
// Then the drift (ruling R13). Until plan 5 queues it as a release attempt,
// the converge is the one path that moves a worker onto an installed build,
// so a recycle must not starve it: an exited worker never reaches the
// sweep's drift check, and a max_run shorter than ConvergeGrace would recycle
// before the check fired. So the recycle reads live/<host> as the sweep does
// (ConvergeDecision over a.liveVer and spawnVersion), starting the drift
// clock if this is the first sight of it, and once the drift has lasted
// ConvergeGrace it converges instead (onto nil).
//
// That converge is paced like the sweep's own (ruling R17): it spends the
// sweep's converge budget (a.convergeBudget, reset each sweep), so a box
// whose workers' recycles fall due together still moves at most
// maxConvergeCyclesPerSweep of them per sweep. And it happens only while no
// scheduler is present: when CoalesceRecycle is set (plan 5), that
// scheduler's admission owns every build change, so a recycle it did not
// coalesce never converges. Either way, a converge that is not admitted
// recycles the captured build, and the drift keeps its age for a later
// sweep.
//
// On a box that declined CARLOS_SWAP=overlap and honours failure records or
// hold-ran markers, a move onto the live link is also asked of the live
// exec gate first (SetLiveExecGate, Task 9), before anything is stopped: a
// converge it refuses recycles the captured build instead, and a worker
// with no captured build is not respawned from the link, so a live child
// keeps running and a dead one stays down until a wake starts it.
func (a *Service) recycleWorkerLocked(e *actEntry, stopOld Proc, why string) {
	if hook := a.CoalesceRecycle; hook != nil && hook(e.host) {
		log.Printf("activator: worker %s %s; a queued release attempt makes the spawn instead of a recycle", e.host, why)
		return
	}
	var onto *release.StagedIdentity
	if e.workerServing.StagedPath != "" {
		cur, readErr := a.liveVer(e.host)
		converge, next := ConvergeDecision(cur, readErr, e.spawnVersion, e.staleSince, a.now(), ConvergeGrace)
		e.staleSince = next
		switch {
		case !converge:
		case a.CoalesceRecycle != nil:
			log.Printf("activator: worker %s %s; live/%s names %s, but the release scheduler owns build changes, so it recycles its captured build %s", e.host, why, e.host, cur, e.spawnVersion)
		case a.convergeBudget <= 0:
			log.Printf("activator: worker %s %s; live/%s names %s, but this sweep's converge budget is spent, so it recycles its captured build %s and converges on a later sweep", e.host, why, e.host, cur, e.spawnVersion)
		case a.liveGateRefusesLocked(e, cur, "it recycles its captured build "+e.spawnVersion):
			// A declined box that honours a record or marker holding cur
			// (Task 9): the captured build is recycled, as for a converge
			// not admitted, and the drift keeps its age.
		default:
			a.convergeBudget--
			log.Printf("activator: worker %s %s; it serves %s but live/%s has named %s for at least %s, so it converges onto the new build instead of recycling", e.host, why, e.spawnVersion, e.host, cur, ConvergeGrace)
			a.respawnWorkerLocked(e, stopOld, "converging onto a new build", nil)
			return
		}
		id := e.workerServing
		onto = &id
	} else if a.CoalesceRecycle != nil {
		if !a.uncapturedSaid[e.host] {
			if a.uncapturedSaid == nil {
				a.uncapturedSaid = map[string]bool{}
			}
			a.uncapturedSaid[e.host] = true
			left := uncapturedLeft(stopOld)
			log.Printf("activator: worker %s %s; it has no captured build, and the release scheduler owns its recycles, so it is not respawned from live/%s: %s", e.host, why, e.host, left)
		}
		return
	} else if a.liveLinkRefusedLocked(e, uncapturedLeft(stopOld)) {
		// No captured build, and the build the live link names is one this
		// declined box will not start (Task 9): asked before anything is
		// stopped, so a live child keeps running and a dead one stays down
		// until a wake starts it, as on an armed box.
		return
	} else {
		log.Printf("activator: worker %s %s; no captured build to recycle onto, so it respawns from live/%s", e.host, why, e.host)
	}
	a.respawnWorkerLocked(e, stopOld, why, onto)
}

// liveLinkRefusedLocked is liveGateRefusesLocked for the build live/<host>
// names now. With no live exec gate (every box but a declined one that
// honours something) it reads nothing and refuses nothing, so the recycle is
// today's. A link that cannot be read is left to the spawn's own gate
// (StartWorkerStaged). The caller holds a.mu.
func (a *Service) liveLinkRefusedLocked(e *actEntry, left string) bool {
	if a.liveExecGate == nil {
		return false
	}
	cur, err := a.liveVer(e.host)
	return err == nil && a.liveGateRefusesLocked(e, cur, left)
}

// uncapturedLeft is what becomes of a worker with no captured build whose
// recycle is not made: a live child (stopOld) keeps running, a dead one
// stays down.
func uncapturedLeft(stopOld Proc) string {
	if stopOld == nil {
		return "it stays down until a release attempt or a wake starts it"
	}
	return "the running child keeps running"
}

// WorkerReportedVersion is what host's worker last said it runs, over the
// agent socket: display metadata only (spec §7.9). It may be empty or an
// app-chosen string, and it never identifies the executable. "" when no
// worker entry exists.
func (a *Service) WorkerReportedVersion(host string) string {
	a.mu.Lock()
	defer a.mu.Unlock()
	if e := a.entries[host]; e != nil && e.worker {
		return e.workerVersion
	}
	return ""
}