package activator

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

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

// ReplaceWorker runs one socketless attempt (zero-downtime spec §6.5, §7.9)
// on an awake worker or native sidecar member: stop-then-start on the SAME
// entry, through respawnWorkerLocked, so the lease and the replicator stay
// put. The new child must still be running firstWakeSettle (10 s) after it
// started; a clean exit inside that window is a one-shot's success. On a
// failure:
//   - restart or config: the build it replaced is respawned (spec §6.5);
//   - release, install or retry in hold mode: nothing falls back, and the
//     entry stops, so the instance stays down (spec §6.4, §8.3);
//   - release, install or retry in overlap mode: fallback, the host's good
//     build, is respawned in its place; with none, the entry stops, as a
//     failed wake does today (spec §8.2).
//
// Every respawn here asks plan 3's ownership check at its exec, and keeps
// its termination proof (respawnWorkerLocked). The attempt's own respawn
// also asks the attempt's §7.4 checks there, at all three of its points
// (before the preparation, at the exec, before the install). A refusal
// that is a supersession brings back the build the attempt replaced, unless
// the attempt is a hold whose build was exec'd before the refusal (its
// install check shed it): that build may have migrated the database, so
// nothing is restored and the entry stops (final review F2; spec §6.4,
// §8.3). Any other refusal (no ownership) spawns nothing at all (spec
// §7.4).
//
// The scheduler runs the cohort-failure transition from the result, then
// calls PublishHeldServing for a fallback's identity. newest is called with
// a.mu held, as Swap's is.
func (a *Service) ReplaceWorker(ctx context.Context, at Attempt, fallback *release.StagedIdentity, newest func(host string) uint64) (out SwapResult) {
	if a.cfg.HostWrites.Enabled {
		return abandon(ErrClaimIsolationUnavailable)
	}
	// The execution evidence (SwapResult.Executed, final review F8): the
	// attempt's respawn started a process of its build (attemptExec),
	// installed or shed. Read once the respawn is over; false before it
	// starts. A caller that stops waiting still gets it (final review wave
	// 2, N4): the respawn is joined, or reported executed-or-unknown.
	var executed bool
	defer func() { out.Executed = executed }()
	if newest == nil {
		return abandon(errSwapSuperseded)
	}
	if _, ok := a.backend.(stagedWorkerStarter); !ok {
		// A backend that cannot exec a named build would run whatever the
		// live link names: nothing of the attempt's choosing (as a planned
		// worker wake is refused, spawnPlannedWorker).
		return abandon(errSwapNoGens)
	}
	onto := release.StagedIdentity{Version: at.Build, StagedPath: at.StagedPath, Digest: at.Digest, Demand: at.Demand}
	a.mu.Lock()
	e := a.entries[at.Host]
	if err := a.swapChecksLocked(e, at, newest); err != nil {
		a.mu.Unlock()
		return abandon(err)
	}
	if !e.worker {
		a.mu.Unlock()
		return abandon(errSwapKind)
	}
	if e.workerRespawning || e.firstWatch != nil {
		a.mu.Unlock()
		return abandon(errSwapBusy)
	}
	prev := e.workerServing
	// The attempt's build and its fallback, for the generations file (Task
	// 12). startWorker's pending generation already names the build about
	// to exec at each spawn; listing the attempt's own here too is harmless.
	e.attemptBuilds = []string{onto.Version}
	if fallback != nil {
		e.attemptBuilds = append(e.attemptBuilds, fallback.Version)
	}
	// The same checks, carried into the respawn: it runs them again before
	// its preparation, at the exec and before the install, after the old
	// child has retired (spec §6.3, §7.4, §8.1).
	e.attemptGate = func() error { return a.swapChecksLocked(e, at, newest) }
	e.attemptVeto, e.attemptExec = nil, false
	// Spec §9.4: the bounce event names the attempt's cause.
	emitEvent(a.Logs, e.route, "info", bounceText(true, at.Cause))
	// The incarnation before the respawn: the attempt's child counts only if
	// the respawn installed a NEW one (P5 review finding 5). A failed exec
	// keeps the old e.proc and its identity, which for a restart or a config
	// attempt is the same artifact, so the artifact alone passed it.
	inc0 := e.workerInc
	a.respawnWorkerLocked(e, e.proc, "attempt ("+string(at.Cause)+")", &onto)
	done := e.respawnDone
	a.mu.Unlock()
	defer func() {
		a.mu.Lock()
		e.attemptBuilds = nil
		a.mu.Unlock()
		// The attempt's builds are no longer pinned on its account; what
		// the worker runs now is its captured build, published at install.
		_ = a.persistGenerations()
	}()

	if !a.awaitRespawn(ctx, done) {
		executed = a.joinAttemptRespawn(e, done)
		return abandon(errSwapStopped)
	}
	a.mu.Lock()
	veto := e.attemptVeto
	executed = e.attemptExec
	e.attemptVeto, e.attemptExec = nil, false
	if a.entries[at.Host] != e || e.state != actAwake {
		a.mu.Unlock()
		return abandon(errSwapStopped)
	}
	if veto != nil {
		// The respawn's re-check refused the attempt's build: before its
		// exec nothing of it ran; at the install check its child was shed.
		// Only a supersession brings back the build it replaced (index P5
		// #1); without ownership (a lost lease, a fence, shutdown) nothing
		// spawns, a restore included, and the entry stops through the
		// sweep's lease-loss path.
		switch {
		case errors.Is(veto, errBuildStateUnwritten) && !executed:
			// The attempt's build state could not be written, so its
			// exec was refused (settings spec §4.2): nothing of it ran,
			// and ownership is intact. verdictOf reads it as NoOwnership
			// (so a plain respawn logs a refusal), but here the replaced
			// build comes back at once, as for an unpublished build (the
			// incarnation-unchanged arm below). Its restore records its
			// own build first, and a refusal there leaves the dead child
			// to the sweep's crash restart.
			a.restoreLocked(e, prev)
		case verdictOf(veto) != gateSuperseded:
		case executed && at.Mode == ModeHold:
			// A hold build that was exec'd may have migrated the database,
			// and nothing falls back from a hold (spec §6.4, §8.3; final
			// review F2): no restore of the replaced build. The entry stops
			// and stays down until its next attempt or wake. Abandoned, not
			// a failure of the build: it was superseded, not seen to fail.
			log.Printf("activator: worker %s: hold attempt onto %s was superseded after its exec; the build it replaced (%s) is not restored, and the worker stops", at.Host, onto.Version, prev.Version)
			a.beginStopLocked(e, StopRelease)
		default:
			a.restoreLocked(e, prev)
		}
		a.mu.Unlock()
		return abandon(fmt.Errorf("re-checked at the spawn: %w", veto))
	}
	if e.workerInc == inc0 || !sameArtifact(e.workerServing, onto) {
		// Nothing of the attempt's build ran: no new incarnation was
		// installed (the exec failed, its build could not be published
		// first, the old child is not proven gone), or the backend exec'd
		// other bytes. Put the build it replaced back, as a failed restart
		// does (spec §6.5), through a respawn that asks ownership and the
		// termination proof at its own exec. This is a staging refusal,
		// never a failure of the build, and never a success: a restart or
		// config attempt whose child never ran acknowledges nothing (spec
		// §8.1).
		a.restoreLocked(e, prev)
		a.mu.Unlock()
		return abandon(fmt.Errorf("staging: worker not respawned onto %s", onto.Version))
	}
	// The observation. firstWatch hides the worker from Current and lets an
	// intentional stop (beginStopLocked) cancel it. unexpectedExit is
	// cleared here, for this child: the arms that set it never clear it, and
	// a mark left by an earlier crash would make beginStopLocked leave an
	// intentional stop to be read as this build's exit.
	obsCtx, cancel := context.WithCancel(ctx)
	e.firstWatch, e.unexpectedExit = cancel, false
	proc, drain, inc1 := e.proc, e.retireTook, e.workerInc
	a.mu.Unlock()

	r := observeSettle(obsCtx, proc, firstWakeSettle)
	cancel()

	a.mu.Lock()
	e.firstWatch = nil
	unexpected := e.unexpectedExit
	// mine: the entry, awake, still running the attempt's child. With the
	// scheduler's CoalesceRecycle nothing recycles it meanwhile; without it
	// (or after a stop), the attempt does not touch what replaced it.
	mine := a.entries[at.Host] == e && e.state == actAwake && e.workerInc == inc1 && !e.workerRespawning
	if r == settleOK || (r == settleExited && !procCrashed(proc)) {
		// The checks once more before the result counts (spec §6.3, §8.1):
		// a newer release or a halted cohort since the spawn means this
		// build must not be recorded good. Its child stays until the newer
		// attempt replaces it. A one-shot that finished may have been parked
		// meanwhile: no longer awake, but still not superseded. Only the
		// sweep's clean park of this very child counts (markCleanParkLocked,
		// past the sweep's own fence and lease checks), and only while no
		// fence or shutdown stands now. Any other stop of the exited child
		// (a lost lease, a fence, the agent's shutdown) leaves it not awake
		// too, with no ownership behind the exit, and is abandoned.
		err := a.swapChecksLocked(e, at, newest)
		if r == settleExited && errors.Is(err, errSwapNotAwake) &&
			e.parkedInc == inc1 && a.fences[at.Host] == 0 && !a.shuttingDown {
			err = nil
			if newest(at.Host) != at.SchedRev {
				err = errSwapSuperseded
			}
		}
		if err != nil {
			a.mu.Unlock()
			return abandon(err)
		}
		g := a.workerResultGen(e, onto)
		a.mu.Unlock()
		// A clean exit inside the window is a one-shot's success (spec
		// §6.5): nothing keeps running to be acknowledged later, so the
		// result says so and the scheduler records the build good from it.
		return SwapResult{Outcome: SwapDone, Gen: g, Drain: drain, CleanExit: r == settleExited}
	}
	if r == settleCancelled && !unexpected {
		// An intentional stop cancels the observation and records nothing
		// (spec §6.4).
		a.mu.Unlock()
		return abandon(errSwapStopped)
	}
	reason := "exited within " + firstWakeSettle.String() + " of starting"
	adoption := at.Cause == CauseRelease || at.Cause == CauseInstall || at.Cause == CauseRetry
	switch {
	case !adoption:
		if mine {
			a.restoreLocked(e, prev)
		}
		a.mu.Unlock()
		return SwapResult{Outcome: SwapFailed, Reason: reason}
	case at.Mode == ModeHold:
		if mine {
			a.beginStopLocked(e, StopRelease)
		}
		a.mu.Unlock()
		return SwapResult{Outcome: SwapDown, Reason: reason}
	case !mine:
		// The entry stopped, or another child replaced B, while it was
		// observed: there is nothing here to fall back on.
		a.mu.Unlock()
		return SwapResult{Outcome: SwapFailed, Reason: reason}
	case fallback == nil:
		a.beginStopLocked(e, StopRelease)
		a.mu.Unlock()
		return SwapResult{Outcome: SwapFailed, Reason: reason + "; no good build to fall back to"}
	}
	// The fallback's publication waits for the failure transition (spec
	// §8.1; P5 review finding 7): the respawn stores it (holdServing), and
	// the scheduler delivers it with PublishHeldServing once done() has
	// recorded B's failure. A synchronous publication here would sit in
	// front of that record, and a blocked callback would leave B unheld.
	// Like every respawn, the fallback asks ownership and the termination
	// proof at its exec: a lost lease spawns nothing (index P6 new#1).
	e.holdServing = true
	inc2 := e.workerInc
	a.respawnWorkerLocked(e, e.proc, "falling back to "+fallback.Version, fallback)
	done = e.respawnDone
	a.mu.Unlock()
	if !a.awaitRespawn(ctx, done) {
		return SwapResult{Outcome: SwapFailed, Reason: reason}
	}
	a.mu.Lock()
	defer a.mu.Unlock()
	if a.entries[at.Host] == e && e.workerInc != inc2 && sameArtifact(e.workerServing, *fallback) {
		return SwapResult{Outcome: SwapFailed, Gen: a.workerResultGen(e, *fallback), Reason: reason}
	}
	return SwapResult{Outcome: SwapFailed, Reason: reason + "; the good build did not start either"}
}

// restoreLocked respawns prev, the build an attempt replaced, on e (spec
// §6.5), retiring whatever child e holds. Like every respawn it asks
// ownership and the termination proof at its exec, so a host no longer
// owned gets nothing. Nothing is done when another respawn is in flight,
// or when prev was never captured (the sweep's crash restart then
// respawns from the live link, as a recycle with no captured build does).
// The caller holds a.mu.
func (a *Service) restoreLocked(e *actEntry, prev release.StagedIdentity) {
	if e.workerRespawning || prev.StagedPath == "" {
		return
	}
	a.respawnWorkerLocked(e, e.proc, "restoring "+prev.Version, &prev)
}

// workerResultGen is the generation a socketless attempt reports: id's
// artifact, marked Worker so the scheduler compares it by its artifact and
// leaves its serving version to publishWorkerServing, with the entry's
// generation number. The caller holds a.mu.
func (a *Service) workerResultGen(e *actEntry, id release.StagedIdentity) Gen {
	g := workerGenOf(id)
	g.N = e.gen.N
	return g
}

// PublishHeldServing delivers a worker's publication that a fallback held
// back (holdServing) until the caller had recorded the failure it falls
// back from (spec §8.1). Incarnation-ordered like any other: a newer install
// since makes it a no-op. Called with no lock held.
func (a *Service) PublishHeldServing(host string) {
	a.mu.Lock()
	e := a.entries[host]
	var held *heldServing
	if e != nil {
		held, e.heldServing = e.heldServing, nil
	}
	a.mu.Unlock()
	if held != nil {
		a.publishWorkerServing(e, held.id, held.inc)
	}
}

// heldServing is a worker publication held for PublishHeldServing.
type heldServing struct {
	id  release.StagedIdentity
	inc uint64
}

// publishWorkerServing delivers a worker's captured identity for
// incarnation inc of entry e (zero-downtime spec §7.9). It is delivered only
// while e is the host's entry and inc its current incarnation, checked and
// delivered under servingMu, so an older publication can never land after a
// newer one. Called with no lock held.
//
// An empty version is published too (P5 review finding 8): a child exec'd
// from an external live-link target has no version to capture, and the
// row's previous ServingVersion must be cleared, not left naming a build
// that no longer runs.
func (a *Service) publishWorkerServing(e *actEntry, id release.StagedIdentity, inc uint64) {
	if a.OnWorkerServing == nil {
		return
	}
	a.servingMu.Lock()
	defer a.servingMu.Unlock()
	a.mu.Lock()
	current := a.entries[e.host] == e && e.workerInc == inc
	a.mu.Unlock()
	if !current {
		return
	}
	a.OnWorkerServing(e.host, id)
}

// sameArtifact reports whether the backend exec'd exactly the chosen bytes:
// the staged path and its digest (the release demand is not the backend's
// to report).
func sameArtifact(got, want release.StagedIdentity) bool {
	return got.StagedPath == want.StagedPath && got.Digest == want.Digest
}

// joinAttemptRespawn is ReplaceWorker's answer for SwapResult.Executed when
// its caller stopped waiting before the attempt's respawn finished (final
// review wave 2, N4). The respawn runs on regardless: it is the entry's, on
// the service lifetime, and may still exec the attempt's build. So it is
// joined, bounded by its own worst case end to end (respawnJoinBound, the
// bound a terminal stop's join uses), and its evidence read as on the
// normal path. One still running past the bound is reported executed or
// unknown (true): only proof that nothing ran may count as unexecuted.
func (a *Service) joinAttemptRespawn(e *actEntry, done chan struct{}) bool {
	bound := time.NewTimer(respawnJoinBound)
	defer bound.Stop()
	select {
	case <-done:
	case <-bound.C:
		log.Printf("activator: worker %s: the attempt's respawn did not finish within %v of its caller leaving; reporting its build as possibly executed", e.host, respawnJoinBound)
		return true
	}
	a.mu.Lock()
	defer a.mu.Unlock()
	executed := e.attemptExec
	e.attemptVeto, e.attemptExec = nil, false
	return executed
}

// awaitRespawn waits for a respawn's goroutine to finish with every child it
// made (respawnDone closes last). False when ctx ended first.
func (a *Service) awaitRespawn(ctx context.Context, done chan struct{}) bool {
	select {
	case <-done:
		return true
	case <-ctx.Done():
		return false
	}
}