//! `on gc`: cleaning up what this machine keeps. Saves of changes that
//! landed or were dropped more than `--older` ago (30 days) go, with the
//! undo steps from before then, and then every object nothing kept still
//! needs. Hosted projects keep main's latest version, the history in the
//! host refetches when a command asks for it; local projects keep all of
//! main, since this machine has the only copy. The saver runs it once a day.

use std::collections::HashSet;
use std::time::{Duration, SystemTime};

use anyhow::Result;
use onset_core::{Id, tree};

use crate::app::App;
use crate::ids;
use crate::objects::Store;
use crate::out::{Fail, Out, Report};
use crate::ws;

/// Objects written this recently stay: a command may be about to use them
/// (ONSET_GC_FRESH, in seconds, for tests).
fn fresh() -> Duration {
    Duration::from_secs(crate::paths::var("GC_FRESH").and_then(|s| s.parse().ok()).unwrap_or(3600))
}

pub struct Cleaned {
    pub saves: usize,
    pub steps: usize,
    pub objects: usize,
    pub bytes: u64,
}

pub fn run(app: &App, older: Option<String>, dry_run: bool) -> Out {
    let words = older.unwrap_or_else(|| "30d".into());
    let cutoff = ids::parse_time(&format!("{words} ago"))
        .ok_or_else(|| Fail::new("bad_usage", format!("can't read --older {words:?}; try 30d, 12h or 2w")).state("nothing changed"))?;
    let done = clean(app, cutoff, dry_run)?;
    let mut r = Report::new();
    let verb = if dry_run { "would remove" } else { "removed" };
    r.line(verb, format!("{} saves and {} undo steps from before {}", done.saves, done.steps, ids::clock(cutoff)));
    r.line(verb, format!("{} objects nothing needs ({})", done.objects, size(done.bytes)));
    r.set("saves", done.saves).set("steps", done.steps).set("objects", done.objects).set("bytes", done.bytes).set("dry_run", dry_run);
    if dry_run {
        r.next("on gc to remove them");
    }
    Ok(r)
}

pub(crate) fn size(bytes: u64) -> String {
    match bytes {
        b if b >= 1 << 30 => format!("{:.1} GB", b as f64 / (1u64 << 30) as f64),
        b if b >= 1 << 20 => format!("{:.1} MB", b as f64 / (1u64 << 20) as f64),
        b => format!("{} KB", b.div_ceil(1024)),
    }
}

/// Removes saves, undo steps and objects from before `cutoff` (seconds since the epoch).
pub fn clean(app: &App, cutoff: i64, dry_run: bool) -> Result<Cleaned> {
    let db = &app.db;
    // Saves of changes that are over: the change's content is in main, or it was dropped.
    let over = OVER;
    let saves: usize = db.conn.query_row(&format!("SELECT count(*) FROM saves WHERE change IN ({over})"), [cutoff], |r| r.get(0))?;
    let steps: usize = db.conn.query_row("SELECT count(*) FROM ops WHERE at <= ?1", [cutoff], |r| r.get(0))?;
    if !dry_run {
        db.conn.execute(&format!("DELETE FROM saves WHERE change IN ({over})"), [cutoff])?;
        db.conn.execute("DELETE FROM ops WHERE at <= ?1", [cutoff])?;
    }
    let live = live_objects(app, cutoff)?;
    let (objects, bytes) = sweep(&app.store, &live, dry_run)?;
    if objects > 0 && !dry_run {
        // Which trees are whole here is worked out again as needed.
        db.conn.execute("DELETE FROM complete", [])?;
    }
    Ok(Cleaned { saves, steps, objects, bytes })
}

/// Changes that are over: landed or dropped by `?1`.
const OVER: &str = "SELECT id FROM changes WHERE state IN ('landed', 'dropped') AND updated <= ?1";

/// The trees that are kept, and everything they reach that's here: what
/// changes that aren't over, and undo steps after `cutoff`, still need.
pub(crate) fn live_objects(app: &App, cutoff: i64) -> Result<HashSet<Id>> {
    let db = &app.db;
    let mut roots: Vec<Id> = Vec::new();
    let mut hex = |text: &str| {
        if let Some(id) = Id::parse(text) {
            roots.push(id);
        }
    };
    // Main: all of it for a project on this machine only, the latest version for a hosted one.
    for row in db.conn.prepare("SELECT v.tree FROM versions v JOIN projects p ON p.id = v.project WHERE p.kind != 'hosted'")?.query_map([], |r| r.get::<_, String>(0))? {
        hex(&row?);
    }
    for row in db
        .conn
        .prepare("SELECT v.tree FROM versions v JOIN projects p ON p.id = v.project WHERE p.kind = 'hosted' AND v.n = (SELECT max(n) FROM versions WHERE project = v.project)")?
        .query_map([], |r| r.get::<_, String>(0))?
    {
        hex(&row?);
    }
    for row in db
        .conn
        .prepare(&format!("SELECT base_tree, head_tree FROM changes WHERE id NOT IN ({OVER})"))?
        .query_map([cutoff], |r| Ok((r.get::<_, String>(0)?, r.get::<_, Option<String>>(1)?)))?
    {
        let (base, head) = row?;
        hex(&base);
        if let Some(h) = head {
            hex(&h);
        }
    }
    let saves = db
        .conn
        .prepare(&format!("SELECT tree, base_tree FROM saves WHERE change NOT IN ({OVER})"))?
        .query_map([cutoff], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)))?
        .collect::<rusqlite::Result<Vec<_>>>()?;
    for (t, b) in saves {
        hex(&t);
        hex(&b);
    }
    // Undo steps name the trees they'd go back to.
    let steps: Vec<String> = db.conn.prepare("SELECT data FROM ops WHERE at > ?1")?.query_map([cutoff], |r| r.get(0))?.collect::<rusqlite::Result<_>>()?;
    for data in steps {
        for word in data.split(|c: char| !c.is_ascii_hexdigit()) {
            if word.len() == 64 {
                hex(word);
            }
        }
    }
    for w in db.all_workspaces()? {
        if let Some(t) = ws::Index::load(&w.path).tree {
            roots.push(t);
        }
    }
    let mut live: HashSet<Id> = HashSet::new();
    // Agent sessions kept with changes stay until they're deleted.
    for id in db.session_files()? {
        add_file(&app.store, &id, &mut live);
    }
    for row in db.conn.prepare("SELECT file FROM mine")?.query_map([], |r| r.get::<_, String>(0))? {
        if let Some(id) = Id::parse(&row?) {
            add_file(&app.store, &id, &mut live);
        }
    }
    for root in roots {
        walk(&app.store, &root, &mut live);
    }
    Ok(live)
}

fn add_file(store: &Store, id: &Id, live: &mut HashSet<Id>) {
    for part in store.file_parts(id).unwrap_or_else(|_| vec![*id]) {
        live.insert(part);
    }
}

/// A tree and what it reaches, as far as this machine has it.
fn walk(store: &Store, root: &Id, live: &mut HashSet<Id>) {
    let mut stack = vec![*root];
    while let Some(t) = stack.pop() {
        if !live.insert(t) {
            continue;
        }
        let Ok(entries) = tree::read(store, &t) else { continue };
        for e in entries {
            match e.kind {
                tree::EntryKind::Tree => stack.push(e.id),
                tree::EntryKind::Gitlink => {}
                _ => add_file(store, &e.id, live),
            }
        }
    }
}

/// Removes objects that aren't live (and weren't written in the last hour).
fn sweep(store: &Store, live: &HashSet<Id>, dry_run: bool) -> Result<(usize, u64)> {
    let now = SystemTime::now();
    let (mut count, mut bytes) = (0, 0);
    for (id, path) in store.all()? {
        if live.contains(&id) {
            continue;
        }
        let Ok(meta) = std::fs::metadata(&path) else { continue };
        if meta.modified().ok().and_then(|m| now.duration_since(m).ok()).is_none_or(|age| age < fresh()) {
            continue;
        }
        count += 1;
        bytes += meta.len();
        if !dry_run {
            let _ = std::fs::remove_file(&path);
        }
    }
    // Packed objects nothing needs: their packs are written again without them.
    let dead: HashSet<Id> = store.packed_ids().into_iter().filter(|id| !live.contains(id)).collect();
    count += dead.len();
    if dry_run {
        bytes += dead.iter().map(|id| store.stored_len(id)).sum::<u64>();
    } else if !dead.is_empty() {
        bytes += store.remove_packed(&dead)?;
    }
    // What's left loose (and not just written) goes into a pack: one file, not thousands.
    if !dry_run && store.all()?.len() >= *PACK_AT {
        store.pack_loose(fresh())?;
    }
    Ok((count, bytes))
}

/// Loose objects this many or more are packed (ONSET_PACK_AT, for tests).
static PACK_AT: std::sync::LazyLock<usize> = std::sync::LazyLock::new(|| crate::paths::var("PACK_AT").and_then(|s| s.parse().ok()).unwrap_or(1000));

/// The saver's daily run, with the defaults; quiet unless something fails.
pub fn daily(app: &App) {
    let mark = crate::paths::home().join("gc-at");
    let last = std::fs::read_to_string(&mark).ok().and_then(|t| t.trim().parse::<i64>().ok()).unwrap_or(0);
    if ids::now() - last < 86_400 {
        return;
    }
    let _ = std::fs::write(&mark, ids::now().to_string());
    if let Err(e) = clean(app, ids::now() - 30 * 86_400, false) {
        eprintln!("on: cleaning up: {e:#}");
    }
}