felhom.eu/hub/internal/pbsdrheal/reconciler.go

// Package pbsdrheal is the hub-side PBS-DR self-heal reconciler (TASK 2026-07-15, from
// SPIKE-pbsdr-selfheal-2026-07-15 / e8f8c44).
//
// Root cause the spike proved (not reasoned): a customer box re-installed / restored / rolled back
// onto its STABLE host_id loses its agent-side converged marker; the hub still holds the durable
// pbs_dr descriptor (enabled) and the durable *consumed* one-time secret, and the WG peer still
// exists (same pubkey → changed==false → the provision cascade cannot re-fire). The agent gets the
// descriptor, verifies over the tunnel, but ConsumePBSToken returns "no secret" — it sits in
// pbs_dr.state="waiting_secret" forever. PBS-DR never converges, so escrow can't run and offsite
// never arms. The MISSING PIECE IS A CONSUMABLE SECRET, NOT THE DESCRIPTOR (spike SQ-2b′:
// re-staging the stored secret converged the box in one ~30 s agent tick, using the existing ep0
// token, zero churn).
//
// This reconciler, for each host whose descriptor is enabled+provisioned and whose LATEST report is
// a stuck state sustained across a debounce, RE-STAGES the stored secret (store.RestageHostPBSSecret
// — no ep0 call, no generation bump) and escalates to the existing Re-issue only when there is no
// stored secret to re-stage or the agent burned one (consumed_failed). It NEVER re-runs the whole
// provision atom (which refuses ErrTokenExists) and NEVER blind-timer-reissues (hash/gen thrash).
// A converged/healthy/disabled host is a pure no-op (idempotency — Scenario C).
package pbsdrheal

import (
	"context"
	"log"
	"time"

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

// The reported agent pbs_dr.state strings (felhom-agent internal/pbsdr/manager.go). Only the two
// STUCK states below are acted on; everything else (applied, adopted, disabled, verify_failed, "")
// is a no-op. verify_failed is DELIBERATELY not healed: it is a descriptor/PBS-reachability problem
// (the secret is untouched by verify-before-consume), self-heals when the tunnel recovers, and must
// stay LOUD for the operator if it does not — re-staging a secret would not help it.
const (
	stateWaitingSecret = "waiting_secret"
	stateConsumedFailed = "consumed_failed"
)

// Audit event types (store.SaveEvent; hub-internal, not gated by allowedEventTypes). Distinct per
// remediation so the operator sees exactly what was auto-done (Scenario E wants a distinct signal).
const (
	eventRestaged       = "pbsdr_selfheal_restaged"        // re-armed the stored secret (routine)
	eventReissued       = "pbsdr_selfheal_reissued"        // no stored secret → minted a fresh one
	eventConsumedFailed = "pbsdr_selfheal_consumed_failed" // burned secret → minted a fresh one (a real problem was remediated)
)

// Actions is the mutation seam — fakes in tests count calls without SSH/ep0. Restage flips a stored
// secret's consumed flag (returns restaged=false when NO row exists → the caller escalates). Reissue
// mints a fresh ep0 token + stores a fresh consume-once secret + bumps the descriptor.
type Actions interface {
	Restage(hostID string) (restaged bool, err error)
	Reissue(ctx context.Context, customerID string) error
}

// Reissuer is satisfied by *web.Server (its ReissuePBSDR). Kept here so main.go can wire the server
// as the escalation path without an import cycle.
type Reissuer interface {
	ReissuePBSDR(ctx context.Context, customerID string) error
}

// storeActions is the production Actions: Restage → the store primitive; Reissue → the web server.
type storeActions struct {
	st       *store.Store
	reissuer Reissuer
}

func (a storeActions) Restage(hostID string) (bool, error) { return a.st.RestageHostPBSSecret(hostID) }
func (a storeActions) Reissue(ctx context.Context, customerID string) error {
	return a.reissuer.ReissuePBSDR(ctx, customerID)
}

// NewActions builds the production mutation seam.
func NewActions(st *store.Store, reissuer Reissuer) Actions { return storeActions{st: st, reissuer: reissuer} }

// debounceState tracks, per host, the last DISTINCT report observed and how many consecutive
// distinct reports it has held the current stuck state — so a fresh box that briefly shows
// waiting_secret between provision and its first consume (resolved within one 60 s agent tick, well
// inside one ~15 min report cycle) is NOT healed on a single report (Scenario D).
type debounceState struct {
	reportID int64
	state    string
	streak   int
}

// Reconciler re-arms stuck PBS-DR hosts. DECLARATIVE + IDEMPOTENT: a tick over a converged fleet
// writes nothing (Scenario C). It reads the hub DB (the source of truth) — never the box.
type Reconciler struct {
	store    *store.Store
	act      Actions
	interval time.Duration
	debounceReports int // distinct stuck reports required before healing (default 2)
	onlyHost string // "" = whole fleet; non-empty restricts the work set to one host (supervised rollout)
	trigger  chan struct{}
	logger   *log.Logger

	deb map[string]debounceState
}

// NewReconciler builds the reconciler. interval defaults to 5m, debounceReports to 2.
func NewReconciler(st *store.Store, act Actions, logger *log.Logger) *Reconciler {
	if logger == nil {
		logger = log.Default()
	}
	return &Reconciler{
		store:           st,
		act:             act,
		interval:        5 * time.Minute,
		debounceReports: 2,
		trigger:         make(chan struct{}, 1),
		logger:          logger,
		deb:             map[string]debounceState{},
	}
}

// RestrictToHost scopes the reconciler's work set to a single host_id (empty = whole fleet). Used
// for a supervised first rollout (PBSDRHEAL_ONLY_HOST): validate the converged-host no-op live on the
// drill guest before widening to the fleet. Set before Run.
func (r *Reconciler) RestrictToHost(hostID string) { r.onlyHost = hostID }

// Trigger requests an immediate reconcile (tests + the mutation handlers). Non-blocking.
func (r *Reconciler) Trigger() {
	select {
	case r.trigger <- struct{}{}:
	default:
	}
}

// Run loops until ctx is done, reconciling on each tick or Trigger. Never exits on an error.
func (r *Reconciler) Run(ctx context.Context) {
	ticker := time.NewTicker(r.interval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-r.trigger:
		case <-ticker.C:
		}
		r.reconcileOnce(ctx)
	}
}

// reconcileOnce reads the whole fleet's DR-heal state and acts on the stuck, debounce-confirmed,
// enabled+provisioned hosts. Errors are logged and retried next tick.
func (r *Reconciler) reconcileOnce(ctx context.Context) {
	rows, err := r.store.PBSDRHealStates()
	if err != nil {
		r.logger.Printf("[ERROR] pbsdrheal: read heal states: %v (retry next tick)", err)
		return
	}
	seen := make(map[string]bool, len(rows))
	for _, row := range rows {
		if r.onlyHost != "" && row.HostID != r.onlyHost {
			continue // supervised rollout scope: only this host is in the work set
		}
		seen[row.HostID] = true
		// Work set: descriptor enabled AND provisioned (namespace set). A DR-OFF / disabled / bare-
		// enabled-but-never-provisioned host is out of scope (Scenario F) — forget any debounce.
		if !row.DescriptorEnabled || !row.DescriptorProvisioned {
			delete(r.deb, row.HostID)
			continue
		}
		switch row.ReportedState {
		case stateWaitingSecret:
			if r.confirm(row) {
				r.healWaitingSecret(ctx, row)
			}
		case stateConsumedFailed:
			if r.confirm(row) {
				r.healConsumedFailed(ctx, row)
			}
		default:
			// applied | adopted | disabled | verify_failed | "" (no report) | anything else → no-op.
			delete(r.deb, row.HostID)
		}
	}
	// Drop debounce state for hosts that vanished from the fleet.
	for h := range r.deb {
		if !seen[h] {
			delete(r.deb, h)
		}
	}
}

// confirm advances the per-host debounce and reports whether the stuck state has held across
// >= debounceReports DISTINCT reports. A re-observed same report (same reportID) never advances the
// streak — the debounce counts fresh evidence, not reconciler ticks.
func (r *Reconciler) confirm(row store.PBSDRHealRow) bool {
	st := r.deb[row.HostID]
	if row.ReportID != st.reportID {
		if row.ReportedState == st.state {
			st.streak++
		} else {
			st.streak = 1
		}
		st.state = row.ReportedState
		st.reportID = row.ReportID
		r.deb[row.HostID] = st
	}
	return st.streak >= r.debounceReports
}

// healWaitingSecret re-stages the stored secret; if none is stored, escalates to Re-issue.
func (r *Reconciler) healWaitingSecret(ctx context.Context, row store.PBSDRHealRow) {
	restaged, err := r.act.Restage(row.HostID)
	if err != nil {
		r.logger.Printf("[ERROR] pbsdrheal: re-stage %s: %v (retry next tick)", row.HostID, err)
		return
	}
	if restaged {
		r.logger.Printf("[INFO] pbsdrheal: re-staged the stored one-time secret for host %s (customer %s) stuck in waiting_secret — the agent re-consumes on its next tick (no ep0 token minted, no generation bump)", row.HostID, row.CustomerID)
		r.event(row.CustomerID, eventRestaged, "info",
			"PBS-DR self-heal: re-staged the stored one-time credential for a box stuck awaiting a secret (re-install/rollback recovery). No endpoint token was minted.")
		r.resetAfterHeal(row)
		return
	}
	// No stored secret to re-stage → escalate to a fresh mint.
	r.logger.Printf("[INFO] pbsdrheal: host %s (customer %s) is waiting_secret with NO stored secret — escalating to Re-issue", row.HostID, row.CustomerID)
	if r.reissue(ctx, row) {
		r.event(row.CustomerID, eventReissued, "warning",
			"PBS-DR self-heal: re-issued endpoint credentials for a box awaiting a secret that the hub no longer had stored.")
		r.resetAfterHeal(row)
	}
}

// healConsumedFailed escalates a burned-secret box to a fresh mint — a re-stage of the SAME secret
// would only re-feed the credential the agent already burned into a failed apply (Scenario E).
func (r *Reconciler) healConsumedFailed(ctx context.Context, row store.PBSDRHealRow) {
	r.logger.Printf("[WARN] pbsdrheal: host %s (customer %s) reports consumed_failed (burned secret) — escalating to Re-issue", row.HostID, row.CustomerID)
	if r.reissue(ctx, row) {
		r.event(row.CustomerID, eventConsumedFailed, "warning",
			"PBS-DR self-heal: a box reported consumed_failed (it burned a one-time credential into a failed apply); re-issued fresh endpoint credentials so the agent can converge.")
		r.resetAfterHeal(row)
	}
}

// reissue runs the escalation; returns true on success (the caller then records the audit event).
func (r *Reconciler) reissue(ctx context.Context, row store.PBSDRHealRow) bool {
	if err := r.act.Reissue(ctx, row.CustomerID); err != nil {
		r.logger.Printf("[ERROR] pbsdrheal: re-issue for customer %s (host %s): %v (retry next tick)", row.CustomerID, row.HostID, err)
		return false
	}
	return true
}

// resetAfterHeal clears the streak (keeping the report id) so the same stuck report does not re-heal
// on the next tick — a FRESH report must re-confirm the host is still stuck before acting again.
func (r *Reconciler) resetAfterHeal(row store.PBSDRHealRow) {
	r.deb[row.HostID] = debounceState{reportID: row.ReportID, state: row.ReportedState, streak: 0}
}

// event records an audit row; a failure to write it must never break the heal loop.
func (r *Reconciler) event(customerID, eventType, severity, message string) {
	if _, err := r.store.SaveEvent(customerID, eventType, severity, message, "", "hub"); err != nil {
		r.logger.Printf("[WARN] pbsdrheal: audit event %s for %s not stored: %v", eventType, customerID, err)
	}
}