package host

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"log"
	"math"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"

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

// TestSamplerTickAndFlush is the brief's Step-1 test, verbatim: two ticks
// across a delta (first has no CPU reading, second derives a fraction from
// both /proc/stat and a unit's cgroup usage_usec counters), then a flush
// within the hour followed by a flush across an hour boundary, checking the
// raw chunk and the day rollup it produces.
func TestSamplerTickAndFlush(t *testing.T) {
	ctx := context.Background()
	store := release.NewDirStore(t.TempDir())
	at := time.Date(2026, 8, 2, 10, 59, 30, 0, time.UTC)
	c0, c1 := &metrics.CPUCounters{Busy: 100, Total: 1000}, &metrics.CPUCounters{Busy: 160, Total: 1100}
	mem, disk := 0.5, 0.3
	usec0, usec1 := int64(0), int64(3_000_000)
	sm := &sampler{}
	probes := func(c *metrics.CPUCounters, usec int64, now time.Time) sampleProbes {
		return sampleProbes{
			procStat: func() *metrics.CPUCounters { return c },
			loadavg:  func() (*float64, *float64, *float64) { return nil, nil, nil },
			memFrac:  func() *float64 { return &mem },
			diskFrac: func() *float64 { return &disk },
			units: func() map[string]unitRaw {
				return map[string]unitRaw{"carlos-app@a.example.com.service": {cpuUsec: &usec}}
			},
			now: func() time.Time { return now },
		}
	}
	sm.tick(probes(c0, usec0, at))
	sm.tick(probes(c1, usec1, at.Add(time.Minute)))
	if len(sm.buf) != 2 {
		t.Fatalf("buf=%d, want 2", len(sm.buf))
	}
	if sm.buf[0].Host.CPUFrac != nil {
		t.Fatal("first tick has no delta — CPU must be nil, not 0")
	}
	if f := sm.buf[1].Host.CPUFrac; f == nil || math.Abs(*f-0.6) > 1e-9 {
		t.Fatalf("second tick CPU=%v, want 0.6", f)
	}
	if f := sm.buf[1].Units["carlos-app@a.example.com.service"].CPUFrac; f == nil || math.Abs(*f-0.05) > 1e-9 {
		t.Fatalf("unit CPU=%v, want 0.05 (3s of CPU over 60s)", f)
	}

	// This flush call's buf already spans the 10->11 hour boundary (one
	// sample in each hour): hour 10 is the "old" hour relative to `now` and
	// is written to its own chunk AND rolled up into the day rollup right
	// here, on this first flush — not deferred to the next one. Only hour
	// 11's sample stays buffered.
	if err := metricsFlush(ctx, store, "box-a", sm, at.Add(time.Minute)); err != nil {
		t.Fatal(err)
	}
	if _, _, err := store.Get(ctx, "metrics/boxes/box-a/raw/2026-08-02-10.json"); err != nil {
		t.Fatalf("raw chunk missing: %v", err)
	}
	// Both samples now buffered are in hour 11: this second flush has
	// nothing to roll over, and the day rollup checked below is simply what
	// the first flush already wrote.
	sm.tick(probes(c1, usec1, at.Add(2*time.Minute))) // 11:01:30
	if err := metricsFlush(ctx, store, "box-a", sm, at.Add(2*time.Minute)); err != nil {
		t.Fatal(err)
	}
	body, _, err := store.Get(ctx, "metrics/boxes/box-a/days/2026-08-02.json")
	if err != nil {
		t.Fatalf("day rollup missing: %v", err)
	}
	var day metrics.DayRollup
	if err := json.Unmarshal(body, &day); err != nil {
		t.Fatal(err)
	}
	if _, ok := day.Hours["10"]; !ok {
		t.Fatalf("hour 10 not rolled up: %v", day.Hours)
	}
}

// ---- Task 5 (customer-fleets plan) fix round: metrics loop honors the fleet prefix ----

// TestMetricsLoopStoreHonorsFleetPrefix pins the review fix: metricsLoop's
// store must be wrapped in release.NewPrefixStore(CARLOS_STATUS_PREFIX)
// exactly like statusLoop's, so a fleet box's raw samples, day rollup, and
// alert state land under its own fleets/<name>/ namespace — not the shared
// metrics/boxes/ + alerts/boxes/ paths a platform box uses. Without this, a
// customer's Pi would surface on the platform operator's monitoring page
// (console metricsview lists metrics/boxes/) and a threshold breach would
// page the platform's own alert webhook for customer hardware. Exercises
// metricsFlush and alertTick through a release.NewPrefixStore-wrapped
// store — the same composition metricsLoop's store-construction line now
// produces — rather than reimplementing the wrapping.
func TestMetricsLoopStoreHonorsFleetPrefix(t *testing.T) {
	ctx := context.Background()
	under := release.NewDirStore(t.TempDir())
	s := release.NewPrefixStore(under, "fleets/pi/")
	box := "pi-box"

	// --- metricsFlush: raw chunk + day rollup land under the prefix -------
	sm := &sampler{}
	at := time.Date(2026, 8, 2, 10, 59, 30, 0, time.UTC)
	mem, disk := 0.5, 0.3
	probes := func(now time.Time) sampleProbes {
		return sampleProbes{
			procStat: func() *metrics.CPUCounters { return nil },
			loadavg:  func() (*float64, *float64, *float64) { return nil, nil, nil },
			memFrac:  func() *float64 { return &mem },
			diskFrac: func() *float64 { return &disk },
			units:    func() map[string]unitRaw { return nil },
			now:      func() time.Time { return now },
		}
	}
	sm.tick(probes(at))
	sm.tick(probes(at.Add(2 * time.Minute))) // crosses the 10 -> 11 hour boundary
	if err := metricsFlush(ctx, s, box, sm, at.Add(2*time.Minute)); err != nil {
		t.Fatal(err)
	}
	if _, _, err := under.Get(ctx, "fleets/pi/metrics/boxes/pi-box/raw/2026-08-02-10.json"); err != nil {
		t.Errorf("raw chunk not under the fleet prefix: %v", err)
	}
	if _, _, err := under.Get(ctx, "metrics/boxes/pi-box/raw/2026-08-02-10.json"); err == nil {
		t.Error("raw chunk leaked into the shared (unprefixed) metrics namespace")
	}
	if _, _, err := under.Get(ctx, "fleets/pi/metrics/boxes/pi-box/days/2026-08-02.json"); err != nil {
		t.Errorf("day rollup not under the fleet prefix: %v", err)
	}

	// --- alertTick: published alert state lands under the prefix too ------
	rules := metrics.DefaultRules()
	st := &metrics.AlertState{V: 1, Box: box, Rules: map[string]metrics.RuleState{}}
	firing := 0.95
	latest := metrics.Sample{At: at, Host: metrics.HostSample{DiskFrac: &firing}}
	statePath := filepath.Join(t.TempDir(), "monitor", "alerts.json")
	if err := alertTick(ctx, s, box, rules, st, latest, statePath, nil); err != nil {
		t.Fatal(err)
	}
	if _, _, err := under.Get(ctx, "fleets/pi/"+metrics.AlertKey(box)); err != nil {
		t.Errorf("alert state not under the fleet prefix: %v", err)
	}
	if _, _, err := under.Get(ctx, metrics.AlertKey(box)); err == nil {
		t.Error("alert state leaked into the shared (unprefixed) alerts namespace — would page the platform webhook for customer hardware")
	}

	// --- empty prefix (every platform box today) is a pure passthrough ----
	if got := release.NewPrefixStore(under, ""); got != release.Store(under) {
		t.Errorf("NewPrefixStore with empty prefix = %v, want the underlying store unchanged", got)
	}
}

// TestMetricsFlushPrunesOldChunks pins the 48h prune: a raw chunk seeded
// well past the retention window must be gone after a flush that crosses an
// hour boundary (the only time pruning runs), while a fresh chunk survives.
func TestMetricsFlushPrunesOldChunks(t *testing.T) {
	ctx := context.Background()
	store := release.NewDirStore(t.TempDir())
	box := "box-a"

	oldHour := time.Date(2026, 7, 30, 9, 0, 0, 0, time.UTC) // 3 days before "now" below
	if err := metrics.WriteRaw(ctx, store, metrics.RawChunk{
		V: 1, Box: box, Hour: metrics.HourOf(oldHour), Samples: []metrics.Sample{{At: oldHour}},
	}); err != nil {
		t.Fatal(err)
	}
	freshHour := time.Date(2026, 8, 2, 9, 0, 0, 0, time.UTC)
	if err := metrics.WriteRaw(ctx, store, metrics.RawChunk{
		V: 1, Box: box, Hour: metrics.HourOf(freshHour), Samples: []metrics.Sample{{At: freshHour}},
	}); err != nil {
		t.Fatal(err)
	}

	// A sampler buffer holding one sample from the hour just before "now" —
	// this is what triggers the hour-rollover branch (rollup + prune).
	sm := &sampler{buf: []metrics.Sample{
		{At: time.Date(2026, 8, 2, 9, 30, 0, 0, time.UTC)},
	}}
	now := time.Date(2026, 8, 2, 10, 0, 5, 0, time.UTC)
	if err := metricsFlush(ctx, store, box, sm, now); err != nil {
		t.Fatal(err)
	}

	if _, _, err := store.Get(ctx, metrics.RawKey(box, metrics.HourOf(oldHour))); !errors.Is(err, os.ErrNotExist) {
		t.Errorf("old raw chunk should be pruned, got err=%v", err)
	}
	if _, _, err := store.Get(ctx, metrics.RawKey(box, metrics.HourOf(freshHour))); err != nil {
		t.Errorf("fresh raw chunk should survive pruning: %v", err)
	}
}

// TestAlertTickPersistsPublishesAndPosts pins the fire path: a disk reading
// over threshold persists local state, publishes the new AlertState to the
// store, and posts exactly one webhook event for the transition — a second,
// identical tick (still firing, no transition) must not post again.
func TestAlertTickPersistsPublishesAndPosts(t *testing.T) {
	ctx := context.Background()
	store := release.NewDirStore(t.TempDir())
	statePath := filepath.Join(t.TempDir(), "monitor", "alerts.json")
	box := "box-a"
	rules := metrics.DefaultRules()
	st := &metrics.AlertState{V: 1, Box: box, Rules: map[string]metrics.RuleState{}}

	disk := 0.95
	latest := metrics.Sample{
		At:   time.Date(2026, 8, 2, 12, 0, 0, 0, time.UTC),
		Host: metrics.HostSample{DiskFrac: &disk},
	}

	var posts [][]byte
	post := func(url string, body []byte) error {
		posts = append(posts, body)
		return nil
	}

	if err := alertTick(ctx, store, box, rules, st, latest, statePath, webhookSinks("http://example.test/webhook", post)); err != nil {
		t.Fatal(err)
	}

	if _, err := os.Stat(statePath); err != nil {
		t.Fatalf("state file not written: %v", err)
	}
	body, _, err := store.Get(ctx, metrics.AlertKey(box))
	if err != nil {
		t.Fatalf("published alert state missing: %v", err)
	}
	var published metrics.AlertState
	if err := json.Unmarshal(body, &published); err != nil {
		t.Fatal(err)
	}
	if !published.Rules["disk"].Firing {
		t.Errorf("disk rule not firing in published state: %+v", published.Rules["disk"])
	}
	if len(posts) != 1 {
		t.Fatalf("posts = %d, want 1", len(posts))
	}
	if !strings.Contains(string(posts[0]), `"rule":"disk"`) {
		t.Errorf("post body = %s, want it to contain the disk rule", posts[0])
	}

	// Second, identical tick: already firing, no new transition -> no post.
	if err := alertTick(ctx, store, box, rules, st, latest, statePath, webhookSinks("http://example.test/webhook", post)); err != nil {
		t.Fatal(err)
	}
	if len(posts) != 1 {
		t.Errorf("posts after second tick = %d, want still 1 (no new transition)", len(posts))
	}
}

// TestAlertTickWebhookFailureDoesNotBlock pins the fire-and-forget webhook
// contract: a post failure is logged, never returned — state must still be
// persisted locally and published to the store.
func TestAlertTickWebhookFailureDoesNotBlock(t *testing.T) {
	ctx := context.Background()
	store := release.NewDirStore(t.TempDir())
	statePath := filepath.Join(t.TempDir(), "monitor", "alerts.json")
	box := "box-a"
	rules := metrics.DefaultRules()
	st := &metrics.AlertState{V: 1, Box: box, Rules: map[string]metrics.RuleState{}}

	disk := 0.95
	latest := metrics.Sample{
		At:   time.Date(2026, 8, 2, 12, 0, 0, 0, time.UTC),
		Host: metrics.HostSample{DiskFrac: &disk},
	}
	post := func(url string, body []byte) error { return errors.New("webhook unreachable") }

	if err := alertTick(ctx, store, box, rules, st, latest, statePath, webhookSinks("http://example.test/webhook", post)); err != nil {
		t.Fatalf("alertTick must not fail on a webhook error: %v", err)
	}
	if _, err := os.Stat(statePath); err != nil {
		t.Errorf("state file must still be written: %v", err)
	}
	if _, _, err := store.Get(ctx, metrics.AlertKey(box)); err != nil {
		t.Errorf("state must still be published: %v", err)
	}
}

// TestAlertTickPersistFailureDoesNotSuppressPublish pins the review fix: a
// failed local state write (statePath's parent is a file, not a directory,
// so os.MkdirAll fails deterministically) must not suppress the alert
// itself — the store publish and the webhook post are the reliable
// channels, and the local write is most likely to fail exactly when the
// disk rule is the one firing (disk full).
func TestAlertTickPersistFailureDoesNotSuppressPublish(t *testing.T) {
	ctx := context.Background()
	store := release.NewDirStore(t.TempDir())
	box := "box-a"
	rules := metrics.DefaultRules()
	st := &metrics.AlertState{V: 1, Box: box, Rules: map[string]metrics.RuleState{}}

	// statePath's parent directory is actually a file: MkdirAll must fail.
	blocker := filepath.Join(t.TempDir(), "not-a-dir")
	if err := os.WriteFile(blocker, []byte("x"), 0o644); err != nil {
		t.Fatal(err)
	}
	statePath := filepath.Join(blocker, "monitor", "alerts.json")

	disk := 0.95
	latest := metrics.Sample{
		At:   time.Date(2026, 8, 2, 12, 0, 0, 0, time.UTC),
		Host: metrics.HostSample{DiskFrac: &disk},
	}
	var posted bool
	post := func(url string, body []byte) error {
		posted = true
		return nil
	}

	if err := alertTick(ctx, store, box, rules, st, latest, statePath, webhookSinks("http://example.test/webhook", post)); err != nil {
		t.Fatalf("alertTick must not fail just because the local state write failed: %v", err)
	}
	if !posted {
		t.Error("webhook must still be posted when only the local persist failed")
	}
	body, _, err := store.Get(ctx, metrics.AlertKey(box))
	if err != nil {
		t.Fatalf("alert state must still be published to the store: %v", err)
	}
	var published metrics.AlertState
	if err := json.Unmarshal(body, &published); err != nil {
		t.Fatal(err)
	}
	if !published.Rules["disk"].Firing {
		t.Errorf("disk rule not firing in published state: %+v", published.Rules["disk"])
	}
}

// failingPutStore wraps a release.Store and makes every Put fail — the
// fixture for TestSamplerBufCappedUnderSustainedFlushFailure's "IAM allows
// PutObject... no wait, denies it" scenario: reads and Deletes still work
// (delegated via embedding), only Put is broken.
type failingPutStore struct {
	release.Store
}

func (f *failingPutStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	return "", errors.New("put always fails (test)")
}

// TestSamplerBufCappedUnderSustainedFlushFailure pins the review fix: a
// store whose Put always fails means metricsFlush never gets far enough to
// trim sm.buf (it returns early on the first failing WriteRaw), so without
// a cap, buf would grow one sample per tick forever. tick itself enforces
// samplerMaxBuf regardless of flush's fate; this ticks well past the cap,
// confirms it held, confirms a failing flush leaves buf untouched, and then
// confirms a subsequent flush against a real, working store still produces
// a coherent (single-hour) chunk from whatever survived in buf.
func TestSamplerBufCappedUnderSustainedFlushFailure(t *testing.T) {
	ctx := context.Background()
	box := "box-a"
	realStore := release.NewDirStore(t.TempDir())
	failStore := &failingPutStore{Store: realStore}

	sm := &sampler{}
	start := time.Date(2026, 8, 2, 0, 0, 0, 0, time.UTC)
	mem, disk := 0.5, 0.3
	for i := 0; i < samplerMaxBuf+30; i++ {
		now := start.Add(time.Duration(i) * time.Minute)
		sm.tick(sampleProbes{
			procStat: func() *metrics.CPUCounters { return nil },
			loadavg:  func() (*float64, *float64, *float64) { return nil, nil, nil },
			memFrac:  func() *float64 { return &mem },
			diskFrac: func() *float64 { return &disk },
			units:    func() map[string]unitRaw { return nil },
			now:      func() time.Time { return now },
		})
	}
	if len(sm.buf) != samplerMaxBuf {
		t.Fatalf("buf=%d after %d ticks, want capped at %d", len(sm.buf), samplerMaxBuf+30, samplerMaxBuf)
	}

	lastAt := sm.buf[len(sm.buf)-1].At
	if err := metricsFlush(ctx, failStore, box, sm, lastAt); err == nil {
		t.Fatal("expected metricsFlush to fail against a Put-always-fails store")
	}
	if len(sm.buf) != samplerMaxBuf {
		t.Fatalf("buf=%d after a failed flush, want still capped at %d (flush must not further grow or corrupt it)", len(sm.buf), samplerMaxBuf)
	}

	// Recovery: a flush against a real, working store must still succeed
	// and write a coherent (single-hour) chunk from what's left in buf.
	if err := metricsFlush(ctx, realStore, box, sm, lastAt); err != nil {
		t.Fatalf("flush against a working store: %v", err)
	}
	body, _, err := realStore.Get(ctx, metrics.RawKey(box, metrics.HourOf(lastAt)))
	if err != nil {
		t.Fatalf("current-hour chunk missing after recovery flush: %v", err)
	}
	var chunk metrics.RawChunk
	if err := json.Unmarshal(body, &chunk); err != nil {
		t.Fatal(err)
	}
	if len(chunk.Samples) == 0 {
		t.Fatal("recovered chunk has no samples")
	}
	for _, samp := range chunk.Samples {
		if metrics.HourOf(samp.At) != metrics.HourOf(lastAt) {
			t.Errorf("chunk mixes in a sample from a different hour: %v", samp.At)
		}
	}
}

// TestMetricsFlushMergesExistingChunkAfterRestart pins the review fix: a
// restart resets sm.buf to empty, but the still-open current hour's raw
// chunk may already hold samples this same box wrote before the crash.
// WriteRaw is a wholesale replace, so the first flush after a restart must
// merge the pre-existing chunk's samples in rather than clobber them — that
// loss would otherwise become permanent the moment the hour rolls up.
func TestMetricsFlushMergesExistingChunkAfterRestart(t *testing.T) {
	ctx := context.Background()
	store := release.NewDirStore(t.TempDir())
	box := "box-a"

	hourStart := time.Date(2026, 8, 2, 14, 0, 0, 0, time.UTC)
	preRestart := []metrics.Sample{
		{At: hourStart.Add(1 * time.Minute)},
		{At: hourStart.Add(2 * time.Minute)},
	}
	if err := metrics.WriteRaw(ctx, store, metrics.RawChunk{
		V: 1, Box: box, Hour: metrics.HourOf(hourStart), Samples: preRestart,
	}); err != nil {
		t.Fatal(err)
	}

	// A fresh sampler, as if the process just restarted, with one sample
	// collected after the restart — still within the same hour.
	sm := &sampler{buf: []metrics.Sample{
		{At: hourStart.Add(3 * time.Minute)},
	}}
	now := hourStart.Add(3 * time.Minute)
	if err := metricsFlush(ctx, store, box, sm, now); err != nil {
		t.Fatal(err)
	}

	body, _, err := store.Get(ctx, metrics.RawKey(box, metrics.HourOf(hourStart)))
	if err != nil {
		t.Fatalf("chunk missing after merge flush: %v", err)
	}
	var chunk metrics.RawChunk
	if err := json.Unmarshal(body, &chunk); err != nil {
		t.Fatal(err)
	}
	if len(chunk.Samples) != 3 {
		t.Fatalf("chunk has %d samples, want 3 (2 pre-restart + 1 post-restart)", len(chunk.Samples))
	}
}

// TestMetricsFlushMergesOldHourAfterRestartAcrossBoundary pins the
// round-2 review fix: mergeExistingChunk must apply to EVERY hour this
// process writes for the first time, not just whichever hour happens to be
// "current" at the moment of the first flush. Reproduces the exact
// scenario the review found: 58 pre-crash samples for hour 09, a restart
// whose very first flush already spans the 09->10 boundary (so hour 09
// lands in metricsFlush's "old" bucket on this process's first-ever write
// to it) — the hour-09 chunk must still come out merged (59 samples, not
// 2), and the day rollup's "09" entry must be computed from that merged
// set, not just the post-restart tick.
func TestMetricsFlushMergesOldHourAfterRestartAcrossBoundary(t *testing.T) {
	ctx := context.Background()
	store := release.NewDirStore(t.TempDir())
	box := "box-a"

	hour09Start := time.Date(2026, 8, 2, 9, 0, 0, 0, time.UTC)
	preCrashCPU := 0.1
	preCrash := make([]metrics.Sample, 0, 58)
	for i := 0; i < 58; i++ {
		v := preCrashCPU
		preCrash = append(preCrash, metrics.Sample{
			At:   hour09Start.Add(time.Duration(i) * time.Minute),
			Host: metrics.HostSample{CPUFrac: &v},
		})
	}
	if err := metrics.WriteRaw(ctx, store, metrics.RawChunk{
		V: 1, Box: box, Hour: metrics.HourOf(hour09Start), Samples: preCrash,
	}); err != nil {
		t.Fatal(err)
	}

	// Restart at 09:58: a fresh sampler whose buf, by the time of its
	// first-ever flush, already holds one hour-09 tick and one hour-10
	// tick — the flush call itself spans the boundary.
	postRestartCPU := 0.2
	sm := &sampler{buf: []metrics.Sample{
		{At: time.Date(2026, 8, 2, 9, 58, 0, 0, time.UTC), Host: metrics.HostSample{CPUFrac: &postRestartCPU}},
		{At: time.Date(2026, 8, 2, 10, 3, 0, 0, time.UTC), Host: metrics.HostSample{CPUFrac: &postRestartCPU}},
	}}
	now := time.Date(2026, 8, 2, 10, 3, 0, 0, time.UTC)
	if err := metricsFlush(ctx, store, box, sm, now); err != nil {
		t.Fatal(err)
	}

	body, _, err := store.Get(ctx, metrics.RawKey(box, metrics.HourOf(hour09Start)))
	if err != nil {
		t.Fatalf("hour-09 chunk missing: %v", err)
	}
	var chunk metrics.RawChunk
	if err := json.Unmarshal(body, &chunk); err != nil {
		t.Fatal(err)
	}
	if len(chunk.Samples) != 59 {
		t.Fatalf("hour-09 chunk has %d samples, want 59 (58 pre-crash + 1 post-restart) — the pre-crash samples were clobbered", len(chunk.Samples))
	}

	dayBody, _, err := store.Get(ctx, metrics.DayKey(box, "2026-08-02"))
	if err != nil {
		t.Fatalf("day rollup missing: %v", err)
	}
	var day metrics.DayRollup
	if err := json.Unmarshal(dayBody, &day); err != nil {
		t.Fatal(err)
	}
	if n := day.Hours["09"].Host["cpu"].N; n != 59 {
		t.Fatalf("day rollup hour 09 cpu N=%d, want 59 — the rollup must be computed from the merged set, not just the post-restart tick", n)
	}
}

// TestAlertTickRetriesTransitionAfterPublishFailure pins the round-2 review
// fix: a publish failure must NOT advance *st, so the next identical
// observation reproduces the same transition and retries both the store
// publish and the webhook post — committing *st unconditionally would
// otherwise make Evaluate believe the rule is already firing, emit no
// further events, and permanently swallow the alert.
func TestAlertTickRetriesTransitionAfterPublishFailure(t *testing.T) {
	ctx := context.Background()
	box := "box-a"
	rules := metrics.DefaultRules()
	st := &metrics.AlertState{V: 1, Box: box, Rules: map[string]metrics.RuleState{}}
	statePath := filepath.Join(t.TempDir(), "monitor", "alerts.json")

	disk := 0.95
	latest := metrics.Sample{
		At:   time.Date(2026, 8, 2, 12, 0, 0, 0, time.UTC),
		Host: metrics.HostSample{DiskFrac: &disk},
	}

	posts := 0
	post := func(url string, body []byte) error {
		posts++
		return nil
	}

	failStore := &failingPutStore{Store: release.NewDirStore(t.TempDir())}
	if err := alertTick(ctx, failStore, box, rules, st, latest, statePath, webhookSinks("http://example.test/webhook", post)); err == nil {
		t.Fatal("expected alertTick to return the publish error against a Put-always-fails store")
	}
	if st.Rules["disk"].Firing {
		t.Fatal("st must NOT be mutated to firing when publish failed — the transition must remain pending for a retry")
	}
	if posts != 1 {
		t.Errorf("posts = %d, want 1 (webhook still attempted despite the publish failure)", posts)
	}

	// Retry with the SAME observation against a working store: the pending
	// transition must reproduce, publish, and post again.
	workingStore := release.NewDirStore(t.TempDir())
	if err := alertTick(ctx, workingStore, box, rules, st, latest, statePath, webhookSinks("http://example.test/webhook", post)); err != nil {
		t.Fatalf("retry against a working store: %v", err)
	}
	if !st.Rules["disk"].Firing {
		t.Fatal("st must be firing after the retried publish succeeds")
	}
	if posts != 2 {
		t.Errorf("posts = %d, want 2 (event re-emitted and re-posted on retry)", posts)
	}
	body, _, err := workingStore.Get(ctx, metrics.AlertKey(box))
	if err != nil {
		t.Fatalf("AlertKey must be published on the retry: %v", err)
	}
	var published metrics.AlertState
	if err := json.Unmarshal(body, &published); err != nil {
		t.Fatal(err)
	}
	if !published.Rules["disk"].Firing {
		t.Errorf("published state disk rule not firing: %+v", published.Rules["disk"])
	}
}

// putFailsOnceForKey wraps a release.Store and makes Put fail exactly once
// for one specific key — the fixture for
// TestMetricsFlushPartialFailureDoesNotRegressEarlierHour's "a later hour's
// write fails, an earlier hour's succeeded" scenario. Every other key, and
// every call to failKey after the first, passes through unchanged.
type putFailsOnceForKey struct {
	release.Store
	failKey string
	failed  bool
}

func (f *putFailsOnceForKey) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if key == f.failKey && !f.failed {
		f.failed = true
		return "", errors.New("put fails once for this key (test)")
	}
	return f.Store.Put(ctx, key, body, ifMatch)
}

// TestMetricsFlushPartialFailureDoesNotRegressEarlierHour pins the
// round-3 review's Critical fix: writtenHours must only be marked (and a
// hour retired from buf) once ALL of that hour's durable writes actually
// succeeded. Two old hours (A older, B less old) each already hold a
// pre-crash sample in the store; a freshly "restarted" sampler holds one
// post-restart sample for each in buf. The first flush call's Put fails
// only for hour B's raw key (hour A, processed first, succeeds in full).
// Before the round-3 fix, hour A would have been marked written and
// dropped from buf only via the final `sm.buf = curSamples` at the very
// end of metricsFlush, which this failure never reaches — so the retry
// would reprocess hour A too, see it (wrongly) already marked, skip the
// merge, and overwrite its correct 2-sample chunk with just the bare
// 1-sample in-memory set. This test asserts hour A's chunk survives the
// retry intact, and hour B ends up correctly merged too.
func TestMetricsFlushPartialFailureDoesNotRegressEarlierHour(t *testing.T) {
	ctx := context.Background()
	realStore := release.NewDirStore(t.TempDir())
	box := "box-a"

	hourAStart := time.Date(2026, 8, 2, 8, 0, 0, 0, time.UTC)
	hourBStart := time.Date(2026, 8, 2, 9, 0, 0, 0, time.UTC)

	cpuA := 0.1
	preA := []metrics.Sample{{At: hourAStart.Add(10 * time.Minute), Host: metrics.HostSample{CPUFrac: &cpuA}}}
	if err := metrics.WriteRaw(ctx, realStore, metrics.RawChunk{
		V: 1, Box: box, Hour: metrics.HourOf(hourAStart), Samples: preA,
	}); err != nil {
		t.Fatal(err)
	}
	cpuB := 0.2
	preB := []metrics.Sample{{At: hourBStart.Add(10 * time.Minute), Host: metrics.HostSample{CPUFrac: &cpuB}}}
	if err := metrics.WriteRaw(ctx, realStore, metrics.RawChunk{
		V: 1, Box: box, Hour: metrics.HourOf(hourBStart), Samples: preB,
	}); err != nil {
		t.Fatal(err)
	}

	// A fresh (post-restart) sampler with one post-restart tick in each old
	// hour — no current-hour sample, so the whole flush is "old hours".
	postA, postB := 0.3, 0.4
	sm := &sampler{buf: []metrics.Sample{
		{At: hourAStart.Add(20 * time.Minute), Host: metrics.HostSample{CPUFrac: &postA}},
		{At: hourBStart.Add(20 * time.Minute), Host: metrics.HostSample{CPUFrac: &postB}},
	}}
	now := time.Date(2026, 8, 2, 10, 5, 0, 0, time.UTC) // current hour 10 — both A and B are old

	hourBKey := metrics.RawKey(box, metrics.HourOf(hourBStart))
	failingStore := &putFailsOnceForKey{Store: realStore, failKey: hourBKey}

	if err := metricsFlush(ctx, failingStore, box, sm, now); err == nil {
		t.Fatal("expected the first flush to fail (hour B's Put fails once)")
	}

	// Hour A must already be fully durable: merged (2-sample) chunk.
	bodyA, _, err := realStore.Get(ctx, metrics.RawKey(box, metrics.HourOf(hourAStart)))
	if err != nil {
		t.Fatalf("hour A chunk missing after the partial failure: %v", err)
	}
	var chunkA metrics.RawChunk
	if err := json.Unmarshal(bodyA, &chunkA); err != nil {
		t.Fatal(err)
	}
	if len(chunkA.Samples) != 2 {
		t.Fatalf("hour A chunk has %d samples after the partial failure, want 2 (merged)", len(chunkA.Samples))
	}

	// Retry: hour B's Put now succeeds.
	if err := metricsFlush(ctx, failingStore, box, sm, now); err != nil {
		t.Fatalf("retry flush: %v", err)
	}

	// Hour A must STILL be the merged 2-sample chunk — the retry must not
	// have reprocessed and regressed it.
	bodyA2, _, err := realStore.Get(ctx, metrics.RawKey(box, metrics.HourOf(hourAStart)))
	if err != nil {
		t.Fatal(err)
	}
	var chunkA2 metrics.RawChunk
	if err := json.Unmarshal(bodyA2, &chunkA2); err != nil {
		t.Fatal(err)
	}
	if len(chunkA2.Samples) != 2 {
		t.Fatalf("hour A chunk regressed to %d samples after the retry, want still 2", len(chunkA2.Samples))
	}

	// Hour B must now be correctly merged.
	bodyB, _, err := realStore.Get(ctx, metrics.RawKey(box, metrics.HourOf(hourBStart)))
	if err != nil {
		t.Fatalf("hour B chunk missing after retry: %v", err)
	}
	var chunkB metrics.RawChunk
	if err := json.Unmarshal(bodyB, &chunkB); err != nil {
		t.Fatal(err)
	}
	if len(chunkB.Samples) != 2 {
		t.Fatalf("hour B chunk has %d samples, want 2 (merged)", len(chunkB.Samples))
	}

	dayBody, _, err := realStore.Get(ctx, metrics.DayKey(box, "2026-08-02"))
	if err != nil {
		t.Fatalf("day rollup missing: %v", err)
	}
	var day metrics.DayRollup
	if err := json.Unmarshal(dayBody, &day); err != nil {
		t.Fatal(err)
	}
	if n := day.Hours["08"].Host["cpu"].N; n != 2 {
		t.Errorf("day rollup hour 08 cpu N=%d, want 2 (merged count)", n)
	}
	if n := day.Hours["09"].Host["cpu"].N; n != 2 {
		t.Errorf("day rollup hour 09 cpu N=%d, want 2 (merged count)", n)
	}
}

// TestMetricsFlushBackfillsPrevHourRollupAfterRestart pins the round-3 review
// fix: rollups were derived only from in-memory buf, so an agent that stopped
// at 09:58 and came back at 10:07 left hour 09's raw chunk — durably written
// before the stop — with no day-rollup entry, ever. Nothing else rolls that
// hour up afterwards, and the 48h prune deletes the chunk two days later, so
// the hour was permanently lost from the trend history. The first flush of a
// process now looks back one hour and rolls up the stored chunk if the day
// object lacks it.
func TestMetricsFlushBackfillsPrevHourRollupAfterRestart(t *testing.T) {
	ctx := context.Background()
	store := release.NewDirStore(t.TempDir())
	box := "box-a"

	hour09Start := time.Date(2026, 8, 2, 9, 0, 0, 0, time.UTC)
	cpu := 0.4
	preStop := make([]metrics.Sample, 0, 10)
	for i := 0; i < 10; i++ {
		v := cpu
		preStop = append(preStop, metrics.Sample{
			At:   hour09Start.Add(time.Duration(i) * time.Minute),
			Host: metrics.HostSample{CPUFrac: &v},
		})
	}
	if err := metrics.WriteRaw(ctx, store, metrics.RawChunk{
		V: 1, Box: box, Hour: metrics.HourOf(hour09Start), Samples: preStop,
	}); err != nil {
		t.Fatal(err)
	}
	// No day rollup at all: the agent stopped before hour 09 ever closed.
	if _, _, err := store.Get(ctx, metrics.DayKey(box, "2026-08-02")); err == nil {
		t.Fatal("fixture invalid: a day rollup already exists")
	}

	// Restart at 10:07: a fresh sampler with one hour-10 tick, so the flush
	// itself never touches hour 09.
	postCPU := 0.9
	sm := &sampler{buf: []metrics.Sample{
		{At: time.Date(2026, 8, 2, 10, 7, 0, 0, time.UTC), Host: metrics.HostSample{CPUFrac: &postCPU}},
	}}
	now := time.Date(2026, 8, 2, 10, 7, 0, 0, time.UTC)
	if err := metricsFlush(ctx, store, box, sm, now); err != nil {
		t.Fatal(err)
	}

	body, _, err := store.Get(ctx, metrics.DayKey(box, "2026-08-02"))
	if err != nil {
		t.Fatalf("day rollup missing — hour 09 was never rolled up: %v", err)
	}
	var day metrics.DayRollup
	if err := json.Unmarshal(body, &day); err != nil {
		t.Fatal(err)
	}
	agg, ok := day.Hours["09"]
	if !ok {
		t.Fatalf("day rollup has no hour 09: %+v", day.Hours)
	}
	if n := agg.Host["cpu"].N; n != 10 {
		t.Fatalf("hour 09 cpu N=%d, want 10 (rolled up from the stored raw chunk)", n)
	}
	if avg := agg.Host["cpu"].Avg; avg < 0.399 || avg > 0.401 {
		t.Fatalf("hour 09 cpu avg=%v, want ~0.4 (the stored chunk's readings)", avg)
	}
}

// TestMetricsFlushBackfillLeavesExistingRollupAlone: the backfill is repair
// work for a gap, not a rewrite. An hour whose rollup is already present —
// the normal case, written when the hour closed under the previous process —
// must be left exactly as it is, and the backfill must not run again on
// later flushes of the same process.
func TestMetricsFlushBackfillLeavesExistingRollupAlone(t *testing.T) {
	ctx := context.Background()
	store := release.NewDirStore(t.TempDir())
	box := "box-a"

	hour09Start := time.Date(2026, 8, 2, 9, 0, 0, 0, time.UTC)
	cpu := 0.4
	if err := metrics.WriteRaw(ctx, store, metrics.RawChunk{
		V: 1, Box: box, Hour: metrics.HourOf(hour09Start),
		Samples: []metrics.Sample{
			{At: hour09Start, Host: metrics.HostSample{CPUFrac: &cpu}},
			{At: hour09Start.Add(time.Minute), Host: metrics.HostSample{CPUFrac: &cpu}},
		},
	}); err != nil {
		t.Fatal(err)
	}
	// A rollup that plainly did NOT come from the two samples above, so a
	// clobber is unmistakable.
	sentinel := metrics.HourAgg{Host: map[string]metrics.Agg{"cpu": {Min: 0.1, Avg: 0.2, Max: 0.3, N: 57}}}
	if err := metrics.UpsertRollup(ctx, store, box, hour09Start, sentinel); err != nil {
		t.Fatal(err)
	}

	postCPU := 0.9
	sm := &sampler{buf: []metrics.Sample{
		{At: time.Date(2026, 8, 2, 10, 7, 0, 0, time.UTC), Host: metrics.HostSample{CPUFrac: &postCPU}},
	}}
	if err := metricsFlush(ctx, store, box, sm, time.Date(2026, 8, 2, 10, 7, 0, 0, time.UTC)); err != nil {
		t.Fatal(err)
	}

	readHour09 := func() metrics.HourAgg {
		t.Helper()
		body, _, err := store.Get(ctx, metrics.DayKey(box, "2026-08-02"))
		if err != nil {
			t.Fatal(err)
		}
		var day metrics.DayRollup
		if err := json.Unmarshal(body, &day); err != nil {
			t.Fatal(err)
		}
		return day.Hours["09"]
	}
	if got := readHour09().Host["cpu"]; got != sentinel.Host["cpu"] {
		t.Fatalf("hour 09 cpu = %+v, want the pre-existing %+v untouched", got, sentinel.Host["cpu"])
	}

	// A second flush must not reconsider it either.
	sm.buf = []metrics.Sample{{At: time.Date(2026, 8, 2, 10, 12, 0, 0, time.UTC), Host: metrics.HostSample{CPUFrac: &postCPU}}}
	if err := metricsFlush(ctx, store, box, sm, time.Date(2026, 8, 2, 10, 12, 0, 0, time.UTC)); err != nil {
		t.Fatal(err)
	}
	if got := readHour09().Host["cpu"]; got != sentinel.Host["cpu"] {
		t.Fatalf("second flush changed hour 09 to %+v", got)
	}
}

// TestLogRejectedAlertOverrides: a threshold override that RulesFromEnv
// rejects (it wants a percent integer, and "0.95" — the fraction the README
// used to describe — is not one) must not vanish silently. An accepted value
// and an unset one say nothing.
func TestLogRejectedAlertOverrides(t *testing.T) {
	env := map[string]string{
		"CARLOS_ALERT_DISK_PCT": "0.95", // a fraction: rejected
		"CARLOS_ALERT_MEM_PCT":  "85",   // accepted
		// CARLOS_ALERT_CPU_PCT unset
	}
	var buf bytes.Buffer
	prev := log.Writer()
	log.SetOutput(&buf)
	defer log.SetOutput(prev)

	logRejectedAlertOverrides(func(k string) string { return env[k] })

	out := buf.String()
	if !strings.Contains(out, "CARLOS_ALERT_DISK_PCT") {
		t.Errorf("a rejected override logged nothing: %q", out)
	}
	if strings.Contains(out, "CARLOS_ALERT_MEM_PCT") || strings.Contains(out, "CARLOS_ALERT_CPU_PCT") {
		t.Errorf("an accepted or unset override was logged: %q", out)
	}
}

// TestMetricsBoxLabelUsesFleetIdentity is I5 (final customer-fleets
// review): metricsLoop's published label must be the SAME resolved
// fleet-box identity statusLoop uses (fleetidentity.go's
// FleetIdentity.Label), never status.Label(machineID)'s hashed pseudonym,
// whenever CARLOS_FLEET is set — otherwise a fleet box's metrics/alerts
// would land under a different key than its own heartbeat.
func TestMetricsBoxLabelUsesFleetIdentity(t *testing.T) {
	const machineID = "test-machine-id"

	t.Run("no fleet env falls back to the hashed pseudonym", func(t *testing.T) {
		got := metricsBoxLabel(envFunc(map[string]string{}), machineID)
		if want := status.Label(machineID); got != want {
			t.Fatalf("box label = %q, want the hashed pseudonym %q", got, want)
		}
	})

	t.Run("fleet env resolves to the fleet-box label", func(t *testing.T) {
		got := metricsBoxLabel(envFunc(map[string]string{
			"CARLOS_FLEET":     "pi",
			"CARLOS_FLEET_BOX": "pi-box",
		}), machineID)
		if got != "pi-box" {
			t.Fatalf("box label = %q, want %q (the fleet-box label, not a hash of %q)", got, "pi-box", machineID)
		}
		if got == status.Label(machineID) {
			t.Fatalf("box label collided with the hashed pseudonym — fleet identity was not used")
		}
	})

	t.Run("invalid fleet identity degrades to the hashed pseudonym and warns", func(t *testing.T) {
		var buf bytes.Buffer
		prev := log.Writer()
		log.SetOutput(&buf)
		defer log.SetOutput(prev)

		got := metricsBoxLabel(envFunc(map[string]string{
			"CARLOS_FLEET": "pi",
			// no CARLOS_FLEET_BOX, no CARLOS_TUNNEL_NAME to derive one from.
		}), machineID)
		if want := status.Label(machineID); got != want {
			t.Fatalf("box label = %q, want the hashed pseudonym %q on invalid identity", got, want)
		}
		if !strings.Contains(buf.String(), "hashed pseudonym") {
			t.Errorf("invalid fleet identity logged nothing: %q", buf.String())
		}
	})
}

// webhookSinks is the sink set the pre-SNS tests meant: a webhook and
// nothing else.
func webhookSinks(url string, post func(url string, body []byte) error) *alertSinks {
	return newAlertSinks(url, post, "", nil, &alertCounters{})
}