vtc-service 0.12.0

//! 30-day retention sweeper for `Rejected` + `Withdrawn` join
//! requests (spec §5.5).
//!
//! VP contents may include PII; the sole control on inadvertent
//! PII retention is this sweeper. Runs on a daemon-wide tokio
//! task spawned at startup; default cadence is hourly.

use std::time::Duration;

use chrono::{DateTime, Duration as ChronoDuration, Utc};
use serde::{Deserialize, Serialize};
use tokio::sync::watch;
use tracing::{debug, info, warn};
use vti_common::error::AppError;
use vti_common::store::KeyspaceHandle;

use super::{delete_join_request, list_join_requests};

/// Operator-controllable retention window for terminal join
/// requests (Rejected + Withdrawn).
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct JoinRequestsConfig {
    /// Retention window in days. After this many days from
    /// `submitted_at`, terminal-state rows are purged on the next
    /// sweep. Default 30.
    #[serde(default = "default_retention_days")]
    pub retention_days: u32,
    /// How often the sweeper runs, in seconds. Default 3600 (1
    /// hour). Lower in tests via the config knob if needed.
    #[serde(default = "default_sweep_interval_secs")]
    pub sweep_interval_secs: u64,
}

pub fn default_retention_days() -> u32 {
    30
}

fn default_sweep_interval_secs() -> u64 {
    3600
}

impl Default for JoinRequestsConfig {
    fn default() -> Self {
        Self {
            retention_days: default_retention_days(),
            sweep_interval_secs: default_sweep_interval_secs(),
        }
    }
}

/// Owns the sweeper background task.
pub struct RetentionSweeper;

impl RetentionSweeper {
    /// Spawn the sweeper. Returns immediately; the task runs
    /// until the daemon's shutdown watcher fires.
    ///
    /// Sweeps five kinds of stale row each tick:
    /// - `Rejected` / `Withdrawn` join requests past the retention window
    ///   (PII in the submitted VP — the headline);
    /// - expired `present-challenge:` + `credx-pending:` rows (TTLs are
    ///   otherwise enforced only on the read path) — both in `join_requests_ks`;
    /// - `Failed` registry sync jobs past the retention window
    ///   (`sync_queue_ks`);
    /// - accepted-document-id records past their retention deadline
    ///   (`accepted_ids_ks`). Unlike the others this one is a **storage**
    ///   bound, not a correctness one: an expired record is already treated as
    ///   absent by `crate::trust_tasks::accepted_ids::AcceptedIds::claim`. The
    ///   in-memory guard it replaced was bounded by capacity eviction; a
    ///   keyspace is not.
    /// - backup bundles past their TTL (bytes deleted, record marked
    ///   `Expired`) or past retention once terminal
    ///   ([`vti_common::backup_transfer::sweeper`]);
    /// - operation-bound step-up marks past their five-minute life
    ///   (`step_up_marks_ks`). Also a storage bound: `crate::acl::bound_step_up`
    ///   treats an expired mark as absent on both reads.
    /// - step-up passkey invites and redemption/revocation ceremonies past
    ///   their life (`step_up_passkeys_ks`), the same storage bound for
    ///   `crate::step_up_passkey`. The credentials themselves are kept.
    #[allow(clippy::too_many_arguments)]
    pub fn spawn(
        join_requests_ks: KeyspaceHandle,
        sync_queue_ks: KeyspaceHandle,
        accepted_ids_ks: KeyspaceHandle,
        step_up_marks_ks: KeyspaceHandle,
        step_up_passkeys_ks: KeyspaceHandle,
        backup_bundles_ks: KeyspaceHandle,
        backup_blob_dir: std::path::PathBuf,
        config: JoinRequestsConfig,
        mut shutdown_rx: watch::Receiver<bool>,
    ) -> tokio::task::JoinHandle<()> {
        tokio::spawn(async move {
            let interval = Duration::from_secs(config.sweep_interval_secs.max(60));
            info!(
                retention_days = config.retention_days,
                interval_secs = interval.as_secs(),
                "retention sweeper started"
            );
            // Sweep once on startup so a freshly-restarted daemon
            // catches up on rows that aged out while it was down.
            if let Err(e) = sweep_all(
                &join_requests_ks,
                &sync_queue_ks,
                &accepted_ids_ks,
                &step_up_marks_ks,
                config.retention_days,
                Utc::now(),
            )
            .await
            {
                warn!(error = %e, "initial retention sweep failed");
            }
            sweep_backup_bundles(&backup_bundles_ks, &backup_blob_dir).await;
            sweep_step_up_passkeys(&step_up_passkeys_ks).await;
            loop {
                tokio::select! {
                    _ = shutdown_rx.changed() => {
                        info!("retention sweeper shutting down");
                        return;
                    }
                    _ = tokio::time::sleep(interval) => {
                        if let Err(e) = sweep_all(
                            &join_requests_ks,
                            &sync_queue_ks,
                            &accepted_ids_ks,
                            &step_up_marks_ks,
                            config.retention_days,
                            Utc::now(),
                        )
                        .await
                        {
                            warn!(error = %e, "retention sweep failed");
                        }
                        sweep_backup_bundles(&backup_bundles_ks, &backup_blob_dir).await;
                        sweep_step_up_passkeys(&step_up_passkeys_ks).await;
                    }
                }
            }
        })
    }
}

/// Expire backup bundles past their TTL and drop terminal ones past retention.
/// Separate from [`sweep_all`] because its failures are its own: a bundle whose
/// bytes could not be deleted is retried next pass without holding up the rest.
async fn sweep_backup_bundles(ks: &KeyspaceHandle, blob_dir: &std::path::Path) {
    if let Err(e) = vti_common::backup_transfer::sweeper::sweep_bundles(ks, blob_dir).await {
        warn!(error = %e, "backup-bundle sweep failed");
    }
}

/// Drop lapsed step-up passkey invites and ceremonies. Its own pass for the
/// same reason as [`sweep_backup_bundles`].
async fn sweep_step_up_passkeys(ks: &KeyspaceHandle) {
    match crate::step_up_passkey::sweep_expired(ks, Utc::now()).await {
        Ok(0) => {}
        Ok(n) => info!(
            expired = n,
            "retention sweep purged lapsed step-up passkey invites and ceremonies"
        ),
        Err(e) => warn!(error = %e, "step-up passkey sweep failed"),
    }
}

/// One full retention pass across all five stale-row kinds. Each sub-sweep is
/// independent; an error in one is propagated but the others on the same tick
/// have already run (sweeps are ordered, not transactional). `present-challenge:`
/// and `credx-pending:` share `join_requests_ks` with the join rows.
async fn sweep_all(
    join_requests_ks: &KeyspaceHandle,
    sync_queue_ks: &KeyspaceHandle,
    accepted_ids_ks: &KeyspaceHandle,
    step_up_marks_ks: &KeyspaceHandle,
    retention_days: u32,
    now: DateTime<Utc>,
) -> Result<(), AppError> {
    sweep_once(join_requests_ks, retention_days, now).await?;
    let challenges =
        crate::credentials::present_challenge::sweep_expired(join_requests_ks, now).await?;
    // Hidden-vetter admission (`zkp-pcs`): its challenges share this keyspace under a prefix of
    // their own, and an applicant who asks for one and walks away leaves a row behind exactly
    // as a holder who never presents does.
    #[cfg(feature = "vetting-pcs")]
    let challenges =
        challenges + crate::vetting::pcs_challenge::sweep_expired(join_requests_ks, now).await?;
    let offers = crate::credentials::exchange::sweep_expired_pending(join_requests_ks, now).await?;
    let failed_jobs =
        crate::registry::storage::sweep_failed_sync_jobs(sync_queue_ks, retention_days, now)
            .await?;
    // VTI-OPS-026: the accepted-id record is bounded by the acceptance window.
    // Its deadline (~11 minutes) is far shorter than this sweep's cadence, which
    // costs only storage — `claim` already treats an expired record as absent.
    let accepted_ids =
        crate::trust_tasks::accepted_ids::sweep_expired(accepted_ids_ks, now).await?;
    let step_up_marks = crate::acl::bound_step_up::sweep_expired(step_up_marks_ks, now).await?;
    if challenges + offers + failed_jobs + accepted_ids + step_up_marks > 0 {
        info!(
            expired_challenges = challenges,
            expired_offers = offers,
            failed_sync_jobs = failed_jobs,
            expired_accepted_ids = accepted_ids,
            expired_step_up_marks = step_up_marks,
            "retention sweep purged auxiliary stale rows"
        );
    }
    Ok(())
}

/// One pass: scan every row, delete those whose status is
/// `Rejected` / `Withdrawn` AND `submitted_at` is older than the
/// retention window.
async fn sweep_once(
    ks: &KeyspaceHandle,
    retention_days: u32,
    now: DateTime<Utc>,
) -> Result<(), AppError> {
    let cutoff = now - ChronoDuration::days(retention_days as i64);
    let rows = list_join_requests(ks).await?;
    let mut purged = 0usize;
    for row in rows {
        if row.status.is_terminal_retainable() && row.submitted_at < cutoff {
            delete_join_request(ks, row.id).await?;
            purged += 1;
        }
    }
    if purged > 0 {
        info!(
            purged,
            retention_days, "join-request retention sweep complete"
        );
    } else {
        debug!(
            retention_days,
            "join-request retention sweep: nothing to purge"
        );
    }
    Ok(())
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;
    use crate::join::storage::store_join_request;
    use crate::join::{JoinRequest, JoinStatus};
    use chrono::{Duration as ChronoDuration, Utc};
    use vti_common::config::StoreConfig;
    use vti_common::store::Store;

    async fn temp_ks() -> (KeyspaceHandle, tempfile::TempDir) {
        let dir = tempfile::tempdir().unwrap();
        let store = Store::open(&StoreConfig {
            data_dir: dir.path().to_path_buf(),
        })
        .unwrap();
        let ks = store.keyspace("join_requests").unwrap();
        (ks, dir)
    }

    fn at(submitted_at: chrono::DateTime<Utc>, status: JoinStatus) -> JoinRequest {
        JoinRequest {
            submitted_at,
            status,
            ..JoinRequest::new("did:key:z", serde_json::json!({}))
        }
    }

    #[tokio::test]
    async fn sweep_purges_old_rejected_rows() {
        let (ks, _dir) = temp_ks().await;
        let old = at(Utc::now() - ChronoDuration::days(31), JoinStatus::Rejected);
        let recent = at(Utc::now() - ChronoDuration::days(7), JoinStatus::Rejected);
        let pending_old = at(Utc::now() - ChronoDuration::days(60), JoinStatus::Pending);
        for r in [&old, &recent, &pending_old] {
            store_join_request(&ks, r).await.unwrap();
        }

        sweep_once(&ks, 30, Utc::now()).await.unwrap();

        let remaining = list_join_requests(&ks).await.unwrap();
        let dids: Vec<_> = remaining.iter().map(|r| r.id).collect();
        assert!(!dids.contains(&old.id), "old Rejected row must be purged");
        assert!(
            dids.contains(&recent.id),
            "recent Rejected row must be retained"
        );
        assert!(
            dids.contains(&pending_old.id),
            "Pending rows must never be swept"
        );
    }

    #[tokio::test]
    async fn sweep_purges_old_withdrawn_rows() {
        let (ks, _dir) = temp_ks().await;
        let old = at(Utc::now() - ChronoDuration::days(45), JoinStatus::Withdrawn);
        store_join_request(&ks, &old).await.unwrap();
        sweep_once(&ks, 30, Utc::now()).await.unwrap();
        assert!(
            list_join_requests(&ks).await.unwrap().is_empty(),
            "old Withdrawn rows must be purged"
        );
    }

    #[tokio::test]
    async fn sweep_does_not_purge_approved_rows() {
        let (ks, _dir) = temp_ks().await;
        let approved = at(Utc::now() - ChronoDuration::days(365), JoinStatus::Approved);
        store_join_request(&ks, &approved).await.unwrap();
        sweep_once(&ks, 30, Utc::now()).await.unwrap();
        assert_eq!(list_join_requests(&ks).await.unwrap().len(), 1);
    }

    /// `sweep_all` orchestrates all four kinds: a stale row of each is purged
    /// and a not-yet-expired row of each survives.
    #[tokio::test]
    async fn sweep_all_purges_every_stale_kind_and_keeps_fresh() {
        use crate::credentials::{exchange, present_challenge};
        use crate::registry::model::{SyncJob, SyncJobKind, SyncJobState};
        use crate::registry::storage::{list_sync_jobs, store_sync_job};

        let dir = tempfile::tempdir().unwrap();
        let store = Store::open(&StoreConfig {
            data_dir: dir.path().to_path_buf(),
        })
        .unwrap();
        let join_ks = store.keyspace("join_requests").unwrap();
        let sync_ks = store.keyspace("sync_queue").unwrap();
        let accepted_ids_ks = store
            .keyspace(crate::store::keyspaces::ACCEPTED_IDS)
            .unwrap();
        let step_up_marks_ks = store
            .keyspace(crate::store::keyspaces::STEP_UP_MARKS)
            .unwrap();
        let now = Utc::now();

        // --- stale rows (all must be purged) ---
        store_join_request(
            &join_ks,
            &at(now - ChronoDuration::days(31), JoinStatus::Rejected),
        )
        .await
        .unwrap();
        present_challenge::issue(
            &join_ks,
            "stale-thread",
            "did:web:v",
            None,
            present_challenge::DEFAULT_CHALLENGE_TTL,
            now - ChronoDuration::minutes(10),
        )
        .await
        .unwrap();
        exchange::make_offer(
            &join_ks,
            "https://issuer.example",
            vec!["VMC".into()],
            serde_json::json!({ "vc": 1 }),
            "did:key:zHolder",
            exchange::DEFAULT_OFFER_TTL,
            now - ChronoDuration::hours(1),
        )
        .await
        .unwrap();
        let mut failed = SyncJob::fresh(SyncJobKind::PublishMember, "did:key:zMember");
        failed.state = SyncJobState::Failed;
        failed.last_attempted_at = Some(now - ChronoDuration::days(40));
        store_sync_job(&sync_ks, &failed).await.unwrap();

        // --- fresh rows (all must survive) ---
        let fresh_join = at(now, JoinStatus::Rejected);
        store_join_request(&join_ks, &fresh_join).await.unwrap();
        present_challenge::issue(
            &join_ks,
            "fresh-thread",
            "did:web:v",
            None,
            present_challenge::DEFAULT_CHALLENGE_TTL,
            now,
        )
        .await
        .unwrap();

        sweep_all(
            &join_ks,
            &sync_ks,
            &accepted_ids_ks,
            &step_up_marks_ks,
            30,
            now,
        )
        .await
        .unwrap();

        // Stale join purged, fresh join survives.
        let join_ids: Vec<_> = list_join_requests(&join_ks)
            .await
            .unwrap()
            .into_iter()
            .map(|r| r.id)
            .collect();
        assert_eq!(
            join_ids,
            vec![fresh_join.id],
            "only the fresh Rejected row survives"
        );
        // Fresh challenge still consumable; stale one gone.
        assert!(
            present_challenge::consume(&join_ks, "fresh-thread", now)
                .await
                .is_ok()
        );
        assert!(
            present_challenge::consume(&join_ks, "stale-thread", now)
                .await
                .is_err()
        );
        // Expired credx-pending offer purged.
        assert!(
            join_ks
                .prefix_iter_raw(b"credx-pending:".to_vec())
                .await
                .unwrap()
                .is_empty()
        );
        // Old Failed sync job reaped.
        assert!(list_sync_jobs(&sync_ks).await.unwrap().is_empty());
    }
}