// The box half of a NARROWED restart (discussion #16): the same two-process
// dance restart.go describes, reading the request log at
// release.RestartScopeKey instead of the app-wide generation, and cycling only
// the routes a request actually names.
//
// Everything load-bearing in restart.go still holds here, and this file is
// deliberately built to inherit rather than restate it:
//
//   - It rides the SERIAL. No poll of its own; the agent's existing 2s serial
//     read is what makes either object get looked at, and an idle box reads
//     neither. The cost of this feature at rest is zero, and on a serial MOVE
//     it is one extra GET per app on this box.
//   - SEED-DON'T-REPLAY, per app, on this object's own mark. A box that has
//     never looked records where the log stands and cycles nothing.
//   - HIBERNATING routes are never woken. The root half skips them outright;
//     the agent's bounce is a no-op on a sleeping instance and cycles a warm
//     one. That is restart.go's rule verbatim, and narrowing the route set
//     cannot weaken it.
//   - A selector this binary does not understand addresses NOTHING. The
//     request is stepped over and the mark advances past it, so a future
//     `--region` cannot make today's box cycle an app.
//
// The one thing that is genuinely different: the CONTENT is read, not just the
// etag. See internal/release/restartscope.go for why that grants nothing —
// the account, the app and the candidate routes all come from this box's own
// registry, and a selector can only ever pick a subset of them.
package host

import (
	"context"
	"crypto/ed25519"
	"fmt"
	"log"
	"sort"
	"time"

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

// restartMemory is the agent's in-process memory for BOTH restart signals: the
// app-wide generation's etag, and the id of the last narrowed request acted
// on. One struct rather than two loose maps because they are seeded together,
// compared together on the same serial move, and are only ever meaningful as a
// pair — a leg holding one and not the other would replay or drop half a
// request set.
//
// It lives for the life of the agent process (fastConvergeLoop owns it,
// restartSeed fills it), which is the whole point: there is nowhere else to
// remember what "changed" means between two serial moves that may be days
// apart.
type restartMemory struct {
	gen   map[[2]string]string // app -> restart-gen.json etag, or genAbsent
	scope map[[2]string]string // app -> last applied request id, or release.RestartScopeNone
	// retries is app -> the ids of the host-retry log's entries this
	// process has handled (hostRetryPass). It is not a seed mark: an empty
	// memory, at agent start, handles every entry in the log.
	retries map[[2]string]map[string]bool

	// retryHost clears one host's failure hold, for a RestartScopeRetry
	// request whose root-signed grant verified: the grant's build, and the
	// restriction's book and epoch it was issued for (rolloutRetryHost;
	// final review round 1, finding 3). It is the
	// rollout manager's (rollout.Manager.RetryHost) when armed, and a
	// declined box's that honours its failure book (failureHolds.retryHost,
	// Task 10). nil otherwise: there is no hold to clear, and the request is
	// dropped.
	retryHost func(host, build, book string, epoch uint64) bool
	// pub is the deployment root's public key (CARLOS_RELEASE_PUBKEY), which
	// every retry's grant is verified against (spec §11). Nil drops every
	// retry: a box that cannot tell a grant from a forgery clears nothing.
	pub ed25519.PublicKey
}

func newRestartMemory() *restartMemory {
	return &restartMemory{gen: map[[2]string]string{}, scope: map[[2]string]string{}, retries: map[[2]string]map[string]bool{}}
}

// scopedRestart is one app's outstanding narrowed work: the requests this box
// has not acted on yet, newest last.
type scopedRestart struct {
	app      [2]string
	requests []release.RestartRequest
	// history marks retries handed over because the box first saw the app or
	// lost its place in the log, not because somebody asked for them since
	// the last look (restartScopePass). A box with nothing that could clear
	// a hold reads them as history, silently, as it always has
	// (withoutHistory).
	history bool
}

// restartScopePass is the agent's comparison for narrowed requests — the twin
// of restartGenPass, with the same first-sight, absent-forever and
// read-error rules and for the same reasons. What it reports is not "this app
// moved" but WHICH REQUESTS are outstanding, because that is the only thing
// that says which routes to touch.
//
// Two requests inside one poll window are two entries in the log, so the pair
// is not collapsed into whichever landed last — the failure a bare generation
// object would have here, and the reason the object is a log at all.
//
// A mark that is no longer IN the log (this box was away long enough for its
// position to roll off) re-baselines and asks for nothing, with a line saying
// so. Restart requests are point-in-time events: a dropped one is recovered by
// typing the verb again, while a replayed one cycles production for something
// somebody asked for last week.
//
// A retry an earlier console left in this log is the one exception to both
// (Astra plan 9): it is not a restart, and a grant is not seed-skipped. On
// a first sight and on a lost position the log's retry entries are
// reported anyway, for the caller to verify and apply exactly as on a move
// (retryRequests, applyRetries): a grant spent or made stale since is a
// no-op under the failure book's generation rule, and an unsigned one is
// dropped with a line. withoutRetries keeps them out of every restart.
func restartScopePass(ctx context.Context, routes []registry.Route, store release.Store, cache map[[2]string]string) (out []scopedRestart, readErrs []error) {
	apps := restartApps(routes)
	live := make(map[[2]string]bool, len(apps))
	for _, k := range apps {
		live[k] = true
		g, _, err := release.LoadRestartScopes(ctx, store, k[0], k[1])
		if err != nil {
			// Unreadable OR unparseable, one answer: leave the mark exactly as
			// it was and evaluate the rest. Clearing it would turn a recovered
			// blip into a bogus first sight; advancing it past a log this
			// binary cannot read would drop the requests in it for good.
			readErrs = append(readErrs, fmt.Errorf("%s/%s: %w", k[0], k[1], err))
			continue
		}
		newest := g.Newest()
		prev, seen := cache[k]
		if !seen {
			cache[k] = newest // first sight: record, cycle nothing
			out = appendHistory(out, k, g.Requests)
			continue
		}
		if prev == newest {
			continue
		}
		pending, found := g.PendingSince(prev)
		cache[k] = newest
		if !found {
			log.Printf("agent: narrowed restarts for %s/%s: this box's position is no longer in the log — re-baselined, nothing cycled (run the verb again if something is still waiting)", k[0], k[1])
			out = appendHistory(out, k, g.Requests)
			continue
		}
		if len(pending) > 0 {
			out = append(out, scopedRestart{app: k, requests: pending})
		}
	}
	for k := range cache {
		if !live[k] {
			delete(cache, k)
		}
	}
	return out, readErrs
}

// scopedHosts resolves outstanding requests against this box's routes: which
// hosts of that app the requests actually name, deduplicated.
//
// Deduplicated across the whole set, not per request, because two pending
// requests that both name a host mean one cycle of it, not two — and a verb
// whose entire job is stopping processes must not manufacture a second stop
// out of an operator typing the same thing twice.
//
// An UNKNOWN selector kind is stepped over with a log line and addresses
// nothing. That is the forward-compatibility rule this feature is built on:
// a box that cannot understand a narrowing must never fall back to the wider
// thing.
//
// keep is the caller's own route filter — the root pass and the agent reach
// different processes and each passes its own — so this function decides
// membership and nothing else.
func scopedHosts(routes []registry.Route, app [2]string, requests []release.RestartRequest, keep func(registry.Route) bool) []string {
	seen := map[string]bool{}
	var out []string
	for _, req := range requests {
		// A retry is not a restart: it is handled by retryRequests, and must
		// neither cycle anything nor be logged as a selector this build does
		// not understand.
		if req.Scope.Kind == release.RestartScopeRetry {
			continue
		}
		// Asked of the KIND alone, once, before any route is considered: an
		// app with no routes on this box must not look like an app whose
		// selector we failed to understand.
		if _, understood := req.Scope.Selects(registry.Route{}); !understood {
			log.Printf("restart: narrowed request for %s/%s names selector %q, which this build does not understand — nothing cycled for it (upgrade carlos on this box)",
				app[0], app[1], req.Scope.Kind)
			continue
		}
		for _, rt := range routes {
			if !release.Adoptable(rt) || rt.App != app[1] || release.RouteAccount(rt) != app[0] {
				continue
			}
			if sel, _ := req.Scope.Selects(rt); !sel || seen[rt.Host] || !keep(rt) {
				continue
			}
			seen[rt.Host] = true
			out = append(out, rt.Host)
		}
	}
	sort.Strings(out)
	return out
}

// retryRequest is one outstanding retry entry naming a host this box
// serves: the entry's account and app (the log's own), its host, and the
// grant it carries, unverified.
type retryRequest struct {
	acct, app, host string
	signed          string
}

// retryRequests is the retry entries of work, from the host-retry log
// (hostRetryPass) or a legacy retry in the narrowed log (restartScopePass):
// the entries, oldest first, whose host is one of this box's routes in the
// requesting app. A host on another box names nothing here. Entries are
// not merged by host: each carries its own grant, and the box verifies
// each (a forged entry must not hide a real one behind it). A grant spent
// by an earlier entry moves its restriction's epoch, so a second one for
// the same failure clears nothing.
func retryRequests(routes []registry.Route, work []scopedRestart) []retryRequest {
	var out []retryRequest
	for _, w := range work {
		for _, req := range w.requests {
			if req.Scope.Kind != release.RestartScopeRetry {
				continue
			}
			for _, rt := range routes {
				if rt.Host == req.Scope.Value && rt.App == w.app[1] && release.RouteAccount(rt) == w.app[0] {
					out = append(out, retryRequest{acct: w.app[0], app: w.app[1], host: rt.Host, signed: req.Scope.Signed})
					break
				}
			}
		}
	}
	return out
}

// appendRetries adds app's retry entries among requests to out, as one
// scopedRestart, when there are any.
func appendRetries(out []scopedRestart, app [2]string, requests []release.RestartRequest) []scopedRestart {
	var rs []release.RestartRequest
	for _, req := range requests {
		if req.Scope.Kind == release.RestartScopeRetry {
			rs = append(rs, req)
		}
	}
	if len(rs) == 0 {
		return out
	}
	return append(out, scopedRestart{app: app, requests: rs})
}

// appendHistory is appendRetries for retries that are history: the box first
// saw the app, or lost its place in the log.
func appendHistory(out []scopedRestart, app [2]string, requests []release.RestartRequest) []scopedRestart {
	n := len(out)
	out = appendRetries(out, app, requests)
	if len(out) > n {
		out[n].history = true
	}
	return out
}

// withoutHistory drops the retries that are history (scopedRestart.history)
// when this box has nothing that could clear a hold (retryHost nil), so the
// flag-off box does not log, for every legacy retry in a log it has just
// met, that its rollout manager is not armed. A retry somebody asked for
// since the last look is not history and still gets its line.
func withoutHistory(work []scopedRestart, mem *restartMemory) []scopedRestart {
	if mem.retryHost != nil {
		return work
	}
	var out []scopedRestart
	for _, w := range work {
		if !w.history {
			out = append(out, w)
		}
	}
	return out
}

// hostRetryPass reads each app's host-retry log (release.HostRetryKey) and
// returns the entries this process has not handled yet, for retryRequests
// and applyRetries. It rides the serial like restartScopePass, one GET per
// app on a move, and none at all on a box with nothing that could clear a
// hold (retryHost nil), which no grant could change while the agent runs.
//
// It is not seed-skipped (Astra plan 9): a memory with no entry for an app,
// at agent start or for an app this box has just gained, hands over every
// entry in the log, so a grant written before an agent restart is applied
// after it. That is safe because a grant is checked, not trusted: one past
// its window, unsigned, or forged is dropped with a line, and one spent or
// made stale since is a no-op under the failure book's generation rule
// (RetryHostAt). Each entry is handled once per process: cache is the ids
// handled, rebuilt from the log on each read so it stays as small as the
// log, and an unchanged log hands over nothing, so nothing is verified
// twice. A read or parse error leaves the app's ids as they were.
func hostRetryPass(ctx context.Context, routes []registry.Route, store release.Store, mem *restartMemory) (out []scopedRestart, readErrs []error) {
	if mem.retryHost == nil {
		return nil, nil
	}
	apps := restartApps(routes)
	live := make(map[[2]string]bool, len(apps))
	for _, k := range apps {
		live[k] = true
		g, _, err := release.LoadHostRetries(ctx, store, k[0], k[1])
		if err != nil {
			readErrs = append(readErrs, fmt.Errorf("%s/%s host retries: %w", k[0], k[1], err))
			continue
		}
		done := mem.retries[k]
		next := make(map[string]bool, len(g.Requests))
		var pending []release.RestartRequest
		for _, req := range g.Requests {
			if !done[req.ID] && !next[req.ID] {
				pending = append(pending, req)
			}
			next[req.ID] = true
		}
		mem.retries[k] = next
		out = appendRetries(out, k, pending)
	}
	for k := range mem.retries {
		if !live[k] {
			delete(mem.retries, k)
		}
	}
	return out, readErrs
}

// applyRetries verifies and applies each retry in work naming a host this
// box serves (retryRequests), the agent's one path for a grant from either
// log. A retry clears a hold only on the console's root-signed grant for
// its entry (spec §11, Astra impl-6 finding 5): every box can write both
// logs, so an unsigned or foreign entry is dropped here, with a line,
// before anything is cleared. The grant's epoch is checked against the
// book where the clear is (rollout.Manager.RetryHost, or
// failureHolds.retryHost on a box that declined the flag but honours its
// book).
func applyRetries(routes []registry.Route, work []scopedRestart, mem *restartMemory) {
	for _, rq := range retryRequests(routes, work) {
		if relayRootRetry(mem.pub, rq) {
			continue
		}
		if mem.retryHost == nil {
			log.Printf("agent: retry requested for %s, dropped: rollout manager not armed on this box, so nothing is held to clear", rq.host)
			continue
		}
		g, why := verifyRetry(mem.pub, rq, time.Now())
		if why != "" {
			log.Printf("agent: retry requested for %s, dropped: %s", rq.host, why)
			continue
		}
		if mem.retryHost(rq.host, g.Build, g.Book, g.Epoch) {
			next := "its next attempt runs as a canary"
			if armedRollout() == nil {
				next = "it starts the build as today" // a declined box runs no canary
			}
			log.Printf("agent: retry requested for %s: its failure hold on %s is cleared and %s", rq.host, g.Build, next)
		} else {
			log.Printf("agent: retry for %s: nothing cleared for %s at epoch %d of book %q (the restriction has changed since the grant, it was another book's grant, or this box has not saved its failure book yet)", rq.host, g.Build, g.Epoch, g.Book)
		}
	}
}

// verifyRetry is the box's check of one retry entry (spec §11, Paul's
// ruling on plan 6 Q4): the grant must verify against the root key and
// name this entry's account, app and host, within its window
// (release.VerifyHostRetry). The reason it gives is an operator's log line.
func verifyRetry(pub ed25519.PublicKey, rq retryRequest, now time.Time) (release.HostRetryGrant, string) {
	switch {
	case len(pub) == 0:
		return release.HostRetryGrant{}, "this box has no root public key to verify the grant with"
	case rq.signed == "":
		return release.HostRetryGrant{}, "unsigned (only a console's signed grant clears a hold)"
	}
	g, err := release.VerifyHostRetry(pub, rq.signed, rq.acct, rq.app, rq.host, now)
	if err != nil {
		return release.HostRetryGrant{}, fmt.Sprintf("the grant did not verify for this entry: %v", err)
	}
	return g, ""
}

// withoutRetries drops retry requests, and apps left with none, so a serial
// move that carried only retries asks root for no restart pass.
func withoutRetries(work []scopedRestart) []scopedRestart {
	var out []scopedRestart
	for _, w := range work {
		var keep []release.RestartRequest
		for _, req := range w.requests {
			if req.Scope.Kind != release.RestartScopeRetry {
				keep = append(keep, req)
			}
		}
		if len(keep) > 0 {
			out = append(out, scopedRestart{app: w.app, requests: keep})
		}
	}
	return out
}

// execBounceable is the complement of what rootRestartRoute's systemctl arms
// can reach — no named unit, no unit backing — so the agent's half and the
// root half never cycle the same process twice. Split out of bounceExecRoutes
// so the narrowed half cannot drift from the app-wide one on which processes
// belong to whom.
func execBounceable(rt registry.Route) bool {
	return rt.Unit == "" && rt.Backing != registry.BackingUnit
}

// bounceScopedRoutes cycles the agent's OWN children for the routes the
// outstanding narrowed requests name. bounceExecRoutes' argument, one selector
// narrower — including on hibernation: Bounce cannot start anything, so a
// sleeping instance is untouched and a warm one gets the cycle that was asked
// for.
func bounceScopedRoutes(routes []registry.Route, work []scopedRestart, bounce func(host string) bool) int {
	if bounce == nil || len(work) == 0 {
		return 0
	}
	n := 0
	for _, w := range work {
		for _, host := range scopedHosts(routes, w.app, w.requests, execBounceable) {
			if bounce(host) {
				n++
				log.Printf("agent: narrowed restart requested: bounced exec instance %s", host)
			}
		}
	}
	return n
}

// restartScopeLeg is the ROOT pass's narrowed half, called from inside
// restartPass under the SAME lock and against the same marks file: for every
// app whose request log has moved past this box's mark, cycle the routes those
// requests name.
//
// already is the hosts the app-wide half has cycled on this pass. A route that
// has just been restarted by a wider request must not be restarted a second
// time by a narrower one that arrived in the same window — one pass, one cycle
// per host.
//
// The rules are restartPass's, applied to this object:
//
//   - An app this box has not recorded is recorded and NOT cycled.
//   - An unknown baseline records and dispatches nothing.
//   - A mark that rolled off the log re-baselines and dispatches nothing.
//   - HIBERNATING routes are skipped and not counted as failures: restarting a
//     process that is not running is meaningless, waking it to restart it
//     defeats hibernation for no gain, and it execs fresh on its next wake.
//   - A dispatch FAILURE still advances the mark. A request is a point-in-time
//     event, not a desired state; retrying it every hour forever is how one
//     wrong unit name becomes a permanent restart loop.
//   - The mark advances only AFTER the dispatch, so a pass that dies mid-cycle
//     retries rather than swallowing the request.
func restartScopeLeg(ctx context.Context, routes []registry.Route, store release.Store, apps [][2]string,
	marks *restartMarks, known bool, restart func(host string) error, already map[string]bool) (changed, readFailed bool) {
	for _, k := range apps {
		g, _, err := release.LoadRestartScopes(ctx, store, k[0], k[1])
		if err != nil {
			log.Printf("adopt: narrowed restarts %s/%s: %v (skipped this pass)", k[0], k[1], err)
			readFailed = true
			continue
		}
		newest := g.Newest()
		key := restartMarkKey(k[0], k[1])
		prev, seen := marks.Scopes[key]
		if !known || !seen {
			if !seen || prev != newest {
				marks.Scopes[key] = newest
				changed = true
			}
			continue
		}
		if prev == newest {
			continue
		}
		pending, found := g.PendingSince(prev)
		if !found {
			log.Printf("adopt: narrowed restarts for %s/%s: this box's position is no longer in the log — re-baselined, nothing cycled", k[0], k[1])
			marks.Scopes[key] = newest
			changed = true
			continue
		}
		dispatched := 0
		for _, host := range scopedHosts(routes, k, pending, func(registry.Route) bool { return true }) {
			if already[host] {
				continue
			}
			rt, ok := routeByHost(routes, host)
			if !ok {
				continue
			}
			if rt.Hibernate {
				log.Printf("adopt: narrowed restart for %s/%s: %s is hibernating — left asleep (it execs fresh on its next wake)", k[0], k[1], host)
				continue
			}
			if err := restart(host); err != nil {
				log.Printf("adopt: narrowed restart for %s/%s: %s did not cycle: %v", k[0], k[1], host, err)
				continue
			}
			already[host] = true
			dispatched++
		}
		// One line per APP, saying DISPATCHED rather than cycled, for the
		// reason the app-wide half says it: an exec-backed route reaching
		// rootRestartRoute lands on a try-restart of a carlos-app@ unit that
		// does not exist — a deliberate no-op this pass cannot tell apart from
		// a real one.
		log.Printf("adopt: narrowed restart for %s/%s: %d request(s), dispatched to %d route(s)", k[0], k[1], len(pending), dispatched)
		marks.Scopes[key] = newest
		changed = true
	}
	return changed, readFailed
}

// routeByHost finds a route by host among the ones this box serves. Linear
// because the route set is a box's own table, and a map built per call would
// be the same walk with an allocation on top.
func routeByHost(routes []registry.Route, host string) (registry.Route, bool) {
	for _, rt := range routes {
		if rt.Host == host {
			return rt, true
		}
	}
	return registry.Route{}, false
}