feat(archive): BAK-08 Checksum-basierte Objekt-Integritaetspruefung

Stichprobenbasierter Scrub-Job: nimmt BAK-05s existing_in_storage,
priorisiert nach eigenem scrub_state.last_scrubbed_at (nicht
file_revisions.created_at, sonst kein echtes Rotationsverhalten),
prueft Inhalt per SHA-256 gegen file_revisions.checksum_sha256.
Meldung ueber echten dauerhaften /metrics-Endpunkt (Pull-Modell,
OPS-03 scrapt, kein Push), Counter monoton steigend. Real registriert
in Core metrics_sources, End-zu-Ende ueber OPS-03-Aggregator bestaetigt,
realer Befund-Durchlauf mit absichtlich falscher Pruefsumme durchgefuehrt.

archive/internal/scrub/state.go

@@ -0,0 +1,74 @@
+package scrub
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// LoadLastScrubbed liefert je storage_key den Zeitpunkt der letzten
+// Pruefung — Grundlage fuer Sample's Cooldown-Filter.
+func LoadLastScrubbed(ctx context.Context, pool *pgxpool.Pool) (map[string]time.Time, error) {
+	rows, err := pool.Query(ctx, `SELECT storage_key, last_scrubbed_at FROM scrub_state`)
+	if err != nil {
+		return nil, fmt.Errorf("scrub: scrub_state lesen: %w", err)
+	}
+	defer rows.Close()
+
+	out := make(map[string]time.Time)
+	for rows.Next() {
+		var key string
+		var ts time.Time
+		if err := rows.Scan(&key, &ts); err != nil {
+			return nil, fmt.Errorf("scrub: scrub_state-zeile lesen: %w", err)
+		}
+		out[key] = ts
+	}
+	return out, rows.Err()
+}
+
+// MarkScrubbed vermerkt Ergebnis und Zeitpunkt der Pruefung eines
+// Objekts — idempotent (ON CONFLICT), damit ein unterbrochener und neu
+// gestarteter Lauf keinen inkonsistenten Zustand hinterlaesst
+// (Akzeptanzkriterium: Lauf ist unterbrechbar ohne inkonsistenten
+// Zustand).
+func MarkScrubbed(ctx context.Context, pool *pgxpool.Pool, storageKey string, ok bool, at time.Time) error {
+	result := "ok"
+	if !ok {
+		result = "failed"
+	}
+	_, err := pool.Exec(ctx, `
+		INSERT INTO scrub_state (storage_key, last_scrubbed_at, last_result)
+		VALUES ($1, $2, $3)
+		ON CONFLICT (storage_key) DO UPDATE SET last_scrubbed_at = $2, last_result = $3
+	`, storageKey, at, result)
+	if err != nil {
+		return fmt.Errorf("scrub: scrub_state schreiben: %w", err)
+	}
+	return nil
+}
+
+// RecordFinding erhoeht den monoton steigenden Befund-Zaehler
+// (scrub_counters.findings_total) um genau 1 — als gueltiger Prometheus-
+// Counter darf dieser Wert nur steigen, niemals sinken, auch wenn ein
+// Befund spaeter behoben wird.
+func RecordFinding(ctx context.Context, pool *pgxpool.Pool) error {
+	_, err := pool.Exec(ctx, `UPDATE scrub_counters SET findings_total = findings_total + 1 WHERE id = 1`)
+	if err != nil {
+		return fmt.Errorf("scrub: befund-zaehler erhoehen: %w", err)
+	}
+	return nil
+}
+
+// FindingsTotal liest den aktuellen Zaehlerstand — genutzt vom
+// /metrics-Endpunkt (cmd/scrub-metrics).
+func FindingsTotal(ctx context.Context, pool *pgxpool.Pool) (int64, error) {
+	var total int64
+	err := pool.QueryRow(ctx, `SELECT findings_total FROM scrub_counters WHERE id = 1`).Scan(&total)
+	if err != nil {
+		return 0, fmt.Errorf("scrub: befund-zaehler lesen: %w", err)
+	}
+	return total, nil
+}