package activator

import (
	"log"
	"path/filepath"
	"sort"

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

// Worker pin publication (zero-downtime deploys spec §4.2, §8.5; rulings
// R17, R19).
//
// A worker recycles onto the build it was exec'd from, its captured serving
// identity, long after live/<host> and the registry row have moved on. Root's
// staging and root's version sweep must never overwrite or remove that staged
// copy, and root is a separate process that cannot see this one's memory, so
// the activator publishes it: the generations file (release.WriteGenerations,
// read by root through release.GenerationPins and release.GenerationVersions)
// lists, for every worker entry it holds with a captured serving build and
// for every build a spawn is about to exec, that build under the worker's
// host, by version (when it is a valid version name) and by its staged path
// (when it has one). It does so whatever CARLOS_SWAP says; this
// flag-independent writer is the one plan 5 extends.
//
// Plan 5 (Task 12) adds every live or pending INSTANCE generation (spec
// §8.5: current, successor, retiring, spawning): the current generation, a
// hold's pinned one, a planned wake's build and its fallback, every
// operation-set member, a failed wake retained until its end is proven, and
// a socketless attempt's builds. An instance generation is listed with
// generations on, or whenever it was exec'd from a staged path
// (genIdentity). Since settings Task 5c every exec-backed wake runs its
// recorded file by path, flag off included, so a flag-off file names those
// instances' files too; only a generation started from the live link (a
// backend that cannot pin a file) stays unlisted, the link being root's
// pin already.
//
// A cold wake (no plan, no pin) captures the live link once after its
// restore and publishes it as a pending build before it records it
// (settleWakeIdentity, ruling R-COLDPUBLISH); wake() drops that pending
// build once the entry's generation names the file.
//
// Every staged instance spawn (a planned wake and its fallback, a hold's
// pinned wake, an overlap successor) publishes the entry state that names
// its build BEFORE the exec, and does not exec when that write fails
// (errStagedBuildNotPublished): doWake and doClaimWake before the lease, checkPlan
// for a replanned build, swapOverlap before its Step 3a. The entry state is
// set first (e.plan, e.pin, the operation set) and then written, rather than
// recorded as a pending build: that state already lives as long as the
// spawn's process may, so nothing has to be dropped when the spawn installs.
//
// A hold is the exception (zdd-readiness Task 13, Astra R1 T12, R3 §5 item
// 9). It stops its incumbent before its pinned entry exists, and from that
// stop until the pinned wake publishes nothing serves and nothing names the
// hold's build, so root could sweep or restage it under the wake about to
// run it. swapHold records the build as a pending generation and writes the
// file BEFORE the stop, and drops the record when it returns: by then the
// pinned entry, or a retained failed wake, names the build for as long as a
// process of it may run.
//
// The file is written at Start, and before the first worker spawn (so it exists, even
// empty, before any worker runs), before every staged spawn's exec, naming
// the build about to run (startWorker's before-exec hook, ruling R19 item 3),
// after every spawn or respawn installs its captured identity in place of
// that pending one, and after every worker stop removes one. A write that
// fails is logged and marks the publication dirty, and the sweep retries it.
// A failed write before an exec refuses that spawn; elsewhere it blocks
// nothing, because root reads a publication that did not finish as an
// incomplete pin set (release.ReadGenerations) and then neither restages nor
// sweeps.

// pendingGeneration names one build a spawn is about to exec, recorded by
// startWorker's before-exec hook until the spawn installs or fails. 0 is
// none.
type pendingGeneration uint64

type pendingBuild struct {
	host string
	id   release.StagedIdentity
}

// addPendingGeneration records id as a build host's spawn is about to exec,
// and returns its name. Never call it with a.mu held.
func (a *Service) addPendingGeneration(host string, id release.StagedIdentity) pendingGeneration {
	a.mu.Lock()
	defer a.mu.Unlock()
	return a.addPendingGenerationLocked(host, id)
}

// addPendingGenerationLocked is addPendingGeneration for a caller that holds
// a.mu: a hold registers its build in the same critical section as its
// admission (swapHold).
func (a *Service) addPendingGenerationLocked(host string, id release.StagedIdentity) pendingGeneration {
	if a.genPending == nil {
		a.genPending = map[pendingGeneration]pendingBuild{}
	}
	a.genPendingSeq++
	k := pendingGeneration(a.genPendingSeq)
	a.genPending[k] = pendingBuild{host: host, id: id}
	return k
}

// dropPendingGenerationLocked forgets a pending build: its spawn failed, its
// child was shed, or its captured identity has been installed on the entry.
// 0 is a no-op. The caller holds a.mu.
func (a *Service) dropPendingGenerationLocked(k pendingGeneration) {
	if k != 0 {
		delete(a.genPending, k)
	}
}

// generationsSnapshotLocked is what the generations file should say now: for
// every entry (waking, awake, mid-swap, mid-respawn or stopping, since its
// processes may still run) what entryGenerationsLocked names, the build of
// every retained failed wake (unprovenWakes), and every pending build, each
// listed under its host by version when the version is a valid version name,
// and by staged path (made absolute) when it has one. A build with neither
// names nothing root could restage or sweep. Each host's lists are sorted and
// hold each entry once. The caller holds a.mu.
func (a *Service) generationsSnapshotLocked() release.Generations {
	hosts, paths := map[string]map[string]bool{}, map[string]map[string]bool{}
	add := func(host string, id release.StagedIdentity) {
		if v := id.Version; v != "" && release.ValidVersion(v) {
			if hosts[host] == nil {
				hosts[host] = map[string]bool{}
			}
			hosts[host][v] = true
		}
		if p := id.StagedPath; p != "" {
			// A relative path is published as the exec resolved it, from
			// this process's working directory; if even that fails, the
			// relative path stays and WriteGenerations refuses it, so the
			// spawn does not happen.
			if abs, err := filepath.Abs(p); err == nil {
				p = abs
			}
			if paths[host] == nil {
				paths[host] = map[string]bool{}
			}
			paths[host][p] = true
		}
	}
	for host, e := range a.entries {
		a.entryGenerationsLocked(e, func(id release.StagedIdentity) { add(host, id) })
	}
	// A failed wake retained until its termination is proven: its entry is
	// gone, but its process may still run the build (F12; spec §4.2, §8.5).
	// A worker's is its captured identity, an instance's its generation.
	for host, u := range a.unprovenWakes {
		add(host, u.build)
		if id, ok := a.genIdentity(u.gen); ok {
			add(host, id)
		}
	}
	for _, pb := range a.genPending {
		add(pb.host, pb.id)
	}
	return release.Generations{Hosts: sortedLists(hosts), Paths: sortedLists(paths)}
}

// entryGenerationsLocked calls add with every build e's processes run or may
// be about to run (spec §8.5):
//   - a worker's captured serving build;
//   - the current generation, and every operation-set member's (successor,
//     retiring, spawning), and the build of a shed worker child the set (or,
//     with generations off, the entry) retains (final review F12, R2-1);
//   - a hold's pinned generation, from the moment its waking entry exists;
//   - a planned wake's build and its fallback, from the moment the plan is
//     stored, so both are on disk before either execs;
//   - a socketless attempt's build and fallback (Task 11);
//   - the generation every run not yet proven settled was admitted
//     against (merge ruling M2).
//
// The plan's builds stay listed while the entry lives, as e.plan does
// (afterWake reads it after the wake): a few more pins than the processes
// strictly need, never fewer. The caller holds a.mu.
func (a *Service) entryGenerationsLocked(e *actEntry, add func(release.StagedIdentity)) {
	gen := func(g Gen) {
		if id, ok := a.genIdentity(g); ok {
			add(id)
		}
	}
	if e.worker {
		add(e.workerServing)
	}
	gen(e.gen)
	if e.pin != nil {
		gen(*e.pin)
	}
	if e.plan.planned() {
		add(e.plan.Boot)
		if e.plan.Fallback != nil {
			add(*e.plan.Fallback)
		}
	}
	for _, m := range e.ops.members {
		add(m.build)
		gen(m.gen)
	}
	if e.shed != nil {
		add(e.shed.build)
	}
	for _, v := range e.attemptBuilds {
		add(release.StagedIdentity{Version: v})
	}
	// A run's generation, until the run is proven settled (merge ruling
	// M2): its command execs that build, and may outlive the generation's
	// own process once a swap retires it.
	for _, h := range cmdHoldsLocked(e) {
		gen(h.gen)
	}
}

// genIdentity is g as the generations file names it, and whether it is
// listed at all. A generation exec'd from a staged path is listed whatever
// the switch says. Otherwise it is listed only with generations on: with
// them off an instance's generation is the live link's build, which root
// pins as the row's Version already, and plan 4's flag-off file never
// named it.
func (a *Service) genIdentity(g Gen) (release.StagedIdentity, bool) {
	if g.StagedPath == "" && (!a.gens || g.Build == "") {
		return release.StagedIdentity{}, false
	}
	return release.StagedIdentity{Version: g.Build, StagedPath: g.StagedPath}, true
}

// publishesLocked reports whether e names anything in the generations file,
// so a change that removes e must rewrite it. The caller holds a.mu.
func (a *Service) publishesLocked(e *actEntry) bool {
	named := false
	a.entryGenerationsLocked(e, func(id release.StagedIdentity) {
		if id.StagedPath != "" || release.ValidVersion(id.Version) {
			named = true
		}
	})
	return named
}

// publishBeforeExec writes the generations file before a staged instance
// spawn's exec (spec §4.2, §8.5; the index's generations-file contract).
// The caller has already put the build about to run on its entry (e.plan,
// e.pin, the operation set), so the snapshot names it. When the write fails
// the caller does not exec: root reads only what the file names, and could
// sweep or restage a build it does not name under the process. The refusal
// is errStagedBuildNotPublished, which classifyStep reads as abandonment,
// never a failure of the build. The write error itself goes to the operator
// log (persistGenerations), not into the returned error, which may reach a
// member's wake event (as the fence's wording, memberSafeWakeReason). Never
// call it with a.mu held.
func (a *Service) publishBeforeExec() error {
	if err := a.persistGenerations(); err != nil {
		return errStagedBuildNotPublished
	}
	return nil
}

func sortedLists(m map[string]map[string]bool) map[string][]string {
	out := make(map[string][]string, len(m))
	for k, set := range m {
		l := make([]string, 0, len(set))
		for v := range set {
			l = append(l, v)
		}
		sort.Strings(l)
		out[k] = l
	}
	return out
}

// persistGenerations writes the generations file from a snapshot of the
// entries. Never call it with a.mu held: it takes genMu first, then a.mu only
// for the snapshot, and does the file I/O off a.mu. genMu serialises the
// writers, and each one snapshots after taking it, so an older snapshot can
// never land over a newer one. A no-op when GenerationsFile is empty.
func (a *Service) persistGenerations() error {
	if a.GenerationsFile == "" {
		return nil
	}
	a.genMu.Lock()
	defer a.genMu.Unlock()
	return a.persistGenerationsGenLocked()
}

// persistGenerationsGenLocked is persistGenerations' body. The caller holds
// genMu and not a.mu.
func (a *Service) persistGenerationsGenLocked() error {
	a.mu.Lock()
	g := a.generationsSnapshotLocked()
	a.mu.Unlock()
	if err := release.WriteGenerations(a.GenerationsFile, g); err != nil {
		a.genDirty = true
		log.Printf("activator: publish generations to %s: %v; the sweep retries", a.GenerationsFile, err)
		return err
	}
	a.genWritten = true
	a.genDirty = false
	return nil
}

// publishGenerationsBeforeSpawn makes sure the generations file has been
// written once before a worker is spawned. A failure is logged and left to
// the sweep; this write does not decide the spawn, the staged spawn's own
// before-exec publication does (startWorker). Never call it with a.mu held.
func (a *Service) publishGenerationsBeforeSpawn() {
	if a.GenerationsFile == "" {
		return
	}
	a.genMu.Lock()
	defer a.genMu.Unlock()
	if a.genWritten {
		return
	}
	_ = a.persistGenerationsGenLocked()
}

// retryGenerations rewrites the generations file if the last write failed.
// The sweep calls it after releasing a.mu.
func (a *Service) retryGenerations() {
	if a.GenerationsFile == "" {
		return
	}
	a.genMu.Lock()
	defer a.genMu.Unlock()
	if !a.genDirty {
		return
	}
	_ = a.persistGenerationsGenLocked()
}