gunnar-server 2.0.0

//! **The account audit logs' retention sweep** (2026-10-05): every few hours,
//! each account's log drops the days past its retention (a year by default),
//! so an account nothing happens in for a while does not keep entries the
//! promise says are gone. An append does the same for its own account first;
//! this is the rest. Never on a twin: its control store is a replica, and the
//! primary's sweep reaches it as transactions.

use std::sync::Arc;

use crate::stores::ServerStores;

/// How often the sweep runs.
pub const EVERY: std::time::Duration = std::time::Duration::from_secs(6 * 3600);

/// **One pass**: every account (a person who is not an employee), pruned at
/// `now`. Returns how many entries went. A failure is one WARN per account.
pub fn sweep_once(stores: &ServerStores, now: i64) -> u64 {
    if stores.is_twin() {
        return 0;
    }
    let auth = stores.auth();
    let Ok(principals) = auth.principals() else {
        return 0;
    };
    let mut dropped = 0;
    for p in principals {
        if p.kind != gunnar_core::PrincipalKind::User || !matches!(auth.employer_of(&p.id), Ok(None)) {
            continue;
        }
        match stores.account_audit().prune(&p.id, now) {
            Ok(n) => dropped += n,
            Err(err) => tracing::warn!(
                target: "gunnar::audit",
                event = "account_audit.sweep_failed",
                account = %p.id,
                error = %err,
                "an account's audit log could not be swept"
            ),
        }
    }
    if dropped > 0 {
        tracing::info!(
            target: "gunnar::audit",
            event = "account_audit.swept",
            dropped,
            "account audit entries past their retention were dropped"
        );
    }
    dropped
}

/// The jera job that sweeps the account audit logs (A32-14).
pub const SWEEP_JOB: &str = "account-audit-sweep";

/// **The sweep as a jera job**: a pass at once, then every `every`, while the stores live (it
/// holds them weakly). A pass that panics is said (ALERT) and a FAILED row, and the next runs; the
/// detached task it replaces dropped the panic without a word.
pub fn start(stores: &Arc<ServerStores>, every: std::time::Duration) -> Result<Arc<gunnar_common::jobs::Job>, String> {
    let weak = Arc::downgrade(stores);
    let say: gunnar_common::jobs::Say = Arc::new(|l: &str| tracing::error!(target: "gunnar::audit", "{l}"));
    gunnar_common::jobs::start(SWEEP_JOB, every, say, move || match weak.upgrade() {
        Some(stores) => {
            sweep_once(&stores, gunnar_core::now_ms());
            gunnar_common::jobs::Round::Again
        }
        None => gunnar_common::jobs::Round::Done,
    })
}

#[cfg(test)]
mod tests {
    use super::*;

    /// ★ **The sweep is a jera job that runs while the stores live and ends with them** (A32-14).
    #[test]
    fn the_sweep_is_a_jera_job_that_ends_with_the_stores() {
        let dir = tempfile::tempdir().expect("tempdir");
        let stores = Arc::new(ServerStores::open(dir.path()).expect("stores"));
        let job = start(&stores, std::time::Duration::from_millis(10)).expect("the sweep starts as a jera job");
        assert_eq!(job.kind(), SWEEP_JOB);
        let t = std::time::Instant::now();
        while job.queue().stats().done < 2 && t.elapsed() < std::time::Duration::from_secs(10) {
            std::thread::sleep(std::time::Duration::from_millis(10));
        }
        assert!(job.queue().stats().done >= 2, "the passes are done jera jobs: {:?}", job.queue().stats());
        drop(stores);
        let t = std::time::Instant::now();
        while !job.is_stopped() && t.elapsed() < std::time::Duration::from_secs(10) {
            std::thread::sleep(std::time::Duration::from_millis(10));
        }
        assert!(job.is_stopped(), "the job ended with the stores");
    }
}