package host

import (
	"context"
	"errors"
	"log"
	"sync"
	"sync/atomic"
	"time"

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

// rolloutBookHonoured is RolloutReport.FailureBook's code for a box that
// declined CARLOS_SWAP=overlap and honours the failure records an earlier
// armed run left (honourDeclinedFailures). A code, not member-facing text.
const rolloutBookHonoured = "honoured"

// refuseNoSwaps is the wake refusal for a managed instance whose build needs
// the scheduler this box cannot run: refuseManaged's text for a reason it has
// no words for, and a declined box's for a build a failure record holds.
const refuseNoSwaps = "this box cannot run two builds side by side, which this instance's build needs"

// failureHolds is the failure book a box that declined CARLOS_SWAP=overlap
// still honours (final review F6, which replaced PF15's "today's paths"). An
// earlier armed run recorded these failures, and losing the delegation the
// scheduler needs is not an operator clearing them: a hold build whose
// migration failed must not run again on the live link. So, with generations
// off and no scheduler, the box keeps today's paths except that a managed
// wake onto a build the book holds for that instance is refused (Planner),
// and refused again at the real launch, against the build that launch runs
// (gate, Task 9); and the converge leg, the adopt bounce, the restart leg and
// config sync's restart leave such an instance running rather than stop it
// into a refused wake. The book holds records and hold-ran markers.
//
// The book is read once, at arming, and nothing on a declined box records a
// failure or a marker, or prunes (Astra R2 ruling 194, "startup-only
// records"). But it is cleared while the agent runs, as an armed box's is
// (Task 10): a verified signed `carlos instances retry` (retryHost), a
// `carlos promote --retry` pointer and a signed rollback pointer
// (applyPointers) each clear in this one book and save it. The planner, the
// launch gate and the legs read it live, so a clear lets the instance start
// at once, with no agent restart. It is also written at arming when it owes
// a save (an armed run left intents in it, or an earlier agent wrote it),
// and by the periodic pass while that save keeps failing (pass; the
// crash-epoch hard limit, rule 6).
type failureHolds struct {
	book *rollout.FailureBook
	// get reads a route by host, for the legs that are handed only a host.
	get func(host string) (*registry.Route, error)
	// list reads every route, for the box's status report.
	list func() ([]registry.Route, error)
	// optDir is where the live links are: a wake execs the build
	// live/<host> names, which root repoints before it stamps the row.
	optDir string
	// said is the (leg, host, build) triples already logged, so a leg that
	// runs every two seconds says once that it is leaving a host alone.
	said sync.Map
}

// declinedHolds is set by honourDeclinedFailures, before any leg is built,
// and only when the book holds at least one record or hold-ran marker. Nil:
// the flag is off, armed, refusing, or declined with an empty book, and
// every leg is as its own branch says.
var declinedHolds atomic.Pointer[failureHolds]

// heldHolds is declinedHolds' value.
func heldHolds() *failureHolds { return declinedHolds.Load() }

// honourDeclinedFailures is armRollout's branch for a box that declined the
// flag (spec §§6.4, 8.4; final review F6). An absent book, or one with no
// records and no hold-ran markers, leaves everything exactly as today: no
// hook, no leg changed, LiveExecGate nil. An unreadable one fails closed as
// refuseManaged does for an armed box, since any record in it could be a
// hold-down. A readable one with records or markers is honoured, read-only,
// through the Planner, the launch gate and the legs (failureHolds).
func honourDeclinedFailures(svc *activator.Service, reg *registry.Registry, dd, optDir string) {
	book, err := rollout.OpenFailureBook(dd)
	if err != nil {
		log.Printf("agent: %s=overlap declined, and the failure book is unreadable: %v", rollout.SwapEnv, err)
		refuseManaged(svc, rolloutBookUnreadable)
		return
	}
	if n := book.Reconciled(); n > 0 {
		// An armed run's executions left unrecorded (Task 5): their records
		// are honoured below whether or not the save below lands, since the
		// book in memory holds them either way, and the next restart
		// reconciles the intents again if it did not.
		log.Printf("agent: %s=overlap declined: %d execution(s) an armed run left unrecorded are now failure records", rollout.SwapEnv, n)
	}
	// Records or hold-ran markers (Task 9): a marker alone still bars the good
	// build from an instance whose database a hold build may have migrated.
	recs, marks := len(book.List()), book.HoldRanCount()
	if recs == 0 && marks == 0 {
		return
	}
	// A book that owes a save is saved now, whatever reconciliation found
	// (the crash-epoch hard limit, rule 6): reconciled records, a new
	// identity for a book an earlier agent wrote (rule 4), epochs given to
	// older entries. Until it lands, status publishes no generation and no
	// retry grant clears anything; the periodic pass (pass) retries it. A
	// book this rule wrote and the open did not change owes nothing, and is
	// not written.
	if err := book.SaveIfDirty(); err != nil {
		log.Printf("agent: %s=overlap declined: save the failure book after opening it: %v (retried by the periodic pass)", rollout.SwapEnv, err)
	}
	h := &failureHolds{book: book, get: reg.Get, list: reg.List, optDir: optDir}
	declinedHolds.Store(h)
	svc.Planner = h.plan
	// And at the real launch, with the build actually started (Task 9): the
	// plan reads the live link before the lease and the restore, and root
	// can repoint it at a held build in between.
	svc.SetLiveExecGate(h.gate)
	log.Printf("agent: %s=overlap declined, but %d failure record(s) and %d hold-ran marker(s) from an armed run are honoured: "+
		"no managed instance wakes, starts, or is stopped, onto a build they hold for it", rollout.SwapEnv, recs, marks)
}

// refusal is the wake refusal for starting build on rt, "" when it may
// start. Only a managed route can be held: the scheduler records nothing
// else.
//
// The rule is the armed planner's (rollout.decideWake and decideHoldRan,
// spec §8.5, §6.4), with the one difference that a declined box cannot boot
// the good build instead, so it refuses where the armed box would recover.
// A hold-down holds whatever the instance served, in that build's words.
// Then, with a hold-ran marker of rt's app standing for the instance (Task
// 9; Astra R3 N5/N7, ruling 205):
//   - the marked build may start, unless the instance has a good build and
//     a record holds the marked build for it. A record never holds a build
//     the instance served, so a marker naming the good build itself lets it
//     start (the marked-build rule), whether the stamp or the live link
//     names it: unlike the armed planner, a declined box never boots the
//     good build as a recovery, so the build being started is always the
//     installed one, stamped or linked (fix round 1, ruling 4a);
//   - otherwise the good build is refused: an installed good build is no
//     proof of a signed rollback, since root's automatic sibling repoint
//     installs it too, and only the pointer pass clears the marker on one
//     (Task 10);
//   - any other build a record holds for the instance is refused.
//
// All three refuse in HeldDownRefusal's words for the marked build, as the
// armed planner does, so "roll back or retry" stays true here. Without a
// marker, a plain record holds a build only when the instance has a good
// build to fall back to (a build it has served is never held by a record
// alone); an instance with none boots its installed build as today, because
// refusing it would turn one sibling's failure into an outage for every
// sleeping sibling. It reads only memory, so the activator may ask it under
// its own lock (gate).
func (h *failureHolds) refusal(rt registry.Route, build string) string {
	return h.holdOf(rt, build).refuse
}

// hold is why the book holds an instance off one build: the wake refusal,
// the build whose restriction refuses it (by), which is what a retry of
// the instance must name (RetryHostAt), and whether the instance stays down
// for it (a hold-down, or a hold-ran marker) rather than on what it runs.
type hold struct {
	refuse, by string
	down       bool
}

// holdOf is refusal with the build it holds by: the hold-down's build, the
// marker's build when the marker refuses the marked build or the good
// build, and otherwise the build a record holds, which a marker may refuse
// in its own words while only that record's retry lifts it.
//
// Eligibility is rollout.Restrictable, not Managed (final review round 1,
// finding 2): a row a move holds (landing, restoring) is the row a moves
// probe or landing launches, and the book's restriction of the instance
// applies there as on its ready row.
func (h *failureHolds) holdOf(rt registry.Route, build string) hold {
	if !rollout.Restrictable(rt) || build == "" {
		return hold{}
	}
	acct, app, host := release.RouteAccount(rt), rt.App, rt.Host
	hasGood := rt.GoodVersion != ""
	if h.book.HeldDown(acct, app, build, host) {
		return hold{rollout.HeldDownRefusal(build), build, true}
	}
	if mark, ok := h.book.HoldRanOn(host); ok && mark.Account == acct && mark.App == app {
		isGood := hasGood && build == rt.GoodVersion
		switch {
		case build == mark.Build:
			if hasGood && h.book.Blocks(acct, app, build, host, isGood) {
				return hold{rollout.HeldDownRefusal(mark.Build), mark.Build, true}
			}
			return hold{}
		case isGood:
			return hold{rollout.HeldDownRefusal(mark.Build), mark.Build, true}
		case hasGood && h.book.Blocks(acct, app, build, host, false):
			return hold{rollout.HeldDownRefusal(mark.Build), build, true}
		}
		return hold{}
	}
	if hasGood && h.book.Blocks(acct, app, build, host, build == rt.GoodVersion) {
		return hold{refuseNoSwaps, build, false}
	}
	return hold{}
}

// heldBuild is the build the book holds rt off and its hold (holdOf), ""
// when nothing holds it. Both the installed build (root's stamp) and the
// build live/<host> names are asked, because a wake execs the live link and
// root repoints it before it stamps the row; an unreadable link is left to
// the stamp.
func (h *failureHolds) heldBuild(rt registry.Route) (string, hold) {
	if !rollout.Restrictable(rt) {
		return "", hold{}
	}
	builds := []string{rt.Version}
	if live, err := release.LiveVersion(h.optDir, rt.Host); err == nil && live != "" && live != rt.Version {
		builds = append(builds, live)
	}
	for _, b := range builds {
		if hd := h.holdOf(rt, b); hd.refuse != "" {
			return b, hd
		}
	}
	return "", hold{}
}

// plan is the declined box's Planner: a managed wake onto a held build is
// refused with fixed text that names the build at most, never an error; any
// other wake gets the zero plan, which is today's wake from the live link.
// rt is the row doWake has just re-read (Service.Routes).
func (h *failureHolds) plan(_ context.Context, rt registry.Route) activator.WakePlan {
	_, hd := h.heldBuild(rt)
	return activator.WakePlan{Refuse: hd.refuse}
}

// report is the declined box's per-host rollout status (Task 10, plan
// review #5): each managed instance the book holds, with the build that
// holds it (HeldBy) and that restriction's generation as published
// (HeldEpoch, rollout.FailureBook.PublishedEpoch: 0 while the book owes a
// save), which the console's retry doors sign into their grant, attempt
// down when it stays down for the hold and failed otherwise. Nil when
// nothing is held or the routes cannot be read. A declined box runs no
// scheduler, so it reports no serving build and no attempt in progress.
func (h *failureHolds) report() map[string]status.HostRollout {
	routes, err := h.list()
	if err != nil {
		return nil
	}
	var out map[string]status.HostRollout
	for _, rt := range routes {
		build, hd := h.heldBuild(rt)
		if build == "" {
			continue
		}
		acct := release.RouteAccount(rt)
		attempt := status.AttemptFailed
		if hd.down {
			attempt = status.AttemptDown
		}
		if out == nil {
			out = map[string]status.HostRollout{}
		}
		out[rt.Host] = status.HostRollout{Account: acct, App: rt.App, Installed: rt.Version, Good: rt.GoodVersion,
			Attempt: attempt, AttemptBuild: hd.by, HeldBy: hd.by, HeldEpoch: h.book.PublishedEpoch(acct, rt.App, hd.by, rt.Host),
			HeldBook: h.book.ID(),       // the book the epoch counts in (final review round 1, finding 3)
			RetryV:   status.HostRetryV} // grants ride the host-retry log (readiness round 2, Task 3)
	}
	return out
}

// retryHost is restartMemory.retryHost on a declined box (Task 10, ruling
// 194): a `carlos instances retry` whose root-signed grant the restart leg
// has verified, applied to this box's book exactly as the armed box applies
// it (RetryHostAt): while build's restriction of host is still at epoch, in
// this box's book (book, final review round 1, finding 3), it
// clears build's record for host and host's marker when the marker names
// build, and saves. A grant for another build, or from before the
// restriction last changed, clears nothing. There is no scheduler to run a
// canary: the next wake, or the converge leg, starts the build as today.
func (h *failureHolds) retryHost(host, build, book string, epoch uint64) bool {
	rt, err := h.get(host)
	// Restrictable, not Managed (finding 2): a grant that arrives while a
	// move holds the row (draining, moving, landing) is applied, not read
	// once by the restart leg and dropped.
	if err != nil || rt == nil || !rollout.Restrictable(*rt) || build == "" {
		return false
	}
	acct := release.RouteAccount(*rt)
	rec, mark, now, err := h.book.RetryHostAt(acct, rt.App, build, host, book, epoch)
	if err != nil && (rec || mark) {
		log.Printf("agent: retry %s on %s: save the failure book: %v (cleared in memory)", build, host, err)
	} else if err != nil {
		// The book owed a save and it failed again: nothing is cleared until
		// one lands (the crash-epoch hard limit, rule 2).
		log.Printf("agent: retry %s on %s: the failure book cannot be saved (%v), so nothing was cleared", build, host, err)
	} else if !rec && !mark && now != epoch {
		log.Printf("agent: retry %s on %s: the grant is for epoch %d, the restriction's is %d, so nothing was cleared", build, host, epoch, now)
	} else if !rec && !mark && book != h.book.ID() {
		log.Printf("agent: retry %s on %s: the grant is for book %q, not this box's, so nothing was cleared", build, host, book)
	}
	if rec || mark {
		nudgeStatus()
	}
	return rec || mark
}

// applyPointers is the declined box's pointer pass (Task 10, ruling 194;
// Correction 3): the armed box's two pointer rules (rollout.ApplyPointers)
// on this box's book, for every app with a record or a marker a signed
// rollback could clear. read is release.Channel with the agent's store and
// root key. The caller paces it (pointerLoop).
func (h *failureHolds) applyPointers(ctx context.Context, read rollout.PointerRead, routes []registry.Route) {
	if rollout.ApplyPointers(ctx, h.book, read, routes, log.Printf) {
		nudgeStatus()
	}
}

// pass is one turn of the declined box's periodic pass: the save the book
// owes, if any (the crash-epoch hard limit, rule 6), so a box whose disk
// recovers publishes its generations again with no pointer or grant
// arriving, then the pointer pass over the registry's routes.
func (h *failureHolds) pass(ctx context.Context, read rollout.PointerRead) {
	if h.book.SaveOwed() {
		if err := h.book.SaveIfDirty(); err != nil {
			log.Printf("agent: declined pass: save the failure book: %v", err)
		} else {
			nudgeStatus()
		}
	}
	if routes, err := h.list(); err != nil {
		log.Printf("agent: declined pointer pass: list routes: %v", err)
	} else {
		h.applyPointers(ctx, read, routes)
	}
}

// pointerLoop runs pass, at most every rollout.PointerReadEvery, for the
// life of the agent. An app with nothing the pointer pass could clear costs
// no read.
func (h *failureHolds) pointerLoop(ctx context.Context, read rollout.PointerRead) {
	for {
		h.pass(ctx, read)
		select {
		case <-ctx.Done():
			return
		case <-time.After(rollout.PointerReadEvery):
		}
	}
}

// rolloutRetryHost is restartMemory.retryHost for this box: the armed
// manager's, a declined box's that honours its book (failureHolds.retryHost,
// Task 10), or nil, which drops every retry because nothing is held.
func rolloutRetryHost() func(host, build, book string, epoch uint64) bool {
	if m := armedRollout(); m != nil {
		return m.RetryHost
	}
	if h := heldHolds(); h != nil {
		return h.retryHost
	}
	return nil
}

// gate is the declined box's LiveExecGate (Task 9; Astra R1 #6, the R2/R3
// declined residuals): the plan's refusal asked again at the real launch,
// about the build that launch is about to run, which root may have
// repointed the live link to since the plan. The backend asks it at a flat
// or nested instance's live-link start and a worker's spawn from the live
// link, the activator before a worker's converge or a recycle's move onto
// the live link, and a claims-backed wake once its snapshot has resolved the
// link (claimServingLaunchSource). An empty build (a link outside the
// version store, or one that could not be read) is left to the stamp, as
// heldBuild leaves it. It reads only memory and rt, and writes nothing: the
// activator asks it under its own lock.
func (h *failureHolds) gate(rt registry.Route, build string) error {
	if build == "" {
		build = rt.Version
	}
	if r := h.refusal(rt, build); r != "" {
		return errors.New(r)
	}
	return nil
}

// heldRoute reports whether rt is held, logging it once per leg and build.
func (h *failureHolds) heldRoute(leg string, rt registry.Route) bool {
	build, _ := h.heldBuild(rt)
	if build == "" {
		return false
	}
	if _, said := h.said.LoadOrStore(leg+"\x00"+rt.Host+"\x00"+build, true); !said {
		log.Printf("agent: %s: leaving %s running: a failure record or hold-ran marker from an armed run holds build %s for it, and %s=overlap is declined on this box",
			leg, rt.Host, build, rollout.SwapEnv)
	}
	return true
}

// heldHost is heldRoute for a leg handed only a host. A row that cannot be
// read counts as held: the leg leaves the instance running, which is what
// makeRestartRoute does with such a row anyway.
func (h *failureHolds) heldHost(leg, host string) bool {
	rt, err := h.get(host)
	if err != nil {
		return true
	}
	return rt != nil && h.heldRoute(leg, *rt)
}

// skipGet wraps the converge leg's row read: a held row reads as no row, so
// convergeAdopted passes the host by.
func (h *failureHolds) skipGet(get func(string) (*registry.Route, error)) func(string) (*registry.Route, error) {
	return func(host string) (*registry.Route, error) {
		rt, err := get(host)
		if err == nil && rt != nil && h.heldRoute("converge", *rt) {
			return nil, nil
		}
		return rt, err
	}
}

// skipHost wraps a bounce handed only a host (the adopt bounce): a held host
// is not stopped, which its caller reads as nothing running to cycle.
func (h *failureHolds) skipHost(leg string, bounce func(string) bool) func(string) bool {
	return func(host string) bool {
		if h.heldHost(leg, host) {
			return false
		}
		return bounce(host)
	}
}

// skipRoutes wraps the restart leg's bounce over the routes it was handed,
// so it reads no row of its own.
func (h *failureHolds) skipRoutes(routes []registry.Route, bounce func(string) bool) func(string) bool {
	if bounce == nil {
		return nil
	}
	byHost := make(map[string]registry.Route, len(routes))
	for _, rt := range routes {
		byHost[rt.Host] = rt
	}
	return func(host string) bool {
		if rt, ok := byHost[host]; ok && h.heldRoute("restart", rt) {
			return false
		}
		return bounce(host)
	}
}

// configCallbacks is config sync's callbacks on a declined box that honours
// records: today's restart and ack for every host but a held one, which is
// not stopped into a refused wake. Its ack must not claim a restart: an
// awake held instance still runs its old config, so the restart answers
// rolloutRestartRefused (the ack's RestartError); a sleeping one reads the
// file when it next wakes, so its ack says delivered on wake.
func (h *failureHolds) configCallbacks(ack func(appconfig.ConfigAck), legs configLegs) appconfig.Callbacks {
	restart := func(host string) error {
		if h.heldHost("config sync", host) {
			if legs.awake != nil && legs.awake(host) {
				return errors.New(rolloutRestartRefused)
			}
			return appconfig.ErrDeliveredOnWake
		}
		if legs.restart == nil {
			return nil
		}
		return legs.restart(host)
	}
	return appconfig.Callbacks{Restart: restart, Ack: ack}
}

// declinedHoldsStay is fastConvergeLoop's one line, at arming, when a box
// that honours its failure book has no deploy store or no release key (Task
// 10 rule 2): with neither a pointer can be read nor a retry's grant
// verified, so no pointer pass runs and no retry is read, and what the book
// holds stays held until the agent restarts with both.
func declinedHoldsStay(why string) {
	if heldHolds() == nil {
		return
	}
	log.Printf("agent: %s=overlap declined: %s, so no signed retry or channel pointer can clear the failure records and hold-ran markers honoured here",
		rollout.SwapEnv, why)
}