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

package monitor

import (
	"fmt"
	"log"
	"strings"
	"sync"
	"time"

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

// EventNotifyFunc is called after a hub-generated event is saved,
// to trigger notification dispatch. Keeps monitor decoupled from notify.
type EventNotifyFunc func(customerID, eventType, severity, message, detailsJSON, source string)

// StateDisabled is the state of a customer whose box we DELIBERATELY told to stop reporting.
//
// R-321. Switching a box's hub reporting off is a supported product state: the controller sends one
// final report carrying health.status = "disabled" (felhom-controller
// `cmd/controller/main.go:1253`) and then goes quiet BY DESIGN. That status is parsed and persisted
// (`store.go:953-955` → `reports.health_status`), reaches this checker on every pass inside
// `CustomerSummary.HealthStatus` (`store.go:40`), and the operator roll-up already renders such a
// customer as `disabled` (`web/rollup.go:25`).
//
// This checker ignored it and measured only the AGE of the last report — so a machine we asked to be
// quiet went stale at 30 minutes, down at 60, and e-mailed the operator twice about an outage we
// caused on purpose. That is the false alarm that teaches people to ignore the channel, and it is a
// sibling of R-195, where the guard that should have covered a customer was keyed off the wrong fact.
//
// It is a STATE here rather than a deletion (the `blocked` branch below deletes) because other
// checkers consult GetState: CheckBackupDeadlines skips a customer that is "down", and a DELETED
// state returns "" — which is not "down", so a disabled box would have gone on alarming
// `expected_backup_missed` from a different function. Exactly the R-195 shape returning through a
// second door.
const StateDisabled = "disabled"

// StalenessChecker monitors customer report freshness and generates
// node_stale / node_down / node_recovered events on state transitions.
type StalenessChecker struct {
	store     *store.Store
	threshold time.Duration // "stale" after this duration (default 30m)
	downAfter time.Duration // "down" after this duration (2x threshold)
	logger    *log.Logger
	onEvent   EventNotifyFunc

	mu            sync.Mutex
	states        map[string]string    // customerID → "ok" | "stale" | "down"
	downtimeStart map[string]time.Time // customerID → when node first became unreachable
}

// NewStalenessChecker creates a checker and initializes state from current data.
// No events are generated during initialization (binding #12).
// onEvent is called after each hub-generated event is saved (may be nil).
func NewStalenessChecker(s *store.Store, threshold time.Duration, onEvent EventNotifyFunc, logger *log.Logger) *StalenessChecker {
	sc := &StalenessChecker{
		store:         s,
		threshold:     threshold,
		downAfter:     2 * threshold,
		logger:        logger,
		onEvent:       onEvent,
		states:        make(map[string]string),
		downtimeStart: make(map[string]time.Time),
	}

	// Seed states from current report timestamps — no events on init
	customers, err := s.GetCustomers()
	if err != nil {
		logger.Printf("[WARN] Staleness checker: failed to seed states: %v", err)
		return sc
	}

	var okCount, staleCount, downCount int
	for _, c := range customers {
		if s.IsCustomerBlocked(c.CustomerID) {
			continue
		}
		age := time.Since(c.ReceivedAt)
		switch {
		case age > sc.downAfter:
			sc.states[c.CustomerID] = "down"
			downCount++
		case age > sc.threshold:
			sc.states[c.CustomerID] = "stale"
			staleCount++
		default:
			sc.states[c.CustomerID] = "ok"
			okCount++
		}
	}
	logger.Printf("[INFO] Staleness checker initialized: %d ok, %d stale, %d down (node_stale after %s, node_down after %s)", okCount, staleCount, downCount, threshold, 2*threshold)

	return sc
}

// Check evaluates all customers and emits events on state transitions.
// Should be called periodically (every 60s).
func (sc *StalenessChecker) Check() {
	customers, err := sc.store.GetCustomers()
	if err != nil {
		sc.logger.Printf("[WARN] Staleness check failed: %v", err)
		return
	}

	sc.mu.Lock()
	defer sc.mu.Unlock()

	// Track which customers are still present (to clean up removed ones)
	seen := make(map[string]bool, len(customers))

	for _, c := range customers {
		seen[c.CustomerID] = true

		if sc.store.IsCustomerBlocked(c.CustomerID) {
			delete(sc.states, c.CustomerID)
			continue
		}

		// R-321 — a machine we told to be quiet is not a machine that died.
		//
		// The age-based transition is skipped entirely; no stale, no down, no e-mail. The state is
		// RECORDED rather than deleted so the deliberate silence is visible to anything that asks
		// (see StateDisabled). `downtimeStart` is cleared on ENTRY so that a later, genuine outage
		// cannot compute its duration from a clock that started before we asked for the silence.
		//
		// The discriminator is the box's OWN last word, not an inference: it said `disabled` on the
		// way out. LIMIT, stated rather than hidden: if reporting is re-enabled and the box then
		// fails to report at all, the hub still sees only that final `disabled` report and keeps
		// suppressing. The hub cannot distinguish that from "still switched off" — its view changes
		// only when a report arrives. This is why the state is made VISIBLE: an operator who
		// re-enabled a box and still sees `disabled` is being told it has not come back.
		if strings.EqualFold(c.HealthStatus, StateDisabled) {
			if sc.states[c.CustomerID] != StateDisabled {
				sc.states[c.CustomerID] = StateDisabled
				delete(sc.downtimeStart, c.CustomerID)
			}
			continue
		}

		age := time.Since(c.ReceivedAt)
		var newState string
		switch {
		case age > sc.downAfter:
			newState = "down"
		case age > sc.threshold:
			newState = "stale"
		default:
			newState = "ok"
		}

		oldState := sc.states[c.CustomerID]
		if oldState == "" || oldState == StateDisabled {
			// New customer — set state without event.
			//
			// R-321 adds the re-enabled case to the SAME branch, deliberately. A box coming back
			// from a deliberate silence is a first observation, not a recovery: emitting here would
			// send `node_recovered` for an outage that never happened — and would do it every time
			// an operator switched reporting back on.
			//
			// THE RE-ENABLEMENT CLOCK, and this is the judgement: it is the age of the report the
			// hub can actually see. For a box that reports on re-enabling, that report IS the
			// re-enablement, so the clock starts there and scenario D (re-enabled, then genuinely
			// quiet) alarms on a normal schedule timed from the moment it came back. Timing from the
			// last report BEFORE the switch-off would fire an instant stale/down for a quiet period
			// we asked for — a false alarm produced by fixing false alarms.
			sc.states[c.CustomerID] = newState
			continue
		}

		if oldState == newState {
			continue
		}

		// State transition — emit event
		sc.states[c.CustomerID] = newState
		if newState == "stale" && oldState == "ok" {
			// Record when node first became unreachable
			sc.downtimeStart[c.CustomerID] = time.Now()
		}
		downtimeDur := age
		if newState == "ok" {
			if t, ok := sc.downtimeStart[c.CustomerID]; ok {
				downtimeDur = time.Since(t)
			}
			delete(sc.downtimeStart, c.CustomerID)
		}
		sc.emitTransition(c.CustomerID, oldState, newState, downtimeDur)
	}

	// Clean up customers that no longer have reports
	for id := range sc.states {
		if !seen[id] {
			delete(sc.states, id)
			delete(sc.downtimeStart, id)
		}
	}
}

// GetState returns the current staleness state for a customer.
func (sc *StalenessChecker) GetState(customerID string) string {
	sc.mu.Lock()
	defer sc.mu.Unlock()
	s := sc.states[customerID]
	if s == "" {
		return "unknown"
	}
	return s
}

func (sc *StalenessChecker) emitTransition(customerID, oldState, newState string, age time.Duration) {
	var eventType, severity, message string

	switch {
	case newState == "stale":
		eventType = "node_stale"
		severity = "warning"
		message = "No report received for " + formatDuration(age)
	case newState == "down":
		eventType = "node_down"
		severity = "error"
		message = "No report received for " + formatDuration(age)
	case newState == "ok" && (oldState == "stale" || oldState == "down"):
		eventType = "node_recovered"
		severity = "info"
		message = "Reports resumed (was " + oldState + " for " + formatDuration(age) + ")"
	default:
		return
	}

	sc.logger.Printf("[INFO] Staleness: %s %s → %s (%s)", customerID, oldState, newState, eventType)

	if _, err := sc.store.SaveEvent(customerID, eventType, severity, message, "{}", "hub"); err != nil {
		sc.logger.Printf("[WARN] Failed to save staleness event for %s: %v", customerID, err)
		return
	}

	if sc.onEvent != nil {
		sc.onEvent(customerID, eventType, severity, message, "{}", "hub")
	}
}

func formatDuration(d time.Duration) string {
	if d < time.Hour {
		return fmt.Sprintf("%dm", int(d.Minutes()))
	}
	h := int(d.Hours())
	m := int(d.Minutes()) % 60
	if m == 0 {
		return fmt.Sprintf("%dh", h)
	}
	return fmt.Sprintf("%dh%dm", h, m)
}