package activator

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

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

// errBuildStateUnwritten refuses a spawn whose build could not be recorded
// as the one about to touch the instance's database (settings spec §4.2).
// Nothing of the build ran, so it is never a failure of the build: the
// classifiers read it as a refusal (classifyStep, verdictOf), and a member
// sees the maintenance fence's wording (memberSafeWakeReason).
var errBuildStateUnwritten = errors.New("the build state could not be written, so this build was not started")

// recordBuild runs Service.BeforeExec, if set, for a spawn about to run
// build against host's database (settings spec §4.2): the build state names
// the build before the process opens the data, and a write that fails
// refuses the spawn with errBuildStateUnwritten. Callers hold the
// instance's lease and call it after their own gate, immediately before the
// exec. A nil BeforeExec is today's behaviour.
//
// Not to be confused with withBeforeExec / runBeforeExec (workerbuild.go):
// those carry startWorker's exec-time hook on a worker spawn's context, and
// that hook is one of recordBuild's callers.
//
// A cold wake's own record is pending until its first exec
// (settleWakeIdentity, amadan #272): a call for that build writes nothing
// here, and the exec seam writes it (runPendingRecord), after every refusal
// the backend makes. A call for another build writes now and supersedes
// it, so the exec seam never writes the wake's build over this one. A build
// with no name (an executable outside the version store) is refused while
// the hook is set: no build state could name it (ruling R-EXEC-NOVERSION).
//
// With host claims off the lease is the hibernate lease, so a context that
// names no location gets that one (Config.Bucket, HibernatePrefix), and the
// agent writes beside its lock.json. A claimed start attaches its own
// (startAdmitted); with claims on and none attached (a claimed instance's
// legacy sidecar restart), the agent locates the host through its claim.
func (a *Service) recordBuild(ctx context.Context, host, build string) error {
	if a.BeforeExec == nil {
		return nil
	}
	if p := pendingRecordFrom(ctx); p != nil && p.defers(build) {
		return nil
	}
	if build == "" {
		log.Printf("activator: %s not started: the executable it would run is outside the version store, so no build state can name it (settings spec 4.2); move the instance onto a promoted build", host)
		return fmt.Errorf("%w: the executable is outside the version store", errBuildStateUnwritten)
	}
	if _, ok := BuildStateLocationFrom(ctx); !ok && !a.cfg.HostWrites.Enabled {
		ctx = WithBuildStateLocation(ctx, BuildStateLocation{Bucket: a.cfg.Bucket, Prefix: a.cfg.hibernatePrefix(host)})
	}
	if err := a.BeforeExec(ctx, host, build); err != nil {
		log.Printf("activator: %s: %s not started: its build state could not be written: %v", host, build, err)
		return fmt.Errorf("%w: %v", errBuildStateUnwritten, err)
	}
	return nil
}

// askLiveExecGate asks the live exec gate (SetLiveExecGate) whether rt may
// start build, a build this branch captured from the live link and will run
// by path: a cold wake's step 2b capture (settleWakeIdentity) and a sidecar
// on an unstaged generation (startSidecar). The backend asks the gate only
// on its live-link arms (exe == "", onto == nil), which those starts never
// take, so without this a declined box's exec-time refusal (zdd-readiness
// Task 9) would not be asked at all. It is asked before the publication and
// the record, so a build the box will not start is never named as the one
// about to run (settings spec §4.2) and completes no request. No gate (every
// box but a declined one that honours records or markers) refuses nothing.
// The gate reads only memory, the row and the link.
func (a *Service) askLiveExecGate(rt registry.Route, build string) error {
	a.mu.Lock()
	gate := a.liveExecGate
	a.mu.Unlock()
	if gate == nil {
		return nil
	}
	return gate(rt, build)
}

// claimedBuild is the claimed path's version check before its record
// (ruling R-EXEC-CLAIMVERSION). The record must name the bytes that run:
// s.Staged is the file the snapshot hashed, and the launch verifies the
// file it opens against that digest. s.Version is a separate read of the
// live link, so a repoint between the hash and that read leaves the start
// naming a build other than the one it would run, and it is refused. With
// no hook nothing is recorded, and the start is today's.
func (a *Service) claimedBuild(host string, s ClaimStart) error {
	if a.BeforeExec == nil || s.Staged.Version == s.Version {
		return nil
	}
	log.Printf("activator: %s not started: the executable it hashed (%s) is build %q, but the start names build %q (the live link moved between the two reads); the next wake reads it again", host, s.Staged.StagedPath, s.Staged.Version, s.Version)
	return fmt.Errorf("%w: the executable hashed for this start is build %q, but the start names build %q", errBuildStateUnwritten, s.Staged.Version, s.Version)
}

// settleWakeIdentity is doWake's step 2b (settings spec §4.2, Astra I1
// #5): the one file this wake runs, its seed included, chosen under the
// wake's lease after the restore and before anything execs. Its record is
// made pending on the context it returns, and written at the first exec.
//
//   - A planned wake runs its plan's boot build, and a pinned wake (a hold
//     swap's) its pin. Step 0.8 published both before the lease.
//   - A held wake with a retained, verified copy of its target (heldID;
//     the held wake check, settings spec §4.3 rule 3, with no planner) runs
//     that copy.
//   - Any other wake (flag off; flag on with no plan; a worker) captures the
//     live link once (LiveIdentity, which materializes a missing link
//     first). That capture, and a retained copy, is published as a pending
//     generation before it is recorded (ruling R-COLDPUBLISH; anchors D18:
//     no plan or pin published a retained copy): the generations file names
//     a build before its build state does, and root's sweep keeps the file
//     until the process that runs it is gone. A write that fails refuses the
//     wake as a worker spawn's refuses its exec. The pending record is
//     returned so the wake drops it once the entry names the build itself
//     (wake), or the wake failed. Before that publication, a declined box's
//     live exec gate is asked about the captured build (askLiveExecGate),
//     so a build the box will not start is neither published nor recorded.
//   - A held wake (held, its target) refuses an identity that is neither
//     its target nor the planner's recovery boot (heldIdentityOK; Astra I2
//     #4): root repointed between the wait and the capture. That refusal
//     comes before the publication and the record, so nothing names it. An
//     identity with no version is never the target, so a held wake never
//     takes the unversioned live-link fallback below (ruling
//     R-EXEC-NOVERSION-2).
//   - A live link outside the version store (Version "") names no build a
//     build state could hold, and root refuses a staged executable with no
//     version (instancectl.validateStagedExe). With BeforeExec set it is
//     refused here (recordBuild, ruling R-EXEC-NOVERSION). With no hook (no
//     replica bucket) the wake is today's (ruling R-EXEC-NOVERSION-2): an
//     identity with no file comes back, nothing is published or recorded,
//     and the caller takes the live-link calls in every layout.
//
// The record is not written here (amadan #272). The context returned
// carries it pending (withPendingRecord), and the seed's or the spawn's
// exec seam writes it (runPendingRecord): after the backend's own refusals
// (the replica-key gate, and the cgroup's: a killed cgroup kept, a
// populated legacy parent, a limit not applied, a placement that failed),
// immediately before the process execs. A launch refused before its exec
// leaves no record naming a build that never ran, and completes no request
// (settings spec §4.2: the record is what completes a request, at
// execution). A recovery plan's boot (the good build in place of the
// installed one the failure book blocks) is recorded with the planner's
// selection (ruling R-PLANNED-RECOVERY), so the agent's revalidation admits
// it before root's consumed file names the installed build failed. With no
// hook nothing is pending, as nothing is recorded. Any error means nothing
// was seeded or spawned; the caller releases the lease.
func (a *Service) settleWakeIdentity(ctx context.Context, ib identityBackend, e *actEntry, rt registry.Route, held string, heldID *release.StagedIdentity) (release.StagedIdentity, pendingGeneration, context.Context, error) {
	a.mu.Lock()
	pin, plan := e.pin, e.plan
	a.mu.Unlock()
	var (
		id       release.StagedIdentity
		pend     pendingGeneration
		captured bool // neither planned nor pinned: published here, below
	)
	switch {
	case plan.planned():
		id = plan.Boot
	case pin != nil:
		id = release.StagedIdentity{Version: pin.Build, StagedPath: pin.StagedPath, Digest: pin.Digest, Demand: pin.Demand}
	case heldID != nil:
		id, captured = *heldID, true
	default:
		var err error
		if id, err = ib.LiveIdentity(ctx, rt); err != nil {
			return release.StagedIdentity{}, 0, ctx, err
		}
		captured = true
	}
	if !heldIdentityOK(held, id, plan, rt) {
		log.Printf("activator: %s not woken: it is held on %s, and this wake would run %q (the live link moved after the check); the next wake checks again", rt.Host, held, id.Version)
		return release.StagedIdentity{}, 0, ctx, fmt.Errorf("%w: %s would run %q, wants %s", errStaleLink, rt.Host, id.Version, held)
	}
	if captured {
		if id.Version == "" {
			return release.StagedIdentity{}, 0, ctx, a.recordBuild(ctx, rt.Host, "")
		}
		// A declined box's exec-time gate, asked about the file this wake
		// will run, before it is published or recorded (askLiveExecGate):
		// the spawn below runs it by path, past the backend's live-link
		// check. Its refusal comes back as it is, as the backend's would.
		if err := a.askLiveExecGate(rt, id.Version); err != nil {
			return release.StagedIdentity{}, 0, ctx, err
		}
		pend = a.addPendingGeneration(rt.Host, id)
		if err := a.persistGenerations(); err != nil {
			log.Printf("activator: %s not woken: the build it would run (%s %s) could not be published first: %v", rt.Host, id.Version, id.StagedPath, err)
			if e.worker {
				// A worker's wake reads as its exec-time refusal always has
				// (startWorker's hook, final review F9).
				return id, pend, ctx, fmt.Errorf("%w: %v", ErrBuildNotPublished, err)
			}
			return id, pend, ctx, errStagedBuildNotPublished
		}
	}
	if a.BeforeExec == nil {
		return id, pend, ctx, nil
	}
	rctx := ctx
	if plan.planned() {
		rctx = withRecoverySelection(ctx, plan, rt, id.Version)
	}
	build := id.Version
	write := func() error { return a.recordBuild(rctx, rt.Host, build) }
	// The wake's own admission, asked again once the write is done (Astra
	// review of zdd-owed-items, P1): the bucket write takes time, and a
	// fence raised or the wake's context ended meanwhile must start
	// nothing, as doWake's own check before the spawn refuses it.
	recheck := func() error {
		if ctx.Err() != nil || !a.wakeAllowed(rt.Host) {
			return ErrFenced
		}
		return nil
	}
	return id, pend, withPendingRecord(ctx, build, write, recheck), nil
}

// pendingRecord is one wake's record of build, written at most once, at
// the first exec that asks for it (runPendingRecord).
//
// The write is a bucket PUT, and the lease can be lost, a fence raised or
// an attempt superseded while it runs. So every exec that consults the
// record asks recheck (the wake's own admission: its context and the
// maintenance fence) after it, and an exec seam that has an exec gate asks
// that gate again once the write has run (runExecGate, launchGateWith).
// Either refusal starts nothing.
//
// A refusal there leaves the record naming a build that did not run. It is
// not undone. An undo would be a second bucket write, made after the lease
// may already be lost, so it could land under another holder's lease; the
// object's CAS would stop it overwriting that holder's record, but writing
// outside the lease is exactly what the build state's single-writer rule
// (ruling R-BS-UPDATE) forbids. The window is the one main had before
// amadan #272, where step 2b wrote the record and doWake then checked the
// fence: the record names a build that passed every refusal of its own and
// was about to run, the next wake records again, and a request it completed
// was the request for that build.
type pendingRecord struct {
	build   string
	write   func() error
	recheck func() error

	mu sync.Mutex
	// done is set once the write ran; err is its answer, which every later
	// exec of the wake gets too.
	done bool
	err  error
	// superseded is set once a record of another build was written in this
	// wake (recordBuild): the pending one would name the wrong build.
	superseded bool
}

type pendingWakeKey struct{}

// withPendingRecord returns ctx carrying build's pending record, written by
// write and followed by recheck (nil asks nothing) at every exec.
func withPendingRecord(ctx context.Context, build string, write, recheck func() error) context.Context {
	return context.WithValue(ctx, pendingWakeKey{}, &pendingRecord{build: build, write: write, recheck: recheck})
}

func pendingRecordFrom(ctx context.Context) *pendingRecord {
	p, _ := ctx.Value(pendingWakeKey{}).(*pendingRecord)
	return p
}

// defers reports whether a record of build is the pending one's to write
// (recordBuild then writes nothing). A record of another build supersedes
// it, and once it is superseded nothing defers to it.
func (p *pendingRecord) defers(build string) bool {
	if build == "" {
		return false
	}
	p.mu.Lock()
	defer p.mu.Unlock()
	if build != p.build {
		p.superseded = true
		return false
	}
	return !p.superseded
}

// run writes the record once, then asks recheck. wrote reports that this
// call made the write. A superseded record writes and asks nothing.
func (p *pendingRecord) run() (wrote bool, err error) {
	p.mu.Lock()
	if p.superseded {
		p.mu.Unlock()
		return false, nil
	}
	if !p.done {
		p.err, p.done, wrote = p.write(), true, true
	}
	err = p.err
	p.mu.Unlock()
	if err != nil {
		return wrote, err
	}
	if p.recheck != nil {
		if err := p.recheck(); err != nil {
			return wrote, err
		}
	}
	return wrote, nil
}

// runPendingRecord writes ctx's pending wake record, if it carries one that
// is still due. Every exec seam a cold wake reaches asks it immediately
// before its exec: the flat seed's and spawn's cgroup before-start hook,
// a worker's, and, nested, right before root is asked to run the seed or
// the run unit. An error refuses that exec, with nothing run. wrote
// reports that this call made the write, so a seam with an exec gate asks
// it again.
func runPendingRecord(ctx context.Context) (wrote bool, err error) {
	if p := pendingRecordFrom(ctx); p != nil {
		return p.run()
	}
	return false, nil
}

// pendingRecordHook is runPendingRecord as a cgroup before-start hook for
// a spawn that has no other: nil when ctx carries no pending record, so
// that spawn is exactly as it was.
func pendingRecordHook(ctx context.Context) func() error {
	if pendingRecordFrom(ctx) == nil {
		return nil
	}
	return func() error {
		_, err := runPendingRecord(ctx)
		return err
	}
}