//! 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, 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, 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, verifications: Arc, password_resets: Arc, invitations: Arc, ) { 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"); } }