retain: core retention sweep (RETAIN-001)

Prunes aged rows from append-only history tables on a schedule.
Three operator-tunable horizons (metrics/events/audit), 0 = keep
forever. ctid-based batched DELETE, closed-table whitelist.

Signed-off-by: Fimeg <casey.tunturi@gmail.com>

server/internal/database/queries/retention.go

@@ -0,0 +1,55 @@
+package queries
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/jmoiron/sqlx"
+)
+
+// RetentionQueries deletes aged rows from append-only history tables
+// (RETAIN-001). The table -> timestamp-column map below is a closed
+// whitelist: table and column names never reach SQL from caller-supplied
+// strings, so the dynamic DELETE is injection-free by construction.
+type RetentionQueries struct {
+	db *sqlx.DB
+}
+
+// prunableTables maps each table the retention sweep may touch to the
+// timestamp column that defines row age. Adding a table here is a
+// reviewable act — current-state tables (agents, update_packages,
+// current_package_state, docker_*) must never appear; only append-only
+// history belongs.
+var prunableTables = map[string]string{
+	"metrics":                "created_at",
+	"storage_metrics":        "created_at",
+	"system_events":          "created_at",
+	"agent_commands":         "created_at",
+	"update_logs":            "executed_at",
+	"update_version_history": "update_completed_at",
+}
+
+func NewRetentionQueries(db *sqlx.DB) *RetentionQueries {
+	return &RetentionQueries{db: db}
+}
+
+// PruneBatch deletes up to limit rows older than cutoff from table.
+// Returns the number of rows deleted. ctid-based so it works regardless
+// of the table's primary key shape; bounded so a first sweep over months
+// of backlog cannot hold a long transaction.
+func (q *RetentionQueries) PruneBatch(table string, cutoff time.Time, limit int) (int64, error) {
+	column, ok := prunableTables[table]
+	if !ok {
+		return 0, fmt.Errorf("retention: table %q is not on the prunable whitelist", table)
+	}
+
+	query := fmt.Sprintf(
+		`DELETE FROM %s WHERE ctid IN (SELECT ctid FROM %s WHERE %s < $1 LIMIT $2)`,
+		table, table, column,
+	)
+	res, err := q.db.Exec(query, cutoff, limit)
+	if err != nil {
+		return 0, fmt.Errorf("retention: prune %s: %w", table, err)
+	}
+	return res.RowsAffected()
+}

server/internal/services/retention.go

@@ -0,0 +1,88 @@
+package services
+
+import (
+	"log"
+	"time"
+
+	"github.com/Fimeg/RedFlag/server/internal/database/queries"
+)
+
+// RetentionService (RETAIN-001) prunes aged rows from append-only history
+// tables on a schedule. Without it, metrics/system_events/agent_commands
+// grow unbounded — at 1,000 agents on 60-second heartbeats that is real
+// churn. Three operator-tunable horizons, resolved through the layered
+// settings (env > config > DB > default); a value of 0 means keep forever
+// (sovereignty: deletion is opt-out-able, never forced).
+//
+// Deliberately NOT covered: current-state tables, and the process snapshot
+// tables (those carry their own 10-snapshot cap). Pruned rows are gone —
+// this runs against history the operator has aged out, and every sweep
+// logs what it removed.
+type RetentionService struct {
+	queries  *queries.RetentionQueries
+	settings *SecuritySettingsService
+}
+
+// retentionCategory groups tables that age out on one shared horizon.
+type retentionCategory struct {
+	settingKey  string // operational.<key>, int days; 0 = keep forever
+	defaultDays int
+	tables      []string
+}
+
+var retentionCategories = []retentionCategory{
+	// High-churn telemetry, low audit value once superseded.
+	{settingKey: "retention_metrics_days", defaultDays: 30,
+		tables: []string{"metrics", "storage_metrics"}},
+	// Operational event stream — feeds History/notification bell.
+	{settingKey: "retention_events_days", defaultDays: 90,
+		tables: []string{"system_events"}},
+	// Audit trail — command history and update lifecycle. Long default;
+	// this is the "what happened on that machine" record.
+	{settingKey: "retention_history_days", defaultDays: 365,
+		tables: []string{"agent_commands", "update_logs", "update_version_history"}},
+}
+
+const (
+	pruneBatchSize        = 5000
+	maxPruneBatchesPerRun = 20 // caps one sweep at 100k rows/table; backlog drains over successive sweeps
+)
+
+func NewRetentionService(q *queries.RetentionQueries, settings *SecuritySettingsService) *RetentionService {
+	return &RetentionService{queries: q, settings: settings}
+}
+
+// Sweep prunes every category once. Errors on one table are logged and do
+// not stop the rest of the sweep (assume failure; the next sweep retries).
+// Intended to run on the taskrunner periodic registry.
+func (s *RetentionService) Sweep() {
+	for _, cat := range retentionCategories {
+		days := cat.defaultDays
+		if s.settings != nil {
+			days = s.settings.GetOperationalInt(cat.settingKey, cat.defaultDays)
+		}
+		if days <= 0 {
+			// 0 (or negative) = retention disabled for this category.
+			continue
+		}
+		cutoff := time.Now().UTC().AddDate(0, 0, -days)
+
+		for _, table := range cat.tables {
+			var total int64
+			for i := 0; i < maxPruneBatchesPerRun; i++ {
+				n, err := s.queries.PruneBatch(table, cutoff, pruneBatchSize)
+				if err != nil {
+					log.Printf("[WARNING] [server] [retention] prune_failed table=%s error=%q", table, err)
+					break
+				}
+				total += n
+				if n < pruneBatchSize {
+					break
+				}
+			}
+			if total > 0 {
+				log.Printf("[INFO] [server] [retention] pruned table=%s rows=%d retention_days=%d", table, total, days)
+			}
+		}
+	}
+}

server/internal/services/security_settings_service.go

@@ -280,6 +280,17 @@ func (s *SecuritySettingsService) ValidateSetting(category, key string, value in
 			return fmt.Errorf("backoff_max_seconds must be a number")
 		}
 
+	case "operational.retention_metrics_days",
+		"operational.retention_events_days",
+		"operational.retention_history_days":
+		if days, ok := value.(float64); ok {
+			if days < 0 || days > 3650 {
+				return fmt.Errorf("retention days must be between 0 (keep forever) and 3650")
+			}
+		} else {
+			return fmt.Errorf("retention days must be a number")
+		}
+
 	case "supply_chain.min_package_age_hours":
 		if hours, ok := value.(float64); ok {
 			if hours < 0 || hours > 8760 {