// The metrics sampler loop — the box-side half of host-process monitoring.
// Every 60s it samples the whole host plus every monitored systemd unit
// (sampler.tick), evaluates on-box alert rules against the freshest sample
// (alertTick), and every fifth tick flushes the buffered samples out to the
// deployment bucket as raw chunks, rolling up and pruning across hour
// boundaries (metricsFlush). The sampler/probes split mirrors statusLoop's
// own status.Probes seam: sampleProbes is the injectable fixture bundle so
// host_test.go never touches /proc, a cgroup file, or a real clock.
package host

import (
	"context"
	"encoding/json"
	"errors"
	"log"
	"os"
	"path/filepath"
	"time"

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

// sampleProbes is the injectable fixture bundle sampler.tick reads from each
// tick, like status.Probes for statusTick. now is its own seam (rather than
// time.Now called directly) so tests control the elapsed time between ticks
// that unit CPU deltas depend on.
type sampleProbes struct {
	procStat func() *metrics.CPUCounters
	loadavg  func() (l1, l5, l15 *float64)
	memFrac  func() *float64
	diskFrac func() *float64
	units    func() map[string]unitRaw
	now      func() time.Time
}

// samplerMaxBuf hard-caps sm.buf at ~2h of 60s ticks. metricsFlush only
// trims buf on a fully successful flush, so a store that's up for reads but
// persistently failing writes (an IAM policy that allows PutObject but
// denies DeleteObject, a full disk, whatever) would otherwise grow buf by
// one sample every tick forever, re-marshaling the whole (ever-growing)
// buffer on every retry — this bounds that regardless of how flush is
// doing, since the cap is enforced in tick, not in flush.
const samplerMaxBuf = 120

// sampler carries the delta state between ticks: the previous /proc/stat
// reading (for host CPU), the previous per-unit cgroup usage_usec reading
// (for unit CPU), the wall-clock time of the previous tick (unit CPU needs
// elapsed wall time, not just the counter delta), the buffered samples
// awaiting their next flush (hard-capped at samplerMaxBuf), and which hours
// this process has already written/merged at least once — see
// writeHourChunk's doc comment for why a set (not a single bool) is
// required: a restart's first-ever flush can span an hour boundary, in
// which case BOTH the old and the current hour are being written by this
// process for the first time in the same call.
type sampler struct {
	prevCPU      *metrics.CPUCounters
	prevUnit     map[string]*int64 // unit -> cpu usec at previous tick
	prevAt       time.Time
	buf          []metrics.Sample
	writtenHours map[string]bool // hour -> this process has written/merged it at least once
	backfilled   bool            // the one-shot previous-hour rollup repair has run (see backfillPrevHourRollup)
}

// tick reads p once and appends one metrics.Sample to sm.buf. The very
// first tick on a fresh sampler has no previous reading to diff against, so
// its host and unit CPU fractions are nil — not a guessed zero — exactly
// like metrics.CPUFrac/UnitCPUFrac's own nil-on-no-prior-reading contract.
func (sm *sampler) tick(p sampleProbes) {
	now := p.now()
	curCPU := p.procStat()
	l1, l5, l15 := p.loadavg()
	mem := p.memFrac()
	disk := p.diskFrac()
	units := p.units()

	var elapsed time.Duration
	if !sm.prevAt.IsZero() {
		elapsed = now.Sub(sm.prevAt)
	}

	hostCPU := metrics.CPUFrac(sm.prevCPU, curCPU)

	var unitSamples map[string]metrics.UnitSample
	var newPrevUnit map[string]*int64
	if units != nil {
		unitSamples = make(map[string]metrics.UnitSample, len(units))
		newPrevUnit = make(map[string]*int64, len(units))
		for name, u := range units {
			var prevUsec *int64
			if sm.prevUnit != nil {
				prevUsec = sm.prevUnit[name]
			}
			unitSamples[name] = metrics.UnitSample{
				CPUFrac:      metrics.UnitCPUFrac(prevUsec, u.cpuUsec, elapsed),
				MemBytes:     u.memBytes,
				MemPeakBytes: u.memPeakBytes,
				ReadBytes:    u.readBytes,
				WriteBytes:   u.writeBytes,
			}
			newPrevUnit[name] = u.cpuUsec
		}
	}

	sm.buf = append(sm.buf, metrics.Sample{
		At: now,
		Host: metrics.HostSample{
			CPUFrac:  hostCPU,
			Load1:    l1,
			Load5:    l5,
			Load15:   l15,
			MemFrac:  mem,
			DiskFrac: disk,
		},
		Units: unitSamples,
	})

	if len(sm.buf) > samplerMaxBuf {
		drop := len(sm.buf) - samplerMaxBuf
		log.Printf("host: sampler buffer exceeded %d samples (flush stuck?) — dropping %d oldest unflushed sample(s)", samplerMaxBuf, drop)
		sm.buf = sm.buf[drop:]
	}

	sm.prevCPU = curCPU
	sm.prevUnit = newPrevUnit
	sm.prevAt = now
}

// metricsRawPrefix is the bucket prefix under which a box's raw chunks live
// — metrics.RawKey's own prefix, duplicated here only for List's sake (List
// takes a bare prefix, not a templated key).
func metricsRawPrefix(box string) string {
	return "metrics/boxes/" + box + "/raw/"
}

// metricsFlush writes sm.buf out as raw chunks grouped by each sample's own
// hour — never mixing two hours into one chunk, even when a single flush
// call spans a boundary. Samples in the current hour (relative to now) are
// written as that hour's chunk and kept in buf (the current hour's chunk
// gets rewritten wholesale on every flush as more samples accumulate,
// mirroring WriteRaw's own "replace whatever is there" contract). Samples
// from any older hour are written to their own chunk, rolled up via
// RollupHour+UpsertRollup, and then dropped from buf — that hour is done,
// buf must not grow unbounded by carrying it forever. Pruning of raw chunks
// older than 48h runs only when an hour actually rolled over (there is
// nothing new to prune otherwise).
//
// Every hour written here — whether it lands in the "old" bucket or is the
// current one — goes through writeHourChunk, which merges in any
// pre-existing chunk the FIRST time this process writes that hour. That
// matters for old hours too, not just the current one: after a restart, a
// sampler's very first flush can itself span an hour boundary (a crash
// mid-hour, then a restart that ticks a couple of times before its first
// flush lands in the next hour) — in that case the OLD hour is also being
// written by this process for the first time, and skipping the merge there
// would silently clobber that hour's pre-crash samples and permanently
// corrupt its day rollup the moment RollupHour runs over the truncated set.
func metricsFlush(ctx context.Context, s release.Store, box string, sm *sampler, now time.Time) error {
	// Once per process, before anything else: repair the hour a restart may
	// have left un-rolled-up (see backfillPrevHourRollup). It is deliberately
	// not allowed to fail the flush — it is repair work for an hour that is
	// already over, and this flush's own samples matter more. A failure just
	// leaves the flag unset so the next flush tries again.
	if !sm.backfilled {
		if err := backfillPrevHourRollup(ctx, s, box, now); err != nil {
			log.Printf("host: previous-hour rollup backfill for %s failed, retrying next flush: %v", box, err)
		} else {
			sm.backfilled = true
		}
	}

	if len(sm.buf) == 0 {
		return nil
	}
	curHour := metrics.HourOf(now)

	curSamples := make([]metrics.Sample, 0, len(sm.buf))
	oldByHour := map[string][]metrics.Sample{}
	var oldHours []string
	for _, samp := range sm.buf {
		h := metrics.HourOf(samp.At)
		if h == curHour {
			curSamples = append(curSamples, samp)
			continue
		}
		if _, ok := oldByHour[h]; !ok {
			oldHours = append(oldHours, h)
		}
		oldByHour[h] = append(oldByHour[h], samp)
	}

	for _, hour := range oldHours {
		samples, merged, err := sm.writeHourChunk(ctx, s, box, hour, oldByHour[hour])
		if err != nil {
			return err
		}
		// RollupHour runs over the (possibly merged) full set, never just
		// what this process itself collected — otherwise a restart would
		// permanently truncate that hour's day rollup even though the raw
		// chunk itself was correctly merged above.
		agg := metrics.RollupHour(samples)
		if err := metrics.UpsertRollup(ctx, s, box, samples[0].At, agg); err != nil {
			return err
		}
		// Both durable writes for this hour succeeded — NOW it's safe to
		// mark it written (only meaningful when this call actually merged;
		// see writeHourChunk/markHourWritten) and retire its samples from
		// buf immediately, rather than waiting for the whole flush call to
		// finish. If a LATER hour's write fails below, buf must not still
		// be carrying THIS hour's already-durably-written samples — a
		// retry that reprocessed them would see writtenHours already true,
		// skip the merge, and overwrite this hour's correctly-merged chunk
		// with just the bare in-memory set, silently regressing it.
		if merged {
			sm.markHourWritten(hour)
		}
		sm.dropHourFromBuf(hour)
	}

	if len(oldHours) > 0 {
		if err := pruneStaleRaw(ctx, s, box, now); err != nil {
			return err
		}
	}

	if len(curSamples) > 0 {
		merged, wasMerged, err := sm.writeHourChunk(ctx, s, box, curHour, curSamples)
		if err != nil {
			return err
		}
		curSamples = merged
		if wasMerged {
			sm.markHourWritten(curHour)
		}
	}

	sm.buf = curSamples
	return nil
}

// backfillPrevHourRollup rolls up the PREVIOUS hour's stored raw chunk if the
// day object has no entry for that hour yet. It runs once per process, on the
// first flush.
//
// The gap it closes: rollups are otherwise derived only from what this
// process itself buffered. An agent that stops at 09:58 and starts again at
// 10:07 never has an hour-09 sample in buf, so nothing ever calls RollupHour
// for hour 09 — the raw chunk written before the stop sits there with no day
// entry, and 48h later pruneStaleRaw deletes it. The hour is then gone from
// the trend history permanently, silently, and only for the hours that
// contain a restart (i.e. exactly the interesting ones).
//
// Deliberately narrow, in three ways. Only the immediately previous hour is
// considered: a longer outage has no raw chunk to recover from anyway, since
// the agent that would have written it was down. An hour whose rollup already
// exists is left completely alone — this repairs a gap, it never rewrites a
// closed hour. And a missing chunk, an empty one, or an unreadable one is a
// no-op rather than an error: there is simply nothing to recover.
func backfillPrevHourRollup(ctx context.Context, s release.Store, box string, now time.Time) error {
	prev := now.Add(-time.Hour)
	hour := metrics.HourOf(prev)
	key := metrics.RawKey(box, hour)

	body, _, err := s.Get(ctx, key)
	if err != nil {
		if errors.Is(err, os.ErrNotExist) {
			return nil // no chunk for that hour: nothing to roll up
		}
		return err
	}
	var chunk metrics.RawChunk
	if err := json.Unmarshal(body, &chunk); err != nil {
		log.Printf("host: raw chunk %s unreadable, skipping rollup backfill: %v", key, err)
		return nil
	}
	if len(chunk.Samples) == 0 {
		return nil
	}

	has, err := rollupHasHour(ctx, s, box, prev)
	if err != nil {
		return err
	}
	if has {
		return nil
	}

	log.Printf("host: hour %s has a raw chunk but no day rollup (restart across the boundary) — rolling it up now", hour)
	return metrics.UpsertRollup(ctx, s, box, prev, metrics.RollupHour(chunk.Samples))
}

// rollupHasHour reports whether box's day object already carries hour's UTC
// hour bucket. A missing day object is a plain "no". An unreadable one
// reports "yes": UpsertRollup would fail on the same unmarshal anyway, and a
// corrupt day object is a separate problem the normal rollover path will
// surface — the backfill must not spin on it every flush.
func rollupHasHour(ctx context.Context, s release.Store, box string, hour time.Time) (bool, error) {
	key := metrics.DayKey(box, hour.UTC().Format("2006-01-02"))
	body, _, err := s.Get(ctx, key)
	if err != nil {
		if errors.Is(err, os.ErrNotExist) {
			return false, nil
		}
		return false, err
	}
	var day metrics.DayRollup
	if err := json.Unmarshal(body, &day); err != nil {
		log.Printf("host: day rollup %s unreadable, skipping rollup backfill: %v", key, err)
		return true, nil
	}
	_, ok := day.Hours[hour.UTC().Format("15")]
	return ok, nil
}

// writeHourChunk writes samples as hour's raw chunk, merging in whatever
// chunk already exists at that key the first time THIS sampler writes that
// particular hour (tracked in sm.writtenHours) — see metricsFlush's doc
// comment for why both old and current hours need this, not just the
// current one. Every write after the first for a given hour skips the
// merge: by then sm.buf already carries forward everything this process
// itself wrote there, so re-merging would be redundant (and for an old
// hour, over by definition — it's dropped from buf right after this call).
//
// Deliberately does NOT mark sm.writtenHours itself — only that the merge
// happened this call (the returned merged bool). Marking is the caller's
// job, done only once every durable write this hour's data depends on has
// actually succeeded (for an old hour that's this WriteRaw AND its
// UpsertRollup; for the current hour, just this WriteRaw). Marking here,
// before the caller even knows whether those later steps succeed, would
// let a downstream failure retry with the merge silently skipped — exactly
// the bug this shape avoids. Returns the actual sample set written (merged
// or not) so the caller uses the right data for RollupHour / for keeping
// in buf.
func (sm *sampler) writeHourChunk(ctx context.Context, s release.Store, box, hour string, samples []metrics.Sample) (out []metrics.Sample, merged bool, err error) {
	if !sm.writtenHours[hour] {
		m, err := mergeExistingChunk(ctx, s, box, hour, samples)
		if err != nil {
			return nil, false, err
		}
		samples = m
		merged = true
	}
	if err := metrics.WriteRaw(ctx, s, metrics.RawChunk{V: 1, Box: box, Hour: hour, Samples: samples}); err != nil {
		return nil, false, err
	}
	return samples, merged, nil
}

// markHourWritten records that hour's chunk has been durably written (and,
// where applicable, rolled up) at least once by this sampler — see
// writeHourChunk's doc comment for why the caller, not writeHourChunk
// itself, decides when that's true.
func (sm *sampler) markHourWritten(hour string) {
	if sm.writtenHours == nil {
		sm.writtenHours = map[string]bool{}
	}
	sm.writtenHours[hour] = true
}

// dropHourFromBuf removes every sample belonging to hour from sm.buf. Old
// hours are retired from buf the moment their write is fully durable
// (metricsFlush calls this right after, per hour, inside its oldHours
// loop) rather than in one pass at the very end — so that if a LATER old
// hour's write fails within the same flush call, THIS hour's samples are
// already gone from buf and cannot be reprocessed (and, worse, rewritten
// unmerged — see metricsFlush's doc comment at the call site) on the next
// flush attempt.
func (sm *sampler) dropHourFromBuf(hour string) {
	kept := make([]metrics.Sample, 0, len(sm.buf))
	for _, samp := range sm.buf {
		if metrics.HourOf(samp.At) != hour {
			kept = append(kept, samp)
		}
	}
	sm.buf = kept
}

// mergeExistingChunk reads whatever raw chunk already exists for hour (most
// likely written by this same box before a restart or crash) and merges it
// under inMemory: existing samples first, then the in-memory ones, deduping
// on exact At-timestamp collisions with the in-memory sample winning (it's
// the fresher read of that instant). A fresh sampler's buf always starts
// empty after a restart, and WriteRaw is a wholesale replace — without
// this, the first flush that touches a given hour after a restart would
// silently clobber everything the box measured earlier in that hour, and
// that loss would propagate permanently into the hour's day rollup once it
// closes (or immediately, if the hour in question is already "old" by the
// time this process gets to it — see metricsFlush's doc comment). A
// missing chunk (no prior write this hour — the common case) is not an
// error; a corrupt one is logged and skipped rather than failing the flush,
// since a garbled pre-restart file must not block ongoing monitoring.
func mergeExistingChunk(ctx context.Context, s release.Store, box, hour string, inMemory []metrics.Sample) ([]metrics.Sample, error) {
	key := metrics.RawKey(box, hour)
	body, _, err := s.Get(ctx, key)
	if err != nil {
		if errors.Is(err, os.ErrNotExist) {
			return inMemory, nil
		}
		return nil, err
	}
	var existing metrics.RawChunk
	if err := json.Unmarshal(body, &existing); err != nil {
		log.Printf("host: existing raw chunk %s unreadable, flushing without it: %v", key, err)
		return inMemory, nil
	}

	inMemoryAt := make(map[time.Time]bool, len(inMemory))
	for _, samp := range inMemory {
		inMemoryAt[samp.At] = true
	}

	merged := make([]metrics.Sample, 0, len(existing.Samples)+len(inMemory))
	for _, samp := range existing.Samples {
		if !inMemoryAt[samp.At] {
			merged = append(merged, samp)
		}
	}
	merged = append(merged, inMemory...)
	return merged, nil
}

// pruneStaleRaw deletes this box's raw chunks older than 48h — List +
// metrics.StaleRawKeys (which never returns a key it cannot parse — see its
// own doc comment) + Delete.
func pruneStaleRaw(ctx context.Context, s release.Store, box string, now time.Time) error {
	keys, err := s.List(ctx, metricsRawPrefix(box))
	if err != nil {
		return err
	}
	for _, key := range metrics.StaleRawKeys(keys, now, 48*time.Hour) {
		if err := s.Delete(ctx, key); err != nil {
			return err
		}
	}
	return nil
}

// alertTick evaluates rules against latest and, when that produces a
// firing/clearing transition (Evaluate emits Events exactly on those — see
// its own doc comment), publishes the new state to metrics.AlertKey(box)
// and delivers each event to sinks (alertsink.go: the webhook, the SNS topic,
// either, both or neither; nil is none).
//
// Retry semantics on a publish failure are the crux of this function: *st
// is the caller's durable record of "what has been reliably told to the
// world", so it is only ever advanced to newState when the publish that
// matches it actually succeeded (or there was nothing to publish — no
// events this tick). If putAlertState fails, *st is left exactly as it was:
// the SAME transition is still "pending" as far as this sampler is
// concerned, so the next tick's Evaluate call starts from the same
// not-yet-committed state and — as long as the underlying condition is
// still there — reproduces the identical event and retries both the
// publish and the webhook post. Committing newState unconditionally (this
// function's first-round shape) would silently swallow a one-tick store
// blip forever: Evaluate only emits an event on an actual transition, so
// once *st already says "firing", every later tick just confirms it's
// still firing with no new event and nothing left to retry.
//
// Delivery to the sinks is still attempted this tick regardless of whether
// publish succeeded — it's independently fire-and-forget, logged and counted
// on failure, never returned; it does not gate whether *st advances. Local persistence to statePath is also unconditionally
// best-effort (a failure is logged, never returned or allowed to block
// publish/webhook — see round 1's fix, since the local write is most likely
// to fail exactly when the disk rule is the one firing), but it always
// persists whichever state ends up committed to *st, so a restart loads
// exactly what this process believes right now — never a state the store
// was never told about.
//
// Honest limit: for a sustain-window rule (mem, cpu — sustain > 0), the
// Since bookkeeping itself is NOT at risk from a publish outage — every
// tick before the rule is actually ready to fire carries zero events, so
// publish is never even attempted on those ticks and *st (Since included)
// commits normally regardless of the store's health. The outage only bites
// on the tick(s) where the rule is ready to fire: if publish fails exactly
// then, *st doesn't advance, so the next tick re-evaluates the same
// already-overdue transition and retries. The alert ends up firing (and
// published) as soon as the store recovers, delayed by at most the
// outage's own duration — it does not compound beyond that. Immediate
// rules (disk, lag; sustain 0) retry the same way, just starting from
// their very first over-threshold tick instead of a sustain deadline.
func alertTick(ctx context.Context, s release.Store, box string, rules metrics.Rules,
	st *metrics.AlertState, latest metrics.Sample, statePath string,
	sinks *alertSinks) error {

	obs := metrics.Observation{
		At:       latest.At,
		CPUFrac:  latest.Host.CPUFrac,
		MemFrac:  latest.Host.MemFrac,
		DiskFrac: latest.Host.DiskFrac,
		LagSecs:  nil, // v1: no lag reading feeds the agent loop yet
		GeoIP:    geoIPAlertReading(latest.At),
	}
	newState, events := metrics.Evaluate(rules, *st, obs)

	var publishErr error
	if len(events) > 0 {
		body, err := json.MarshalIndent(newState, "", "  ")
		if err != nil {
			publishErr = err
		} else if err := putAlertState(ctx, s, metrics.AlertKey(box), body); err != nil {
			publishErr = err
		}
	}

	// Fire-and-forget, independent of publish's outcome: an event still
	// deserves a delivery attempt even if the store write for it failed.
	for _, ev := range events {
		sinks.deliver(ctx, rules, latest.At, ev)
	}

	committed := *st
	if publishErr == nil {
		committed = newState
		*st = newState
	}
	// else: *st stays exactly as it was — see doc comment above.

	if err := persistAlertState(statePath, committed); err != nil {
		log.Printf("host: alert state persist to %s failed, continuing: %v", statePath, err)
	}

	return publishErr
}

// persistAlertState writes st to path as indented JSON via the repo's
// standard temp+rename idiom (internal/appconfig/sync.go's renderEnv,
// internal/release/store.go's DirStore.Put): a crashed write must never
// leave a half-written state file behind for the next boot to load and
// choke on. The parent directory (<dataDir>/monitor/) is created if
// missing — this is the first thing that ever writes into it.
func persistAlertState(path string, st metrics.AlertState) error {
	if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
		return err
	}
	body, err := json.MarshalIndent(st, "", "  ")
	if err != nil {
		return err
	}
	tmp := path + ".tmp"
	if err := os.WriteFile(tmp, body, 0o644); err != nil {
		return err
	}
	if err := os.Rename(tmp, path); err != nil {
		os.Remove(tmp)
		return err
	}
	return nil
}

// putAlertState upserts key with a replace-or-create Put, retrying once on
// an ErrCASConflict — the same read-etag/Put/retry-once dance WriteRaw and
// UpsertRollup use, since this box is the only writer that should ever be
// targeting its own alert key.
func putAlertState(ctx context.Context, s release.Store, key string, body []byte) error {
	etag, err := currentAlertEtag(ctx, s, key)
	if err != nil {
		return err
	}
	if _, err := s.Put(ctx, key, body, etag); err != nil {
		if !errors.Is(err, release.ErrCASConflict) {
			return err
		}
		etag, err = currentAlertEtag(ctx, s, key)
		if err != nil {
			return err
		}
		if _, err := s.Put(ctx, key, body, etag); err != nil {
			return err
		}
	}
	return nil
}

// currentAlertEtag reads key's current etag for a replace-only Put, folding
// "doesn't exist yet" into "" (create-only) — the same convention
// metrics.WriteRaw's own currentRawEtag uses.
func currentAlertEtag(ctx context.Context, s release.Store, key string) (string, error) {
	_, etag, err := s.Get(ctx, key)
	if err != nil {
		if errors.Is(err, os.ErrNotExist) {
			return "", nil
		}
		return "", err
	}
	return etag, nil
}

// httpPostWebhook is the production webhook post func: a 5s-timeout POST
// with a JSON content type. Any non-2xx status is an error — the caller
// (alertTick) only logs it, but it still needs to know delivery failed.
func httpPostWebhook(url string, body []byte) error { return alerting.PostJSON(url, body) }

// realSampleProbes wires sampleProbes to the actual OS/cgroup readers in
// metrics_probes.go and probes.go — the production seam metricsLoop uses,
// never called from a test (host_test.go's convention: tests inject
// fixtures, never touch /proc or a syscall).
//
// stampedUnits is re-read inside the units probe rather than captured as a
// slice, because sampleProbes is built ONCE and read every tick: a snapshot
// taken here would pin the stamped set to whatever the registry held at agent
// start, and a route stamped this afternoon would go unsampled until the next
// restart. A nil func means "no stamped units" — the darwin/dev shape, and
// what every test double passes.
func realSampleProbes(execRoot string, stampedUnits func() []string) sampleProbes {
	return sampleProbes{
		procStat: readProcStat,
		loadavg:  readLoadAvg,
		memFrac:  readMemFrac,
		diskFrac: readDiskFrac,
		units: func() map[string]unitRaw {
			var stamped []string
			if stampedUnits != nil {
				stamped = stampedUnits()
			}
			return realUnitReadings(execRoot, stamped)
		},
		now: time.Now,
	}
}

// realUnitReadings lists the box's currently active CARLOS units and reads
// each one's cgroup usage — the production units func sampleProbes.units
// wraps. Keyed by UNIT NAME (what the console matches against), read from the
// unit's own cgroup DIRECTORY (which for a tenant instance is a level down;
// see listCarlosUnits) — plus the operator-stamped bespoke units named in
// stamped (registry Route.Unit; see listStampedUnits) and, when delegation is
// armed, the per-host exec-child cgroups under execRoot (keyed
// carlos-exec@<host>; see listExecCgroups).
//
// The three scans may overlap — a stamped unit that also matches the carlos-
// convention is discovered twice — and the map write below is what dedupes
// them: both entries name the same unit and the same cgroup dir, so the
// second read is redundant, never contradictory.
func realUnitReadings(execRoot string, stamped []string) map[string]unitRaw {
	units := listCarlosUnits(cgroupSliceDir)
	// carlos-edge.service moved to carlos.slice (isolated-exec design §4.3;
	// see cgroupCarlosSliceDir's doc comment), so it is found only by also
	// scanning that second root. system.slice is still scanned above for
	// every other platform unit — litestream, adopt, relay and the rest —
	// which have not moved.
	units = append(units, listCarlosUnits(cgroupCarlosSliceDir)...)
	units = append(units, listStampedUnits(cgroupSliceDir, stamped)...)
	units = append(units, listExecCgroups(execRoot)...)
	if len(units) == 0 {
		return nil
	}
	out := make(map[string]unitRaw, len(units))
	for _, u := range units {
		out[u.name] = readUnitRaw(u.dir)
	}
	return out
}

// alertPctEnvVars are the percent-integer threshold overrides
// metrics.RulesFromEnv reads.
var alertPctEnvVars = []string{"CARLOS_ALERT_DISK_PCT", "CARLOS_ALERT_MEM_PCT", "CARLOS_ALERT_CPU_PCT"}

// logRejectedAlertOverrides says so when one of those is SET but not
// accepted. RulesFromEnv is pure and silently keeps the default, which is
// the right behaviour — an operator's typo must never stop the agent — but
// silence is the wrong report: someone who writes 0.95 (a fraction, the
// units this took before the README was corrected) would otherwise believe
// they had lowered the CPU threshold while the box quietly ran on 0.95's
// rejection and the built-in default.
func logRejectedAlertOverrides(getenv func(string) string) {
	for _, name := range alertPctEnvVars {
		v := getenv(name)
		if v == "" {
			continue
		}
		if _, ok := metrics.PctFrac(v); !ok {
			log.Printf("agent: %s=%q ignored: want a percent integer 0-100 (e.g. 95) — keeping the built-in default", name, v)
		}
	}
}

// loadAlertState reads a persisted AlertState from path, falling back to a
// fresh state for box on anything that isn't a clean read — a missing file
// (first boot) or a corrupt one (a previous crash mid-write, or a hand
// edit) must never crash the agent loop; monitoring starting from a blank
// slate is always a safe, honest state to resume from.
func loadAlertState(path, box string) metrics.AlertState {
	fresh := metrics.AlertState{V: 1, Box: box, Rules: map[string]metrics.RuleState{}}
	body, err := os.ReadFile(path)
	if err != nil {
		return fresh
	}
	var st metrics.AlertState
	if err := json.Unmarshal(body, &st); err != nil {
		return fresh
	}
	if st.Rules == nil {
		st.Rules = map[string]metrics.RuleState{}
	}
	return st
}

// metricsBoxLabel resolves the label metrics/alerts publish under — pulled
// out of metricsLoop as its own pure seam (I5, final customer-fleets review)
// so it is testable without a bucket, mirroring how statusTick takes an
// explicit labelOverride rather than re-deriving it inline.
//
// It must be the SAME resolved identity statusLoop uses
// (fleetidentity.go's FleetIdentity.Label, the operator-chosen
// fleet_boxes.label the edge actually dials), NEVER status.Label(machineID)'s
// hashed pseudonym, when this is a fleet box. Diverging from statusLoop here
// would put this box's heartbeat under one label
// (fleets/<fleet>/status/boxes/<label>.json) and its metrics under a
// DIFFERENT one (status.Label's hash), which breaks the fleet-channel
// identity a customer's monitoring page keys everything by — see
// fleetidentity.go's doc comment for why the hashed pseudonym is wrong here
// at all. An invalid (not merely absent) identity degrades to the hashed
// label rather than disabling metrics outright, mirroring statusLoop's own
// degrade-don't-disable choice.
func metricsBoxLabel(getenv func(string) string, machineID string) string {
	if groupReferenced(getenv) {
		// A box its region's group launched (boxreference.go).
		return groupBoxReference()
	}
	if identity, err := resolveFleetIdentity(getenv); err == nil {
		return identity.Label
	} else if !errors.Is(err, errNotFleetBox) {
		log.Printf("agent: %v — metrics will publish under the hashed pseudonym instead", err)
	}
	return status.Label(machineID)
}

// BoxLabel is metricsBoxLabel for callers outside this package — the relay
// (`carlos relay serve`, relay spec §3) meters and heartbeats under the same
// box label the metrics loop publishes under, so the two agree about which
// box a row belongs to. A thin wrapper deliberately: the selection rule
// stays in one place, tested there.
func BoxLabel() string { return metricsBoxLabel(os.Getenv, readMachineID()) }

// metricsLoop runs the metrics sampler's 60s tick forever, once armed — thin
// and untested by the same convention as statusLoop/adoptLoop: everything
// with behavior lives in sampler.tick, metricsFlush, alertTick, and
// metricsBoxLabel, all exercised above with fixtures. Missing
// status/deployment bucket config disables the loop with one log line, the
// same "a box with no store is still a valid deployment" shape every other
// optional loop in this package uses. Flush runs every 5th tick (5 minutes
// at the 60s tick cadence); alertTick runs every tick, since a threshold
// breach deserves faster detection than a metrics flush does.
//
// stampedUnits is the sweep's one registry seam (agent.go's
// stampedUnitNames): the bespoke units to sample alongside the carlos- ones,
// re-read once per tick so a freshly stamped route is picked up within a
// minute. It is a func, not the registry handle, so this loop keeps its "no
// state but a bucket and /proc" shape; nil = none.
//
// sinks are the agent's alert sinks, built and armed for paging by runAgent
// at startup (armAgentAlerts); the loop delivers its alerts on them and
// never builds or arms its own. nil delivers nowhere.
func metricsLoop(execRoot string, stampedUnits func() []string, sinks *alertSinks) {
	// This loop is always on and shares a process with the agent that serves
	// this box's traffic, so a panic in it — an unreadable cgroup shape, a
	// nil no one expected — must not take the box down with it. Monitoring
	// stopping is a bad day; the box stopping is an outage.
	defer func() {
		if r := recover(); r != nil {
			log.Printf("agent: metrics loop panic, monitoring stopped for this process: %v", r)
		}
	}()

	ctx := context.Background()
	// Same selection rule statusLoop uses, in one place for both (agent.go's
	// openStatusStore): a status/deployment bucket wins and keeps direct S3;
	// a fleet box with neither set publishes through the edge's store door.
	raw, target, err := openStatusStore(ctx, os.Getenv)
	if err != nil {
		if errors.Is(err, errNoStatusStore) {
			log.Printf("agent: metrics loop disabled: neither CARLOS_STATUS_BUCKET nor CARLOS_DEPLOYMENT_BUCKET is set")
		} else {
			log.Printf("agent: metrics loop disabled: %v", err)
		}
		return
	}
	// s is what metricsFlush/alertTick actually read/write through: the same
	// CARLOS_STATUS_PREFIX rebase statusLoop applies (release.NewPrefixStore),
	// so a fleet box's metrics/alerts land under its own fleets/<name>/
	// namespace instead of the shared metrics/boxes/ + alerts/boxes/ paths a
	// platform box uses — without this, a customer's raw metrics would
	// surface on the platform operator's monitoring page (console
	// metricsview lists metrics/boxes/) and an alert would page the
	// platform's own webhook for customer hardware. Empty prefix (every
	// platform box today) is a pure passthrough — see the package doc
	// comment's CARLOS_STATUS_PREFIX entry. resolveStatusPrefix, not a bare
	// env read: on a door-backed fleet box an empty prefix would put every
	// flush outside the door's write ACL, so that branch defaults to
	// fleets/<fleet>/.
	s := release.NewPrefixStore(raw, resolveStatusPrefix(os.Getenv))

	box := metricsBoxLabel(os.Getenv, readMachineID())
	statePath := filepath.Join(dataDir(), "monitor", "alerts.json")
	st := loadAlertState(statePath, box)
	rules := metrics.RulesFromEnv(os.Getenv)
	logRejectedAlertOverrides(os.Getenv)

	log.Printf("agent: metrics loop armed (60s ticks, 5min flush), publishing to %s", target)

	sm := &sampler{}
	probes := realSampleProbes(execRoot, stampedUnits)
	tick := 0
	// One watch for the life of the loop: its memory of what it has already
	// alerted is what keeps a standing refusal from paging every minute.
	// dataDir() is the bake dir the agent hands release.BakeDir.
	var refusals refusalWatch
	for {
		time.Sleep(60 * time.Second)
		tick++

		rows := refusals.tick(ctx, dataDir(), sinks, time.Now())
		releaseRefusals.Store(&rows)

		sm.tick(probes)
		latest := sm.buf[len(sm.buf)-1]
		if err := alertTick(ctx, s, box, rules, &st, latest, statePath, sinks); err != nil {
			log.Printf("agent: alert tick: %v", err)
		}

		if tick%5 == 0 {
			if err := metricsFlush(ctx, s, box, sm, time.Now()); err != nil {
				log.Printf("agent: metrics flush: %v", err)
			}
		}
	}
}