XQKqueue/server/internal/database/maintenance.go

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
	HistoryAnonymized  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.HistoryAnonymized, err = boundedUpdate(ctx, tx, `
		WITH candidates AS (
			SELECT id
			FROM queue_tickets
			WHERE history_anonymized_at IS NULL
			  AND joined_at <= ($1::timestamptz - interval '90 days')
			ORDER BY joined_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 = COALESCE(ticket.personal_data_purged_at, $1),
			history_anonymized_at = $1,
			updated_at = $1
		FROM candidates
		WHERE ticket.id = candidates.id`, now, batchSize)
	if err != nil {
		return rollback(fmt.Errorf("mark historical data anonymized: %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.HistoryAnonymized+stats.IdempotencyDeleted+stats.SessionsDeleted+stats.AuditDeleted > 0 {
				logger.Info("database maintenance completed",
					"personal_data_purged", stats.PersonalDataPurged,
					"history_anonymized", stats.HistoryAnonymized,
					"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()
			}
		}
	}()
}