// The box half of a sidecar wake request: when a sidecar's wake object
// (release.SidecarWakeKey) MOVES, wake exactly that sidecar. Pull-based like
// the restart signal it is modelled on (restart.go) — no push channel, no
// box access, and the box need not even be reachable from anywhere for this
// to work.
//
// It exists because a sidecar has no socket and nothing dials it: an
// instance wakes on its next inbound request, but a hibernating sidecar has
// no "next request" to wake on, so waking one from outside the box needs an
// explicit signal a pull-based box can notice on its own.
//
// Split into the same two shapes restart.go uses for exactly the same
// reason: sidecarWakeChanges is the cheap, synchronous etag comparison
// (restartGenPass's counterpart), while dispatchSidecarWakes hands the
// WakeWorker fan-out to restart.go's shared dispatchBoundedWakes — which
// can block up to wakeWaitBound (30s, activator.go) per sidecar waiting
// out the settle window, and runs the WHOLE bounded fan-out inside one
// goroutine it launches itself, so the caller returns immediately
// regardless of how many sidecars are queued (review round 2: an app
// shaped like Eleven's, a sidecar per enrolled member, is routinely more
// than workerBootConcurrency). sidecarWakeLeg is the two glued together and
// wired into fastConvergeLoop beside the restart leg (agent.go's
// fastDeps.wakeLeg), so it rides the same 2s serial-driven convergence
// `carlos restart` already gets.
package host

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"os"

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

// sidecarsServed maps every sidecar route this box serves to the (account,
// app, name) triple its wake object lives at — name comes from Slug, the
// field a sidecar route carries its short name on (registry.Route's own
// doc comment; provision's reconciler stamps it from the sidecar record).
//
// Gated on rt.IsSidecar() explicitly, not just a non-empty Slug: Slug is
// NOT sidecar-exclusive (it is also an ordinary app's root-domain short
// name), so an instance route that happens to carry the same Slug as a real
// sidecar must never be mistaken for that sidecar.
func sidecarsServed(routes []registry.Route) map[[3]string]registry.Route {
	out := make(map[[3]string]registry.Route)
	for _, rt := range routes {
		if !rt.IsSidecar() || rt.Slug == "" {
			continue
		}
		out[[3]string{release.RouteAccount(rt), rt.App, rt.Slug}] = rt
	}
	return out
}

// sidecarWakeMemory is the wake leg's memory across ticks: the etag last
// seen for each served sidecar, plus the one bit that says whether the
// arming BASELINE has been established.
//
// The two live together deliberately. "First sight" means two completely
// different things either side of that baseline — history a booting box must
// not replay, versus a route that has just appeared under a running agent —
// and a bare map cannot tell them apart, which is exactly how the
// create-then-wake flow came to swallow its own first wake.
type sidecarWakeMemory struct {
	claimsEnabled bool
	identities    map[string]release.SidecarWake
	etags         map[[3]string]string
	// seeded is set by the first completed pass (the arming one in
	// production, sidecarWakeSeed; the first leg pass when arming's route
	// list could not be read). Before it, first sight is the baseline; after
	// it, first sight is a NEW route.
	seeded bool
}

func newSidecarWakeMemory(claimsEnabled bool) *sidecarWakeMemory {
	return &sidecarWakeMemory{claimsEnabled: claimsEnabled, identities: map[string]release.SidecarWake{}, etags: map[[3]string]string{}}
}

// sidecarWakeChanges is sidecar-wake's counterpart to restartGenPass: read
// the wake-object etag of every sidecar this box serves and report which
// ones MOVED, as their routes. It performs no wake itself — see
// dispatchSidecarWakes for the half that can block.
//
//   - First sight BEFORE the baseline exists records silently and is never
//     reported as moved. A box (or an agent process) that has just started
//     must not wake every sidecar it serves because it has never seen these
//     etags before — that is the seed-don't-replay rule the serial watcher
//     already applies (agent.go's fastTick). Without it, an agent restart
//     would replay every wake this app's sidecars have ever received — a
//     thundering herd on boot, and for a hibernating fleet a real cost, not
//     a cosmetic one. sidecarWakeSeed runs one such silent pass at arming so
//     the FIRST request after a boot is acted on rather than swallowed.
//   - First sight AFTER it, with a wake object already present, is reported
//     as MOVED. Such a route did not exist when this agent last looked, so
//     it has just been minted — and a wake object standing at a route this
//     very pass minted is a wake somebody already asked for and the CLI
//     already reported as accepted. That is the create-then-wake flow, and
//     it is the feature's PRIMARY one: an operator wake is a hibernating
//     sidecar's only first-run mechanism (the timer needs a next-due that
//     only a running child can set), and `carlos sidecars create` followed
//     by `carlos sidecars wake` routinely lands both inside a single tick —
//     the tick that mints the route would otherwise seed the wake's etag and
//     silently drop the wake. A spurious extra wake is idempotent and cheap;
//     a lost one is a bot that never runs. (A route that vanished and came
//     back is a first sight too, and is woken once for the same reason: it
//     has just been re-minted, and its wake object outlived it.)
//   - An unchanged etag does nothing.
//   - Absent that stays absent does nothing, forever: a sidecar nobody has
//     ever woken this way has no object and never will until someone asks.
//     A brand-new route with no wake object is therefore silent too — only a
//     PRESENT object makes a first sight a wake.
//   - Any movement counts: the etag changed, or absent -> present (the
//     sidecar's first ever wake request).
//   - A wake object naming a sidecar this box does not serve is never even
//     read: this function only ever builds keys from sidecarsServed, which
//     comes from THIS box's own route table. Such an object, however it
//     moves in the bucket, is simply never asked about.
//   - A read error for one sidecar leaves its cache entry untouched and the
//     rest of the pass keeps going; the error comes back for the caller to
//     aggregate into its own log line, same discipline as restartGenPass.
//     After the baseline that means its next sighting is a first sight, so
//     it may be woken once — the cheap direction of an unknown.
//
// The memory is mutated in place and is expected to live for the process
// lifetime, exactly like restartGenPass's cache.
func sidecarWakeChanges(ctx context.Context, routes []registry.Route, store release.Store, mem *sidecarWakeMemory) (moved []registry.Route, readErrs []error) {
	served := sidecarsServed(routes)
	live := make(map[[3]string]bool, len(served))
	baseline := !mem.seeded
	for k, rt := range served {
		live[k] = true
		raw, etag, err := store.Get(ctx, release.SidecarWakeKey(k[0], k[1], k[2]))
		switch {
		case errors.Is(err, os.ErrNotExist):
			etag = genAbsent
		case err != nil:
			readErrs = append(readErrs, fmt.Errorf("%s/%s/%s: %w", k[0], k[1], k[2], err))
			continue // cache entry untouched: unknown this tick, not changed
		}
		if mem.claimsEnabled {
			var signal release.SidecarWake
			if etag == genAbsent || json.Unmarshal(raw, &signal) != nil {
				continue
			}
			if err := verifySidecarWake(ctx, store, rt, signal); err != nil {
				readErrs = append(readErrs, err)
				continue
			}
			if mem.identities == nil {
				mem.identities = map[string]release.SidecarWake{}
			}
			mem.identities[rt.Host] = signal
			etag = signal.AppID + "/" + signal.Incarnation + "/" + signal.RequestID
		}
		prev, seen := mem.etags[k]
		mem.etags[k] = etag
		switch {
		case seen:
			if prev != etag {
				moved = append(moved, rt)
			}
		case baseline:
			// The agent-restart protection: history, recorded silently.
		case etag != genAbsent:
			// A route minted since the last pass, with a wake already
			// standing at it — the create-then-wake flow.
			moved = append(moved, rt)
		}
	}
	for k := range mem.etags {
		if !live[k] {
			delete(mem.etags, k)
		}
	}
	// One completed pass IS the baseline, whichever caller ran it: arming's
	// silent seed in production, or the first leg pass on a box whose arming
	// route list could not be read.
	mem.seeded = true
	return moved, readErrs
}

// dispatchSidecarWakes wakes every route in moved through the shared
// dispatchBoundedWakes helper (restart.go) — the SAME idiom Task 8's
// runWorkerBootPass and cycleWorkers' own wake half use, deliberately not a
// third shape. WakeWorker can block up to wakeWaitBound (30s) per sidecar
// waiting out the settle window, and this function's only caller
// (sidecarWakeLeg) runs synchronously on the box's single per-tick
// convergence goroutine (fastTick) — dispatchBoundedWakes' own doc comment
// covers why the WHOLE bounded fan-out, not just each individual wake,
// must be off that goroutine (review round 2: an app shaped like Eleven's,
// a sidecar per enrolled member, routinely queues more than
// workerBootConcurrency sidecars behind one wake burst). Safe because
// WakeWorker is idempotent and a wake request's contract is "wake the
// process", not "return once it's up" — a sidecar that comes up a moment
// after this tick has still been woken. A wake that errors is logged and
// does not stop, delay, or fail the others.
//
// wake may be nil (no activator armed on this box): a working no-op.
func dispatchSidecarWakes(moved []registry.Route, wake func(rt registry.Route) error) {
	dispatchBoundedWakes(moved, wake, func(rt registry.Route, err error) {
		if err != nil {
			log.Printf("agent: sidecar wake requested: wake %s: %v", rt.Host, err)
			return
		}
		log.Printf("agent: sidecar wake requested: woke %s", rt.Host)
	})
}

// sidecarWakeLeg is sidecarWakeChanges and dispatchSidecarWakes glued
// together — the leg fastConvergeLoop wires into fastDeps.wakeLeg beside
// the restart leg, so a wake request rides the same 2s serial-driven
// convergence `carlos restart` already gets. Reports whether anything
// moved (for logging/tests); the actual wake calls happen off this
// function's return path (dispatchSidecarWakes), so it never blocks on a
// slow or unresponsive sidecar.
func sidecarWakeLeg(ctx context.Context, routes []registry.Route, store release.Store, mem *sidecarWakeMemory, wake func(rt registry.Route) error) (moved []registry.Route, readErrs []error) {
	moved, readErrs = sidecarWakeChanges(ctx, routes, store, mem)
	if len(readErrs) > 0 {
		log.Printf("agent: sidecar wake check: %d sidecars unreadable: %v", len(readErrs), readErrs[0])
	}
	if len(moved) == 0 {
		return moved, readErrs
	}
	if mem.claimsEnabled {
		captured := make(map[string]release.SidecarWake, len(moved))
		for _, rt := range moved {
			captured[rt.Host] = mem.identities[rt.Host]
		}
		original := wake
		wake = func(rt registry.Route) error {
			if err := verifySidecarWake(ctx, store, rt, captured[rt.Host]); err != nil {
				return err
			}
			rt.WakeIncarnation = captured[rt.Host].Incarnation
			if original == nil {
				return nil
			}
			return original(rt)
		}
	}
	dispatchSidecarWakes(moved, wake)
	log.Printf("agent: sidecar wake requested for %d sidecar(s) on this box", len(moved))
	return moved, readErrs
}

// sidecarWakeSeed is sidecarWakeChanges' counterpart to restartSeed: at
// arming, read each served sidecar's wake-object etag once, silently, so
// the memory sidecarWakeLeg compares against exists before the first serial
// move. Without it the first wake request after a boot is the sweep that
// establishes the baseline, and is therefore the one request that does
// nothing.
//
// This pass IS the agent-restart protection: only a sidecar this box already
// served when the agent armed has its history recorded silently. Everything
// that appears afterwards is a route minted under a running agent, and a
// wake object already standing at one of those is acted on (see
// sidecarWakeChanges).
//
// A LIST failure leaves the memory unseeded on purpose: the first leg pass
// then establishes the baseline itself rather than reading an empty cache as
// "every sidecar on this box is brand new" and waking the lot. A per-sidecar
// read error is narrower — that one sidecar is left out of the cache and may
// be woken once on its next sighting.
func sidecarWakeSeed(ctx context.Context, reg *registry.Registry, store release.Store, mem *sidecarWakeMemory) {
	routes, err := reg.List()
	if err != nil {
		log.Printf("agent: sidecar wake baseline: list routes: %v (the first request may need a second try)", err)
		return
	}
	_, readErrs := sidecarWakeChanges(ctx, routes, store, mem)
	if len(readErrs) > 0 {
		log.Printf("agent: sidecar wake baseline: %d sidecars unreadable: %v", len(readErrs), readErrs[0])
	}
}