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"
	// stateAuthFailed (R-39, agent >= 0.91.0) — the box HAS a credential, the descriptor is applied,
	// and PBS rejects it with 401. Before 0.91.0 this state could not exist: the agent's verify loop
	// read the credential file directly as non-root, always failed with "permission denied", and
	// skipped — so an applied-and-dead tier was invisible to both tiers. It is healed like
	// consumed_failed (escalate to a fresh mint), never by a re-stage: re-staging re-feeds the SAME
	// secret PBS just rejected.
	stateAuthFailed = "auth_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)
	eventAuthFailed     = "pbsdr_selfheal_auth_failed"     // PBS rejected the box's credential (401) → minted a fresh one
	eventUnconsumed     = "pbsdr_unconsumed_secret"        // staged secret never consumed under an `applied` box (surfaced, NOT auto-healed)
)

// unconsumedGrace bounds how long a staged-but-unconsumed secret is NORMAL before it is a lie.
//
// A fresh mint (or a deliberate re-stage) is consumed on the agent's next tick — well inside one
// ~15-minute report cycle. Beyond this window, an unconsumed secret sitting under a box that reports
// `applied` is the exact fingerprint of the 2026-07-18 N100 failure: hub minted, agent
// short-circuited, both tiers green, box serving a revoked credential.
const unconsumedGrace = 15 * time.Minute

// 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
	// unconsumedSeen remembers the last report id already surfaced per host (Scenario F), so a
	// sustained disagreement produces one event per fresh report rather than one per tick.
	unconsumedSeen map[string]int64
}

// 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{},
		unconsumedSeen:  map[string]int64{},
	}
}

// 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
		}
		// R-39 consumed_at HONESTY (Scenario F). Deliberately a SURFACE, not a heal: the remediation
		// for a genuinely stuck box is the auth_failed / waiting_secret / consumed_failed machinery
		// above, driven by what the BOX reports. This check catches the disagreement itself — the hub
		// staged a credential the box never took, while the box claims to be applied — which is a
		// state no single tier can detect alone. Auto-re-issuing on it would mint a second secret on
		// top of an unconsumed one: exactly the mint/consume race R-39(a) already recorded.
		r.checkUnconsumed(row)

		switch row.ReportedState {
		case stateWaitingSecret:
			if r.confirm(row) {
				r.healWaitingSecret(ctx, row)
			}
		case stateConsumedFailed:
			if r.confirm(row) {
				r.healConsumedFailed(ctx, row)
			}
		case stateAuthFailed:
			if r.confirm(row) {
				r.healAuthFailed(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)
		}
	}
}

// checkUnconsumed surfaces the applied-but-never-consumed disagreement (Scenario F). One event per
// distinct report, so a sustained state does not spam the audit log every tick.
func (r *Reconciler) checkUnconsumed(row store.PBSDRHealRow) {
	if row.SecretUnconsumedFor <= unconsumedGrace {
		return // no staged secret, already consumed, or still inside the normal pickup window
	}
	// Only meaningful when the box claims to be converged. A box that honestly reports
	// waiting_secret / consumed_failed / auth_failed is already being healed above and its unconsumed
	// secret is the SYMPTOM being fixed, not a contradiction.
	if row.ReportedState != "applied" && row.ReportedState != "adopted" {
		return
	}
	if st, ok := r.unconsumedSeen[row.HostID]; ok && st == row.ReportID {
		return // already surfaced for this exact report
	}
	if r.unconsumedSeen == nil {
		r.unconsumedSeen = map[string]int64{}
	}
	r.unconsumedSeen[row.HostID] = row.ReportID
	r.logger.Printf("[WARN] pbsdrheal: host %s (customer %s) reports pbs_dr=%s while a staged one-time secret has been UNCONSUMED for %s — the box is not using the credential the hub issued (R-39 fingerprint)",
		row.HostID, row.CustomerID, row.ReportedState, row.SecretUnconsumedFor.Round(time.Minute))
	r.event(row.CustomerID, eventUnconsumed, "warning",
		"PBS-DR honesty check: the box reports the DR tier as applied, but a one-time credential issued by the hub has never been consumed. The tier may be authenticating with a superseded credential.")
}

// 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)
	}
}

// healAuthFailed escalates a box whose credential PBS rejects (401) to a fresh mint. This is the leg
// that closes the R-39 loop end to end: the agent now PROVES the credential is dead instead of
// silently skipping, the hub re-keys, the fresh mint advances the secret generation, the descriptor
// hash moves, and the agent finally re-consumes (Scenario A). Before this, an applied-and-401 tier
// stayed green forever and would have surfaced first at a real restore.
//
// A re-stage is deliberately NOT attempted: the stored secret IS the one PBS just rejected, so
// re-arming it would burn a tick and change nothing. Damping is the SHARED confirm() — a 401 flap
// must not turn into a secret-minting chain.
func (r *Reconciler) healAuthFailed(ctx context.Context, row store.PBSDRHealRow) {
	r.logger.Printf("[WARN] pbsdrheal: host %s (customer %s) reports auth_failed (PBS rejects its credential) — escalating to Re-issue", row.HostID, row.CustomerID)
	if r.reissue(ctx, row) {
		r.event(row.CustomerID, eventAuthFailed, "warning",
			"PBS-DR self-heal: a box reported auth_failed (the DR endpoint rejected its stored credential); re-issued fresh endpoint credentials so the agent can re-consume and 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)
	}
}