felhom-agent/internal/pbs/live_reporter.go

package pbs

import (
	"context"
	"errors"
	"log/slog"
	"time"

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

// DefaultLiveSnapshotTimeout bounds the per-collect live PBS reads so a slow/hung PBS can never
// stall the host-report (and thus the heartbeat). A few seconds is ample for the cheap list GET on
// the LAN; on overrun the reporter falls back to last-known-good rather than blocking.
const DefaultLiveSnapshotTimeout = 8 * time.Second

// LiveSnapshotReporter resolves the PBS snapshot inventory LIVE on each collect (the cheap
// Snapshots() list), so the host-report — and the DR recipe's pbs coord derived from it — is present
// whenever PBS is reachable, INDEPENDENT of the 6 h verify cadence. This closes the gap where a
// one-shot collect (selftest=hub) and the first window of every daemon after a restart saw the
// verify-loop's SnapshotStore still empty (→ no pbs coord, the restore SOURCE missing).
//
// On a per-datastore live error/timeout it falls back to the store's last-known-good; a successful
// (even empty) response is authoritative and updates the store. If targets cannot be resolved at all
// it returns the store's full aggregate. It shares the same *SnapshotStore the verify loop Records
// into, so the two keep each other's last-known-good warm. It NEVER triggers a verify — list only.
type LiveSnapshotReporter struct {
	targets Targets        // pbsTargetsFromPVE(...) closure — re-resolved each collect
	store   *SnapshotStore // shared last-known-good cache (verify loop writes it too)
	timeout time.Duration  // per-collect bound on the live reads
	log     *slog.Logger

	// listSnapshots is the production→PBS seam, overridable in tests so no live PBS is needed.
	// Default = liveListSnapshots (one Snapshots() GET, converted via Snapshot.ToHub()).
	listSnapshots func(ctx context.Context, t Target) ([]hub.PBSSnapshot, error)

	// probeAuth is the R-39 credential probe seam (default = (*Client).ProbeAuth). Overridable so the
	// auth-honesty path is testable with no PBS.
	probeAuth func(ctx context.Context, t Target) error
	// authSink receives every probe result. nil = nobody is listening (the probe is then skipped
	// entirely — no point paying for a request nothing consumes).
	authSink AuthSink
}

// AuthSink receives the result of each per-storage credential probe (R-39 leg c).
//
// It exists so the DR bridge can turn a 401 into a LOUD `auth_failed` state instead of the Warn-and-
// skip that made an applied-but-dead tier invisible. Deliberately a plain interface taking a bool
// rather than the error: the consumer (internal/pbsdr) must not have to import this package just to
// test a sentinel.
type AuthSink interface {
	// NoteAuthResult reports one storage's credential health. unauthorized=true means PBS REJECTED
	// the credential (401) — terminal until it is replaced. A transport error is unauthorized=false
	// with a non-empty detail: unknown, not dead, and never a reason to re-key.
	NoteAuthResult(storageID string, unauthorized bool, detail string)
}

// SetAuthSink wires the credential-probe consumer. Without it the reporter does not probe at all.
func (r *LiveSnapshotReporter) SetAuthSink(s AuthSink) { r.authSink = s }

// NewLiveSnapshotReporter builds a live reporter sharing store with the verify loop. A zero timeout
// falls back to DefaultLiveSnapshotTimeout; a nil logger to slog.Default.
func NewLiveSnapshotReporter(targets Targets, store *SnapshotStore, timeout time.Duration, log *slog.Logger) *LiveSnapshotReporter {
	if timeout <= 0 {
		timeout = DefaultLiveSnapshotTimeout
	}
	if log == nil {
		log = slog.Default()
	}
	return &LiveSnapshotReporter{
		targets:       targets,
		store:         store,
		timeout:       timeout,
		log:           log,
		listSnapshots: liveListSnapshots,
		probeAuth:     func(ctx context.Context, t Target) error { return t.Client.ProbeAuth(ctx) },
	}
}

// liveListSnapshots does ONE cheap snapshot list for a target and converts each record to the hub
// wire shape (reusing Snapshot.ToHub — RFC3339 backup_time, verify_state). List only; no verify.
func liveListSnapshots(ctx context.Context, t Target) ([]hub.PBSSnapshot, error) {
	snaps, err := t.Client.Snapshots(ctx, t.Datastore)
	if err != nil {
		return nil, err
	}
	out := make([]hub.PBSSnapshot, 0, len(snaps))
	for _, s := range snaps {
		out = append(out, s.ToHub())
	}
	return out, nil
}

// probeAuthAndReport runs the credential probe for one target and forwards the verdict to the sink.
// No sink → no probe (nothing would consume it). Never fails the collect: a report must still go out.
func (r *LiveSnapshotReporter) probeAuthAndReport(ctx context.Context, t Target) {
	if r.authSink == nil || r.probeAuth == nil {
		return
	}
	err := r.probeAuth(ctx, t)
	switch {
	case err == nil:
		r.authSink.NoteAuthResult(t.StorageID, false, "")
	case errors.Is(err, ErrUnauthorized):
		// The one case that is TERMINAL and actionable: the credential is rejected, not the network.
		r.log.Error("pbs: the DR endpoint REJECTED this box's credential (401) — the storage entry is "+
			"pinned to a superseded secret and every backup/restore against it will fail",
			"storage", t.StorageID, "datastore", t.Datastore)
		r.authSink.NoteAuthResult(t.StorageID, true, "PBS rejected the stored credential (401)")
	default:
		// Unreachable/timeout/TLS — UNKNOWN, not dead. Reporting this as unauthorized would re-key a
		// perfectly good credential on every network blip.
		r.log.Debug("pbs: credential probe inconclusive (not a rejection)", "storage", t.StorageID, "err", err)
		r.authSink.NoteAuthResult(t.StorageID, false, err.Error())
	}
}

// PBSSnapshots implements hub.PBSReporter with a single bounded, live pass (last-known-good
// fallback). Non-nil result so it marshals as [].
func (r *LiveSnapshotReporter) PBSSnapshots(ctx context.Context) []hub.PBSSnapshot {
	childCtx, cancel := context.WithTimeout(ctx, r.timeout)
	defer cancel()

	targets, err := r.targets(childCtx)
	if err != nil {
		// Cannot even enumerate datastores → fall back to the full last-known-good aggregate.
		r.log.Debug("pbs: live targets resolution failed; using last-known-good aggregate", "err", err)
		return r.store.PBSSnapshots(ctx)
	}

	out := []hub.PBSSnapshot{}
	for _, t := range targets {
		// R-39 leg (c): prove the credential BEFORE interpreting anything else about this datastore.
		// A snapshot list that fails with 401 used to look identical to "PBS is busy" — which is how
		// an applied-and-dead tier stayed green. The probe runs on this 15-minute collect path (not
		// the 6 h verify cadence) because that is how fast the hub can react.
		r.probeAuthAndReport(childCtx, t)

		snaps, err := r.listSnapshots(childCtx, t)
		if err != nil {
			// Per-datastore live failure → that datastore's last-known-good (does NOT clobber it).
			r.log.Debug("pbs: live list failed; using last-known-good", "datastore", t.Datastore, "err", err)
			out = append(out, r.store.Get(t.Datastore)...)
			continue
		}
		// A successful response (INCLUDING empty) is authoritative — record it as the new
		// last-known-good so a later error reuses fresh truth, not a stale set.
		r.store.Record(t.Datastore, snaps)
		out = append(out, snaps...)
	}
	return out
}