package database import ( "context" "database/sql" "fmt" "log/slog" "time" ) const maintenanceLockID int64 = 733081337591911 // CleanupStats describes rows handled by one bounded maintenance pass. // Counts are deliberately bounded so a busy API pod cannot hold a long // transaction while trying to clean an entire historical table. type CleanupStats struct { PersonalDataPurged int64 IdempotencyDeleted int64 SessionsDeleted int64 AuditDeleted int64 } // RunMaintenanceOnce purges expired personal data and bounded operational // records. A PostgreSQL advisory lock makes it safe to run from every API pod // without duplicate work when the service is horizontally scaled. func RunMaintenanceOnce(ctx context.Context, db *sql.DB, now time.Time, batchSize int) (CleanupStats, error) { if db == nil { return CleanupStats{}, fmt.Errorf("maintenance database is nil") } if batchSize < 1 { return CleanupStats{}, fmt.Errorf("maintenance batch size must be positive") } conn, err := db.Conn(ctx) if err != nil { return CleanupStats{}, fmt.Errorf("acquire maintenance connection: %w", err) } defer conn.Close() var acquired bool if err := conn.QueryRowContext(ctx, `SELECT pg_try_advisory_lock($1)`, maintenanceLockID).Scan(&acquired); err != nil { return CleanupStats{}, fmt.Errorf("acquire maintenance lock: %w", err) } if !acquired { return CleanupStats{}, nil } defer func() { _, _ = conn.ExecContext(context.Background(), `SELECT pg_advisory_unlock($1)`, maintenanceLockID) }() tx, err := conn.BeginTx(ctx, nil) if err != nil { return CleanupStats{}, fmt.Errorf("begin maintenance transaction: %w", err) } rollback := func(cause error) (CleanupStats, error) { _ = tx.Rollback() return CleanupStats{}, cause } stats := CleanupStats{} stats.PersonalDataPurged, err = boundedUpdate(ctx, tx, ` WITH candidates AS ( SELECT id FROM queue_tickets WHERE personal_data_purge_at <= $1 AND personal_data_purged_at IS NULL AND status IN ('COMPLETED', 'MISSED', 'CANCELED') ORDER BY personal_data_purge_at, id LIMIT $2 FOR UPDATE SKIP LOCKED ) UPDATE queue_tickets AS ticket SET phone_ciphertext = NULL, phone_nonce = NULL, phone_hmac = NULL, last_name_ciphertext = NULL, last_name_nonce = NULL, honorific = '游客', personal_data_purged_at = $1, updated_at = $1 FROM candidates WHERE ticket.id = candidates.id`, now, batchSize) if err != nil { return rollback(fmt.Errorf("purge personal data: %w", err)) } stats.IdempotencyDeleted, err = boundedDelete(ctx, tx, ` WITH candidates AS ( SELECT id FROM idempotency_keys WHERE expires_at <= $1 ORDER BY expires_at, id LIMIT $2 ) DELETE FROM idempotency_keys AS record USING candidates WHERE record.id = candidates.id`, now, batchSize) if err != nil { return rollback(fmt.Errorf("delete expired idempotency keys: %w", err)) } stats.SessionsDeleted, err = boundedDelete(ctx, tx, ` WITH candidates AS ( SELECT id FROM auth_sessions WHERE expires_at <= $1 OR (revoked_at IS NOT NULL AND revoked_at <= $1 - interval '30 days') ORDER BY expires_at, id LIMIT $2 ) DELETE FROM auth_sessions AS session USING candidates WHERE session.id = candidates.id`, now, batchSize) if err != nil { return rollback(fmt.Errorf("delete expired sessions: %w", err)) } stats.AuditDeleted, err = boundedDelete(ctx, tx, ` WITH candidates AS ( SELECT id FROM audit_entries WHERE retain_until <= $1 ORDER BY retain_until, id LIMIT $2 ) DELETE FROM audit_entries AS entry USING candidates WHERE entry.id = candidates.id`, now, batchSize) if err != nil { return rollback(fmt.Errorf("delete expired audit entries: %w", err)) } if err := tx.Commit(); err != nil { return CleanupStats{}, fmt.Errorf("commit maintenance transaction: %w", err) } return stats, nil } func boundedUpdate(ctx context.Context, tx *sql.Tx, query string, now time.Time, batchSize int) (int64, error) { result, err := tx.ExecContext(ctx, query, now, batchSize) if err != nil { return 0, err } return result.RowsAffected() } func boundedDelete(ctx context.Context, tx *sql.Tx, query string, now time.Time, batchSize int) (int64, error) { return boundedUpdate(ctx, tx, query, now, batchSize) } // StartMaintenance starts a bounded periodic cleanup loop. It returns // immediately; cancellation of ctx stops the loop during graceful shutdown. func StartMaintenance(ctx context.Context, db *sql.DB, logger *slog.Logger, interval time.Duration, batchSize int) { if interval <= 0 { return } if logger == nil { logger = slog.Default() } go func() { run := func() { stats, err := RunMaintenanceOnce(ctx, db, time.Now().UTC(), batchSize) if err != nil { logger.Error("database maintenance failed", "error", err) return } if stats.PersonalDataPurged+stats.IdempotencyDeleted+stats.SessionsDeleted+stats.AuditDeleted > 0 { logger.Info("database maintenance completed", "personal_data_purged", stats.PersonalDataPurged, "idempotency_deleted", stats.IdempotencyDeleted, "sessions_deleted", stats.SessionsDeleted, "audit_deleted", stats.AuditDeleted, ) } } run() ticker := time.NewTicker(interval) defer ticker.Stop() for { select { case <-ctx.Done(): return case <-ticker.C: run() } } }() }