felhom-agent/internal/backup/runner.go

package backup

import (
	"context"
	"fmt"
	"log/slog"
	"sort"
	"strings"
	"sync"
	"time"

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

// BackupAPI is the read+backup proxmox surface the runner needs. *proxmox.Client satisfies it.
type BackupAPI interface {
	Vzdump(ctx context.Context, opts proxmox.VzdumpOptions) (string, error)
	WaitTask(ctx context.Context, upid string, opts proxmox.WaitOptions) (proxmox.TaskStatus, error)
	GuestConfig(ctx context.Context, vmid int) (proxmox.GuestConfig, error)
	StorageContent(ctx context.Context, store string) ([]proxmox.StorageContent, error)
	// ListStorage enumerates storages (name+type) — used to scope local-only retention (never prune PBS).
	ListStorage(ctx context.Context) ([]proxmox.Storage, error)
	// TaskLogTail reads trailing task-log lines — used to read the ACTUAL vzdump mode
	// (PVE may downgrade a requested snapshot to stop for a stopped guest — spike B1).
	TaskLogTail(ctx context.Context, upid string, limit int) ([]string, error)
}

// BackupRunner orchestrates a crash-consistent vzdump to a local target and reports the
// result (incl. the bulk-volume gap). An agent-initiated vzdump is crash-consistent only
// (no fsfreeze); the report says so.
type BackupRunner struct {
	api    BackupAPI
	target string             // backup storage (content=backup)
	mode   proxmox.BackupMode // default ModeSnapshot
	notes  string             // optional notes-template
	// retention is the per-run `--prune-backups` spec (e.g. "keep-last=3") applied to a LOCAL target after
	// each successful backup, so the agent's own backups can't pile up and refill root. Empty → no prune
	// (the legacy behaviour; restore-test/selftest runners pass ""). NEVER applied to a PBS target.
	retention string
	// waitTimeout bounds the WaitTask poll on this runner's vzdump. Per-TIER since R-82: 30m is
	// right for a local vzdump and badly wrong for an offsite PBS upload (see the 2026-07-26 live
	// failure recorded on config.BackupTargetConfig.WaitTimeoutSeconds). 0 → 30m (legacy).
	waitTimeout time.Duration
	// allowPBSPrune permits `--prune-backups` on a PBS-type target. OFF by default and ON only for
	// an ADDITIONAL tier whose keep_last was set explicitly (operator ruling 2026-07-26: keep two
	// weeks of weekly offsite backups).
	//
	// The blanket PBS refusal it replaces existed for a real reason and still applies to the
	// PRIMARY tier: BackupTarget() DEFAULTS to "felhom-pbs" and KeepLast() DEFAULTS to 3, so a box
	// with neither key set would silently prune its offsite DR to 3 restore points. An additional
	// tier cannot have that accident — its keep_last defaults to 0 (never prune), so any value
	// there is a deliberate act.
	allowPBSPrune bool
	logger        *slog.Logger
	now           func() time.Time
	// rejected remembers the volids already announced by warnRejectedArchiveOnce, so an incomplete
	// archive is reported ONCE rather than on every 5-minute due-check. Bounded in practice: one
	// entry per aborted upload, and a process restart clears it. Guarded by rejectedMu because the
	// due-check is served from the local-API handler goroutines.
	rejectedMu sync.Mutex
	rejected   map[string]struct{}
}

// NewBackupRunner builds a runner. mode defaults to snapshot (works for a stopped guest and
// for lvm-thin); the caller may pass ModeStop for storages without snapshot support. retention is the
// per-run prune spec ("keep-last=N", or "" to never prune) — only the periodic local backup sets it.
func NewBackupRunner(api BackupAPI, target string, mode proxmox.BackupMode, notes, retention string, logger *slog.Logger) *BackupRunner {
	return NewBackupRunnerWithWait(api, target, mode, notes, retention, 0, logger)
}

// NewBackupRunnerWithWait is NewBackupRunner plus an explicit vzdump wait bound (0 → 30m).
func NewBackupRunnerWithWait(api BackupAPI, target string, mode proxmox.BackupMode, notes, retention string, waitTimeout time.Duration, logger *slog.Logger) *BackupRunner {
	return NewBackupRunnerFull(api, target, mode, notes, retention, waitTimeout, false, logger)
}

// NewBackupRunnerFull is the full constructor. allowPBSPrune must be true ONLY for an additional
// tier with an explicitly configured keep_last — see BackupRunner.allowPBSPrune.
func NewBackupRunnerFull(api BackupAPI, target string, mode proxmox.BackupMode, notes, retention string, waitTimeout time.Duration, allowPBSPrune bool, logger *slog.Logger) *BackupRunner {
	if mode == "" {
		mode = proxmox.ModeSnapshot
	}
	if logger == nil {
		logger = slog.Default()
	}
	if waitTimeout <= 0 {
		waitTimeout = 30 * time.Minute
	}
	return &BackupRunner{api: api, target: target, mode: mode, notes: notes, retention: retention,
		waitTimeout: waitTimeout, allowPBSPrune: allowPBSPrune, logger: logger,
		now: func() time.Time { return time.Now().UTC() }}
}

// localPruneSpec returns the `--prune-backups` spec to apply to THIS backup, or "" to skip pruning. It
// applies the configured retention ONLY when the target is a non-PBS storage — PBS offsite retention is a
// separate lifecycle and must never be pruned by the per-run flag (§9). Fail-safe: if the target's type
// can't be confirmed (lookup error / not found), it SKIPS pruning rather than risk pruning PBS — the
// host_disk + storage_fill detectors remain the safety net.
func (r *BackupRunner) localPruneSpec(ctx context.Context) string {
	if r.retention == "" {
		return ""
	}
	stores, err := r.api.ListStorage(ctx)
	if err != nil {
		r.logger.Warn("backup: could not resolve target storage type — skipping local prune this run", "target", r.target, "err", err)
		return ""
	}
	for _, s := range stores {
		if s.Storage == r.target {
			if s.Type == "pbs" && !r.allowPBSPrune {
				// Not opted in → never prune the offsite DR (the pre-R-82 rule, and still the rule
				// for the primary tier, whose target+retention both DEFAULT and could prune by
				// accident).
				return ""
			}
			return r.retention
		}
	}
	r.logger.Warn("backup: target storage not found in list — skipping local prune this run (fail-safe)", "target", r.target)
	return ""
}

// snapshotMarker is the vzdump task-log line that signals the storage snapshot has been created
// and the backup is now reading from it — the point after which resuming the guest's app cannot
// affect the backup (slice 8B.2; validated on PVE 9.2.2: `INFO: create storage snapshot 'vzdump'`).
// It only appears in snapshot mode (stop mode takes no storage snapshot), so its presence ⟹
// snapshot mode — the basis for the controller's early resume.
const snapshotMarker = "create storage snapshot"

// snapshotWatchInterval is how often watchForSnapshot polls the task log. A package var so tests
// can shrink it (production: poll once a second — the marker appears in the first ~1s, §0).
var snapshotWatchInterval = time.Second

// Backup runs one vzdump of vmid to the local target and returns the report record.
func (r *BackupRunner) Backup(ctx context.Context, vmid int) (hub.Backup, error) {
	return r.backup(ctx, vmid, nil)
}

// BackupWithSnapshotHook is Backup plus an onSnapshot callback invoked ONCE, mid-backup, when the
// storage snapshot has been taken (snapshot mode only) — the 8B.2 early-resume signal. In
// stop/downgraded mode the marker never appears, so onSnapshot is never called (the caller then
// resumes at completion). onSnapshot must be cheap + non-blocking (it runs on a watcher goroutine).
func (r *BackupRunner) BackupWithSnapshotHook(ctx context.Context, vmid int, onSnapshot func()) (hub.Backup, error) {
	return r.backup(ctx, vmid, onSnapshot)
}

// backup is the shared body. A failure is returned BOTH as an error and as a
// Backup{Success:false,...} so the caller can record the failed attempt.
func (r *BackupRunner) backup(ctx context.Context, vmid int, onSnapshot func()) (hub.Backup, error) {
	start := r.now()
	rec := hub.Backup{
		TargetID:        r.target,
		VMID:            vmid,
		Mode:            string(r.mode),
		CrashConsistent: true, // always, this slice (no controller quiesce)
		StartedAt:       start.Format(time.RFC3339),
	}
	if r.target == "" {
		rec.Error = "no backup target configured"
		return rec, fmt.Errorf("backup: %s", rec.Error)
	}

	// Bulk-volume gap: which mountpoints the vzdump will EXCLUDE (best-effort; a config-read
	// failure just leaves the gap unknown, never fails the backup).
	if cfg, err := r.api.GuestConfig(ctx, vmid); err == nil {
		rec.UncoveredVolumes = uncoveredMountpoints(cfg.MountPoints())
	} else {
		r.logger.Warn("backup: could not read guest config for bulk-gap", "vmid", vmid, "err", err)
		rec.UncoveredVolumes = []string{}
	}

	upid, err := r.api.Vzdump(ctx, proxmox.VzdumpOptions{
		VMID: vmid, Storage: r.target, Mode: r.mode, Notes: r.notes,
		PruneBackups: r.localPruneSpec(ctx), // local target → keep-last=N; PBS/unknown → "" (no prune)
	})
	if err != nil {
		rec.Error = err.Error()
		rec.DurationSeconds = time.Since(start).Seconds()
		return rec, fmt.Errorf("backup: vzdump vmid %d: %w", vmid, err)
	}
	if upid != "" {
		// 8B.2: while the backup runs, watch the task log for the storage-snapshot marker and
		// fire onSnapshot once (snapshot mode only) so the controller can resume its app early.
		if onSnapshot != nil {
			watchCtx, stopWatch := context.WithCancel(ctx)
			defer stopWatch()
			go r.watchForSnapshot(watchCtx, upid, onSnapshot)
		}
		if _, err := r.api.WaitTask(ctx, upid, proxmox.WaitOptions{Timeout: r.waitTimeout}); err != nil {
			rec.Error = err.Error()
			rec.DurationSeconds = time.Since(start).Seconds()
			return rec, fmt.Errorf("backup: vzdump task vmid %d: %w", vmid, err)
		}
		// Report the ACTUAL mode PVE used (it may downgrade snapshot→stop for a stopped
		// guest — spike B1), read from the task log; fall back to the requested mode.
		if lines, err := r.api.TaskLogTail(ctx, upid, 200); err == nil {
			if actual := parseBackupMode(lines); actual != "" {
				rec.Mode = actual
			}
		}
	}

	// Resolve the produced archive (volid + size) — the task status carries no result volid.
	vol, size, err := r.latestArchive(ctx, vmid)
	if err != nil {
		rec.Error = fmt.Sprintf("backup succeeded but archive not resolved: %v", err)
		rec.DurationSeconds = time.Since(start).Seconds()
		return rec, fmt.Errorf("backup: resolve archive vmid %d: %w", vmid, err)
	}
	rec.Archive = vol
	rec.SizeBytes = size
	rec.Success = true
	rec.DurationSeconds = time.Since(start).Seconds()
	r.logger.Info("backup: completed", "vmid", vmid, "target", r.target, "archive", vol, "size_bytes", size,
		"uncovered_volumes", len(rec.UncoveredVolumes))
	return rec, nil
}

// watchForSnapshot polls the running backup's task log until it sees the storage-snapshot marker
// (→ onSnapshot once) or the requested mode is reported as `stop` (→ downgraded; the marker will
// never come, so stop watching) or ctx is cancelled (backup finished). Best-effort: a log-read
// error is retried on the next tick; onSnapshot fires at most once.
func (r *BackupRunner) watchForSnapshot(ctx context.Context, upid string, onSnapshot func()) {
	ticker := time.NewTicker(snapshotWatchInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			lines, err := r.api.TaskLogTail(ctx, upid, 200)
			if err != nil {
				continue
			}
			// A stop-mode (or downgraded) backup never creates a storage snapshot → never resume early.
			if m := parseBackupMode(lines); m != "" && m != string(proxmox.ModeSnapshot) {
				return
			}
			for _, ln := range lines {
				if strings.Contains(ln, snapshotMarker) {
					onSnapshot()
					return
				}
			}
		}
	}
}

// PickRestoreCandidate returns the newest backup archive on the target (any guest), or ""
// when there is none — the restore-test then no-ops cleanly.
func (r *BackupRunner) PickRestoreCandidate(ctx context.Context) (string, error) {
	return r.PickRestoreCandidateOn(ctx, r.target)
}

// PickRestoreCandidateOn is PickRestoreCandidate for an ARBITRARY tier's storage (R-85 1.2), so the
// scheduler can rotate across tiers instead of only ever seeing this runner's own target.
//
// Contract preserved: "" + nil error when the storage holds no archive. **A tier with nothing to
// restore is not an error** — a brand-new offsite tier legitimately has nothing yet, and turning
// that into a failure would make every fresh box look broken for its first week.
func (r *BackupRunner) PickRestoreCandidateOn(ctx context.Context, target string) (string, error) {
	if target == "" {
		return "", nil
	}
	contents, err := r.api.StorageContent(ctx, target)
	if err != nil {
		return "", err
	}
	var best string
	var bestCTime int64 = -1
	for _, e := range contents {
		if e.Content == "backup" && e.CTime > bestCTime {
			bestCTime, best = e.CTime, e.VolID
		}
	}
	return best, nil
}

// latestArchive finds the newest backup archive volid + size for vmid on the target.
func (r *BackupRunner) latestArchive(ctx context.Context, vmid int) (string, int64, error) {
	contents, err := r.api.StorageContent(ctx, r.target)
	if err != nil {
		return "", 0, err
	}
	var vol string
	var size, bestCTime int64 = 0, -1
	for _, e := range contents {
		if e.Content == "backup" && e.VMID == vmid && e.CTime > bestCTime {
			bestCTime, vol, size = e.CTime, e.VolID, e.Size
		}
	}
	if vol == "" {
		return "", 0, fmt.Errorf("no backup archive found for vmid %d on %s", vmid, r.target)
	}
	return vol, size, nil
}

// NewestArchiveTime reports when this guest's newest backup archive LANDED ON THIS TARGET, from the
// storage itself. ok=false means the target genuinely holds no archive for this guest.
//
// R-84: this is the cure for the redundant-backup-after-restart problem. The agent's backup Store is
// in-memory ("lost on restart; the cadence re-populates"), so after every restart /backup/due
// reported "no successful backup recorded yet" and the controller dutifully took another one. On the
// local tier that is wasted minutes; on the OFFSITE tier it is a wasted multi-hour WAN upload after
// every agent deploy — and agent deploys are routine. Three redundant local backups were observed on
// minPlausibleArchiveBytes is the floor below which a storage entry cannot be a real whole-guest
// backup and is therefore treated as an INCOMPLETE artefact rather than a successful one.
//
// MEASURED, not chosen by feel — fleet survey 2026-07-28 (Campaign 8, finding F-CRIT-2):
//
//	smallest REAL backup anywhere on the fleet ...  612,397,450 B  (~584 MiB, a guest-9100 vzdump)
//	demo-hp      local / PBS ..................... 1.59 GB / 4.35-4.37 GB
//	demo-felhom  local / PBS ..................... 5.82-5.84 GB / 14.47-14.51 GB
//	the phantom left by a PBS daemon killed mid-upload ....... 1 B
//
// 1 MiB sits 584x below the smallest real backup and 1,048,576x above the phantom. The two
// populations are nine orders of magnitude apart, so this floor cannot plausibly clip a real
// archive — which is the property that matters, because a floor set too HIGH does not merely lose
// safety margin, it causes fleet-wide backup THRASH (see archivePlausiblyComplete).
const minPlausibleArchiveBytes int64 = 1 << 20

// archivePlausiblyComplete reports whether a storage entry can be a COMPLETE backup, and if not,
// why. Pure, so the contract is unit-testable without a storage.
//
// WHY SIZE, AND NOTHING ELSE. The richer PBS fields look like better discriminators and are all
// traps, because this runner is TIER-AGNOSTIC — the same predicate runs against a PBS datastore and
// against a plain `dir` storage (verified against the live PVE API, 2026-07-28):
//
//   - `verification` is absent on the phantom, but ALSO absent on every local (dir) archive — a dir
//     storage has no verification concept — and absent on a good PBS snapshot until verify-new
//     catches up. Gating on it would reject 100% of local backups and every freshly-taken offsite
//     one: continuous re-backup across the fleet.
//   - `encrypted` fails the same way, and for the same reason.
//   - `notes` happens to be present on both good tiers today only because the agent sets it; an
//     archive written by any other path lacks it. Too fragile to gate freshness on.
//
// Size is the only signal that means the same thing on every tier.
//
// THE FAIL-SAFE DIRECTION, stated explicitly: when completeness cannot be established the entry is
// NOT counted as a successful backup. That errs toward the tier looking LESS fresh, and its worst
// case is one extra backup. Counting an undecidable entry is precisely the F-CRIT-2 defect — a
// failed upload that made its tier look freshly backed up and silenced it for a full cadence.
func archivePlausiblyComplete(e proxmox.StorageContent) (bool, string) {
	if e.Size < minPlausibleArchiveBytes {
		return false, fmt.Sprintf("size %d B is below the %d B plausibility floor — an aborted/incomplete archive, not a successful backup",
			e.Size, minPlausibleArchiveBytes)
	}
	return true, ""
}

// warnRejectedArchiveOnce announces a rejected archive at WARN exactly once per distinct volid.
//
// A rejected archive must never be silent: a tier that quietly ignores the newest entry on its
// storage is a new quiet path, and quiet paths are what F-CRIT-2 was. But the due-check runs every
// 5 minutes and a phantom persists indefinitely — server-side prune does NOT collect it (verified
// by dry-run 2026-07-28: with keep-last 2 it retained two real snapshots PLUS the phantom) — so
// logging per poll would emit ~288 identical lines a day and bury the one that matters.
func (r *BackupRunner) warnRejectedArchiveOnce(e proxmox.StorageContent, why string) {
	r.rejectedMu.Lock()
	if r.rejected == nil {
		r.rejected = map[string]struct{}{}
	}
	_, seen := r.rejected[e.VolID]
	if !seen {
		r.rejected[e.VolID] = struct{}{}
	}
	r.rejectedMu.Unlock()
	if seen {
		return
	}
	r.logger.Warn("backup: ignoring an INCOMPLETE archive when computing tier freshness — it is not a successful backup",
		"target", r.target, "vmid", e.VMID, "volid", e.VolID, "size_bytes", e.Size, "reason", why)
}

// demo-felhom in a single afternoon of deploys (2026-07-26).
//
// Asking the STORAGE rather than persisting the store is deliberate:
//   - it is ground truth, not remembered state — if an archive was pruned or deleted it correctly
//     stops counting, whereas a persisted record would keep claiming a backup that no longer exists;
//   - it needs no new on-disk state and no migration;
//   - it is the same source `latestArchive` already trusts to build the post-backup record.
//
// It answers ONLY "when did a plausibly COMPLETE backup last land", which is exactly what the
// due-check needs. Completeness is not optional here: PBS publishes an aborted upload into the same
// listing (manifest-less, 1 byte, and NEWEST), and counting it made the tier report fresh and go
// silent for a whole cadence — F-CRIT-2. Presence is not validity. The
// richer fields (size, duration, uncovered volumes, error) stay with the real in-memory records — a
// synthesized record would put invented numbers into the host-report.
func (r *BackupRunner) NewestArchiveTime(ctx context.Context, vmid int) (time.Time, bool, error) {
	contents, err := r.api.StorageContent(ctx, r.target)
	if err != nil {
		return time.Time{}, false, err
	}
	var best int64 = -1
	for _, e := range contents {
		if e.Content != "backup" || e.VMID != vmid {
			continue
		}
		if ok, why := archivePlausiblyComplete(e); !ok {
			r.warnRejectedArchiveOnce(e, why)
			continue
		}
		if e.CTime > best {
			best = e.CTime
		}
	}
	if best < 0 {
		return time.Time{}, false, nil
	}
	return time.Unix(best, 0).UTC(), true, nil
}

// parseBackupMode extracts the actual mode from a vzdump task log line `… backup mode: <x>`
// (e.g. "INFO: backup mode: stop"). Returns "" if not found.
func parseBackupMode(lines []string) string {
	const marker = "backup mode:"
	for _, ln := range lines {
		if i := strings.Index(ln, marker); i >= 0 {
			return strings.TrimSpace(ln[i+len(marker):])
		}
	}
	return ""
}

// uncoveredMountpoints returns the mountpoint paths the guest vzdump EXCLUDES. LXC mount
// points are OPT-IN to vzdump: a mpN with `backup=1` is covered; ANY other state — the
// `backup=` token absent OR `backup=0` — is excluded. We deliberately treat unset as
// uncovered (the safe DR direction: never imply an unprotected volume is backed up).
func uncoveredMountpoints(mps map[string]string) []string {
	var out []string
	for key, cfg := range mps {
		if mountpointCovered(cfg) {
			continue
		}
		out = append(out, mountpointLabel(key, cfg))
	}
	sort.Strings(out)
	if out == nil {
		return []string{}
	}
	return out
}

// mountpointCovered reports whether a mpN config string is included in the vzdump — true
// ONLY when it carries an explicit `backup=1`.
func mountpointCovered(cfg string) bool {
	for _, tok := range strings.Split(cfg, ",") {
		if v, ok := strings.CutPrefix(tok, "backup="); ok {
			return v == "1"
		}
	}
	return false // no backup= token → opt-out by default → not covered
}

// mountpointLabel prefers the mp=/path token, falling back to the mpN key.
func mountpointLabel(key, cfg string) string {
	for _, tok := range strings.Split(cfg, ",") {
		if p, ok := strings.CutPrefix(tok, "mp="); ok && p != "" {
			return p
		}
	}
	return key
}

// ToHubRestoreTest maps a reconcile restore-test result to the hub wire record (the backup
// package owns the reconcile→hub mapping so reconcile need not import hub for the result).
func ToHubRestoreTest(res reconcile.RestoreTestResult, testedAt time.Time) hub.RestoreTest {
	rt := hub.RestoreTest{
		SourceArchive:      res.Archive,
		SourceTier:         res.SourceTier,
		ScratchVMID:        res.ScratchVMID,
		Pass:               res.Pass,
		Verified:           res.Verified,
		TestedAt:           testedAt.Format(time.RFC3339),
		DurationSeconds:    res.Duration.Seconds(),
		Warnings:           res.StartWarnings,
		WarningsRecognized: res.WarningsRecognized,
		MountParity:        res.MountParity,
		MountInventory:     res.MountInventory,
	}
	if res.Err != nil {
		rt.Error = res.Err.Error()
	}
	return rt
}