package console

import (
	"context"
	"encoding/json"
	"errors"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

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

// newMetricsFixture builds the minimal Server LoadMetrics's tests need: a
// console Store (NewServer's SigningKey dance requires one) and a bare
// release.DirStore, with none of consoleFixture's registry/keymail/HTTP
// scaffolding — mirrors server_test.go's minimal Config shape, trimmed to
// what a bucket-only read path touches.
func newMetricsFixture(t *testing.T) (*Server, release.Store) {
	t.Helper()
	st, err := Open(filepath.Join(t.TempDir(), "console.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { st.Close() })
	rel := release.NewDirStore(filepath.Join(t.TempDir(), "bucket"))
	srv, err := NewServer(Config{Releases: rel}, st)
	if err != nil {
		t.Fatal(err)
	}
	return srv, rel
}

func p(f float64) *float64 { return &f }
func pi64(i int64) *int64  { return &i }

// writeAlertState publishes an AlertState the way the box agent would,
// create-only since the fixture never re-publishes the same box twice.
func writeAlertState(t *testing.T, rel release.Store, st metrics.AlertState) {
	t.Helper()
	b, err := json.Marshal(st)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := rel.Put(context.Background(), metrics.AlertKey(st.Box), b, ""); err != nil {
		t.Fatal(err)
	}
}

// TestLoadMetricsSweepsAndCaches pins the whole read path in one pass:
// discovery, the 48h raw sweep landing on the newest sample's values,
// non-empty sparklines, CPU-desc unit ordering with nil-CPU last, firing-only
// alerts — and then that a second call inside the 60s TTL returns the SAME
// result without touching the store again (proven by deleting every object
// LoadMetrics would need and asserting the second call doesn't notice).
func TestLoadMetricsSweepsAndCaches(t *testing.T) {
	srv, rel := newMetricsFixture(t)
	ctx := context.Background()

	olderAt := time.Now().UTC().Add(-90 * time.Minute)
	latestAt := time.Now().UTC().Add(-5 * time.Minute)

	older := metrics.RawChunk{
		V:    1,
		Box:  "box-a",
		Hour: metrics.HourOf(olderAt),
		Samples: []metrics.Sample{
			{
				At:   olderAt,
				Host: metrics.HostSample{CPUFrac: p(0.2), MemFrac: p(0.3), DiskFrac: p(0.4)},
				Units: map[string]metrics.UnitSample{
					"carlos-app@a.service": {CPUFrac: p(0.05), MemBytes: pi64(1000)},
				},
			},
		},
	}
	latest := metrics.RawChunk{
		V:    1,
		Box:  "box-a",
		Hour: metrics.HourOf(latestAt),
		Samples: []metrics.Sample{
			{
				At:   latestAt,
				Host: metrics.HostSample{CPUFrac: p(0.5), MemFrac: p(0.6), DiskFrac: p(0.7)},
				Units: map[string]metrics.UnitSample{
					"carlos-app@a.service":      {CPUFrac: p(0.9), MemBytes: pi64(2000)},
					"carlos-app@b.service":      {CPUFrac: p(0.1), MemBytes: pi64(500)},
					"carlos-app@nilcpu.service": {MemBytes: pi64(100)},
				},
			},
		},
	}
	for _, c := range []metrics.RawChunk{older, latest} {
		if err := metrics.WriteRaw(ctx, rel, c); err != nil {
			t.Fatal(err)
		}
	}

	writeAlertState(t, rel, metrics.AlertState{
		V:   1,
		Box: "box-a",
		Rules: map[string]metrics.RuleState{
			"disk": {Firing: true, Value: 0.95, FiredAt: time.Date(2026, 8, 2, 10, 0, 0, 0, time.UTC)},
			"cpu":  {Firing: false, Value: 0.5},
		},
	})

	boxes, alerts, err := srv.LoadMetrics(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if len(boxes) != 1 {
		t.Fatalf("boxes = %d, want 1: %+v", len(boxes), boxes)
	}
	bv := boxes[0]
	if bv.Box != "box-a" {
		t.Fatalf("box = %q, want box-a", bv.Box)
	}
	if bv.CPU == nil || *bv.CPU != 0.5 {
		t.Fatalf("CPU = %v, want 0.5 (the newest sample's value)", bv.CPU)
	}
	if !bv.At.Equal(latestAt) {
		t.Fatalf("At = %v, want %v (the newest sample's time)", bv.At, latestAt)
	}
	if bv.SparkCPU == "" {
		t.Fatal("SparkCPU is empty despite two samples")
	}
	if len(bv.Units) != 3 {
		t.Fatalf("units = %d, want 3: %+v", len(bv.Units), bv.Units)
	}
	if bv.Units[0].Unit != "carlos-app@a.service" || bv.Units[0].CPUFrac == nil || *bv.Units[0].CPUFrac != 0.9 {
		t.Fatalf("units[0] = %+v, want carlos-app@a.service at 0.9 (CPU desc)", bv.Units[0])
	}
	if bv.Units[1].Unit != "carlos-app@b.service" {
		t.Fatalf("units[1] = %+v, want carlos-app@b.service", bv.Units[1])
	}
	if bv.Units[2].Unit != "carlos-app@nilcpu.service" || bv.Units[2].CPUFrac != nil {
		t.Fatalf("units[2] = %+v, want the nil-CPU unit sorted last", bv.Units[2])
	}
	if len(alerts) != 1 {
		t.Fatalf("alerts = %d, want 1 (firing rules only, cpu excluded): %+v", len(alerts), alerts)
	}
	if alerts[0].Box != "box-a" || alerts[0].Rule != "disk" || alerts[0].Value != 0.95 {
		t.Fatalf("alert = %+v", alerts[0])
	}
	if !alerts[0].FiredAt.Equal(time.Date(2026, 8, 2, 10, 0, 0, 0, time.UTC)) {
		t.Fatalf("alert.FiredAt = %v, want the state's FiredAt as-is", alerts[0].FiredAt)
	}

	// Poison the store: delete everything a fresh sweep would need to read.
	if err := rel.Delete(ctx, metrics.RawKey("box-a", metrics.HourOf(latestAt))); err != nil {
		t.Fatal(err)
	}
	if err := rel.Delete(ctx, metrics.RawKey("box-a", metrics.HourOf(olderAt))); err != nil {
		t.Fatal(err)
	}
	if err := rel.Delete(ctx, metrics.AlertKey("box-a")); err != nil {
		t.Fatal(err)
	}

	boxes2, alerts2, err := srv.LoadMetrics(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if len(boxes2) != 1 || boxes2[0].Box != "box-a" || boxes2[0].CPU == nil || *boxes2[0].CPU != 0.5 {
		t.Fatalf("second call within TTL re-read a poisoned store: %+v", boxes2)
	}
	if len(alerts2) != 1 {
		t.Fatalf("second call within TTL re-read alerts from a poisoned store: %+v", alerts2)
	}
}

// TestLoadMetricsEmptyBucket: no box has ever published anything — the
// normal state of a freshly deployed console — is a clean empty result, not
// an error.
func TestLoadMetricsEmptyBucket(t *testing.T) {
	srv, _ := newMetricsFixture(t)
	boxes, alerts, err := srv.LoadMetrics(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	if len(boxes) != 0 || len(alerts) != 0 {
		t.Fatalf("boxes=%+v alerts=%+v, want both empty", boxes, alerts)
	}
}

// ctxStore is a release.Store that honours its context: every read whose
// context is already done fails with that context's error. DirStore ignores
// its ctx entirely (see internal/release/store.go), so a cancelled-caller
// test built on a bare DirStore would pass vacuously no matter what
// LoadMetrics does with the context it is handed.
type ctxStore struct{ release.Store }

func (c ctxStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	if err := ctx.Err(); err != nil {
		return nil, "", err
	}
	return c.Store.Get(ctx, key)
}

func (c ctxStore) List(ctx context.Context, prefix string) ([]string, error) {
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	return c.Store.List(ctx, prefix)
}

// newCtxMetricsFixture is newMetricsFixture with the server reading through a
// context-honouring store. The returned release.Store is the raw one, for
// seeding.
func newCtxMetricsFixture(t *testing.T) (*Server, release.Store) {
	t.Helper()
	st, err := Open(filepath.Join(t.TempDir(), "console.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { st.Close() })
	rel := release.NewDirStore(filepath.Join(t.TempDir(), "bucket"))
	srv, err := NewServer(Config{Releases: ctxStore{rel}}, st)
	if err != nil {
		t.Fatal(err)
	}
	return srv, rel
}

// seedOneBox writes a single box's chunk with one sample, the minimum a sweep
// needs to produce a non-empty result.
func seedOneBox(t *testing.T, rel release.Store, box string) {
	t.Helper()
	at := time.Now().UTC().Add(-time.Minute)
	if err := metrics.WriteRaw(context.Background(), rel, metrics.RawChunk{
		V: 1, Box: box, Hour: metrics.HourOf(at),
		Samples: []metrics.Sample{{At: at, Host: metrics.HostSample{CPUFrac: p(0.5)}}},
	}); err != nil {
		t.Fatal(err)
	}
}

// TestLoadMetricsSweepSurvivesCancelledCaller pins the review fix: the shared
// sweep runs on its own context (WithoutCancel + its own timeout), never the
// request's. Otherwise the first caller to navigate away mid-sweep aborts the
// sweep every other caller is waiting on, and — since a failed sweep is never
// cached — the next request starts over, so a page reloaded impatiently can
// starve the cache indefinitely. A caller whose context is ALREADY cancelled
// is the extreme of that case and must still get real data.
func TestLoadMetricsSweepSurvivesCancelledCaller(t *testing.T) {
	srv, rel := newCtxMetricsFixture(t)
	seedOneBox(t, rel, "box-a")

	ctx, cancel := context.WithCancel(context.Background())
	cancel()

	boxes, _, err := srv.LoadMetrics(ctx)
	if err != nil {
		t.Fatalf("LoadMetrics with a cancelled caller context: %v — the sweep must not inherit the request's cancellation", err)
	}
	if len(boxes) != 1 || boxes[0].Box != "box-a" || boxes[0].CPU == nil || *boxes[0].CPU != 0.5 {
		t.Fatalf("boxes = %+v, want box-a at 0.5", boxes)
	}
}

// TestLoadMetricsConcurrentColdCallers: several callers arriving on a cold
// cache at once — every tenant's app page plus the dashboard on a console
// that just restarted — all get the same correct data, whether they coalesce
// onto one sweep or duplicate it. Run under -race, this is also the guard on
// the cache fields being written while the lock is released across the IO.
func TestLoadMetricsConcurrentColdCallers(t *testing.T) {
	srv, rel := newMetricsFixture(t)
	seedOneBox(t, rel, "box-a")

	const callers = 4
	type result struct {
		boxes []BoxMetricsView
		err   error
	}
	results := make([]result, callers)
	var start sync.WaitGroup
	var done sync.WaitGroup
	start.Add(1)
	for i := 0; i < callers; i++ {
		done.Add(1)
		go func(i int) {
			defer done.Done()
			start.Wait()
			boxes, _, err := srv.LoadMetrics(context.Background())
			results[i] = result{boxes: boxes, err: err}
		}(i)
	}
	start.Done()
	done.Wait()

	for i, r := range results {
		if r.err != nil {
			t.Fatalf("caller %d: %v", i, r.err)
		}
		if len(r.boxes) != 1 || r.boxes[0].Box != "box-a" || r.boxes[0].CPU == nil || *r.boxes[0].CPU != 0.5 {
			t.Fatalf("caller %d got %+v, want box-a at 0.5", i, r.boxes)
		}
	}
}

// writeBoxSample (re)publishes box's current-hour raw chunk with a single
// host CPU reading — metrics.WriteRaw's own "replace whatever is there"
// semantics (see chunk.go) let a later call simulate the agent's next flush
// landing with a different reading, at the same key.
func writeBoxSample(t *testing.T, rel release.Store, box string, cpu float64) {
	t.Helper()
	at := time.Now().UTC().Add(-time.Minute)
	if err := metrics.WriteRaw(context.Background(), rel, metrics.RawChunk{
		V: 1, Box: box, Hour: metrics.HourOf(at),
		Samples: []metrics.Sample{{At: at, Host: metrics.HostSample{CPUFrac: p(cpu)}}},
	}); err != nil {
		t.Fatal(err)
	}
}

// slowCountingStore wraps a release.Store, counting List/Get calls and
// optionally sleeping getDelay before each Get. It is the fixture the
// stale-while-revalidate tests use to (a) prove a background sweep never
// makes its trigger's caller wait for it and (b) count how many sweeps
// actually ran.
type slowCountingStore struct {
	release.Store
	getDelay time.Duration
	lists    atomic.Int64
	gets     atomic.Int64
	// listGate, when non-nil, is received from (blocking) before every List
	// call proceeds to the underlying store. discoverBoxes's List is the
	// very FIRST store operation any sweep makes (before any Get), so
	// gating it deterministically freezes a background sweep before it
	// touches the store at all — letting a test assert "the synchronous
	// caller performed zero store reads" without racing against how fast
	// the background goroutine happens to get scheduled. Set only from the
	// test goroutine strictly before the call that triggers the sweep, so
	// there is a happens-before edge (via the goroutine-creation rule) to
	// whichever goroutine reads it — never written concurrently with a read.
	// A CLOSED gate never blocks again (a receive on a closed channel
	// returns immediately), so once released it stays released for any
	// later List call in the same test, gated or not.
	listGate chan struct{}
	// listErr, when set, is what every List call returns (after clearing
	// listGate, if any) instead of touching the underlying store — the
	// fixture for simulating an outage that starts exactly when a gated
	// sweep is released, so both that sweep AND whatever sweep a coalesced
	// waiter runs on its own afterward see the same failure.
	listErr error
}

func (s *slowCountingStore) List(ctx context.Context, prefix string) ([]string, error) {
	s.lists.Add(1)
	if s.listGate != nil {
		<-s.listGate
	}
	if s.listErr != nil {
		return nil, s.listErr
	}
	return s.Store.List(ctx, prefix)
}

func (s *slowCountingStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	s.gets.Add(1)
	if s.getDelay > 0 {
		time.Sleep(s.getDelay)
	}
	return s.Store.Get(ctx, key)
}

// newSlowMetricsFixture is newMetricsFixture with the server reading through
// a slowCountingStore, for tests that need to observe how many sweeps ran or
// need a sweep slow enough to prove a caller didn't wait on it.
func newSlowMetricsFixture(t *testing.T, getDelay time.Duration) (*Server, *slowCountingStore) {
	t.Helper()
	st, err := Open(filepath.Join(t.TempDir(), "console.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { st.Close() })
	dir := release.NewDirStore(filepath.Join(t.TempDir(), "bucket"))
	slow := &slowCountingStore{Store: dir, getDelay: getDelay}
	srv, err := NewServer(Config{Releases: slow}, st)
	if err != nil {
		t.Fatal(err)
	}
	return srv, slow
}

// waitUntilNotSweeping polls the cache's sweeping flag directly (same
// package, no exported seam needed) until a background sweep finishes or
// the deadline passes.
func waitUntilNotSweeping(t *testing.T, srv *Server, timeout time.Duration) {
	t.Helper()
	deadline := time.Now().Add(timeout)
	for {
		srv.metricsCache.mu.Lock()
		sweeping := srv.metricsCache.sweeping
		srv.metricsCache.mu.Unlock()
		if !sweeping {
			return
		}
		if time.Now().After(deadline) {
			t.Fatalf("background sweep did not finish within %v", timeout)
		}
		time.Sleep(10 * time.Millisecond)
	}
}

// metricsStaleTestBackdate is how far the tests below wind c.at back to
// simulate an expired-but-servable cache: comfortably past metricsCacheTTL
// (60s) so the cache is definitely STALE, and comfortably short of
// metricsStaleServeCeiling (2min) so it's just as definitely still within
// serving range — 30s of margin on each side, generous enough that neither
// boundary is at risk from scheduling jitter.
const metricsStaleTestBackdate = metricsCacheTTL + 30*time.Second

// TestLoadMetricsStaleWhileRevalidate is fix B's core behaviour change: an
// expired-but-still-within-ceiling cache is no longer a reason to make the
// caller wait. The first call after expiry gets the OLD data back
// immediately — proven by a zero Get-count delta on the store, not a wall-
// clock guess — while a sweep runs in the background; once that sweep
// lands, a LATER call sees the NEW data, still without ever blocking.
func TestLoadMetricsStaleWhileRevalidate(t *testing.T) {
	const getDelay = 80 * time.Millisecond
	srv, store := newSlowMetricsFixture(t, getDelay)
	ctx := context.Background()

	writeBoxSample(t, store, "box-a", 0.5)
	boxes, _, err := srv.LoadMetrics(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if len(boxes) != 1 || boxes[0].CPU == nil || *boxes[0].CPU != 0.5 {
		t.Fatalf("initial cold sweep = %+v, want box-a at 0.5", boxes)
	}

	// Expire the cache (within the ceiling) and change what the store
	// holds. From here on a FRESH sweep would see 0.9 — the immediate call
	// below must still answer with the cache's last-known 0.5, proving it
	// didn't wait for a fresh read.
	srv.metricsCache.mu.Lock()
	srv.metricsCache.at = time.Now().Add(-metricsStaleTestBackdate)
	srv.metricsCache.mu.Unlock()
	writeBoxSample(t, store, "box-a", 0.9)

	// Freeze the background sweep this call is about to launch at its very
	// first store operation (discoverBoxes's List, before any Get) — this
	// makes the zero-Gets assertion below deterministic rather than a race
	// against the background goroutine's scheduling.
	store.listGate = make(chan struct{})
	getsBefore := store.gets.Load()
	boxes, _, err = srv.LoadMetrics(ctx)
	getsAfter := store.gets.Load()
	if err != nil {
		t.Fatal(err)
	}
	if len(boxes) != 1 || boxes[0].CPU == nil || *boxes[0].CPU != 0.5 {
		t.Fatalf("stale call = %+v, want the OLD 0.5 (a background refresh must not be awaited)", boxes)
	}
	if getsAfter != getsBefore {
		t.Fatalf("stale call performed %d store Gets, want 0 — it must return the cache without touching the store at all", getsAfter-getsBefore)
	}
	close(store.listGate) // release the frozen background sweep to finish

	waitUntilNotSweeping(t, srv, 5*time.Second)

	boxes, _, err = srv.LoadMetrics(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if len(boxes) != 1 || boxes[0].CPU == nil || *boxes[0].CPU != 0.9 {
		t.Fatalf("after the background sweep landed = %+v, want the NEW 0.9", boxes)
	}
}

// TestLoadMetricsStaleCacheSingleFlight: several callers arriving on an
// EXPIRED-BUT-WITHIN-CEILING (not cold) cache at once must trigger exactly
// one background sweep between them, never one each and never zero. The
// baseline — however many List calls one full sweepAll makes (discoverBoxes
// plus the alerts pass) — is measured from the initial cold sweep rather
// than hard-coded, so this test doesn't have to know sweepAll's internals to
// prove "exactly one more sweep happened."
func TestLoadMetricsStaleCacheSingleFlight(t *testing.T) {
	srv, store := newSlowMetricsFixture(t, 30*time.Millisecond)
	ctx := context.Background()
	writeBoxSample(t, store, "box-a", 0.5)

	// Measured as a delta across just this call (not the cumulative total),
	// so any List traffic from fixture/server construction before this
	// point doesn't get folded into the "one sweep" baseline below.
	beforeCold := store.lists.Load()
	if _, _, err := srv.LoadMetrics(ctx); err != nil {
		t.Fatal(err)
	}
	oneSweepLists := store.lists.Load() - beforeCold // however many List calls one sweepAll makes

	srv.metricsCache.mu.Lock()
	srv.metricsCache.at = time.Now().Add(-metricsStaleTestBackdate)
	srv.metricsCache.mu.Unlock()
	before := store.lists.Load()

	const callers = 8
	var wg sync.WaitGroup
	for i := 0; i < callers; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			if _, _, err := srv.LoadMetrics(ctx); err != nil {
				t.Error(err)
			}
		}()
	}
	wg.Wait()

	waitUntilNotSweeping(t, srv, 5*time.Second)

	if got := store.lists.Load() - before; got != oneSweepLists {
		t.Fatalf("List calls triggered by %d concurrent callers on a stale cache = %d, want exactly %d (one sweep, shared)", callers, got, oneSweepLists)
	}
}

// panicListStore panics on every List call — the fixture for proving a
// panicking BACKGROUND sweep is recovered (startBackgroundSweep's explicit
// recover — a detached goroutine has no net/http per-connection recover
// sitting above it the way a request's own goroutine does) rather than
// taking the whole process down, and that the cache is left usable
// afterward rather than wedged.
type panicListStore struct {
	release.Store
}

func (panicListStore) List(context.Context, string) ([]string, error) {
	panic("simulated bucket panic")
}

// TestLoadMetricsBackgroundSweepPanicRecovers: a background sweep
// (triggered by a stale-but-within-ceiling cache) that panics must not crash
// the test process, must leave the stale cache intact for its trigger's
// caller (no error), must clear sweeping so nothing is wedged, and must not
// prevent a later, healthy sweep from succeeding.
func TestLoadMetricsBackgroundSweepPanicRecovers(t *testing.T) {
	srv, rel := newMetricsFixture(t)
	ctx := context.Background()
	writeBoxSample(t, rel, "box-a", 0.5)

	if _, _, err := srv.LoadMetrics(ctx); err != nil {
		t.Fatal(err)
	}

	srv.metricsCache.mu.Lock()
	srv.metricsCache.at = time.Now().Add(-metricsStaleTestBackdate)
	srv.metricsCache.mu.Unlock()
	srv.cfg.Releases = panicListStore{Store: rel}

	boxes, _, err := srv.LoadMetrics(ctx)
	if err != nil {
		t.Fatalf("stale call with a panicking background sweep returned an error: %v", err)
	}
	if len(boxes) != 1 || boxes[0].CPU == nil || *boxes[0].CPU != 0.5 {
		t.Fatalf("stale call = %+v, want the OLD cached data despite the panic", boxes)
	}

	waitUntilNotSweeping(t, srv, 5*time.Second)

	srv.metricsCache.mu.Lock()
	sweeping := srv.metricsCache.sweeping
	srv.metricsCache.mu.Unlock()
	if sweeping {
		t.Fatal("cache left wedged (sweeping still true) after a panicking background sweep")
	}

	// Nothing is permanently wedged: restore a healthy store and force the
	// COLD path (c.at zeroed) for a clean, unambiguous assertion that a
	// later sweep still works.
	srv.cfg.Releases = rel
	srv.metricsCache.mu.Lock()
	srv.metricsCache.at = time.Time{}
	srv.metricsCache.mu.Unlock()

	boxes, _, err = srv.LoadMetrics(ctx)
	if err != nil {
		t.Fatalf("sweep after a prior background panic failed: %v", err)
	}
	if len(boxes) != 1 || boxes[0].CPU == nil || *boxes[0].CPU != 0.5 {
		t.Fatalf("recovered sweep = %+v, want box-a at 0.5", boxes)
	}
}

// errListStore fails every List call with a fixed error — the fixture for
// proving a background sweep's failure is only logged while its trigger's
// caller sees no error and the stale cache survives (within the ceiling),
// and that the SAME failure surfaces to a blocking caller once the cache
// ages past metricsStaleServeCeiling — restoring outage visibility rather
// than serving indefinitely-stale data with no error forever.
type errListStore struct {
	release.Store
	err error
}

func (e errListStore) List(context.Context, string) ([]string, error) {
	return nil, e.err
}

// TestLoadMetricsBackgroundSweepFailureThenCeilingSurfacesError covers both
// halves of that contract against the one failing store.
func TestLoadMetricsBackgroundSweepFailureThenCeilingSurfacesError(t *testing.T) {
	srv, rel := newMetricsFixture(t)
	ctx := context.Background()
	writeBoxSample(t, rel, "box-a", 0.5)

	if _, _, err := srv.LoadMetrics(ctx); err != nil {
		t.Fatal(err)
	}

	wantErr := errors.New("bucket had a bad day")
	srv.cfg.Releases = errListStore{Store: rel, err: wantErr}

	// Within the ceiling: the stale cache survives, the failing background
	// sweep is only logged, and this caller sees no error at all.
	srv.metricsCache.mu.Lock()
	srv.metricsCache.at = time.Now().Add(-metricsStaleTestBackdate)
	srv.metricsCache.mu.Unlock()

	boxes, _, err := srv.LoadMetrics(ctx)
	if err != nil {
		t.Fatalf("stale call with a failing background sweep returned an error: %v", err)
	}
	if len(boxes) != 1 || boxes[0].CPU == nil || *boxes[0].CPU != 0.5 {
		t.Fatalf("stale call = %+v, want the OLD cached data", boxes)
	}

	waitUntilNotSweeping(t, srv, 5*time.Second)
	srv.metricsCache.mu.Lock()
	sweeping := srv.metricsCache.sweeping
	srv.metricsCache.mu.Unlock()
	if sweeping {
		t.Fatal("cache left wedged after a failing background sweep")
	}

	// Past the ceiling: the SAME failing store must now surface its error
	// to a blocking caller instead of handing back stale data forever.
	srv.metricsCache.mu.Lock()
	srv.metricsCache.at = time.Now().Add(-(metricsStaleServeCeiling + time.Second))
	srv.metricsCache.mu.Unlock()

	if _, _, err := srv.LoadMetrics(ctx); err == nil {
		t.Fatal("call past the ceiling with a failing store returned no error")
	} else if !errors.Is(err, wantErr) {
		t.Fatalf("error = %v, want it to wrap %v", err, wantErr)
	}
}

// TestLoadMetricsPastCeilingBlocksForFreshData: a cache older than
// metricsStaleServeCeiling is no longer eligible for stale-while-revalidate
// even with a perfectly healthy store — the caller blocks on a synchronous
// sweep and gets genuinely fresh data, proven by a non-zero Get-count delta
// (the sweep ran INSIDE this call, not off to the side) rather than a
// timing guess.
func TestLoadMetricsPastCeilingBlocksForFreshData(t *testing.T) {
	srv, store := newSlowMetricsFixture(t, 0)
	ctx := context.Background()
	writeBoxSample(t, store, "box-a", 0.5)

	if _, _, err := srv.LoadMetrics(ctx); err != nil {
		t.Fatal(err)
	}

	writeBoxSample(t, store, "box-a", 0.9)
	srv.metricsCache.mu.Lock()
	srv.metricsCache.at = time.Now().Add(-(metricsStaleServeCeiling + time.Second))
	srv.metricsCache.mu.Unlock()

	getsBefore := store.gets.Load()
	boxes, _, err := srv.LoadMetrics(ctx)
	getsAfter := store.gets.Load()
	if err != nil {
		t.Fatal(err)
	}
	if getsAfter == getsBefore {
		t.Fatal("call past the ceiling performed no store Gets — it must sweep synchronously, not serve stale data")
	}
	if len(boxes) != 1 || boxes[0].CPU == nil || *boxes[0].CPU != 0.9 {
		t.Fatalf("boxes = %+v, want the FRESH 0.9 — this call must have swept, not served the stale 0.5", boxes)
	}
}

// TestLoadMetricsPastCeilingWaiterCoalescesOntoBackgroundSweep pins the one
// state-machine transition the other tests only exercise as a side effect:
// a caller landing in the blocking COLD-or-STALE-PAST-CEILING branch while a
// BACKGROUND sweep (started earlier by a within-ceiling stale caller) is
// already in flight. LoadMetrics's blocking branch waits on c.sweeping/
// c.done regardless of which kind of sweep set them — this proves that
// holds for a background sweep specifically, not just another blocking
// caller's, by gating the background sweep at its very first store call so
// the two callers' timing is deterministic rather than a race.
func TestLoadMetricsPastCeilingWaiterCoalescesOntoBackgroundSweep(t *testing.T) {
	srv, store := newSlowMetricsFixture(t, 0)
	ctx := context.Background()
	writeBoxSample(t, store, "box-a", 0.5)

	// Baseline: however many List calls one full sweepAll makes, measured
	// the same way the single-flight test does.
	beforeCold := store.lists.Load()
	if _, _, err := srv.LoadMetrics(ctx); err != nil {
		t.Fatal(err)
	}
	oneSweepLists := store.lists.Load() - beforeCold

	listsBeforeTrigger := store.lists.Load()

	// Gate the NEXT sweep at its first store call, then trigger it via a
	// within-ceiling stale call — this is startBackgroundSweep's path, not
	// the blocking one.
	store.listGate = make(chan struct{})
	srv.metricsCache.mu.Lock()
	srv.metricsCache.at = time.Now().Add(-metricsStaleTestBackdate)
	srv.metricsCache.mu.Unlock()
	writeBoxSample(t, store, "box-a", 0.9) // what the gated sweep will see once released

	if boxes, _, err := srv.LoadMetrics(ctx); err != nil {
		t.Fatal(err)
	} else if len(boxes) != 1 || boxes[0].CPU == nil || *boxes[0].CPU != 0.5 {
		t.Fatalf("triggering call = %+v, want the OLD 0.5 (background sweep just launched, now gated)", boxes)
	}

	// Age the SAME cache past the ceiling while the background sweep is
	// still stuck at the gate — exactly the state the waiter below lands in.
	srv.metricsCache.mu.Lock()
	srv.metricsCache.at = time.Now().Add(-(metricsStaleServeCeiling + time.Second))
	srv.metricsCache.mu.Unlock()

	waiterDone := make(chan struct{})
	var waiterBoxes []BoxMetricsView
	var waiterErr error
	go func() {
		defer close(waiterDone)
		waiterBoxes, _, waiterErr = srv.LoadMetrics(ctx)
	}()

	select {
	case <-waiterDone:
		t.Fatal("past-ceiling waiter returned before the gated background sweep was released")
	case <-time.After(200 * time.Millisecond):
		// still blocked, as it must be — coalesced onto the background
		// sweep's c.done rather than racing off to start its own.
	}

	close(store.listGate)

	select {
	case <-waiterDone:
	case <-time.After(5 * time.Second):
		t.Fatal("past-ceiling waiter never returned after the gate was released")
	}
	if waiterErr != nil {
		t.Fatalf("waiter error = %v, want nil", waiterErr)
	}
	if len(waiterBoxes) != 1 || waiterBoxes[0].CPU == nil || *waiterBoxes[0].CPU != 0.9 {
		t.Fatalf("waiter boxes = %+v, want the FRESH 0.9 the background sweep it coalesced onto produced", waiterBoxes)
	}
	if got := store.lists.Load() - listsBeforeTrigger; got != oneSweepLists {
		t.Fatalf("List calls across the whole gated-sweep-plus-waiter span = %d, want exactly %d (one sweep, not duplicated by the waiter)", got, oneSweepLists)
	}
	srv.metricsCache.mu.Lock()
	sweeping := srv.metricsCache.sweeping
	srv.metricsCache.mu.Unlock()
	if sweeping {
		t.Fatal("cache left wedged (sweeping still true) after the waiter returned")
	}
}

// TestLoadMetricsPastCeilingWaiterTakesOverFailedBackgroundSweep is the
// failure half of the same coalescing transition: the background sweep the
// waiter coalesced onto fails (rather than lands), so the cache stays past
// the ceiling — the waiter must then take the sweep over ITSELF (the
// c.sweeping check it started blocked on now reads false) rather than
// staying blocked or silently giving up, and surface that second attempt's
// error like any ordinary blocking caller would. Nothing may wedge: the
// flag clears either way, and a later call against a healthy store must
// still succeed.
func TestLoadMetricsPastCeilingWaiterTakesOverFailedBackgroundSweep(t *testing.T) {
	srv, store := newSlowMetricsFixture(t, 0)
	ctx := context.Background()
	writeBoxSample(t, store, "box-a", 0.5)

	if _, _, err := srv.LoadMetrics(ctx); err != nil {
		t.Fatal(err)
	}

	store.listGate = make(chan struct{})
	srv.metricsCache.mu.Lock()
	srv.metricsCache.at = time.Now().Add(-metricsStaleTestBackdate)
	srv.metricsCache.mu.Unlock()

	if _, _, err := srv.LoadMetrics(ctx); err != nil {
		t.Fatal(err)
	} // stale 0.5 returned, background sweep launched and now gated

	srv.metricsCache.mu.Lock()
	srv.metricsCache.at = time.Now().Add(-(metricsStaleServeCeiling + time.Second))
	srv.metricsCache.mu.Unlock()

	waiterDone := make(chan struct{})
	var waiterErr error
	go func() {
		defer close(waiterDone)
		_, _, waiterErr = srv.LoadMetrics(ctx)
	}()

	select {
	case <-waiterDone:
		t.Fatal("past-ceiling waiter returned before the gated background sweep was released")
	case <-time.After(200 * time.Millisecond):
	}

	// The outage starts exactly when the gate opens: the background
	// sweep's blocked List call, and any later List call (including the
	// waiter's own follow-up sweep), all see the same error from here on.
	wantErr := errors.New("bucket had a bad day")
	store.listErr = wantErr
	close(store.listGate)

	select {
	case <-waiterDone:
	case <-time.After(5 * time.Second):
		t.Fatal("past-ceiling waiter never returned after the gate was released")
	}
	if waiterErr == nil {
		t.Fatal("waiter that took over a failed background sweep returned no error")
	}
	if !errors.Is(waiterErr, wantErr) {
		t.Fatalf("waiter error = %v, want it to wrap %v", waiterErr, wantErr)
	}

	waitUntilNotSweeping(t, srv, 5*time.Second)

	// Nothing is permanently wedged: clear the outage and confirm a later
	// call against the same (now healthy) store still succeeds.
	store.listErr = nil
	boxes, _, err := srv.LoadMetrics(ctx)
	if err != nil {
		t.Fatalf("sweep after the prior failures still failed: %v", err)
	}
	if len(boxes) != 1 || boxes[0].CPU == nil || *boxes[0].CPU != 0.5 {
		t.Fatalf("recovered sweep = %+v, want box-a at 0.5", boxes)
	}
}

// errOnKeyStore fails one specific Get with a non-NotExist error and passes
// everything else through — the "bucket is answering, but this object will
// not read" case, which must fail the whole sweep rather than being skipped
// like a missing hour.
type errOnKeyStore struct {
	release.Store
	key string
	err error
}

func (e errOnKeyStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	if key == e.key {
		return nil, "", e.err
	}
	return e.Store.Get(ctx, key)
}

// newMetricsFixtureStore is newMetricsFixture with the server reading through
// a wrapped store. The returned release.Store is the raw one, for seeding.
func newMetricsFixtureStore(t *testing.T, wrap func(release.Store) release.Store) (*Server, release.Store) {
	t.Helper()
	st, err := Open(filepath.Join(t.TempDir(), "console.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { st.Close() })
	rel := release.NewDirStore(filepath.Join(t.TempDir(), "bucket"))
	srv, err := NewServer(Config{Releases: wrap(rel)}, st)
	if err != nil {
		t.Fatal(err)
	}
	return srv, rel
}

// writeChunk publishes one hour's raw chunk for box, the way the box agent
// would.
func writeChunk(t *testing.T, rel release.Store, box, hour string, samples []metrics.Sample) {
	t.Helper()
	if err := metrics.WriteRaw(context.Background(), rel, metrics.RawChunk{
		V: 1, Box: box, Hour: hour, Samples: samples,
	}); err != nil {
		t.Fatal(err)
	}
}

// seedTiedHours writes three hours for box-a whose samples deliberately SHARE
// a timestamp across different hours, plus a second box. Ties are the whole
// point: sort.Slice is not stable, so which tied sample ends up last — and so
// which reading the box's row prints as current — is decided by the order the
// hours were concatenated in, not by the sort. Returns the shared timestamp.
func seedTiedHours(t *testing.T, rel release.Store) time.Time {
	t.Helper()
	now := time.Now().UTC()
	tie := now.Truncate(time.Second)
	unit := func(cpu float64, mem int64) map[string]metrics.UnitSample {
		return map[string]metrics.UnitSample{"carlos-app@a.service": {CPUFrac: p(cpu), MemBytes: pi64(mem)}}
	}
	writeChunk(t, rel, "box-a", metrics.HourOf(now.Add(-2*time.Hour)), []metrics.Sample{
		{At: tie.Add(-2 * time.Hour), Host: metrics.HostSample{CPUFrac: p(0.10)}, Units: unit(0.11, 110)},
		{At: tie, Host: metrics.HostSample{CPUFrac: p(0.20)}, Units: unit(0.22, 220)},
	})
	writeChunk(t, rel, "box-a", metrics.HourOf(now.Add(-time.Hour)), []metrics.Sample{
		{At: tie.Add(-time.Hour), Host: metrics.HostSample{CPUFrac: p(0.30)}, Units: unit(0.33, 330)},
		{At: tie, Host: metrics.HostSample{CPUFrac: p(0.40)}, Units: unit(0.44, 440)},
	})
	writeChunk(t, rel, "box-a", metrics.HourOf(now), []metrics.Sample{
		{At: tie, Host: metrics.HostSample{CPUFrac: p(0.50)}, Units: unit(0.55, 550)},
	})
	writeChunk(t, rel, "box-b", metrics.HourOf(now), []metrics.Sample{
		{At: tie, Host: metrics.HostSample{CPUFrac: p(0.60)}},
	})
	return tie
}

// TestSweepAllDeterministicOrder pins the sweep's output shape against the
// timing of whatever reads it in: boxes come back in sorted name order, and a
// box's samples in hour order (newest hour first, as the loop counts back),
// so tied timestamps resolve the same way on every render. The pinned values
// are the serial sweep's own answers — a sweep that assembled hours or boxes
// in completion order would land on a different reading and the row would
// flap between page loads.
func TestSweepAllDeterministicOrder(t *testing.T) {
	srv, rel := newMetricsFixture(t)
	tie := seedTiedHours(t, rel)

	views, _, err := srv.sweepAll(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	if len(views) != 2 || views[0].Box != "box-a" || views[1].Box != "box-b" {
		t.Fatalf("boxes = %+v, want box-a then box-b (sorted)", views)
	}
	a := views[0]
	if !a.At.Equal(tie) {
		t.Fatalf("At = %v, want the tied timestamp %v", a.At, tie)
	}
	// Hour order decides the tie: the LAST tied sample appended is the oldest
	// hour's (0.20), not the current hour's (0.50).
	if a.CPU == nil || *a.CPU != 0.20 {
		t.Fatalf("CPU = %v, want 0.20 (the tie broken in hour order)", a.CPU)
	}
	if len(a.Units) != 1 || a.Units[0].CPUFrac == nil || *a.Units[0].CPUFrac != 0.22 {
		t.Fatalf("units = %+v, want the same sample's unit reading 0.22", a.Units)
	}
	// The sparkline coordinate string encodes the whole ordered series, so it
	// is the sharpest oracle for "the samples arrived in the same order".
	const wantSpark = "0.00,36.00 60.00,28.00 120.00,20.00 180.00,24.00 240.00,32.00"
	if a.SparkCPU != wantSpark {
		t.Fatalf("SparkCPU = %q, want %q (samples in hour order)", a.SparkCPU, wantSpark)
	}

	// Same seed, repeated sweeps: identical every time, not merely plausible
	// once.
	for i := 0; i < 20; i++ {
		got, _, err := srv.sweepAll(context.Background())
		if err != nil {
			t.Fatal(err)
		}
		if len(got) != 2 || got[0].Box != "box-a" || got[1].Box != "box-b" {
			t.Fatalf("sweep %d box order = %+v", i, got)
		}
		if got[0].CPU == nil || *got[0].CPU != *a.CPU || got[0].SparkCPU != a.SparkCPU {
			t.Fatalf("sweep %d differs: CPU=%v spark=%q, want %v / %q", i, got[0].CPU, got[0].SparkCPU, *a.CPU, a.SparkCPU)
		}
		if got[0].Units[0].SparkCPU != a.Units[0].SparkCPU || got[0].Units[0].SparkMem != a.Units[0].SparkMem {
			t.Fatalf("sweep %d unit sparklines differ: %+v", i, got[0].Units[0])
		}
	}
}

// TestSweepAllFailsOnUnreadableChunk: one hour of one box answering with
// something other than os.ErrNotExist fails the WHOLE sweep — no partial view
// is returned and nothing is cached, so the next caller retries rather than
// inheriting a box's truncated history as fact.
func TestSweepAllFailsOnUnreadableChunk(t *testing.T) {
	boom := errors.New("bucket said no")
	badKey := metrics.RawKey("box-a", metrics.HourOf(time.Now().UTC().Add(-time.Hour)))
	srv, rel := newMetricsFixtureStore(t, func(s release.Store) release.Store {
		return errOnKeyStore{Store: s, key: badKey, err: boom}
	})
	seedTiedHours(t, rel)

	if _, _, err := srv.LoadMetrics(context.Background()); err == nil {
		t.Fatal("LoadMetrics succeeded despite an unreadable chunk")
	} else {
		if !errors.Is(err, boom) {
			t.Fatalf("error = %v, want it to wrap the store's error", err)
		}
		if !strings.Contains(err.Error(), "read "+badKey) {
			t.Fatalf("error = %v, want it to name the key it failed to read", err)
		}
	}
	// A failed sweep is never cached.
	srv.metricsCache.mu.Lock()
	cachedAt := srv.metricsCache.at
	srv.metricsCache.mu.Unlock()
	if !cachedAt.IsZero() {
		t.Fatalf("a failed sweep was cached (at = %v)", cachedAt)
	}
}

// TestOwnerAppMetricsEmptyHosts: an app with no routes asks for nothing and
// gets nothing — empty, no error, and (proven by the poisoned-free fixture
// having no boxes at all) never an error path the app page has to handle.
func TestOwnerAppMetricsEmptyHosts(t *testing.T) {
	srv, _ := newMetricsFixture(t)
	rows, err := srv.OwnerAppMetrics(context.Background(), nil, nil)
	if err != nil {
		t.Fatalf("empty hosts errored: %v", err)
	}
	if len(rows) != 0 {
		t.Fatalf("empty hosts = %+v, want no rows", rows)
	}
	rows, err = srv.OwnerAppMetrics(context.Background(), []string{}, nil)
	if err != nil || len(rows) != 0 {
		t.Fatalf("empty slice = %+v, %v; want no rows, no error", rows, err)
	}
}

// TestOwnerAppMetricsFiltersToOwnUnits is the boundary at the function level:
// one box carries this owner's unit, another tenant's unit, and host-level
// readings; only the owner's unit row comes back, and a host in the allowlist
// with no unit data in the sweep comes back as an unknown (nil) row, never
// zeros and never dropped.
func TestOwnerAppMetricsFiltersToOwnUnits(t *testing.T) {
	srv, rel := newMetricsFixture(t)
	ctx := context.Background()
	// Inside metricsFreshWindow: these are the box's CURRENT readings.
	at := time.Now().UTC().Add(-time.Minute)
	if err := metrics.WriteRaw(ctx, rel, metrics.RawChunk{
		V: 1, Box: "box-secret-a", Hour: metrics.HourOf(at),
		Samples: []metrics.Sample{{
			At:   at,
			Host: metrics.HostSample{CPUFrac: p(0.777), MemFrac: p(0.888), DiskFrac: p(0.999)},
			Units: map[string]metrics.UnitSample{
				"carlos-app@owned.example.com.service":        {CPUFrac: p(0.11), MemBytes: pi64(1024)},
				"carlos-app@other-tenant.example.com.service": {CPUFrac: p(0.55), MemBytes: pi64(4096)},
				"carlos-litestream.service":                   {CPUFrac: p(0.66), MemBytes: pi64(8192)},
			},
		}},
	}); err != nil {
		t.Fatal(err)
	}

	rows, err := srv.OwnerAppMetrics(ctx, []string{"owned.example.com", "asleep.example.com"}, nil)
	if err != nil {
		t.Fatal(err)
	}
	if len(rows) != 2 {
		t.Fatalf("rows = %+v, want exactly 2 (the two allowlisted hosts)", rows)
	}
	byHost := map[string]InstanceMetrics{}
	for _, r := range rows {
		byHost[r.Host] = r
	}
	owned, ok := byHost["owned.example.com"]
	if !ok {
		t.Fatalf("owned.example.com missing from %+v", rows)
	}
	if owned.CPUFrac == nil || *owned.CPUFrac != 0.11 {
		t.Fatalf("owned CPU = %v, want 0.11", owned.CPUFrac)
	}
	if owned.MemBytes == nil || *owned.MemBytes != 1024 {
		t.Fatalf("owned mem = %v, want 1024", owned.MemBytes)
	}
	if owned.SparkCPU == "" || owned.SparkMem == "" {
		t.Fatalf("owned sparklines empty: %+v", owned)
	}
	if !owned.Reporting() {
		t.Fatalf("owned should be reporting: %+v", owned)
	}
	asleep, ok := byHost["asleep.example.com"]
	if !ok {
		t.Fatalf("asleep.example.com dropped instead of rendering as unknown: %+v", rows)
	}
	if asleep.CPUFrac != nil || asleep.MemBytes != nil {
		t.Fatalf("a host with no unit data must be nil (unknown), not zero: %+v", asleep)
	}
	if asleep.Reporting() {
		t.Fatalf("asleep must not report: %+v", asleep)
	}
	// Nothing else can enter the view: no other tenant's unit, no non-app
	// unit, no host row.
	for _, r := range rows {
		if r.Host != "owned.example.com" && r.Host != "asleep.example.com" {
			t.Fatalf("row outside the allowlist: %+v", r)
		}
		if r.CPUFrac != nil && (*r.CPUFrac == 0.55 || *r.CPUFrac == 0.66 || *r.CPUFrac == 0.777) {
			t.Fatalf("row carries another unit's or the host's reading: %+v", r)
		}
	}
}

// TestOwnerAppMetricsSurvivesHealthyFlushCycle pins metricsFreshWindow's
// derivation (flush cadence floor 5min + stale-serve ceiling 2min + 1min
// slack = 8 minutes): a healthy agent's newest VISIBLE sample can be as old
// as flush-cadence-plus-ceiling and still be current, but a sample old
// enough that NOTHING short of a dead agent explains it must still flap to
// no-recent-readings. Three ages, three different hosts/boxes (so the
// "freshest box wins" rule in OwnerAppMetrics can't let one case's fresh
// reading paper over another's staleness):
//
//   - 5m30s: the original healthy-worst-case under the OLD (5min-flush-only)
//     math — comfortably inside both the old and new window.
//   - 7m0s: the NEW healthy worst case this fix widens the window to cover
//     — flush cadence's floor (5min) plus the full stale-serve ceiling
//     (2min) a caller's cache may be behind by.
//   - 8m30s: comfortably past the 8-minute window regardless of derivation
//     — must NOT read as current.
func TestOwnerAppMetricsSurvivesHealthyFlushCycle(t *testing.T) {
	srv, rel := newMetricsFixture(t)
	ctx := context.Background()

	write := func(box, host string, age time.Duration) {
		at := time.Now().UTC().Add(-age)
		if err := metrics.WriteRaw(ctx, rel, metrics.RawChunk{
			V: 1, Box: box, Hour: metrics.HourOf(at),
			Samples: []metrics.Sample{{
				At: at,
				Units: map[string]metrics.UnitSample{
					"carlos-app@" + host + ".service": {CPUFrac: p(0.42), MemBytes: pi64(2048)},
				},
			}},
		}); err != nil {
			t.Fatal(err)
		}
	}
	write("box-healthy-1", "flush1.example.com", 5*time.Minute+30*time.Second)
	write("box-healthy-2", "flush2.example.com", 7*time.Minute)
	write("box-dead", "dead.example.com", 8*time.Minute+30*time.Second)

	rows, err := srv.OwnerAppMetrics(ctx, []string{"flush1.example.com", "flush2.example.com", "dead.example.com"}, nil)
	if err != nil {
		t.Fatal(err)
	}
	byHost := map[string]InstanceMetrics{}
	for _, r := range rows {
		byHost[r.Host] = r
	}
	if r := byHost["flush1.example.com"]; !r.Reporting() {
		t.Fatalf("a 5m30s-old sample flapped to no-recent-readings: %+v", r)
	}
	if r := byHost["flush2.example.com"]; !r.Reporting() {
		t.Fatalf("a 7m0s-old sample (flush floor + stale-serve ceiling) flapped to no-recent-readings: %+v", r)
	}
	if r := byHost["dead.example.com"]; r.Reporting() {
		t.Fatalf("an 8m30s-old sample — past the fresh window — still read as current: %+v", r)
	}
}

// TestSparkPointsInvertsYAndSkipsNil pins the coordinate math the brief
// calls out explicitly: a nil sample contributes no pair at all (not a
// zero), and a higher fraction plots at a SMALLER y (SVG's y axis points
// down, so "up" on screen means a smaller number).
func TestSparkPointsInvertsYAndSkipsNil(t *testing.T) {
	pts := sparkPoints([]*float64{p(0.1), nil, p(0.9)}, 240, 40)
	fields := strings.Fields(pts)
	if len(fields) != 2 {
		t.Fatalf("got %d coordinate pairs, want 2 (the nil sample skipped): %q", len(fields), pts)
	}
	xy := func(pair string) (x, y float64) {
		parts := strings.SplitN(pair, ",", 2)
		if len(parts) != 2 {
			t.Fatalf("malformed pair %q", pair)
		}
		xf, err := strconv.ParseFloat(parts[0], 64)
		if err != nil {
			t.Fatalf("bad x in %q: %v", pair, err)
		}
		yf, err := strconv.ParseFloat(parts[1], 64)
		if err != nil {
			t.Fatalf("bad y in %q: %v", pair, err)
		}
		return xf, yf
	}
	x0, y0 := xy(fields[0])
	x1, y1 := xy(fields[1])
	if x1 <= x0 {
		t.Fatalf("x did not advance across the skipped nil slot: x0=%v x1=%v", x0, x1)
	}
	if y1 >= y0 {
		t.Fatalf("0.9 (fields[1]) should plot ABOVE 0.1 (fields[0]) — smaller y: y0=%v y1=%v", y0, y1)
	}
}

// TestSparkPointsEmptyOrAllNil: no readable data renders no line, per the
// brief ("" when no data) — never a fabricated flat line at zero.
func TestSparkPointsEmptyOrAllNil(t *testing.T) {
	if got := sparkPoints(nil, 240, 40); got != "" {
		t.Fatalf("nil input = %q, want empty", got)
	}
	if got := sparkPoints([]*float64{}, 240, 40); got != "" {
		t.Fatalf("empty slice = %q, want empty", got)
	}
	if got := sparkPoints([]*float64{nil, nil, nil}, 240, 40); got != "" {
		t.Fatalf("all-nil input = %q, want empty", got)
	}
}

// TestSparkPointsDownsamples: 48h of minute samples is ~2880 readings into a
// 240px-wide viewBox — 12 points per pixel of invisible detail, held in the
// shared cache for every unit on every box. sparkPoints buckets down to at
// most the viewBox width, and a bucket's value is the MEAN of its non-nil
// readings, so a spike still lifts its bucket rather than vanishing.
func TestSparkPointsDownsamples(t *testing.T) {
	vals := make([]*float64, 2880)
	for i := range vals {
		vals[i] = p(0.1)
	}
	vals[1000] = p(1.0) // the spike
	pts := strings.Fields(sparkPoints(vals, 240, 40))
	if len(pts) == 0 || len(pts) > 240 {
		t.Fatalf("got %d coordinate pairs, want 1..240", len(pts))
	}
	ys := make([]float64, len(pts))
	for i, pair := range pts {
		parts := strings.SplitN(pair, ",", 2)
		if len(parts) != 2 {
			t.Fatalf("malformed pair %q", pair)
		}
		y, err := strconv.ParseFloat(parts[1], 64)
		if err != nil {
			t.Fatalf("bad y in %q: %v", pair, err)
		}
		ys[i] = y
	}
	// Exactly one bucket contains the spike, and it plots higher (smaller y)
	// than the flat 0.1 baseline every other bucket sits at.
	baseline := ys[0]
	raised := 0
	for _, y := range ys {
		if y < baseline-0.001 {
			raised++
		}
	}
	if raised != 1 {
		t.Fatalf("%d buckets rose above the baseline, want exactly 1 (the spike's): %v", raised, ys)
	}
}

// TestSparkPointsDownsampleKeepsNilUnknown: a bucket with no readable value
// at all contributes NO point — the gap stays a gap, never a zero, even
// after bucketing.
func TestSparkPointsDownsampleKeepsNilUnknown(t *testing.T) {
	vals := make([]*float64, 1000)
	for i := range vals {
		if i < 500 {
			vals[i] = p(0.5)
		}
	}
	pts := strings.Fields(sparkPoints(vals, 240, 40))
	if len(pts) == 0 || len(pts) > 120 {
		t.Fatalf("got %d pairs, want at most the ~120 buckets covering the readable half: %v", len(pts), pts)
	}
	for _, pair := range pts {
		if strings.HasSuffix(pair, ",40.00") {
			t.Fatalf("an all-nil bucket was plotted as zero: %q", pair)
		}
	}
}

// TestOwnerAppMetricsSeesExecBackedInstances pins the second allowlist key:
// an exec-backed host reports under carlos-exec@<host> (the delegated
// cgroup's verbatim name, no .service), and its readings reach the owner
// row exactly like a unit-backed tenant's. Another tenant's exec cgroup
// stays invisible — the allowlist boundary holds for the new key form too.
func TestOwnerAppMetricsSeesExecBackedInstances(t *testing.T) {
	srv, rel := newMetricsFixture(t)
	ctx := context.Background()
	at := time.Now().UTC().Add(-time.Minute)
	if err := metrics.WriteRaw(ctx, rel, metrics.RawChunk{
		V: 1, Box: "box-a", Hour: metrics.HourOf(at),
		Samples: []metrics.Sample{{
			At: at,
			Units: map[string]metrics.UnitSample{
				"carlos-exec@keymail.example.com": {CPUFrac: p(0.22), MemBytes: pi64(2048)},
				"carlos-exec@other.example.com":   {CPUFrac: p(0.44), MemBytes: pi64(4096)},
			},
		}},
	}); err != nil {
		t.Fatal(err)
	}

	rows, err := srv.OwnerAppMetrics(ctx, []string{"keymail.example.com"}, nil)
	if err != nil {
		t.Fatal(err)
	}
	if len(rows) != 1 {
		t.Fatalf("rows = %+v, want exactly the one allowlisted host", rows)
	}
	got := rows[0]
	if got.Host != "keymail.example.com" || !got.Reporting() {
		t.Fatalf("exec-backed host not reporting: %+v", got)
	}
	if got.CPUFrac == nil || *got.CPUFrac != 0.22 {
		t.Fatalf("CPU = %v, want 0.22", got.CPUFrac)
	}
	if got.MemBytes == nil || *got.MemBytes != 2048 {
		t.Fatalf("mem = %v, want 2048", got.MemBytes)
	}
	if got.SparkCPU == "" || got.SparkMem == "" {
		t.Fatalf("sparklines empty: %+v", got)
	}
}

// TestOwnerAppMetricsExecAndUnitKeysNeverDoubleReport pins the spec's
// no-double-report property: a host can't legally report under both key
// forms at once (one route, one backing), but if a sweep ever carries both
// — a backing flip mid-window, two boxes — the existing freshest-box rule
// picks ONE deterministically. No special case, still exactly one row.
func TestOwnerAppMetricsExecAndUnitKeysNeverDoubleReport(t *testing.T) {
	srv, rel := newMetricsFixture(t)
	ctx := context.Background()
	fresher := time.Now().UTC().Add(-time.Minute)
	older := fresher.Add(-2 * time.Minute)
	// The older box reports the host as a systemd unit, the fresher box as
	// an exec cgroup — the flip-mid-window shape.
	if err := metrics.WriteRaw(ctx, rel, metrics.RawChunk{
		V: 1, Box: "box-old", Hour: metrics.HourOf(older),
		Samples: []metrics.Sample{{
			At: older,
			Units: map[string]metrics.UnitSample{
				"carlos-app@flip.example.com.service": {CPUFrac: p(0.99), MemBytes: pi64(9999)},
			},
		}},
	}); err != nil {
		t.Fatal(err)
	}
	if err := metrics.WriteRaw(ctx, rel, metrics.RawChunk{
		V: 1, Box: "box-new", Hour: metrics.HourOf(fresher),
		Samples: []metrics.Sample{{
			At: fresher,
			Units: map[string]metrics.UnitSample{
				"carlos-exec@flip.example.com": {CPUFrac: p(0.11), MemBytes: pi64(1111)},
			},
		}},
	}); err != nil {
		t.Fatal(err)
	}

	rows, err := srv.OwnerAppMetrics(ctx, []string{"flip.example.com"}, nil)
	if err != nil {
		t.Fatal(err)
	}
	if len(rows) != 1 {
		t.Fatalf("rows = %+v, want exactly one row for the host, never two", rows)
	}
	got := rows[0]
	if got.Box != "box-new" {
		t.Fatalf("Box = %q, want the freshest box-new: %+v", got.Box, got)
	}
	if got.CPUFrac == nil || *got.CPUFrac != 0.11 {
		t.Fatalf("CPU = %v, want the freshest box's 0.11", got.CPUFrac)
	}
}

// TestOwnerAppMetricsMatchesStampedUnit pins the third allowlist key: an
// operator-stamped bespoke unit (registry Route.Unit, #110) reports under
// its real systemd unit name, not carlos-app@<host>.service — and the
// sweep can only match it when the caller opts a host into that name via
// unitByHost. Without the entry the same host stays unreached, proving the
// third key is opt-in per host, not a wildcard that reopens the boundary.
func TestOwnerAppMetricsMatchesStampedUnit(t *testing.T) {
	srv, rel := newMetricsFixture(t)
	ctx := context.Background()
	at := time.Now().UTC().Add(-time.Minute)
	if err := metrics.WriteRaw(ctx, rel, metrics.RawChunk{
		V: 1, Box: "b1", Hour: metrics.HourOf(at),
		Samples: []metrics.Sample{{
			At: at,
			Units: map[string]metrics.UnitSample{
				"titogo@vicky.service": {CPUFrac: p(0.33), MemBytes: pi64(3333)},
			},
		}},
	}); err != nil {
		t.Fatal(err)
	}

	rows, err := srv.OwnerAppMetrics(ctx, []string{"vicky.go.tito.io"},
		map[string]string{"vicky.go.tito.io": "titogo@vicky.service"})
	if err != nil {
		t.Fatal(err)
	}
	if len(rows) != 1 {
		t.Fatalf("rows = %+v, want exactly the one allowlisted host", rows)
	}
	got := rows[0]
	if got.Host != "vicky.go.tito.io" || !got.Reporting() {
		t.Fatalf("stamped-unit host not reporting: %+v", got)
	}
	if got.Box != "b1" {
		t.Fatalf("Box = %q, want b1", got.Box)
	}
	if got.CPUFrac == nil || *got.CPUFrac != 0.33 {
		t.Fatalf("CPU = %v, want 0.33", got.CPUFrac)
	}

	// Without the unitByHost entry, the same host must NOT match — the
	// third key is opt-in per host, not a wildcard.
	rows, err = srv.OwnerAppMetrics(ctx, []string{"vicky.go.tito.io"}, nil)
	if err != nil {
		t.Fatal(err)
	}
	if len(rows) != 1 {
		t.Fatalf("rows = %+v, want exactly the one allowlisted host", rows)
	}
	if rows[0].Reporting() {
		t.Fatalf("host reported without an opt-in unitByHost entry: %+v", rows[0])
	}
}