// Console-side read path for host/process monitoring: assembling what the
// operator dashboard (Task 11) and owner app-metrics page (Task 12) render
// out of what the box agent publishes (internal/metrics). Nothing here
// writes anything — it is a bucket sweep plus an in-process cache, and the
// wire shapes it reads (metrics.RawChunk, metrics.AlertState) are read
// defensively: an unparsable object is skipped, never fatal to the sweep,
// mirroring alertwatch.go's "never go silent" posture on the same bucket
// prefixes.
package console

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"os"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/metrics"
)

// UnitRow is one per-instance reading, ready to render, plus its own 48h
// sparklines (the owner app-metrics view renders a unit on its own, with no
// host row to borrow a shape from). Keyed by sample name, which is the
// reading's IDENTITY, not always a systemd unit: an exec-backed instance has
// no unit at all — it is a child of `carlos agent`, accounted in a delegated
// cgroup — and reports under carlos-exec@<host> (no .service; exec-child
// accounting spec). Boxes running the superseded pid sampler (issue #118,
// retired by that spec) published exec children under the
// carlos-app@<host>.service name they WOULD have had, and those keys survive
// in raw chunks for the 48h retention window. Either way the route host
// inside the name is the thing to grep the agent's journal for.
type UnitRow struct {
	Unit                  string // full unit name
	CPUFrac               *float64
	MemBytes              *int64
	ReadBytes, WriteBytes *int64
	SparkCPU, SparkMem    string // SVG polyline points, "" when no data
}

// BoxMetricsView is one box's assembled dashboard row: latest host readings
// plus a 48h sparkline per metric plus its units, sorted.
type BoxMetricsView struct {
	Box                           string
	At                            time.Time // latest sample time
	CPU, Mem, Disk                *float64
	SparkCPU, SparkMem, SparkDisk string    // SVG polyline points, "" when no data
	Units                         []UnitRow // sorted by CPUFrac desc, nil-CPU rows last
}

// AlertRow is one currently-firing rule on one box. LoadMetrics only ever
// returns firing rules — a quiet rule has nothing to show.
type AlertRow struct {
	Box, Rule string
	Value     float64
	FiredAt   time.Time
}

// metricsCacheTTL is how soon after a successful sweep LoadMetrics considers
// the cache FRESH and skips even trying to refresh it. It does NOT bound how
// long a result may be served overall — see metricsStaleServeCeiling for
// that outer bound. Past metricsCacheTTL the cache becomes STALE (still
// served immediately, but a background refresh is triggered) rather than
// unusable. All callers within the FRESH window share the one sweep — the
// operator dashboard and every owner's app-metrics page render off the same
// read.
const metricsCacheTTL = 60 * time.Second

// metricsStaleServeCeiling bounds how old the cache may be and still be
// handed out immediately while a background refresh runs (LoadMetrics's
// STALE-WITHIN-CEILING state). It exists because nothing keeps the cache
// warm between requests — its age tracks the gap since the last REQUEST,
// not a timer — so without a ceiling, the first request after a quiet
// stretch would serve data that is arbitrarily old (a healthy box's actual
// reporting gap, PLUS however long nobody happened to ask), and a
// persistent bucket outage would let LoadMetrics return stale data with a
// nil error forever, since a background sweep's failure is only logged (see
// startBackgroundSweep) and nothing else ever revisits it. Past this
// ceiling, LoadMetrics falls through to the same blocking single-flight
// path the COLD case uses — costing on the order of a second now that
// sweepBox parallelises its GETs, rather than the several seconds a
// sequential sweep used to cost — in exchange for two things a stale answer
// can't give: data no older than one fresh sweep, and a sweep error
// actually reaching a caller instead of vanishing into a log line. No
// separate failure backoff is needed on top of this: a sweep that keeps
// failing never advances c.at, so the cache's age crosses the ceiling on its
// own within a couple of minutes regardless of how long the outage lasts.
const metricsStaleServeCeiling = 2 * time.Minute

// metricsSweepHours is how far back LoadMetrics reads raw chunks: the last
// 48 UTC hours, days/ rollups untouched (a later task's job).
const metricsSweepHours = 48

// metricsFreshWindow is how old a box's newest sample may be before its
// readings stop being anyone's CURRENT figure. The value is derived, not
// chosen freehand, from three terms:
//
//   - Flush cadence, 5 minutes — a FLOOR, not an exact figure: metricsLoop
//     sleeps 60s per tick and does per-tick work on top of that (alertTick
//     every tick, metricsFlush's own bucket writes every 5th), so the actual
//     gap between flushes only ever drifts upward from 5:00, never down.
//   - metricsStaleServeCeiling, 2 minutes — the oldest a cache
//     LoadMetrics hands back under stale-while-revalidate may be, i.e. how
//     far behind the true "latest flush" a box.At a caller sees can sit even
//     with a store that is never slow.
//   - One more minute of slack, covering both the flush-cadence drift above
//     and the ≤metricsSweepTimeout (15s) a background refresh can still be
//     mid-flight for when its result gets read.
//
// 5 + 2 + 1 = 8 minutes. This window must exceed that sum or a perfectly
// healthy, perfectly reachable box flaps between reporting and
// no-recent-readings for the tail of every cycle.
//
// It still has to do the job it always did on top of that: the 48h sweep
// keeps serving a dead box's final sample for two days, so without SOME
// bound an instance that vanished on Monday still reads "11% / 1.0 KB,
// current" on Wednesday. Owners get the no-recent-readings state instead
// (they have no timestamp on the page to judge staleness for themselves; the
// operator dashboard prints "Latest <time>" and so is deliberately left
// unbounded).
//
// What this window does NOT promise: it does not survive even ONE missed
// flush — miss a single 5-minute flush and the gap to the NEXT successful
// one is already ~10 minutes, past this window, and the box drops out
// exactly as it would have before this fix. That was true of the old 5-minute window too; widening
// far enough to also cover a missed flush would mean a truly dead box's
// stale reading survives even longer before it stops being anyone's
// "current" figure — not a trade worth making for one skipped beat.
const metricsFreshWindow = 8 * time.Minute

// metricsBoxesPrefix is where a box publishes its raw chunks and day
// rollups (internal/metrics.RawKey / DayKey); box discovery lists this
// prefix and reads the path segment right after "boxes/".
const metricsBoxesPrefix = "metrics/boxes/"

// metricsSweepBoxes and metricsSweepReads bound how much of a sweep is in
// flight at once: boxes swept in parallel, and hourly chunks read in parallel
// within a box. A cold sweep is boxes x 48 round trips to the bucket and is
// latency-bound, not CPU-bound — serially that is the whole "Instances page
// is slow" complaint — but it is still someone else's rate limit at the far
// end, so the fan-out is bounded rather than "one goroutine per key". The
// worst case is metricsSweepBoxes x metricsSweepReads requests outstanding.
const (
	metricsSweepBoxes = 4
	metricsSweepReads = 12
)

// metricsSweepTimeout bounds one whole bucket sweep: a List plus 48 Gets per
// box plus the alerts pass. It exists because the sweep no longer runs on the
// requesting page's context (see LoadMetrics) and so has no deadline of its
// own to inherit — a store that accepts connections and then hangs would
// otherwise pin the in-flight flag forever and every later caller with it.
// Fifteen seconds is far beyond a healthy sweep of a fleet this size and far
// short of any browser's patience.
const metricsSweepTimeout = 15 * time.Second

// metricsCache holds the result of the last successful bucket sweep. A
// failed sweep never reaches here — a failed BLOCKING sweep (c.at still
// zero, OR the cache is stale past metricsStaleServeCeiling) returns the
// error to its caller and leaves the cache exactly as it was; a failed
// BACKGROUND sweep (c.at set and still within the ceiling) is only logged,
// since nobody is waiting on it and the stale cache it leaves behind is
// still valid data, just not fresher.
//
// No separate failure backoff sits on top of the ceiling: a sweep that keeps
// failing never advances c.at, so the cache's age crosses
// metricsStaleServeCeiling on its own within a couple of minutes regardless
// of how long an outage lasts, and every caller after that lands back on the
// blocking path where the failure is visible again — see
// metricsStaleServeCeiling's own comment.
//
// mu guards the fields and is deliberately NOT held across the sweep's bucket
// IO; sweeping/done are what serialise the IO instead, so that a cold sweep
// delays only the callers that actually need its result rather than every
// caller that touches the cache. sweeping is true while one goroutine (either
// a blocking caller's own, or the detached refresh goroutine LoadMetrics
// launches for stale-while-revalidate) is out at the bucket, and done is the
// channel it closes when it comes back — the SAME channel and flag serve
// both shapes of sweep, so a caller arriving mid-refresh, whichever kind
// started it, coalesces onto it rather than starting a second one.
type metricsCache struct {
	mu       sync.Mutex
	at       time.Time
	boxes    []BoxMetricsView
	alerts   []AlertRow
	sweeping bool
	done     chan struct{}
}

// LoadMetrics reads every box under metrics/boxes/, the raw chunks of the
// last 48 hours (missing hours skipped silently), and alerts/boxes/*.json,
// caching the result in-process so concurrent callers — the dashboard page
// and every owner's app-metrics page — usually share one sweep.
//
// Three cache states:
//
//   - FRESH (c.at within metricsCacheTTL): the cached answer, immediately,
//     no bucket touched at all.
//   - STALE-WITHIN-CEILING (c.at is set, older than metricsCacheTTL, but no
//     older than metricsStaleServeCeiling): the cached answer is STILL
//     returned immediately — stale-while-revalidate — and, if no sweep is
//     already in flight, a fresh one is kicked off on a detached goroutine
//     (startBackgroundSweep) for the NEXT caller to benefit from. This is
//     the fix for the sweep's own cost: sweepBox's 48 hourly GETs for a
//     single box were measured at ~2.5s run sequentially, and that cost
//     used to land inline on whichever request's bad luck it was to arrive
//     right after a cache expiry — the dashboard page and every tenant's
//     app page alike.
//   - COLD (c.at is zero — nothing has ever swept successfully) or STALE
//     PAST THE CEILING (c.at is set but older than
//     metricsStaleServeCeiling — nothing has kept the cache warm for a
//     while, e.g. a quiet console between requests, or every sweep since
//     has been failing): the ORIGINAL blocking behaviour. There is either no
//     servable stale answer (COLD) or the one that exists is too old to
//     hand out one more time (STALE-PAST-CEILING) — either way this caller
//     (or whichever caller's sweep is already in flight) waits for a real
//     sweep and gets ITS error if the sweep fails. The ceiling is what keeps
//     a persistent bucket outage visible: without it, stale-while-revalidate
//     alone would let LoadMetrics return stale data with a nil error
//     forever regardless of how long the store stays down, because a
//     background sweep's failure is only logged (see startBackgroundSweep)
//     — crossing the ceiling forces the next caller back onto the path that
//     actually surfaces the error. Blocking here now costs on the order of
//     a second for a fleet this size rather than the several seconds it
//     cost before sweepBox parallelised its GETs, which is what makes
//     paying that cost here — instead of serving indefinitely-stale data —
//     an acceptable trade.
//
// Two properties matter beyond the cache states themselves, both learned
// from the shape this replaced (mutex held across the whole sweep, on the
// request's own context):
//
//   - The cache mutex is never held across bucket IO. Only the goroutine that
//     actually sweeps is slow; everyone else either takes the cached answer
//     immediately or waits on the in-flight sweep's channel (only the
//     blocking case waits at all), and neither blocks a caller whose cache
//     is fresh or stale-but-servable. handleApp calls this on every tenant's
//     app page, so a cold sweep must not become a queue every page on the
//     console stands in.
//   - Every sweep — blocking or background — runs on a context detached from
//     any particular caller's: context.WithoutCancel of whatever was handed
//     in (context.Background() for the background case, since there is no
//     caller to derive from), plus metricsSweepTimeout. A request cancelled
//     mid-sweep (a reload, a navigate-away) would otherwise abort a sweep
//     other callers are waiting on AND, since a failed blocking sweep is
//     never cached, leave the next request to start from scratch: reload
//     impatiently enough and the cache never populates at all. The caller's
//     own context still governs WAITING in the blocking case (a cancelled
//     waiter gives up rather than hanging), it just never governs the sweep
//     itself.
func (s *Server) LoadMetrics(ctx context.Context) ([]BoxMetricsView, []AlertRow, error) {
	c := &s.metricsCache
	for {
		c.mu.Lock()
		if !c.at.IsZero() && time.Since(c.at) < metricsCacheTTL {
			// FRESH.
			boxes, alerts := c.boxes, c.alerts
			c.mu.Unlock()
			return boxes, alerts, nil
		}
		if !c.at.IsZero() && time.Since(c.at) <= metricsStaleServeCeiling {
			// STALE-WITHIN-CEILING: hand back what we have right now — it
			// is still a real sweep's result, just older than
			// metricsCacheTTL — and let a refresh happen off to the side
			// rather than making this caller pay for it. If one is already
			// running (background OR, in principle, another blocking
			// caller's), don't start a second; this caller just doesn't
			// wait for it either way.
			boxes, alerts := c.boxes, c.alerts
			if !c.sweeping {
				s.startBackgroundSweep(c)
			}
			c.mu.Unlock()
			return boxes, alerts, nil
		}
		// COLD, or STALE-PAST-THE-CEILING: no servable stale answer exists
		// (COLD) or the one that does is too old to hand out again
		// (past metricsStaleServeCeiling) — this caller must wait for real
		// data and see its error if the sweep fails, exactly as the
		// original blocking behaviour did.
		if c.sweeping {
			// Someone is already at the bucket: wait for their result
			// rather than duplicating the sweep, then re-check (their
			// sweep may have failed, in which case this iteration takes
			// the work over).
			wait := c.done
			c.mu.Unlock()
			select {
			case <-wait:
				continue
			case <-ctx.Done():
				return nil, nil, ctx.Err()
			}
		}
		c.sweeping = true
		c.done = make(chan struct{})
		done := c.done
		c.mu.Unlock()

		// err starts non-nil so that if sweepAll panics (unwinding past the
		// assignment below), the deferred bookkeeping still sees a failed
		// sweep: nothing gets cached and the in-flight flag is cleared
		// instead of wedging every future caller. net/http's per-connection
		// recover keeps the process up; this keeps the cache usable.
		var (
			boxes  []BoxMetricsView
			alerts []AlertRow
			err    = errors.New("console: metrics sweep did not complete")
		)
		func() {
			defer func() {
				c.mu.Lock()
				if err == nil {
					// Only a fully successful sweep is cached; a failed
					// one leaves the previous cache exactly as it was.
					c.at = time.Now()
					c.boxes = boxes
					c.alerts = alerts
				}
				c.sweeping = false
				close(done)
				c.mu.Unlock()
			}()
			boxes, alerts, err = s.sweepAll(ctx)
		}()

		if err != nil {
			return nil, nil, err
		}
		return boxes, alerts, nil
	}
}

// startBackgroundSweep launches sweepAll on its own goroutine to refresh a
// cache that has swept successfully at least once and is still within
// metricsStaleServeCeiling, without making any caller wait on it — see
// LoadMetrics's STALE-WITHIN-CEILING case. "Swept successfully at least
// once" is exactly c.at being set: an empty-bucket console (c.at set,
// c.boxes nil) takes this path just like a populated one, since nil boxes
// is itself a complete, valid swept result, not an absent one. Must be
// called with c.mu held: it sets c.sweeping and installs c.done itself, the
// very same fields the blocking path uses, so a caller arriving while this
// is in flight (whether via the STALE-WITHIN-CEILING branch or the blocking
// one) coalesces onto it instead of starting a second sweep. sweepAll is
// handed context.Background() rather than any request's
// context because there IS no request behind this goroutine — it already
// detaches internally via context.WithoutCancel and bounds itself with
// metricsSweepTimeout, so Background() costs nothing extra.
//
// Unlike the blocking path, this goroutine has no net/http per-connection
// recover sitting above it — an unrecovered panic here would take the whole
// process down, not just one request — so it recovers explicitly. A failed
// or panicked background sweep is only logged: no caller is waiting on its
// error, and the stale cache it leaves untouched is still valid data.
func (s *Server) startBackgroundSweep(c *metricsCache) {
	c.sweeping = true
	c.done = make(chan struct{})
	done := c.done
	go func() {
		var (
			boxes  []BoxMetricsView
			alerts []AlertRow
			err    = errors.New("console: background metrics sweep did not complete")
		)
		defer func() {
			rec := recover()
			c.mu.Lock()
			if err == nil {
				c.at = time.Now()
				c.boxes = boxes
				c.alerts = alerts
			}
			c.sweeping = false
			close(done)
			c.mu.Unlock()
			switch {
			case rec != nil:
				log.Printf("console: background metrics sweep panic, cache left at its last good sweep: %v", rec)
			case err != nil:
				log.Printf("console: background metrics sweep: %v", err)
			}
		}()
		boxes, alerts, err = s.sweepAll(context.Background())
	}()
}

// sweepAll is the bucket half of LoadMetrics: discovery, every box's 48h raw
// sweep (metricsSweepBoxes of them at a time), then the alerts pass, on a
// context detached from the caller's (WithoutCancel) and bounded by
// metricsSweepTimeout. It holds no lock — see LoadMetrics's doc comment for
// why that is the point.
//
// Boxes are swept concurrently but land in boxNames order, never completion
// order: the dashboard's row order must not depend on which box's bucket
// answered first. Any box failing fails the whole sweep, with the
// lowest-ordered box's error — the same error the serial sweep returned, so a
// flaky box reads the same in the logs however the reads interleaved.
func (s *Server) sweepAll(ctx context.Context) ([]BoxMetricsView, []AlertRow, error) {
	sweepCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), metricsSweepTimeout)
	defer cancel()

	boxNames, err := s.discoverBoxes(sweepCtx)
	if err != nil {
		return nil, nil, err
	}

	views := make([]BoxMetricsView, len(boxNames))
	errs := make([]error, len(boxNames))
	var (
		wg     sync.WaitGroup
		sem    = make(chan struct{}, metricsSweepBoxes)
		once   sync.Once
		failed = make(chan struct{})
	)
	for i, box := range boxNames {
		wg.Add(1)
		go func() {
			defer wg.Done()
			select {
			case sem <- struct{}{}:
			case <-failed:
				// Another box has already failed the sweep; nothing this one
				// could read would be returned. Boxes already at the bucket
				// still finish — they run on sweepCtx, so the error a failed
				// sweep reports is always a box's own, never one this
				// function inflicted.
				return
			}
			defer func() { <-sem }()
			view, err := s.sweepBox(sweepCtx, box)
			if err != nil {
				errs[i] = err
				once.Do(func() { close(failed) })
				return
			}
			views[i] = view
		}()
	}
	wg.Wait()
	for _, err := range errs {
		if err != nil {
			return nil, nil, err
		}
	}

	alerts, err := s.sweepAlerts(sweepCtx)
	if err != nil {
		return nil, nil, err
	}
	return views, alerts, nil
}

// discoverBoxes lists metrics/boxes/ and returns the deduped, sorted set of
// box names — the path segment right after "boxes/".
func (s *Server) discoverBoxes(ctx context.Context) ([]string, error) {
	keys, err := s.cfg.Releases.List(ctx, metricsBoxesPrefix)
	if err != nil {
		return nil, fmt.Errorf("list %s: %w", metricsBoxesPrefix, err)
	}
	seen := make(map[string]bool)
	var boxes []string
	for _, key := range keys {
		rest := strings.TrimPrefix(key, metricsBoxesPrefix)
		idx := strings.Index(rest, "/")
		if idx <= 0 {
			continue // malformed/unexpected key shape directly under the prefix
		}
		box := rest[:idx]
		if seen[box] {
			continue
		}
		seen[box] = true
		boxes = append(boxes, box)
	}
	sort.Strings(boxes)
	return boxes, nil
}

// sweepBox reads box's raw chunks for the last 48 UTC hours, metricsSweepReads
// at a time, and assembles its BoxMetricsView. A missing hour (os.ErrNotExist)
// or an unparsable chunk is skipped silently — a box that has been up for
// under 48h, or whose agent hiccuped on one hour, still renders on what it
// does have. Any other read error fails the whole sweep (and so is never
// cached), reporting the newest failing hour's error as the serial read did.
//
// Each hour lands in its own slot and the slots are concatenated in hour
// order before sorting, because the sort below is NOT stable: samples sharing
// a timestamp keep whatever relative order they were appended in, so appending
// as the reads complete would let a box's current reading flap between
// renders.
func (s *Server) sweepBox(ctx context.Context, box string) (BoxMetricsView, error) {
	now := time.Now().UTC()
	hourly := make([][]metrics.Sample, metricsSweepHours)
	errs := make([]error, metricsSweepHours)
	var (
		wg     sync.WaitGroup
		sem    = make(chan struct{}, metricsSweepReads)
		once   sync.Once
		failed = make(chan struct{})
	)
	for i := range metricsSweepHours {
		wg.Add(1)
		go func() {
			defer wg.Done()
			select {
			case sem <- struct{}{}:
			case <-failed:
				return // an earlier hour already failed the sweep
			}
			defer func() { <-sem }()
			hour := metrics.HourOf(now.Add(-time.Duration(i) * time.Hour))
			key := metrics.RawKey(box, hour)
			body, _, err := s.cfg.Releases.Get(ctx, key)
			if err != nil {
				if errors.Is(err, os.ErrNotExist) {
					return
				}
				errs[i] = fmt.Errorf("read %s: %w", key, err)
				once.Do(func() { close(failed) })
				return
			}
			var chunk metrics.RawChunk
			if err := json.Unmarshal(body, &chunk); err != nil {
				return // unparsable chunk: skip, don't fail the sweep
			}
			hourly[i] = chunk.Samples
		}()
	}
	wg.Wait()
	for _, err := range errs {
		if err != nil {
			return BoxMetricsView{}, err
		}
	}

	var samples []metrics.Sample
	for _, h := range hourly {
		samples = append(samples, h...)
	}
	sort.Slice(samples, func(i, j int) bool { return samples[i].At.Before(samples[j].At) })
	return buildBoxView(box, samples), nil
}

// buildBoxView assembles one box's view from its swept samples, oldest
// first. The latest sample's own values are used as-is for CPU/Mem/Disk/At
// — nil stays nil; this is deliberately NOT "each field's newest non-nil
// reading independently", which would quietly mix readings from different
// points in time.
func buildBoxView(box string, samples []metrics.Sample) BoxMetricsView {
	view := BoxMetricsView{Box: box}
	if len(samples) == 0 {
		return view
	}
	latest := samples[len(samples)-1]
	view.At = latest.At
	view.CPU = latest.Host.CPUFrac
	view.Mem = latest.Host.MemFrac
	view.Disk = latest.Host.DiskFrac

	cpu := make([]*float64, len(samples))
	mem := make([]*float64, len(samples))
	disk := make([]*float64, len(samples))
	for i, sm := range samples {
		cpu[i] = sm.Host.CPUFrac
		mem[i] = sm.Host.MemFrac
		disk[i] = sm.Host.DiskFrac
	}
	view.SparkCPU = sparkPoints(cpu, 240, 40)
	view.SparkMem = sparkPoints(mem, 240, 40)
	view.SparkDisk = sparkPoints(disk, 240, 40)

	view.Units = buildUnitRows(samples)
	return view
}

// buildUnitRows turns the latest sample's Units map into rows sorted by
// CPUFrac descending, nil-CPU rows last, stable by unit name within ties,
// each carrying its own sparklines drawn across the whole swept window. A
// unit absent from the latest sample's map simply isn't in this slice —
// never rendered, never aggregated as a zero (Task 4's "missing map key is
// no reading" ruling, same map shape here) — even if it appears in older
// samples: the row set is "what is running now", not "what ever ran".
func buildUnitRows(samples []metrics.Sample) []UnitRow {
	if len(samples) == 0 {
		return nil
	}
	units := samples[len(samples)-1].Units
	rows := make([]UnitRow, 0, len(units))
	for name, u := range units {
		rows = append(rows, UnitRow{
			Unit:       name,
			CPUFrac:    u.CPUFrac,
			MemBytes:   u.MemBytes,
			ReadBytes:  u.ReadBytes,
			WriteBytes: u.WriteBytes,
			SparkCPU:   sparkPoints(unitCPUSeries(samples, name), 240, 40),
			SparkMem:   sparkPoints(unitMemSeries(samples, name), 240, 40),
		})
	}
	// Name order first so the CPU-desc sort below, being stable, ties out
	// alphabetically rather than on map iteration order.
	sort.Slice(rows, func(i, j int) bool { return rows[i].Unit < rows[j].Unit })
	sort.SliceStable(rows, func(i, j int) bool {
		a, b := rows[i].CPUFrac, rows[j].CPUFrac
		if a == nil {
			return false // nil never sorts before anything, including another nil
		}
		if b == nil {
			return true
		}
		return *a > *b
	})
	return rows
}

// unitCPUSeries is one unit's CPU fraction at each swept sample index. A
// sample where the unit is absent, or present without a CPU reading, stays
// nil — sparkPoints draws a gap, never a zero.
func unitCPUSeries(samples []metrics.Sample, unit string) []*float64 {
	out := make([]*float64, len(samples))
	for i, sm := range samples {
		if u, ok := sm.Units[unit]; ok {
			out[i] = u.CPUFrac
		}
	}
	return out
}

// unitMemSeries is one unit's memory at each swept sample index, scaled to
// its OWN 48h peak. Unlike CPU, a unit's memory has no 0..1 ceiling the
// console can know — the cgroup limit isn't published, and the box's total
// RAM is deliberately not in an owner's view — so the sparkline shows the
// shape of this instance's memory against its own high-water mark, and the
// number beside it stays the absolute byte count. Absent/nil readings stay
// nil; a unit that never reported memory gets no line at all.
func unitMemSeries(samples []metrics.Sample, unit string) []*float64 {
	var peak int64
	for _, sm := range samples {
		if u, ok := sm.Units[unit]; ok && u.MemBytes != nil && *u.MemBytes > peak {
			peak = *u.MemBytes
		}
	}
	if peak <= 0 {
		return nil
	}
	out := make([]*float64, len(samples))
	for i, sm := range samples {
		u, ok := sm.Units[unit]
		if !ok || u.MemBytes == nil {
			continue
		}
		f := float64(*u.MemBytes) / float64(peak)
		out[i] = &f
	}
	return out
}

// InstanceMetrics is one of an APP OWNER's instances, named the way owners
// name their instances: by route host. It carries that instance's unit
// readings and nothing else — no host totals, no sibling units, no alerts.
// Box is here for internal use (picking the freshest reporting box when a
// host has moved); the app page must not print it, because which box a
// tenant sits on is not an owner's business.
type InstanceMetrics struct {
	ObservedAt time.Time

	Host     string // the route host — how owners name their instances
	Box      string
	CPUFrac  *float64
	MemBytes *int64

	SparkCPU, SparkMem string

	// Hibernate is the route's hibernation setting, copied in by the caller
	// (handleApp) — the sweep cannot see it. It only matters when Reporting
	// is false, where it decides what the absence of readings may claim: a
	// hibernating instance is presumed asleep, anything else is NOT — it may
	// be serving right now from outside per-instance accounting (a bespoke
	// unit like keymail-relay.service, or an exec child on a box whose agent
	// has no delegated subtree yet), and the template must not assert a
	// sleep state nobody observed.
	Hibernate bool
}

// Reporting is the template's one question: did any box report this
// instance's unit in the last sweep? False is the no-readings case, whose
// rendering splits on Hibernate: hibernating instances read "asleep — no
// recent readings", everything else the state-neutral "no per-instance
// readings" — never zeros either way. (Reporting with
// nil readings is a different, honest state: the unit is up but that metric
// was unreadable, which renders as the usual em-dash.)
func (m InstanceMetrics) Reporting() bool { return m.Box != "" }

// OwnerAppMetrics filters the cached 48h sweep down to EXACTLY the given
// hosts' own per-instance readings — carlos-app@<host>.service for a
// unit-backed tenant, carlos-exec@<host> for an exec-backed one, or the
// operator-stamped bespoke unit named in unitByHost[host] (registry
// Route.Unit, #110) when the caller supplies one. It returns unit-level
// rows only — no host totals, no other units, no alert data. The allowlist
// is the hosts slice the caller derived from the app's OWN routes, plus
// unitByHost entries drawn from those same routes; nothing else can enter
// the view.
//
// Every allowlisted host gets exactly one row, in host order, whether or not
// the sweep saw it: a host with no unit data is an UNKNOWN row (nil
// readings, Reporting false), never zeros and never silently dropped — an
// owner whose instance is asleep must still see it listed.
func (s *Server) OwnerAppMetrics(ctx context.Context, hosts []string, unitByHost map[string]string) ([]InstanceMetrics, error) {
	// The allowlist, keyed by the unit names each host can legally match.
	// Matching is exact-string on this map: a unit the caller did not ask
	// for cannot be reached by any path below.
	want := make(map[string]string, 2*len(hosts))
	for _, h := range hosts {
		if h == "" {
			continue
		}
		// Two legal keys per host: the systemd unit a unit-backed tenant
		// reports under, and the delegated exec cgroup an exec-backed one
		// does. One route, one backing — a host can't legally report under
		// both at once; if both ever appear in a sweep, the freshest-box
		// rule below picks deterministically, no special case.
		want[activator.UnitName(h)] = h
		want[activator.ExecName(h)] = h
		// The third legal key: an operator-stamped bespoke unit (registry
		// Route.Unit, #110). Opt-in per host — the caller maps host →
		// stamped unit from the route rows it already holds; an empty map
		// costs nothing and matches nothing new. (A nil map indexes fine
		// in Go — no guard needed.)
		if u := unitByHost[h]; u != "" {
			want[u] = h
		}
	}
	if len(want) == 0 {
		return nil, nil
	}

	boxes, _, err := s.LoadMetrics(ctx)
	if err != nil {
		return nil, err
	}

	found := make(map[string]InstanceMetrics, len(want))
	at := make(map[string]time.Time, len(want))
	now := time.Now()
	for _, box := range boxes {
		// A box that stopped publishing keeps its final sample inside the
		// 48h sweep for two days; those numbers are history, not this
		// instance's current figure, so the whole box drops out of the
		// owner view and its instances read as no-recent-readings.
		if box.At.IsZero() || now.Sub(box.At) > metricsFreshWindow {
			continue
		}
		for _, u := range box.Units {
			host, ok := want[u.Unit]
			if !ok {
				continue // not this app's — the whole boundary is this line
			}
			// A host that reports on two boxes at once (a move in flight)
			// shows the freshest box's reading, not whichever sorted first.
			if prev, taken := at[host]; taken && !box.At.After(prev) {
				continue
			}
			at[host] = box.At
			found[host] = InstanceMetrics{
				Host:       host,
				ObservedAt: box.At,
				Box:        box.Box,
				CPUFrac:    u.CPUFrac,
				MemBytes:   u.MemBytes,
				SparkCPU:   u.SparkCPU,
				SparkMem:   u.SparkMem,
			}
		}
	}

	// Iterate the callers' hosts, not want's values: want now carries TWO
	// keys per host, and "every allowlisted host gets exactly one row" is a
	// promise about hosts, not keys. seen collapses duplicate input hosts,
	// which the old unique-keyed map did implicitly.
	rows := make([]InstanceMetrics, 0, len(hosts))
	seen := make(map[string]bool, len(hosts))
	for _, h := range hosts {
		if h == "" || seen[h] {
			continue
		}
		seen[h] = true
		if row, ok := found[h]; ok {
			rows = append(rows, row)
		} else {
			rows = append(rows, InstanceMetrics{Host: h})
		}
	}
	sort.Slice(rows, func(i, j int) bool { return rows[i].Host < rows[j].Host })
	return rows, nil
}

// sweepAlerts reads every alerts/boxes/*.json and returns one AlertRow per
// currently-firing rule. Unreadable-but-present keys fail the sweep (so it
// is never cached half-done); an unparsable object is skipped, matching
// alertPass's posture on the same prefix.
func (s *Server) sweepAlerts(ctx context.Context) ([]AlertRow, error) {
	keys, err := s.cfg.Releases.List(ctx, alertPrefix)
	if err != nil {
		return nil, fmt.Errorf("list %s: %w", alertPrefix, err)
	}
	var rows []AlertRow
	for _, key := range keys {
		if !strings.HasSuffix(key, ".json") {
			continue
		}
		body, _, err := s.cfg.Releases.Get(ctx, key)
		if err != nil {
			if errors.Is(err, os.ErrNotExist) {
				continue
			}
			return nil, fmt.Errorf("read %s: %w", key, err)
		}
		var st metrics.AlertState
		if err := json.Unmarshal(body, &st); err != nil {
			continue // unparsable: skip, don't fail the sweep
		}
		box := st.Box
		if box == "" {
			// Trust the key over an empty body field, same as alertPass.
			box = strings.TrimSuffix(strings.TrimPrefix(key, alertPrefix), ".json")
		}
		names := make([]string, 0, len(st.Rules))
		for name := range st.Rules {
			names = append(names, name)
		}
		sort.Strings(names)
		for _, name := range names {
			rs := st.Rules[name]
			if !rs.Firing {
				continue
			}
			rows = append(rows, AlertRow{Box: box, Rule: name, Value: rs.Value, FiredAt: rs.FiredAt})
		}
	}
	return rows, nil
}

// sparkPoints maps vals (up to 48h of minute samples, one slot per sample
// index) onto an SVG polyline "points" string on a w x h viewBox. Values are
// fractions and clamp into [0,1] if out of range; y is inverted (SVG's y
// axis points down, so a higher fraction plots at a SMALLER y — visually
// "up"); x is spread evenly across the width by the sample's index, not by
// elapsed time. A nil sample contributes no pair at all — its x slot is
// simply skipped, so the polyline draws a straight line across the gap;
// breaking the line at gaps is not attempted in v1 (documented here per the
// brief). Empty input or input that is nil throughout returns "".
//
// Input longer than w is bucketed down to w points first (see downsample):
// 48h of minute samples is ~2880 readings into a 240px box — twelve points
// per pixel of detail nobody can see, paid for in a coordinate string held
// in the shared cache for every metric of every unit on every box.
func sparkPoints(vals []*float64, w, h int) string {
	if len(vals) == 0 {
		return ""
	}
	vals = downsample(vals, w)
	var b strings.Builder
	n := len(vals)
	for i, v := range vals {
		if v == nil {
			continue
		}
		f := *v
		if f < 0 {
			f = 0
		} else if f > 1 {
			f = 1
		}
		x := 0.0
		if n > 1 {
			x = float64(i) / float64(n-1) * float64(w)
		}
		y := float64(h) - f*float64(h)
		if b.Len() > 0 {
			b.WriteByte(' ')
		}
		b.WriteString(strconv.FormatFloat(x, 'f', 2, 64))
		b.WriteByte(',')
		b.WriteString(strconv.FormatFloat(y, 'f', 2, 64))
	}
	return b.String()
}

// downsample buckets vals into at most max evenly-sized buckets, each
// bucket's value being the MEAN of its non-nil readings — a spike lifts its
// bucket rather than disappearing, and a bucket with nothing readable in it
// stays nil, so "unknown" survives bucketing as a gap and never becomes a
// zero. Input at or under max is returned untouched (the common case: a box
// that has been up for minutes, and every unit test's handful of samples).
func downsample(vals []*float64, max int) []*float64 {
	if max <= 0 || len(vals) <= max {
		return vals
	}
	n := len(vals)
	out := make([]*float64, max)
	for b := 0; b < max; b++ {
		lo := b * n / max
		hi := (b + 1) * n / max
		if hi > n {
			hi = n
		}
		var sum float64
		var cnt int
		for _, v := range vals[lo:hi] {
			if v == nil {
				continue
			}
			sum += *v
			cnt++
		}
		if cnt == 0 {
			continue // nothing readable in this bucket: a gap, not a zero
		}
		mean := sum / float64(cnt)
		out[b] = &mean
	}
	return out
}