PiCloud/crates/manager-core/src/gc.rs

//! Weekly retention sweepers for `dead_letters` + `abandoned_executions`.
//!
//! Both use the `FOR UPDATE SKIP LOCKED` claim pattern so concurrent
//! sweepers (cluster mode v1.3+) don't fight each other. Defaults
//! match design notes §3 / §4: 30 days for DL, 7 days for abandoned.
//! Both env-overridable via `PICLOUD_DEAD_LETTER_RETENTION_DAYS` and
//! `PICLOUD_ABANDONED_EXECUTIONS_RETENTION_DAYS` (loaded by
//! `TriggerConfig::from_env`).
//!
//! Spawned from `build_app` alongside `spawn_session_pruner`.

use std::sync::Arc;
use std::time::Duration;

use chrono::Utc;

use crate::abandoned_repo::AbandonedRepo;
use crate::app_user_invitation_repo::AppUserInvitationRepo;
use crate::app_user_password_reset_repo::AppUserPasswordResetRepo;
use crate::app_user_session_repo::AppUserSessionRepository;
use crate::app_user_verification_repo::AppUserVerificationRepo;
use crate::dead_letter_repo::DeadLetterRepo;

/// Weekly sweep cadence — matches `spawn_session_pruner` shape.
const SWEEP_INTERVAL: Duration = Duration::from_secs(7 * 24 * 60 * 60);

/// Per-tick batch cap so we don't try to delete millions of rows in
/// one transaction. The loop keeps deleting batches until a tick
/// returns 0 rows affected.
const SWEEP_BATCH: i64 = 5_000;

pub fn spawn_dead_letter_gc(repo: Arc<dyn DeadLetterRepo>, retention_days: u32) {
    tokio::spawn(async move {
        let mut ticker = tokio::time::interval(SWEEP_INTERVAL);
        // Skip the immediate first fire — don't sweep at process start.
        ticker.tick().await;
        loop {
            ticker.tick().await;
            sweep_dead_letters(&*repo, retention_days).await;
        }
    });
}

pub fn spawn_abandoned_gc(repo: Arc<dyn AbandonedRepo>, retention_days: u32) {
    tokio::spawn(async move {
        let mut ticker = tokio::time::interval(SWEEP_INTERVAL);
        ticker.tick().await;
        loop {
            ticker.tick().await;
            sweep_abandoned(&*repo, retention_days).await;
        }
    });
}

async fn sweep_dead_letters(repo: &dyn DeadLetterRepo, retention_days: u32) {
    let cutoff = Utc::now() - chrono::Duration::days(i64::from(retention_days));
    let mut total: u64 = 0;
    loop {
        match repo.gc(cutoff, SWEEP_BATCH).await {
            Ok(0) => break,
            Ok(n) => {
                total += n;
                if n < SWEEP_BATCH as u64 {
                    break;
                }
            }
            Err(e) => {
                tracing::warn!(?e, "dead_letters GC sweep errored");
                break;
            }
        }
    }
    if total > 0 {
        tracing::info!(swept = total, "dead_letters GC swept");
    }
}

async fn sweep_abandoned(repo: &dyn AbandonedRepo, retention_days: u32) {
    let cutoff = Utc::now() - chrono::Duration::days(i64::from(retention_days));
    let mut total: u64 = 0;
    loop {
        match repo.gc(cutoff, SWEEP_BATCH).await {
            Ok(0) => break,
            Ok(n) => {
                total += n;
                if n < SWEEP_BATCH as u64 {
                    break;
                }
            }
            Err(e) => {
                tracing::warn!(?e, "abandoned_executions GC sweep errored");
                break;
            }
        }
    }
    if total > 0 {
        tracing::info!(swept = total, "abandoned_executions GC swept");
    }
}

/// v1.1.8: weekly sweep that prunes every app-user token table —
/// expired or revoked sessions, consumed or expired verifications,
/// password resets, and accepted or expired invitations. Each
/// underlying repo's `gc(batch_size)` uses FOR UPDATE SKIP LOCKED so
/// concurrent sweepers don't fight, matching the dead-letter +
/// abandoned-executions pattern.
pub fn spawn_app_user_token_gc(
    sessions: Arc<dyn AppUserSessionRepository>,
    verifications: Arc<dyn AppUserVerificationRepo>,
    password_resets: Arc<dyn AppUserPasswordResetRepo>,
    invitations: Arc<dyn AppUserInvitationRepo>,
) {
    tokio::spawn(async move {
        let mut ticker = tokio::time::interval(SWEEP_INTERVAL);
        ticker.tick().await;
        loop {
            ticker.tick().await;
            sweep_app_user_sessions(&*sessions).await;
            sweep_app_user_verifications(&*verifications).await;
            sweep_app_user_password_resets(&*password_resets).await;
            sweep_app_user_invitations(&*invitations).await;
        }
    });
}

async fn sweep_app_user_sessions(repo: &dyn AppUserSessionRepository) {
    let mut total: u64 = 0;
    loop {
        match repo.gc(SWEEP_BATCH).await {
            Ok(0) => break,
            Ok(n) => {
                total += n;
                if n < SWEEP_BATCH as u64 {
                    break;
                }
            }
            Err(e) => {
                tracing::warn!(?e, "app_user_sessions GC sweep errored");
                break;
            }
        }
    }
    if total > 0 {
        tracing::info!(swept = total, "app_user_sessions GC swept");
    }
}

async fn sweep_app_user_verifications(repo: &dyn AppUserVerificationRepo) {
    let mut total: u64 = 0;
    loop {
        match repo.gc(SWEEP_BATCH).await {
            Ok(0) => break,
            Ok(n) => {
                total += n;
                if n < SWEEP_BATCH as u64 {
                    break;
                }
            }
            Err(e) => {
                tracing::warn!(?e, "app_user_email_verifications GC sweep errored");
                break;
            }
        }
    }
    if total > 0 {
        tracing::info!(swept = total, "app_user_email_verifications GC swept");
    }
}

async fn sweep_app_user_password_resets(repo: &dyn AppUserPasswordResetRepo) {
    let mut total: u64 = 0;
    loop {
        match repo.gc(SWEEP_BATCH).await {
            Ok(0) => break,
            Ok(n) => {
                total += n;
                if n < SWEEP_BATCH as u64 {
                    break;
                }
            }
            Err(e) => {
                tracing::warn!(?e, "app_user_password_resets GC sweep errored");
                break;
            }
        }
    }
    if total > 0 {
        tracing::info!(swept = total, "app_user_password_resets GC swept");
    }
}

async fn sweep_app_user_invitations(repo: &dyn AppUserInvitationRepo) {
    let mut total: u64 = 0;
    loop {
        match repo.gc(SWEEP_BATCH).await {
            Ok(0) => break,
            Ok(n) => {
                total += n;
                if n < SWEEP_BATCH as u64 {
                    break;
                }
            }
            Err(e) => {
                tracing::warn!(?e, "app_user_invitations GC sweep errored");
                break;
            }
        }
    }
    if total > 0 {
        tracing::info!(swept = total, "app_user_invitations GC swept");
    }
}