package host

import (
	"context"
	"errors"
	"log"
	"os"
	"path/filepath"
	"runtime"
	"sync"
	"sync/atomic"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/appconfig"
	"github.com/carlosframework/platform/internal/cgroup"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/rollout"
	"github.com/carlosframework/platform/internal/status"
)

// The reason codes RolloutReport.Unavailable carries (zero-downtime deploys
// spec §15, §8.4). They are codes, not member-facing text: plan 6 words
// them.
const (
	// The flag is on but the box cannot contain generations, so the
	// scheduler is not armed and every leg stays on today's paths (swapArming).
	rolloutNoDelegation   = "no-cgroup-delegation"
	rolloutNoGenControl   = "no-generation-control"
	rolloutGenerationsOff = "generations-off"
	// The flag is on but host claims are on: claims-backed launches cannot
	// yet carry a swap's recovery choice, its execution evidence, the
	// generations file or the good build (#272, "claims-on swaps"), so the
	// box declines and stays on today's paths.
	rolloutHostClaims = "host-claims-on"
	// The flag is on and the box has delegation, but a test placement into
	// a scratch cgroup through the shim failed (placementCheck): generation
	// starts could not be placed inside their limits (#272, the second
	// dev-box check), so the box declines and stays on today's paths.
	rolloutPlacementFailed = "cgroup-placement-failed"
	// The flag is on and generations are on, but the scheduler's required
	// state is unavailable, so every managed start is refused
	// (refuseManaged).
	rolloutNoStore        = "no-deploy-store"
	rolloutNoReleaseKey   = "no-release-key"
	rolloutBookUnreadable = "failure-book-unreadable"
)

// refuseTexts is the wake refusal each refuseManaged code carries. It reaches
// a member through the activator's "wake failed:" event, so it names the
// missing fact and never the raw error, which goes to the agent's log. The
// unreadable book's text is the scheduler's own (rollout Task 5 refuses a
// wake whose intent it cannot write with it), shared rather than copied.
var refuseTexts = map[string]string{
	rolloutNoStore:        "this box cannot start instances right now: the deploy store is unavailable",
	rolloutNoReleaseKey:   "this box cannot start instances right now: the release key is unavailable",
	rolloutBookUnreadable: rollout.BookUnreadableRefusal,
}

var rolloutMgr atomic.Pointer[rollout.Manager]

// rolloutMaterialize is the exec backend's Materialize hook, handed to the
// scheduler so a cold managed host is installed before its wake is planned
// (Task 10). runAgent sets it before armRollout; nil refuses such a wake.
var rolloutMaterialize func(ctx context.Context, host string) error

// armedRollout is the box's rollout manager, nil unless CARLOS_SWAP=overlap
// armed one. Every leg that would bounce an exec child asks it first.
func armedRollout() *rollout.Manager { return rolloutMgr.Load() }

// statusNudge asks the status loop for an early tick: a rollout transition
// should reach status/boxes/<label>.json in seconds, not at the next
// five-minute tick. Never sent to with the flag off. Capacity one, and every
// send is non-blocking, so a burst of transitions is one nudge.
var statusNudge = make(chan struct{}, 1)

// nudgeStatus is the scheduler's OnChange: a non-blocking send that
// coalesces, as Deps.OnChange requires. It never waits for the status loop.
func nudgeStatus() {
	select {
	case statusNudge <- struct{}{}:
	default:
	}
}

// rolloutDown is why CARLOS_SWAP=overlap is set but the scheduler could not
// arm while generations are on, or why a box that declined the flag cannot
// read the failure records it must honour: managed starts are refused and
// the bounce legs stand down (refuseManaged). Nil when it armed, when the
// flag is off, or when the box declined the flag (rolloutDeclined) with a
// readable failure book.
var rolloutDown atomic.Pointer[string]

// rolloutDeclined is why CARLOS_SWAP=overlap is set but the box stayed on
// today's paths: it cannot contain generations, has host claims on, or
// failed its arm-time test placement (swapArming). Unlike
// rolloutDown nothing is refused and no leg changes, except where failure
// records an earlier armed run left hold a build for an instance
// (honourDeclinedFailures, final review F6). The reason is published in the
// box's status.
var rolloutDeclined atomic.Pointer[string]

// rolloutUnavailable is rolloutDown's reason, "" when there is none. A leg
// that bounces exec children stands down while it is set.
func rolloutUnavailable() string {
	if p := rolloutDown.Load(); p != nil {
		return *p
	}
	return ""
}

// rolloutUnarmed is rolloutDeclined's reason, "" when there is none.
func rolloutUnarmed() string {
	if p := rolloutDeclined.Load(); p != nil {
		return *p
	}
	return ""
}

func declineRollout(code string) {
	c := code
	rolloutDeclined.Store(&c)
}

// swapDelegation is what this box lacks to contain generations, "" when
// nothing. A flat or foreign row's generation children live under the
// agent's delegated cgroup subtree (eb.Cgroups): without it a spawn still
// runs, unplaced, but KillGen and the parent raise answer
// cgroup.ErrNoDelegation, so an overlap would be abandoned and a failed first
// wake's kill would be unproven. A nested row's generations are root's
// carlos-instance-gen@ units, reached through the instance control channel,
// which must speak the generation half (activator.GenControl); whether root
// can run them is asked per wake by the backend's nested gate (eb.NestedOK)
// and per call by the channel itself, and a failed StopGen is an unproven
// kill the activator already keeps the lease for. So the box-wide question
// for nested rows is only whether the channel has the generation verbs.
func swapDelegation(eb *activator.ExecBackend) string {
	if eb == nil || eb.Cgroups == nil {
		return rolloutNoDelegation
	}
	if _, ok := eb.Control.(activator.GenControl); !ok {
		return rolloutNoGenControl
	}
	return ""
}

// placementCheck is swapArming's arm-time placement check: one real shim
// placement into a scratch child of the delegated subtree
// (cgroup.Manager.VerifyPlacement). VerifyPlacement bounds itself, the
// shim's waits included, at 4 s, so a decline lands well inside the 20 s
// `carlos system update` gives the restarted edge (systemGateWait). A seam
// host tests replace, since a temp-dir root can never pass it.
var placementCheck = func(m *cgroup.Manager) error {
	return m.VerifyPlacement(context.Background())
}

// swapArming decides, before the activator exists, whether CARLOS_SWAP=overlap
// turns generations on (spec §15; plan 3's review: missing delegation must
// fail closed, CODEX-FINDINGS-ZDD-IMPL-3.md:138). With the flag off it is
// false and says nothing. With the flag on but host claims on, or no usable
// containment, it is false too (claims-on boxes get no swaps, the owner's
// ruling, #272): generations, generation containment and the scheduler all stay
// off, every leg keeps today's paths but for the failure records armRollout
// honours (honourDeclinedFailures), and the refusal is logged loudly and
// published as a reason code in the box's status (rolloutStatusProbe). With
// containment present it asks for one real placement last (placementCheck),
// and declines the same way when it fails: an armed box whose generation
// starts can never be placed fails every one of them (#272).
func swapArming(getenv func(string) string, eb *activator.ExecBackend) bool {
	if !rollout.Enabled(getenv) {
		return false
	}
	if eb != nil && eb.Cfg.HostWrites.Enabled {
		log.Printf("agent: REFUSING %s=overlap (%s): host claims are on, and claims-backed launches cannot yet keep a swap's "+
			"recovery choice, execution evidence, generations file or good build, so this box stays on today's stop-then-wake paths",
			rollout.SwapEnv, rolloutHostClaims)
		declineRollout(rolloutHostClaims)
		return false
	}
	if why := swapDelegation(eb); why != "" {
		log.Printf("agent: REFUSING %s=overlap (%s): generation containment needs the agent's delegated cgroup subtree "+
			"and the control channel's generation units, so this box stays on today's stop-then-wake paths", rollout.SwapEnv, why)
		declineRollout(why)
		return false
	}
	if err := placementCheck(eb.Cgroups); err != nil {
		log.Printf("agent: REFUSING %s=overlap (%s): a test placement into a scratch cgroup failed (%v), so generations could not be started inside their limits; this box stays on today's stop-then-wake paths", rollout.SwapEnv, rolloutPlacementFailed, err)
		declineRollout(rolloutPlacementFailed)
		return false
	}
	return true
}

// refuseManaged is the flag on with the scheduler's required state
// unavailable (spec §6.4, §8.3, §8.4). The legacy paths are NOT a fallback:
// a wake from the live link could exec a build the unreadable failure book
// holds down, including a hold build after its migration failed. So every
// managed wake is refused before the lease, and an unmanaged one (a named
// unit, a unit-backed row) wakes as today. The bounce-based legs stand down
// too (fastConvergeLoop, adoptBounceFn, restartRequestLeg,
// syncExecSerialized): a stop whose next wake is refused would only take an
// instance down. So do the activator's own two moves onto the live link: the
// sweep's convergence stop of an awake instance (OnDrift answers it with
// nothing, as the armed scheduler answers an unmanaged host) and a worker's
// recycle converging onto the live link (CoalesceRecycle present and false:
// the recycle respawns the worker's captured build). The reason code is
// published in the box's status (rolloutStatusProbe) until the agent
// restarts with the state readable.
func refuseManaged(svc *activator.Service, reason string) {
	log.Printf("agent: %s=overlap but %s: refusing every managed start, and every stop and move onto a new build, until the agent restarts with it readable",
		rollout.SwapEnv, reason)
	r := reason
	rolloutDown.Store(&r)
	text := refuseTexts[reason]
	if text == "" {
		text = refuseNoSwaps
	}
	svc.Planner = func(_ context.Context, rt registry.Route) activator.WakePlan {
		if !rollout.Managed(rt) {
			return activator.WakePlan{}
		}
		return activator.WakePlan{Refuse: text}
	}
	svc.OnDrift = func(string) {}
	svc.CoalesceRecycle = func(string) bool { return false }
}

// servingWriter is OnWorkerServing's registry write, off the publisher's
// path. The activator delivers each worker's serving identity under its
// box-wide servingMu, inside the respawn and before the respawn is marked
// done, so a slow registry write there would delay every host's publication,
// a stop and a lease release behind it. publish only records the newest
// version per host and returns; one drain goroutine at a time writes them.
// Deliveries arrive in incarnation order (servingMu), each overwrites the
// host's pending value, and the drain writes a host's values one after
// another, so the last write for a host is always the newest identity it
// was handed: an older incarnation's version can never land last. A
// coalesced intermediate version is never written, which only skips a value
// that was already stale.
type servingWriter struct {
	set     func(host, ver string) error
	mu      sync.Mutex
	pending map[string]string
	running bool
}

func newServingWriter(set func(host, ver string) error) *servingWriter {
	return &servingWriter{set: set, pending: map[string]string{}}
}

func (w *servingWriter) publish(host, ver string) {
	w.mu.Lock()
	w.pending[host] = ver
	start := !w.running
	w.running = true
	w.mu.Unlock()
	if start {
		go w.drain()
	}
}

func (w *servingWriter) drain() {
	for {
		w.mu.Lock()
		host, ver, found := "", "", false
		for h, v := range w.pending {
			host, ver, found = h, v, true
			break
		}
		if !found {
			w.running = false
			w.mu.Unlock()
			return
		}
		delete(w.pending, host)
		w.mu.Unlock()
		if err := w.set(host, ver); err != nil {
			log.Printf("agent: %s: record the worker's serving build %q: %v", host, ver, err)
		}
	}
}

// armAlways wires what the activator needs whatever CARLOS_SWAP says: a
// worker's serving identity (an empty version included) becomes the
// registry's ServingVersion, which `carlos sidecars ls` reads (spec §7.9,
// Task 11). The write goes through a servingWriter so the callback is cheap.
// The generations file is not set here: runAgent sets it flag-independently
// (plan 4 R17).
func armAlways(svc *activator.Service, reg *registry.Registry, dd string) {
	w := newServingWriter(reg.SetServingVersion)
	svc.OnWorkerServing = func(host string, id release.StagedIdentity) { w.publish(host, id.Version) }
}

// rolloutPolicy is the scheduler's policy loader (spec §4.1): the app's
// default config bundle. appconfig.Get answers an error for every store or
// decode failure, and only a confirmed missing object (os.ErrNotExist) as an
// empty manifest, which rollout.PolicyFromManifest reads exactly as nil:
// the defaults. A failed read leaves the policy cache with its last value,
// or hold for an app it has never read, so a bucket failure can never
// become an overlap.
func rolloutPolicy(store release.Store) func(ctx context.Context, acct, app string) (*appconfig.Manifest, error) {
	return func(ctx context.Context, acct, app string) (*appconfig.Manifest, error) {
		return appconfig.Get(ctx, store, acct, app)
	}
}

// touchStageMarker writes the stage-request marker, as stageRequester does.
// A failed write costs latency; carlos-adopt.timer is the floor. Task 15's
// stage relay uses it too.
func touchStageMarker(dd string) {
	if err := os.WriteFile(filepath.Join(dd, stageRequestMarker), []byte(time.Now().UTC().Format(time.RFC3339)+"\n"), 0o644); err != nil {
		log.Printf("agent: stage relay: %v", err)
	}
}

// armRollout builds the scheduler when CARLOS_SWAP=overlap (spec §15) and
// wires it into svc, which must not have started yet. With the flag off it
// returns nil and touches nothing. With the flag on but svc built without
// generations (swapArming declined the flag) it returns nil and touches
// nothing either: a manager is never armed over a Service that cannot run
// its attempts. It does honour the failure records an earlier armed run left
// (honourDeclinedFailures, final review F6): an infrastructure prerequisite
// failing is not an operator clearing a failure hold. With the flag on but
// no deploy store, no public key or an
// unreadable failure book, it returns nil and refuses every managed start
// (refuseManaged): each of those is a fact the scheduler cannot do without
// (the canonical demand, the retry signal, the records), and today's paths
// would ignore the failure records.
func armRollout(ctx context.Context, getenv func(string) string, svc *activator.Service, reg *registry.Registry, dd string,
	sink activator.EventSink, openStore func(context.Context) (release.Store, error)) *rollout.Manager {
	if svc == nil || !rollout.Enabled(getenv) {
		return nil
	}
	if !svc.GenerationsOn() {
		if rolloutUnarmed() == "" {
			log.Printf("agent: REFUSING %s=overlap: the activator was built without generations", rollout.SwapEnv)
			declineRollout(rolloutGenerationsOff)
		}
		honourDeclinedFailures(svc, reg, dd, release.OptDir())
		return nil
	}
	store, err := openStore(ctx)
	if err != nil {
		log.Printf("agent: %s=overlap: deploy store: %v", rollout.SwapEnv, err)
		refuseManaged(svc, rolloutNoStore)
		return nil
	}
	pub, err := release.ParsePublicKey(getenv("CARLOS_RELEASE_PUBKEY"), "CARLOS_RELEASE_PUBKEY")
	if err != nil {
		log.Printf("agent: %s=overlap: release key: %v", rollout.SwapEnv, err)
		refuseManaged(svc, rolloutNoReleaseKey)
		return nil
	}
	book, err := rollout.OpenFailureBook(dd)
	if err != nil {
		log.Printf("agent: %s=overlap: failure book: %v", rollout.SwapEnv, err)
		refuseManaged(svc, rolloutBookUnreadable)
		return nil
	}
	if n := book.Reconciled(); n > 0 {
		// Executions an earlier agent began and never saw recorded (Task 5,
		// spec §8.4): their intents are failure records now, saved before
		// anything runs (the Flush below). A box that cannot save them must
		// not run either: its next restart would find the intents again, and
		// nothing it recorded meanwhile would be kept.
		log.Printf("agent: %s=overlap: %d execution(s) an earlier agent left unrecorded are now failure records", rollout.SwapEnv, n)
	}
	// The book is saved before anything runs: an open may owe a save (a new
	// book's identity, a book an earlier agent wrote given a new one,
	// reconciled intents), and until it lands status publishes no
	// generation and no retry grant clears anything (the crash-epoch hard
	// limit in rollout.FailureBook's doc). A box that cannot save refuses
	// managed starts, as it does for reconciliation.
	if err := book.Flush(); err != nil {
		log.Printf("agent: %s=overlap: save the failure book after opening it: %v", rollout.SwapEnv, err)
		refuseManaged(svc, rolloutBookUnreadable)
		return nil
	}
	platform := runtime.GOOS + "-" + runtime.GOARCH
	m := rollout.NewManager(ctx, rollout.Deps{
		Act: svc, Routes: reg, Book: book, OptDir: release.OptDir(), DataDir: dd, Box: BoxLabel(),
		Demand: func(ctx context.Context, acct, app, ver, path string) (string, string, error) {
			return release.CanonicalDemand(ctx, store, acct, app, ver, platform, path)
		},
		Policy: rolloutPolicy(store),
		// The raw signed read (PF4): only its Version and PromotedAt are
		// used, to clear failure records.
		Pointer: func(ctx context.Context, acct, app, channel string) (*release.ChannelPointer, error) {
			return release.Channel(ctx, store, pub, acct, app, channel)
		},
		Events:      sink,
		Materialize: rolloutMaterialize, // the exec backend's hook, set by runAgent
		Touch:       func() { touchStageMarker(dd) },
		OnChange:    nudgeStatus,
	})
	svc.Planner = m.Plan
	svc.OnGen = m.OnGen
	svc.OnDrift = m.Nudge
	svc.CoalesceRecycle = m.CoalesceRecycle
	if routes, err := reg.List(); err == nil {
		m.Bootstrap(routes) // spec §8.5: awake, healthy instances record their serving build
		m.Pass(routes)
	}
	rolloutMgr.Store(m)
	log.Printf("agent: overlap swaps armed (%s=overlap)", rollout.SwapEnv)
	return m
}

// rolloutLegFor is the converge leg with the flag on (spec §10): the agent
// pass of spec §6.1, every adoptedConvergeEvery, in place of convergeAdopted.
func rolloutLegFor(m *rollout.Manager, reg *registry.Registry) func() {
	return func() {
		routes, err := reg.List()
		if err != nil {
			return
		}
		m.Pass(routes)
	}
}

// fastConvergeLeg is the leg fastConvergeLoop runs every
// adoptedConvergeEvery: the scheduler's pass when it is armed, none while
// managed starts are refused (legacy convergence would stop instances whose
// next wake is refused), and today's convergeAdopted otherwise, the flag off
// or declined included (declined with failure records, it passes by a held
// host: convergeLegFor).
func fastConvergeLeg(svc *activator.Service, reg *registry.Registry) func() {
	if m := armedRollout(); m != nil {
		return rolloutLegFor(m, reg)
	}
	if rolloutUnavailable() != "" {
		return nil
	}
	return convergeLegFor(svc, reg)
}

// rolloutConfigCallbacks is config sync's callbacks when the scheduler, or
// the failure records, have a say, and false when they have none (the flag
// off, or declined with no records), in which case syncExecSerialized runs
// today's SyncExec unchanged.
//
// Armed and refusing, the pass splits as the scheduler does
// (appconfig.Callbacks.Managed, final review F7): an exec route the scheduler
// does not manage gets today's restart (legs.restart, svc.Bounce) and today's
// ack, never a no-op that claims a restart. A managed route:
//
// Armed: the scheduler restarts it, paced, and a revision is acknowledged
// only once it is active (spec §8.1, §10); appconfig calls Delivered after
// the file is written (syncExecHostClaimed), for a managed native sidecar
// too, whose config attempt is the scheduler's. A sleeper's ack is
// Hibernating (configTrack).
//
// Refusing: delivered and acknowledged at delivery, but no restart, because
// the wake after it would be refused. An AWAKE managed host (legs.awake says
// which) still runs its old config, so its restart answers an error and the
// ack carries RestartError with Restarted false rather than claiming a
// restart that did not happen (spec §8.1). A sleeper reads the file on its
// next wake, so its ack says delivered on wake (appconfig.ErrDeliveredOnWake
// sets Hibernating, the unit pass's marker).
//
// Declined with records: today's restart and ack for every host but one a
// record holds (failureHolds.configCallbacks).
func rolloutConfigCallbacks(ctx context.Context, store release.Store, legs configLegs) (appconfig.Callbacks, bool) {
	if m := armedRollout(); m != nil {
		return appconfig.Callbacks{
			Restart:   func(string) error { return nil },
			Delivered: m.ConfigDelivered,
			Ack:       m.DeferAck(configAckWriter(store, "agent")),
			Managed:   rollout.Managed,
			Unmanaged: legs.restart,
		}, true
	}
	if rolloutUnavailable() != "" {
		restart := func(host string) error {
			if legs.awake != nil && legs.awake(host) {
				return errors.New(rolloutRestartRefused)
			}
			return appconfig.ErrDeliveredOnWake
		}
		return appconfig.Callbacks{Restart: restart, Ack: configAckFunc(ctx, store, "agent"),
			Managed: rollout.Managed, Unmanaged: legs.restart}, true
	}
	if h := heldHolds(); h != nil {
		return h.configCallbacks(configAckFunc(ctx, store, "agent"), legs), true
	}
	return appconfig.Callbacks{}, false
}

// feedRestarts is the restart leg's exec half with the flag on (spec §10):
// each moved app's exec instances and sidecars, and each narrowed request's
// hosts, get a restart attempt identified by the request (the app-wide
// generation's etag, or the newest narrowed request's id). The root half,
// named units and unit backing, is unchanged.
func feedRestarts(routes []registry.Route, moved [][2]string, scoped []scopedRestart, gen map[[2]string]string, feed func(host, id string) bool) int {
	want := map[[2]string]bool{}
	for _, k := range moved {
		want[k] = true
	}
	n := 0
	for _, rt := range routes {
		k := [2]string{release.RouteAccount(rt), rt.App}
		if !want[k] || !restartEligible(rt) || !execBounceable(rt) {
			continue
		}
		if feed(rt.Host, "gen:"+gen[k]) {
			n++
		}
	}
	for _, w := range scoped {
		if len(w.requests) == 0 {
			continue
		}
		id := "scope:" + w.requests[len(w.requests)-1].ID
		for _, host := range scopedHosts(routes, w.app, w.requests, execBounceable) {
			if feed(host, id) {
				n++
			}
		}
	}
	return n
}

// configLegs is what config sync's callbacks need from the activator.
type configLegs struct {
	// awake reports a managed instance the activator has a live entry for
	// (managedAwake). Nil: nothing is awake.
	awake func(host string) bool
	// restart is today's restart, bounce(svc): what config sync does to an
	// exec instance with the flag off. Nil: no restart.
	restart func(host string) error
}

// rolloutStatusProbe feeds status.Probes.Rollout: nil with the flag off, the
// scheduler's report when it is armed, and only the reason code when the
// flag is on but the scheduler is not armed (refusing, or declined). A
// declined box's report keeps the decline code and says what it does with
// the failure records (FailureBook): honoured, unreadable (every managed
// start refused), or nothing when there are none. Honoured, it also names
// each instance the book holds, with the build holding it and that
// restriction's epoch (failureHolds.report; Task 10, plan review #5), so
// the console can sign a retry for it.
func rolloutStatusProbe() *status.RolloutReport { return rootRetryStatus(legacyRolloutStatusProbe()) }

func legacyRolloutStatusProbe() *status.RolloutReport {
	m := armedRollout()
	if m == nil {
		if why := rolloutUnarmed(); why != "" {
			r := &status.RolloutReport{Unavailable: why, FailureBook: rolloutUnavailable()}
			if h := heldHolds(); h != nil {
				r.FailureBook = rolloutBookHonoured
				r.Hosts = h.report()
			}
			return r
		}
		if why := rolloutUnavailable(); why != "" {
			return &status.RolloutReport{Unavailable: why}
		}
		return nil
	}
	r := m.Status()
	return &r
}

// rolloutRestartRefused is the RestartError an awake managed host's config
// ack carries while managed starts are refused. It is fixed text: the cause
// goes to the agent's log, never to the member (plan 6 words it).
const rolloutRestartRefused = "restart refused: this box cannot restart instances right now"

// managedAwake reports whether host is a managed exec instance the activator
// has a live entry for (waking, awake or stopping, as Owns counts them), so
// one whose process may still hold its old config. A host that is not in the
// registry, not managed, or not owned is not awake; nor is anything with no
// activator.
func managedAwake(svc *activator.Service, reg *registry.Registry) func(host string) bool {
	return func(host string) bool {
		if svc == nil || reg == nil || !svc.Owns(host) {
			return false
		}
		rt, err := reg.Get(host)
		return err == nil && rt != nil && rollout.Managed(*rt)
	}
}