chore(v1.1.8): GC sweep for app-user sessions + tokens

Extend the existing weekly retention sweeper with a new
spawn_app_user_token_gc function that prunes four tables in one
tick:

  * app_user_sessions — expired (sliding window or absolute cap) or
    explicitly revoked
  * app_user_email_verifications — consumed or expired
  * app_user_password_resets — consumed or expired
  * app_user_invitations — accepted or expired

Each underlying repo's gc(batch_size) uses FOR UPDATE SKIP LOCKED so
concurrent sweepers don't fight (cluster mode v1.3+ ready). No
configurable retention — rows die when their TTL or terminal state
hits, not on an arbitrary clock.

Wired into picloud's build_app alongside spawn_dead_letter_gc and
spawn_abandoned_gc.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

crates/manager-core/src/gc.rs

@@ -15,6 +15,10 @@ 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.
@@ -93,3 +97,116 @@ async fn sweep_abandoned(repo: &dyn AbandonedRepo, retention_days: u32) {
         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");
+    }
+}

crates/manager-core/src/lib.rs

@@ -152,7 +152,7 @@ pub use files_api::{files_admin_router, FilesAdminState};
 pub use files_repo::{FilesConfig, FilesRepo, FilesRepoError, FsFilesRepo};
 pub use files_service::FilesServiceImpl;
 pub use files_sweep::{spawn_files_orphan_sweep, sweep_orphan_tmp_files, SweepStats};
-pub use gc::{spawn_abandoned_gc, spawn_dead_letter_gc};
+pub use gc::{spawn_abandoned_gc, spawn_app_user_token_gc, spawn_dead_letter_gc};
 pub use http_service::{HttpConfig, HttpServiceImpl};
 pub use kv_repo::{KvRepo, KvRepoError, PostgresKvRepo};
 pub use kv_service::KvServiceImpl;

crates/picloud/src/lib.rs

@@ -372,6 +372,15 @@ pub async fn build_app(
         abandoned_repo.clone(),
         trigger_config.abandoned_retention_days,
     );
+    // v1.1.8 weekly sweep for app-user sessions + tokens. No
+    // configurable retention — rows are pruned as soon as they're
+    // either expired (by TTL) or consumed/revoked.
+    picloud_manager_core::spawn_app_user_token_gc(
+        app_user_sessions_repo.clone(),
+        app_user_verifications_repo.clone(),
+        app_user_password_resets_repo.clone(),
+        app_user_invitations_repo.clone(),
+    );
     // v1.1.4: cron scheduler. Polls cron_trigger_details on a tick and
     // enqueues due triggers into the outbox; the dispatcher above
     // delivers them like any other async trigger.