silo-server/internal/policy/decisionlog_cleanup.go

package policy

import (
	"context"
	"log/slog"
	"strconv"
	"strings"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"
)

const decisionLogCleanupBatchSize = 10000

// DecisionLogSettingsStore is satisfied by *catalog.ServerSettingsRepo.
type DecisionLogSettingsStore interface {
	Get(ctx context.Context, key string) (string, error)
}

// DecisionLogPartitionManager is satisfied by *partman.Manager.
type DecisionLogPartitionManager interface {
	EnsureFuturePartitions(ctx context.Context) error
	DropExpiredPartitions(ctx context.Context, cutoff time.Time) ([]string, error)
	DeleteExpiredRowsFromDefault(ctx context.Context, cutoff time.Time) (int64, error)
}

// CleanupDecisionLogsOnce runs one policy decision log retention pass. It
// returns the number of deleted rows plus the first error encountered, so
// callers (the task manager) can report a degraded pass instead of silently
// letting policy_decisions grow unbounded.
func CleanupDecisionLogsOnce(
	ctx context.Context,
	pool *pgxpool.Pool,
	store DecisionLogSettingsStore,
	pm DecisionLogPartitionManager,
) (int64, error) {
	var firstErr error
	days := DefaultDecisionLogRetentionDays
	if store != nil {
		if raw, err := store.Get(ctx, SettingDecisionLogRetentionDays); err == nil && raw != "" {
			if parsed := parsePositiveInt(raw); parsed > 0 {
				days = parsed
			}
		}
	}

	cutoff := time.Now().UTC().AddDate(0, 0, -days)
	if pm != nil {
		if err := pm.EnsureFuturePartitions(ctx); err != nil {
			slog.WarnContext(ctx, "policy decision log ensure future partitions error", "component", "policy", "error", err)
			firstErr = err
		}

		partitionCleanupFailed := false
		totalDeleted := int64(0)
		if dropped, err := pm.DropExpiredPartitions(ctx, cutoff); err != nil {
			slog.WarnContext(ctx, "policy decision log partition cleanup error", "component", "policy", "error", err)
			partitionCleanupFailed = true
			if firstErr == nil {
				firstErr = err
			}
		} else if len(dropped) > 0 {
			slog.InfoContext(ctx, "policy decision log dropped expired partitions", "component", "policy", "partitions", dropped)
		}

		if deleted, err := pm.DeleteExpiredRowsFromDefault(ctx, cutoff); err != nil {
			slog.WarnContext(ctx, "policy decision log default partition cleanup error", "component", "policy", "error", err)
			partitionCleanupFailed = true
			if firstErr == nil {
				firstErr = err
			}
		} else if deleted > 0 {
			totalDeleted += deleted
			slog.InfoContext(ctx, "policy decision log default partition cleanup completed", "component", "policy", "deleted", deleted, "retention_days", days)
		}

		if !partitionCleanupFailed {
			return totalDeleted, firstErr
		}
		slog.WarnContext(ctx, "policy decision log partition cleanup degraded, falling back to row deletes", "component", "policy", "retention_days", days)
	}

	total, err := deleteExpiredDecisionRowsBefore(ctx, pool, cutoff)
	if err != nil && firstErr == nil {
		firstErr = err
	}
	if total > 0 {
		slog.InfoContext(ctx, "policy decision log cleanup completed", "component", "policy", "deleted", total, "retention_days", days)
	}
	return total, firstErr
}

func deleteExpiredDecisionRowsBefore(ctx context.Context, pool *pgxpool.Pool, cutoff time.Time) (int64, error) {
	if pool == nil {
		return 0, nil
	}
	total := int64(0)
	for {
		result, err := pool.Exec(ctx, `
			DELETE FROM policy_decisions
			WHERE (id, "timestamp") IN (
				SELECT id, "timestamp" FROM policy_decisions
				WHERE "timestamp" < $1
				LIMIT $2
			)
			`, cutoff, decisionLogCleanupBatchSize)
		if err != nil {
			slog.WarnContext(ctx, "policy decision log cleanup error", "component", "policy", "error", err)
			return total, err
		}
		deleted := result.RowsAffected()
		total += deleted
		if deleted < int64(decisionLogCleanupBatchSize) {
			break
		}
	}
	return total, nil
}

func parsePositiveInt(s string) int {
	v, err := strconv.Atoi(strings.TrimSpace(s))
	if err != nil {
		return 0
	}
	return v
}