feat: add data retention and auto-cleanup mechanism

Add configurable retention policies for reputation_events, invocations,
and abuse_reports tables. Includes PruneEvents/PruneInvocations/
PruneResolvedReports store methods (SQLite + Postgres), a new
retention.Service that coordinates cleanup, and a ticker goroutine
in main.go. Defaults: 90d reputation events, 30d invocations,
365d resolved abuse reports, 1h cleanup interval.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: gcmsg

cmd/peerclawd/main.go

@@ -25,6 +25,7 @@ import (
 	"github.com/peerclaw/peerclaw-server/internal/observability"
 	"github.com/peerclaw/peerclaw-server/internal/registry"
 	"github.com/peerclaw/peerclaw-server/internal/reputation"
+	"github.com/peerclaw/peerclaw-server/internal/retention"
 	"github.com/peerclaw/peerclaw-server/internal/router"
 	"github.com/peerclaw/peerclaw-server/internal/server"
 	"github.com/peerclaw/peerclaw-server/internal/signaling"
@@ -152,8 +153,9 @@ func main() {
 
 	// Initialize invocation store.
 	var invocationService *invocation.Service
+	var invStore invocation.Store
 	if sqlDB != nil {
-		invStore := invocation.NewStore(cfg.Database.Driver, sqlDB)
+		invStore = invocation.NewStore(cfg.Database.Driver, sqlDB)
 		if err := invStore.Migrate(context.Background()); err != nil {
 			logger.Error("failed to migrate invocation tables", "error", err)
 			os.Exit(1)
@@ -164,8 +166,9 @@ func main() {
 
 	// Initialize review service.
 	var reviewService *review.Service
+	var revStore review.Store
 	if sqlDB != nil {
-		revStore := review.NewStore(cfg.Database.Driver, sqlDB)
+		revStore = review.NewStore(cfg.Database.Driver, sqlDB)
 		if err := revStore.Migrate(context.Background()); err != nil {
 			logger.Error("failed to migrate review tables", "error", err)
 			os.Exit(1)
@@ -440,6 +443,39 @@ func main() {
 		logger.Info("email verification cleanup goroutine started", "interval", "1h")
 	}
 
+	// Start data retention cleanup goroutine.
+	if cfg.Retention.Enabled {
+		retentionInterval, err := time.ParseDuration(cfg.Retention.CleanupInterval)
+		if err != nil {
+			retentionInterval = 1 * time.Hour
+		}
+		retentionSvc := retention.NewService(repStore, invStore, revStore, retention.Config{
+			ReputationEventsDays: cfg.Retention.ReputationEventsDays,
+			InvocationsDays:      cfg.Retention.InvocationsDays,
+			AbuseReportsDays:     cfg.Retention.AbuseReportsDays,
+		}, logger)
+		go func() {
+			ticker := time.NewTicker(retentionInterval)
+			defer ticker.Stop()
+			// Run once on startup.
+			retentionSvc.RunOnce(ctx)
+			for {
+				select {
+				case <-ctx.Done():
+					return
+				case <-ticker.C:
+					retentionSvc.RunOnce(ctx)
+				}
+			}
+		}()
+		logger.Info("data retention cleanup started",
+			"interval", retentionInterval,
+			"reputation_events_days", cfg.Retention.ReputationEventsDays,
+			"invocations_days", cfg.Retention.InvocationsDays,
+			"abuse_reports_days", cfg.Retention.AbuseReportsDays,
+		)
+	}
+
 	var wg sync.WaitGroup
 	errCh := make(chan error, 1)
 

internal/config/config.go

@@ -23,6 +23,7 @@ type Config struct {
 	Auth          AuthConfig          `yaml:"auth"`
 	UserAuth      UserAuthConfig      `yaml:"user_auth"`
 	SMTP          SMTPConfig          `yaml:"smtp"`
+	Retention     RetentionConfig     `yaml:"retention"`
 }
 
 // SMTPConfig holds SMTP email settings.
@@ -35,6 +36,15 @@ type SMTPConfig struct {
 	TLS      bool   `yaml:"tls"`       // default true
 }
 
+// RetentionConfig holds data retention and auto-cleanup settings.
+type RetentionConfig struct {
+	Enabled              bool   `yaml:"enabled"`
+	ReputationEventsDays int    `yaml:"reputation_events_days"`
+	InvocationsDays      int    `yaml:"invocations_days"`
+	AbuseReportsDays     int    `yaml:"abuse_reports_days"`
+	CleanupInterval      string `yaml:"cleanup_interval"`
+}
+
 // AuthConfig holds authentication settings.
 type AuthConfig struct {
 	Required bool `yaml:"required"` // When true, reject unauthenticated requests. Default true.
@@ -196,6 +206,13 @@ func DefaultConfig() *Config {
 			RefreshTTL: "168h",
 			BcryptCost: 12,
 		},
+		Retention: RetentionConfig{
+			Enabled:              true,
+			ReputationEventsDays: 90,
+			InvocationsDays:      30,
+			AbuseReportsDays:     365,
+			CleanupInterval:      "1h",
+		},
 	}
 }
 

internal/invocation/postgres.go

@@ -329,6 +329,18 @@ func (s *PostgresStore) CountInvocations(ctx context.Context) (int, error) {
 	return count, err
 }
 
+// PruneInvocations deletes invocation records older than the given time.
+func (s *PostgresStore) PruneInvocations(ctx context.Context, olderThan time.Time) (int64, error) {
+	res, err := s.db.ExecContext(ctx,
+		`DELETE FROM invocations WHERE created_at < $1`,
+		olderThan.UTC(),
+	)
+	if err != nil {
+		return 0, err
+	}
+	return res.RowsAffected()
+}
+
 func (s *PostgresStore) Close() error {
 	return nil
 }

internal/invocation/sqlite.go

@@ -342,6 +342,18 @@ func (s *SQLiteStore) CountInvocations(ctx context.Context) (int, error) {
 	return count, err
 }
 
+// PruneInvocations deletes invocation records older than the given time.
+func (s *SQLiteStore) PruneInvocations(ctx context.Context, olderThan time.Time) (int64, error) {
+	res, err := s.db.ExecContext(ctx,
+		`DELETE FROM invocations WHERE created_at < ?`,
+		olderThan.UTC().Format(time.RFC3339),
+	)
+	if err != nil {
+		return 0, err
+	}
+	return res.RowsAffected()
+}
+
 func (s *SQLiteStore) Close() error {
 	return nil
 }

internal/invocation/store.go

@@ -83,6 +83,10 @@ type Store interface {
 	// TopAgents returns the top agents by call count since the given time.
 	TopAgents(ctx context.Context, since time.Time, limit int) ([]AgentCallStats, error)
 
+	// PruneInvocations deletes invocation records older than the given time.
+	// Returns the number of deleted rows.
+	PruneInvocations(ctx context.Context, olderThan time.Time) (int64, error)
+
 	// CountInvocations returns the total number of invocations.
 	CountInvocations(ctx context.Context) (int, error)
 

internal/reputation/postgres.go

@@ -197,6 +197,18 @@ func (s *PostgresStore) ListStaleOnlineAgents(ctx context.Context, timeout time.
 	return ids, rows.Err()
 }
 
+// PruneEvents deletes reputation events older than the given time.
+func (s *PostgresStore) PruneEvents(ctx context.Context, olderThan time.Time) (int64, error) {
+	res, err := s.db.ExecContext(ctx,
+		`DELETE FROM reputation_events WHERE created_at < $1`,
+		olderThan.UTC(),
+	)
+	if err != nil {
+		return 0, err
+	}
+	return res.RowsAffected()
+}
+
 // Close is a no-op since the db is shared with the registry store.
 func (s *PostgresStore) Close() error {
 	return nil

internal/reputation/sqlite.go

@@ -216,6 +216,18 @@ func (s *SQLiteStore) ListStaleOnlineAgents(ctx context.Context, timeout time.Du
 	return ids, rows.Err()
 }
 
+// PruneEvents deletes reputation events older than the given time.
+func (s *SQLiteStore) PruneEvents(ctx context.Context, olderThan time.Time) (int64, error) {
+	res, err := s.db.ExecContext(ctx,
+		`DELETE FROM reputation_events WHERE created_at < ?`,
+		olderThan.UTC().Format(time.RFC3339),
+	)
+	if err != nil {
+		return 0, err
+	}
+	return res.RowsAffected()
+}
+
 // Close is a no-op since the db is shared with the registry store.
 func (s *SQLiteStore) Close() error {
 	return nil

internal/reputation/store.go

@@ -60,6 +60,10 @@ type Store interface {
 	// whose last heartbeat is older than the given timeout.
 	ListStaleOnlineAgents(ctx context.Context, timeout time.Duration) ([]string, error)
 
+	// PruneEvents deletes reputation events older than the given time.
+	// Returns the number of deleted rows.
+	PruneEvents(ctx context.Context, olderThan time.Time) (int64, error)
+
 	// Migrate creates the required tables and columns.
 	Migrate(ctx context.Context) error
 

internal/retention/service.go

@@ -0,0 +1,91 @@
+package retention
+
+import (
+	"context"
+	"log/slog"
+	"time"
+
+	"github.com/peerclaw/peerclaw-server/internal/invocation"
+	"github.com/peerclaw/peerclaw-server/internal/reputation"
+	"github.com/peerclaw/peerclaw-server/internal/review"
+)
+
+// Config holds retention policy settings.
+type Config struct {
+	ReputationEventsDays int
+	InvocationsDays      int
+	AbuseReportsDays     int
+}
+
+// PruneResult holds the number of rows deleted from each table.
+type PruneResult struct {
+	ReputationEvents int64
+	Invocations      int64
+	AbuseReports     int64
+}
+
+// Service coordinates data retention cleanup across stores.
+type Service struct {
+	repStore reputation.Store
+	invStore invocation.Store
+	revStore review.Store
+	config   Config
+	logger   *slog.Logger
+}
+
+// NewService creates a new retention service. Any store may be nil.
+func NewService(repStore reputation.Store, invStore invocation.Store, revStore review.Store, cfg Config, logger *slog.Logger) *Service {
+	return &Service{
+		repStore: repStore,
+		invStore: invStore,
+		revStore: revStore,
+		config:   cfg,
+		logger:   logger,
+	}
+}
+
+// RunOnce executes a single cleanup pass, deleting expired data from all configured stores.
+func (s *Service) RunOnce(ctx context.Context) (*PruneResult, error) {
+	now := time.Now().UTC()
+	result := &PruneResult{}
+
+	if s.repStore != nil && s.config.ReputationEventsDays > 0 {
+		cutoff := now.AddDate(0, 0, -s.config.ReputationEventsDays)
+		n, err := s.repStore.PruneEvents(ctx, cutoff)
+		if err != nil {
+			s.logger.Error("retention: failed to prune reputation events", "error", err)
+			return result, err
+		}
+		result.ReputationEvents = n
+	}
+
+	if s.invStore != nil && s.config.InvocationsDays > 0 {
+		cutoff := now.AddDate(0, 0, -s.config.InvocationsDays)
+		n, err := s.invStore.PruneInvocations(ctx, cutoff)
+		if err != nil {
+			s.logger.Error("retention: failed to prune invocations", "error", err)
+			return result, err
+		}
+		result.Invocations = n
+	}
+
+	if s.revStore != nil && s.config.AbuseReportsDays > 0 {
+		cutoff := now.AddDate(0, 0, -s.config.AbuseReportsDays)
+		n, err := s.revStore.PruneResolvedReports(ctx, cutoff)
+		if err != nil {
+			s.logger.Error("retention: failed to prune abuse reports", "error", err)
+			return result, err
+		}
+		result.AbuseReports = n
+	}
+
+	if result.ReputationEvents > 0 || result.Invocations > 0 || result.AbuseReports > 0 {
+		s.logger.Info("retention cleanup completed",
+			"reputation_events", result.ReputationEvents,
+			"invocations", result.Invocations,
+			"abuse_reports", result.AbuseReports,
+		)
+	}
+
+	return result, nil
+}