ipx/src/retention.rs

//! Disk quota and age sweeps. Replaces iPXQuotaManager.py ("SmartSpace").

use anyhow::Result;

use crate::config::Config;
use crate::db::{Candidate, Db, now};

/// Headroom kept below the quota, as getToQuota()'s `1048576 * 50` did.
pub const PAD_BYTES: u64 = 50 * 1024 * 1024;

#[derive(Debug, Default)]
pub struct Report {
    pub reconciled: usize,
    pub aged_out: Vec<Candidate>,
    pub over_quota: Vec<Candidate>,
    pub entries_pruned: usize,
    pub bytes_freed: u64,
}

/// Oldest-first until enough is freed. Pure, so the ordering rule is testable without a disk.
pub fn pick(candidates: &[Candidate], total: u64, limit: u64) -> Vec<Candidate> {
    let ceiling = limit.saturating_sub(PAD_BYTES);
    if total <= ceiling {
        return vec![];
    }
    let mut need = total - ceiling;
    let mut out = vec![];
    for c in candidates {
        if need == 0 {
            break;
        }
        let size = c.bytes.max(0) as u64;
        out.push(c.clone());
        need = need.saturating_sub(size);
    }
    out
}

/// Files older than the cutoff. Flagged ones are already excluded by the query.
pub fn aged(candidates: &[Candidate], cutoff: i64) -> Vec<Candidate> {
    candidates
        .iter()
        .filter(|c| c.age_key > 0 && c.age_key < cutoff)
        .cloned()
        .collect()
}

pub async fn run(cfg: &Config, db: &Db, dry_run: bool) -> Result<Report> {
    let mut report = Report::default();

    // Someone may have deleted a file by hand; the row must stop claiming it exists.
    for (id, path) in db.missing_files().await? {
        if !dry_run {
            db.mark_reaped(id).await?;
        }
        tracing::debug!(path, "file gone, row reaped");
        report.reconciled += 1;
    }

    for id in db.prune_abandoned_failures().await? {
        tracing::info!(feed = id, "forgot a failing feed nobody subscribes to");
    }

    let mut candidates = db.reap_candidates().await?;
    // A file's DeadAir copy is on the disk too, and goes when it goes (#171).
    for c in &mut candidates {
        c.bytes += copy_bytes(&c.path) as i64;
    }

    if cfg.general.max_age_days > 0 {
        let cutoff = now() - (cfg.general.max_age_days * 86_400) as i64;
        report.aged_out = aged(&candidates, cutoff);
        for c in &report.aged_out {
            report.bytes_freed += remove(db, c, dry_run).await?;
        }
        if !dry_run {
            report.entries_pruned = db.prune_entries(cutoff).await?;
        }
    }

    if cfg.general.max_total_gb > 0.0 {
        let limit = (cfg.general.max_total_gb * 1_073_741_824.0) as u64;
        // Re-read: the age sweep may already have freed enough.
        let remaining: Vec<Candidate> = candidates
            .iter()
            .filter(|c| !report.aged_out.iter().any(|a| a.id == c.id))
            .cloned()
            .collect();
        let total: u64 = remaining.iter().map(|c| c.bytes.max(0) as u64).sum();
        report.over_quota = pick(&remaining, total, limit);
        for c in &report.over_quota {
            report.bytes_freed += remove(db, c, dry_run).await?;
        }
    }

    Ok(report)
}

async fn remove(db: &Db, c: &Candidate, dry_run: bool) -> Result<u64> {
    if dry_run {
        return Ok(c.bytes.max(0) as u64);
    }
    // A torrent's folder measures as a directory entry, not what is in it: the recorded size then.
    let size = std::fs::metadata(&c.path)
        .ok()
        .filter(|m| m.is_file())
        .map_or(c.bytes.max(0) as u64, |m| m.len() + copy_bytes(&c.path));
    // Missing is fine: the point is that it is gone.
    if let Err(e) = crate::download::remove(std::path::Path::new(&c.path))
        && e.kind() != std::io::ErrorKind::NotFound
    {
        tracing::warn!(path = c.path, error = %e, "could not delete");
        return Ok(0);
    }
    db.mark_reaped(c.id).await?;
    Ok(size)
}

fn copy_bytes(path: &str) -> u64 {
    std::fs::metadata(crate::deadair::copy_path(std::path::Path::new(path))).map_or(0, |m| m.len())
}

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

    fn cand(id: i64, bytes: i64, age_key: i64, read: bool) -> Candidate {
        Candidate {
            id,
            url: format!("https://x/{id}"),
            path: format!("/tmp/{id}"),
            bytes,
            age_key,
            read,
        }
    }

    #[test]
    fn pick_takes_oldest_first_and_stops_once_under() {
        let gb = 1_073_741_824u64;
        // Ten 1 GB files, oldest first as the query would return them.
        let cands: Vec<Candidate> = (1..=10).map(|i| cand(i, gb as i64, i, true)).collect();

        // 10 GB used against a 6 GB quota: need 4 GB back, plus the 50 MB pad.
        let picked = pick(&cands, 10 * gb, 6 * gb);
        assert_eq!(
            picked.iter().map(|c| c.id).collect::<Vec<_>>(),
            vec![1, 2, 3, 4, 5],
            "oldest first, one extra to clear the headroom pad"
        );
    }

    #[test]
    fn pick_deletes_nothing_when_under_quota() {
        let gb = 1_073_741_824u64;
        let cands: Vec<Candidate> = (1..=3).map(|i| cand(i, gb as i64, i, true)).collect();
        assert!(pick(&cands, 3 * gb, 50 * gb).is_empty());
    }

    #[test]
    fn aged_only_takes_files_past_the_cutoff() {
        let cands = vec![cand(1, 10, 100, true), cand(2, 10, 500, true), cand(3, 10, 0, true)];
        let picked = aged(&cands, 200);
        assert_eq!(picked.iter().map(|c| c.id).collect::<Vec<_>>(), vec![1]);
        // age_key 0 means "never recorded" -- not the same as "infinitely old".
    }

    #[tokio::test]
    async fn query_never_offers_a_file_anyone_starred_and_prefers_ones_everyone_read() {
        // One file serves both subscribers, so it takes both of them to release it.
        let db = Db::memory().await.unwrap();
        db.exec_for_test(
            "INSERT INTO users (id, name, is_admin) VALUES (1,'ray',true),(2,'sam',false);
             INSERT INTO subscriptions (user_id, feed_id) VALUES (1,'f'),(2,'f');
             INSERT INTO entries (feed_id, guid, first_seen) VALUES
                ('f', 'keep',   0),
                ('f', 'half',   0),
                ('f', 'unread', 0),
                ('f', 'read',   0);
             -- Starred by one of the two, so it stays whatever the other thinks.
             INSERT INTO entry_state (user_id, feed_id, guid, read, flagged) VALUES
                (1, 'f', 'keep',   true, true),
                (2, 'f', 'keep',   true, false),
                (1, 'f', 'half',   true, false),
                (1, 'f', 'read',   true, false),
                (2, 'f', 'read',   true, false);
             INSERT INTO enclosures (id, feed_id, guid, url, path, bytes_done, state, downloaded_at) VALUES
                (1, 'f', 'keep',   'u1', '/tmp/keep',   10, 'done', 10),
                (2, 'f', 'half',   'u2', '/tmp/half',   10, 'done', 20),
                (3, 'f', 'unread', 'u3', '/tmp/unread', 10, 'done', 30),
                (4, 'f', 'read',   'u4', '/tmp/read',   10, 'done', 40);",
        ).await
        .unwrap();

        let got: Vec<i64> = db.reap_candidates().await.unwrap().iter().map(|c| c.id).collect();
        assert_eq!(
            got,
            vec![4, 2, 3],
            "starred by anyone is never offered; read by everyone goes first, and one \
             person still having it unread keeps it back with the unread ones"
        );
    }

    #[tokio::test]
    async fn the_marks_of_someone_who_left_count_for_nothing() {
        // ray and sam subscribe; left read one episode and pinned the others before unsubscribing.
        let db = Db::memory().await.unwrap();
        db.exec_for_test(
            "INSERT INTO users (id, name, is_admin) VALUES (1,'ray',true),(2,'sam',false),(3,'left',false);
             INSERT INTO subscriptions (user_id, feed_id) VALUES (1,'f'),(2,'f');
             INSERT INTO entries (feed_id, guid, first_seen) VALUES
                ('f', 'heard', 0), ('f', 'pinned', 0), ('f', 'old', 0);
             INSERT INTO entry_state (user_id, feed_id, guid, read, flagged) VALUES
                (3, 'f', 'heard', true, false),
                (1, 'f', 'heard', true, false),
                (3, 'f', 'pinned', false, true),
                (3, 'f', 'old', false, true);
             INSERT INTO enclosures (id, feed_id, guid, url, path, bytes_done, state, downloaded_at) VALUES
                (1, 'f', 'heard',  'u1', '/tmp/heard',  10, 'done', 10),
                (2, 'f', 'pinned', 'u2', '/tmp/pinned', 10, 'done', 20);",
        ).await
        .unwrap();
        let got: Vec<(i64, bool)> = db.reap_candidates().await.unwrap().iter().map(|c| (c.id, c.read)).collect();
        assert_eq!(got, vec![(1, false), (2, false)], "sam has not heard it, and nobody here pinned it (#156)");
        assert_eq!(db.prune_entries(500).await.unwrap(), 1, "the old item goes, a leaver's pin or not");
    }

    #[tokio::test]
    async fn prune_keeps_entries_that_still_have_a_file() {
        let db = Db::memory().await.unwrap();
        db.exec_for_test(
            "INSERT INTO users (id, name, is_admin) VALUES (1,'ray',true);
             INSERT INTO subscriptions (user_id, feed_id) VALUES (1,'f');
             INSERT INTO entry_state (user_id, feed_id, guid, flagged) VALUES (1,'f','flagged',true);
             INSERT INTO entries (feed_id, guid, first_seen) VALUES
                ('f', 'has-file', 100),
                ('f', 'no-file',  100),
                ('f', 'flagged',  100),
                ('f', 'recent',   900);
             INSERT INTO enclosures (id, feed_id, guid, url, path, state) VALUES
                (1, 'f', 'has-file', 'u1', '/tmp/x', 'done');",
        ).await
        .unwrap();

        assert_eq!(db.prune_entries(500).await.unwrap(), 1, "only the old, fileless, unflagged one");
    }
}