felhom.eu/hub/internal/monitor/restoretest.go

package monitor

import (
	"encoding/json"
	"fmt"
	"log"
	"sync"
	"time"

	"gitea.dooplex.hu/admin/felhom-hub/internal/store"
)

// R-85 Part 2 — the restore-test result becomes a SIGNAL.
//
// Until now a failed restore-test was a `[WARN]` log line in the ingest handler and nothing else:
// no event, no notification, no staleness gauge. That was true for the LOCAL tier that was already
// being tested — so the loudest DR signal this system produces was, in practice, inaudible.
//
// TWO SIGNALS, DELIBERATELY NOT MERGED. They mean different things and warrant different urgency:
//
//	restore_test_failed  — a run completed and did NOT pass. Something is broken NOW.
//	restore_test_stale   — a tier has not been PROVEN within its expected interval. Nothing has
//	                       necessarily broken; we simply no longer know whether it works.
//
// Merging them would collapse "your DR is broken" into "your DR is unverified", and the second is
// the one that quietly becomes the first.
//
// ── THE INVARIANT, inherited from R-81 ────────────────────────────────────────────────────────
//
// Absence of a signal is UNKNOWN, and becomes a fault only once it has outlived an ANCHORED window.
// This monitor family has made the opposite mistake three times (hub v0.12.0, v0.73.0, R-81); this
// is a NEW monitor written straight after the third, so it copies R-81's verdict structure rather
// than re-deriving it. A tier never proven on a newborn box is UNKNOWN, never FAILED.

// ── HOW LONG MAY A TIER GO UNPROVEN? (R-86 Part 2) ───────────────────────────────────────────
//
// This was one flat constant, 7 days, and its comment derived that number like this:
//
//	"the restore-test cadence is 24h and rotation is oldest-first across two tiers, so each tier is
//	 proven roughly every 2 days. 7 days therefore tolerates ~3 consecutive missed opportunities."
//
// **That premise is exactly what R-86 removed.** The agent no longer tests on an interval at all: a
// tier is tested once per ARCHIVE GENERATION — when it holds a settled archive that has not been
// proven. A tier backed up weekly is therefore proved weekly, by design and in perfect health, and
// against a flat 7-day window it would sit on the line and alarm every night about a system that is
// working. Shipping the agent's half alone would have converted the improvement into a false alarm,
// which is why the two ship together.
//
// The window is now derived from **the tier's own backup rhythm**, which the hub can observe from
// the reports it already receives, and it keeps everything the constant had earned:
//
//   - absence is UNKNOWN until an anchored window has passed (R-81's structure, untouched);
//   - the signal stays edge-triggered;
//   - it never exceeds the offsite retention, so a tier is never called stale against an archive
//     that is about to be pruned;
//   - and it is never TIGHTER than the 7 days that were already tolerated.
const (
	// restoreProvenGenerations is how many archive generations may pass unproven before alarming.
	// 4 = the settle lag's own generation plus ~3 missed opportunities — deliberately the same
	// tolerance the flat constant expressed, so the change is to the RHYTHM, not to the patience.
	restoreProvenGenerations = 4

	// restoreProvenWindowFloor is the shortest window that may be applied to any tier. It is the
	// old constant, kept as a FLOOR rather than deleted: a daily tier computes 4 days from its own
	// rhythm, and tightening a live threshold is not what this task is for. A deferral behind a
	// long backup is normal, not a fault.
	restoreProvenWindowFloor = 7 * 24 * time.Hour

	// restoreProvenWindowCap keeps the window strictly inside the 2-week offsite retention
	// (operator ruling 2026-07-26) with two days to spare. Beyond it the hub would be judging a
	// tier against an archive PBS has already pruned — an alarm nobody can act on, and the bound
	// the old constant respected in its own way.
	restoreProvenWindowCap = 12 * 24 * time.Hour

	// restoreWindowRead is how far back the hub reads host-reports for this check: far enough to
	// find proof anywhere inside the widest window, and to see at least two archive generations of
	// a WEEKLY tier so its rhythm is observable at all.
	restoreWindowRead = 2 * restoreProvenWindowFloor
)

// declaredArchiveInterval is the rhythm the hub ALREADY attributes to a tier — the same thresholds
// the backup-freshness checker judges it against (deadline.go / deadline_tiers.go). It is the
// fallback when a box's history is too short to observe a rhythm, and it is the right fallback
// precisely because it is not a second opinion: if these two checkers disagreed about how often a
// tier is expected to receive an archive, one of them would be alarming on the other's model.
//
// It is stated per RESTORE tier name ("local"/"pbs" — what the agent reports as source_tier), which
// is the same split the backup tiers use under different names ("host"/"offsite").
func declaredArchiveInterval(tier string) time.Duration {
	if tier == "pbs" {
		return offsiteBackupStaleAfter // 8 days: the weekly cadence plus a day of headroom
	}
	return backupStaleAfter // 26 hours: the daily cadence plus headroom
}

// restoreProvenWindow is how long THIS tier may go unproven, given its observed archive interval.
//
// observedOK=false means the box's retained history did not contain two archive generations for
// this tier, so the declared rhythm is used. That fallback matters most for exactly the tier this
// task is about: a fresh box with a weekly offsite tier has one snapshot and no observable
// interval, and falling back to the FLOOR there would recreate the false alarm.
//
// OBSERVATION MAY ONLY WIDEN, NEVER TIGHTEN — and this is not caution, it is a live measurement.
// On demo-felhom (2026-08-03) the offsite tier's two retained snapshots are `2026-07-27T19:55:41Z`
// and `2026-07-28T04:49:43Z`: **8 h 54 m apart**, because one is a healing artefact and the other a
// real weekly run. A mean-gap estimate therefore reads a WEEKLY tier as nine-hourly, ×4 gives 36 h,
// the floor lifts it to 7 days — and a weekly tier proved weekly reaches ~8.25 days of proof age, so
// the false alarm this whole task exists to prevent would have returned within a week, on the very
// box it shipped to.
//
// The asymmetry is right on its own terms too. A gap SHORTER than the declared rhythm is routine and
// means nothing — a retry, a manual run, a heal, a catch-up after an outage. A gap LONGER than the
// declared rhythm is real information: this tier genuinely receives archives less often than the
// model says, and its window must widen or it alarms. So observation refines the rhythm upward and
// is ignored downward. The cost is stated plainly: a tier that truly runs FASTER than its declared
// rhythm gets a wider window than it strictly needs, i.e. a slower stale signal. That is the right
// direction for a signal whose message is "unverified" — "broken NOW" is `restore_test_failed`, and
// that one is immediate and unaffected.
func restoreProvenWindow(tier string, observed time.Duration, observedOK bool) time.Duration {
	interval := declaredArchiveInterval(tier)
	if observedOK && observed > interval {
		interval = observed
	}
	w := time.Duration(restoreProvenGenerations) * interval
	if w < restoreProvenWindowFloor {
		w = restoreProvenWindowFloor
	}
	if w > restoreProvenWindowCap {
		w = restoreProvenWindowCap
	}
	return w
}

// Event types. Operator-tier only — see the dispatcher note in RestoreTestChecker.
const (
	EventRestoreTestFailed = "restore_test_failed"
	EventRestoreTestStale  = "restore_test_stale"
)

// hostReportRestoreTests is the slice of a host-report this checker reads.
type hostReportRestoreTests struct {
	RestoreTests []struct {
		SourceArchive string `json:"source_archive"`
		SourceTier    string `json:"source_tier"`
		Pass          bool   `json:"pass"`
		Error         string `json:"error"`
		TestedAt      string `json:"tested_at"`
	} `json:"restore_tests"`
	StorageTargets []struct {
		Name    string `json:"name"`
		Type    string `json:"type"`
		Content string `json:"content"`
	} `json:"storage_targets"`
}

// RestoreTestChecker watches restore-test outcomes per customer.
type RestoreTestChecker struct {
	store   *store.Store
	logger  *log.Logger
	onEvent EventNotifyFunc
	now     func() time.Time

	mu sync.Mutex
	// failStates is edge-trigger state: customerID|archive → already reported. A failing tier is
	// re-reported only when the FAILING RUN changes, so a permanently broken tier does not emit
	// every 60s sweep — the flapping-spam rule every checker here follows.
	failStates map[string]bool
	// staleStates is customerID|tier → "ok"|"stale", so the stale signal is edge-triggered too.
	staleStates map[string]string
}

// NewRestoreTestChecker builds the checker. onEvent may be nil.
func NewRestoreTestChecker(s *store.Store, onEvent EventNotifyFunc, logger *log.Logger) *RestoreTestChecker {
	return &RestoreTestChecker{
		store: s, logger: logger, onEvent: onEvent,
		now:         func() time.Time { return time.Now().UTC() },
		failStates:  map[string]bool{},
		staleStates: map[string]string{},
	}
}

// Check sweeps every active customer. Safe to call on the shared monitor ticker.
func (c *RestoreTestChecker) Check() {
	ids, err := c.store.GetActiveCustomerIDs()
	if err != nil {
		c.logger.Printf("[WARN] restore-test check: failed to list customers: %v", err)
		return
	}
	now := c.now()
	for _, id := range ids {
		latest, rerr := c.store.GetLatestHostReportJSON(id)
		if rerr != nil || latest == "" {
			continue // no agent host-report → nothing to judge (the R-81 legacy-shape branch)
		}
		c.checkFailure(id, latest)
		c.checkStaleness(id, latest, now)
	}
}

// checkFailure raises restore_test_failed when the latest report carries a FAILED run.
func (c *RestoreTestChecker) checkFailure(customerID, reportJSON string) {
	var hr hostReportRestoreTests
	if json.Unmarshal([]byte(reportJSON), &hr) != nil {
		return
	}
	for _, rt := range hr.RestoreTests {
		if rt.Pass {
			// A pass CLEARS the edge state for that archive so a later failure re-reports.
			c.mu.Lock()
			delete(c.failStates, customerID+"|"+rt.SourceArchive)
			c.mu.Unlock()
			continue
		}
		key := customerID + "|" + rt.SourceArchive
		c.mu.Lock()
		already := c.failStates[key]
		c.failStates[key] = true
		c.mu.Unlock()
		if already {
			continue
		}
		tier := rt.SourceTier
		if tier == "" {
			tier = "unknown"
		}
		msg := fmt.Sprintf("Restore-test FAILED on the %s tier: archive %s could not be restored+booted (%s)",
			tier, rt.SourceArchive, rt.Error)
		c.emit(customerID, EventRestoreTestFailed, "error", msg)
	}
}

// checkStaleness raises restore_test_stale for a tier not PROVEN within restoreProvenStaleAfter.
//
// "Proven" means a PASSING run for that tier, found across the hub's retained host-report window —
// the R-81 mechanism. The window is load-bearing here: the agent reports only its LATEST
// restore-test and its store is in-memory, so the latest report alone cannot answer "when was the
// OTHER tier last proven?". The hub's own history can.
func (c *RestoreTestChecker) checkStaleness(customerID, latestJSON string, now time.Time) {
	var latest hostReportRestoreTests
	if json.Unmarshal([]byte(latestJSON), &latest) != nil {
		return
	}
	tiers := expectedRestoreTiers(latest)
	if len(tiers) == 0 {
		return
	}

	rows, err := c.store.GetHostReportsSince(customerID, now.Add(-restoreWindowRead))
	if err != nil {
		c.logger.Printf("[WARN] restore-test check: window read failed for %s: %v", customerID, err)
		return
	}
	proven := lastProvenPerTier(rows)
	intervals := observedArchiveIntervals(rows)

	first, ferr := c.store.GetFirstHostReportAt(customerID)
	if ferr != nil {
		c.logger.Printf("[WARN] restore-test check: first-contact read failed for %s: %v", customerID, ferr)
		return
	}

	for _, tier := range tiers {
		observed, observedOK := intervals[tier]
		window := restoreProvenWindow(tier, observed, observedOK)
		v := assessRestoreProven(tier, proven[tier], first, now, window)
		key := customerID + "|" + tier
		c.mu.Lock()
		prev := c.staleStates[key]
		state := "ok"
		if v.verdict == verdictMissed {
			state = "stale"
		}
		c.staleStates[key] = state
		c.mu.Unlock()

		switch {
		case state == "stale" && prev != "stale":
			c.emit(customerID, EventRestoreTestStale, "warning", v.reason)
		case v.verdict == verdictUnknown && prev == "":
			// Make the deferral VISIBLE exactly once, the v0.73.0 / R-81 precedent: a quiet check
			// must never be indistinguishable from a check that did not run.
			c.logger.Printf("[INFO] restore-test staleness: %s %s", customerID, v.reason)
		}
	}
}

// assessRestoreProven is the per-tier verdict. PURE (now and the window injected) so the policy is
// unit-tested — the property that made R-81 provable, kept deliberately.
//
//	no proof, anchor NOT elapsed → UNKNOWN (newborn box; never an alarm)
//	no proof, anchor elapsed     → MISSED
//	proof older than the window  → MISSED
//	otherwise                    → OK
//
// `window` is now the TIER'S OWN (R-86 Part 2) rather than one constant for every tier, and every
// reason string states the window it was judged against. That is R-100's corollary applied here:
// when a verdict changes what it counts from, the alarm text has to change with it, or an operator
// reads "limit 168h" under a tier that was actually judged at 288h and dismisses a true alarm.
func assessRestoreProven(tier string, provenAt, firstReportAt, now time.Time, window time.Duration) backupAssessment {
	if provenAt.IsZero() {
		if firstReportAt.IsZero() {
			return backupAssessment{verdict: verdictMissed,
				reason: fmt.Sprintf("%s tier: never restore-proven, and no first-contact anchor to defer against", tier)}
		}
		watched := now.Sub(firstReportAt)
		if watched <= window {
			return backupAssessment{verdict: verdictUnknown,
				reason: fmt.Sprintf("%s tier: not restore-proven yet, but only watching for %s (grace %s since first contact %s) — newborn, not a fault",
					tier, watched.Round(time.Hour), window, firstReportAt.Format(time.RFC3339))}
		}
		return backupAssessment{verdict: verdictMissed,
			reason: fmt.Sprintf("%s tier: NEVER successfully restore-proven in %s of watching (limit %s, this tier's own backup rhythm) — the tier is unverified, not known-broken",
				tier, watched.Round(time.Hour), window)}
	}
	if age := now.Sub(provenAt); age > window {
		return backupAssessment{verdict: verdictMissed,
			reason: fmt.Sprintf("%s tier: last successful restore-test was %s ago (limit %s, this tier's own backup rhythm) — the tier is unverified, not known-broken",
				tier, age.Round(time.Hour), window)}
	}
	return backupAssessment{verdict: verdictOK}
}

// observedArchiveIntervals estimates how often each RESTORE tier actually receives an archive, from
// the host-reports the hub already holds. Keyed by restore-tier name ("local"/"pbs").
//
// Evidence is every distinct archive timestamp in the window: `pbs_snapshots[]` for the offsite
// tier (PBS enumerates its whole retention in each report, so one report usually settles the
// question) and successful `backups[]` records attributed by TARGET TYPE for both tiers — the
// slice-A.4 rule, because a PBS-targeted vzdump appears in BOTH arrays and classifying by array
// membership would attribute an offsite archive to the host tier.
//
// The estimate is the MEAN gap: (newest − oldest) / (generations − 1). It needs two generations;
// with fewer, ok=false and the caller falls back to the declared rhythm. It is deliberately crude,
// and can afford to be: restoreProvenWindow clamps the result between a 7-day floor and a 12-day
// cap, so the only discrimination this has to get right is "roughly daily" versus "several days or
// slower" — which is exactly the distinction that turns a healthy weekly tier into a false alarm.
func observedArchiveIntervals(rows []store.HostReportRow) map[string]time.Duration {
	seen := map[string]map[int64]struct{}{ // tier → set of archive unix times
		"local": {},
		"pbs":   {},
	}
	add := func(tier string, t time.Time) {
		if t.IsZero() {
			return
		}
		seen[tier][t.UTC().Unix()] = struct{}{}
	}

	for _, r := range rows {
		var hr hostReportBackups
		if json.Unmarshal([]byte(r.ReportJSON), &hr) != nil {
			continue // one malformed retained report must not blind the scan
		}
		pbs := pbsTargetSet(hr)
		for _, ps := range hr.PBSSnapshots {
			if t, ok := parseBackupTime(ps.BackupTime); ok {
				add("pbs", t)
			}
		}
		for _, b := range hr.Backups {
			if !b.Success {
				continue
			}
			t, ok := parseBackupTime(b.StartedAt)
			if !ok {
				continue
			}
			if pbs[b.TargetID] {
				add("pbs", t)
			} else {
				add("local", t)
			}
		}
	}

	out := map[string]time.Duration{}
	for tier, set := range seen {
		if len(set) < 2 {
			continue // not observable — the caller uses the declared rhythm
		}
		var oldest, newest int64
		first := true
		for ts := range set {
			if first || ts < oldest {
				oldest = ts
			}
			if first || ts > newest {
				newest = ts
			}
			first = false
		}
		span := time.Duration(newest-oldest) * time.Second
		if span <= 0 {
			continue
		}
		out[tier] = span / time.Duration(len(set)-1)
	}
	return out
}

// expectedRestoreTiers names the tiers this box actually HAS, so a box without an offsite tier is
// never reported stale for one. Same gate as Slice C's `expected`, and for the same reason: without
// it every box lacking a tier would alarm once the anchor elapsed — absence-is-not-failure,
// re-introduced one level down.
func expectedRestoreTiers(hr hostReportRestoreTests) []string {
	var out []string
	seenHost, seenOffsite := false, false
	for _, st := range hr.StorageTargets {
		isPBS := st.Type == "pbs"
		if isPBS && !seenOffsite {
			out = append(out, "pbs")
			seenOffsite = true
			continue
		}
		if !isPBS && !seenHost && containsBackupContent(st.Content) {
			out = append(out, "local")
			seenHost = true
		}
	}
	return out
}

func containsBackupContent(content string) bool {
	for i := 0; i+6 <= len(content); i++ {
		if content[i:i+6] == "backup" {
			return true
		}
	}
	return false
}

// lastProvenPerTier walks the retained window for the newest PASSING run per tier.
func lastProvenPerTier(rows []store.HostReportRow) map[string]time.Time {
	out := map[string]time.Time{}
	for _, r := range rows {
		var hr hostReportRestoreTests
		if json.Unmarshal([]byte(r.ReportJSON), &hr) != nil {
			continue // one malformed retained report must not blind the scan
		}
		for _, rt := range hr.RestoreTests {
			if !rt.Pass || rt.SourceTier == "" {
				continue
			}
			t, err := time.Parse(time.RFC3339, rt.TestedAt)
			if err != nil {
				continue
			}
			if cur, ok := out[rt.SourceTier]; !ok || t.After(cur) {
				out[rt.SourceTier] = t.UTC()
			}
		}
	}
	return out
}

// emit saves the event and notifies. OPERATOR-TIER ONLY: neither type has a `customerMessages`
// entry, so the dispatcher cannot route it to a customer. That is deliberate — a customer can take
// no action on a failed restore-test, and „a visszaállítási teszt nem sikerült" would frighten
// without informing. A PERSISTENTLY unproven DR tier may eventually warrant a customer-visible
// statement, but that needs copy review, not a side effect of this task.
func (c *RestoreTestChecker) emit(customerID, eventType, severity, msg string) {
	if _, err := c.store.SaveEvent(customerID, eventType, severity, msg, "{}", "hub"); err != nil {
		c.logger.Printf("[WARN] restore-test check: failed to save %s for %s: %v", eventType, customerID, err)
		return
	}
	c.logger.Printf("[%s] restore-test: %s %s", severityLabel(severity), customerID, msg)
	if c.onEvent != nil {
		c.onEvent(customerID, eventType, severity, msg, "{}", "hub")
	}
}

func severityLabel(sev string) string {
	if sev == "error" {
		return "ERROR"
	}
	return "WARN"
}