//! Local blob-store garbage collection.
//!
//! The content-addressed blob store under `<data_dir>/blobs` accumulates files
//! that lose every reference when attachments go away, either through
//! `delete_attachment`, or through the `ON DELETE CASCADE` on
//! `attachments.task_id` / `attachments.project_id`, which removes attachment
//! rows without ever calling `delete_attachment`. Left alone the orphaned blobs
//! are a permanent disk leak.
//!
//! [`reconcile`] walks the blob directory once at startup and unlinks any file
//! that nothing references. References come from two places, and BOTH must be
//! consulted, deleting a blob an email still points at would be data loss:
//!   1. `attachments.blob_hash`, task / project attachments.
//!   2. `emails.attachment_meta`, email attachments not yet converted to rows.
//!
//! Running at startup (before the sync/email schedulers spawn and before the UI
//! can create attachments) keeps it free of races with concurrent blob writers.

use std::collections::HashSet;
use std::path::Path;

use goingson_core::AttachmentMeta;
use sqlx::SqlitePool;
use tracing::{info, warn};

/// Collect every blob hash currently referenced by an attachment row or an
/// email's `attachment_meta` JSON.
pub async fn collect_referenced_hashes(pool: &SqlitePool) -> Result<HashSet<String>, sqlx::Error> {
    let mut referenced = HashSet::new();

    let attachment_hashes: Vec<(String,)> =
        sqlx::query_as("SELECT DISTINCT blob_hash FROM attachments")
            .fetch_all(pool)
            .await?;
    referenced.extend(attachment_hashes.into_iter().map(|(h,)| h));

    let metas: Vec<(Option<String>,)> = sqlx::query_as(
        "SELECT attachment_meta FROM emails WHERE attachment_meta IS NOT NULL AND attachment_meta != ''",
    )
    .fetch_all(pool)
    .await?;
    for (meta,) in metas {
        let Some(json) = meta else { continue };
        match serde_json::from_str::<Vec<AttachmentMeta>>(&json) {
            Ok(list) => referenced.extend(list.into_iter().map(|m| m.blob_hash)),
            // A malformed row must never widen the delete set: keep its blobs.
            Err(e) => warn!("Blob GC: skipping unparseable attachment_meta: {e}"),
        }
    }

    Ok(referenced)
}

/// Remove blob files in `blobs_dir` whose name is not in `referenced`.
///
/// Returns the number of files unlinked. Non-files and unreadable entries are
/// skipped; a failed unlink is logged and does not abort the sweep.
pub fn reclaim_orphans<S: std::hash::BuildHasher>(
    blobs_dir: &Path,
    referenced: &HashSet<String, S>,
) -> std::io::Result<usize> {
    if !blobs_dir.exists() {
        return Ok(0);
    }

    let mut removed = 0;
    for entry in std::fs::read_dir(blobs_dir)? {
        let entry = entry?;
        if !entry.file_type()?.is_file() {
            continue;
        }
        let name = entry.file_name();
        let Some(name) = name.to_str() else { continue };
        if referenced.contains(name) {
            continue;
        }
        match std::fs::remove_file(entry.path()) {
            Ok(()) => removed += 1,
            Err(e) => warn!("Blob GC: failed to remove orphan {name}: {e}"),
        }
    }
    Ok(removed)
}

/// Reconcile the blob store against live references and unlink orphans.
///
/// Best-effort: any failure is logged and swallowed so startup is never blocked.
pub async fn reconcile(pool: &SqlitePool, data_dir: &Path) {
    let referenced = match collect_referenced_hashes(pool).await {
        Ok(set) => set,
        Err(e) => {
            warn!("Blob GC skipped: could not collect references: {e}");
            return;
        }
    };

    // read_dir + remove_file are blocking; run on the blocking pool so the
    // startup GC doesn't stall the reactor / delay first paint (Perf M4).
    let blobs_dir = data_dir.join("blobs");
    match tokio::task::spawn_blocking(move || reclaim_orphans(&blobs_dir, &referenced)).await {
        Ok(Ok(0)) => {}
        Ok(Ok(n)) => info!("Blob GC reclaimed {n} orphaned blob(s)"),
        Ok(Err(e)) => warn!("Blob GC: could not scan blob dir: {e}"),
        Err(e) => warn!("Blob GC task panicked: {e}"),
    }
}

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

    #[test]
    fn reclaim_removes_only_unreferenced_files() {
        let dir = std::env::temp_dir().join(format!("goingson-blobgc-test-{}", std::process::id()));
        let blobs = dir.join("blobs");
        std::fs::create_dir_all(&blobs).unwrap();

        std::fs::write(blobs.join("keep_me"), b"a").unwrap();
        std::fs::write(blobs.join("orphan_1"), b"b").unwrap();
        std::fs::write(blobs.join("orphan_2"), b"c").unwrap();

        let mut referenced = HashSet::new();
        referenced.insert("keep_me".to_string());
        // A referenced-but-missing hash must not error (blob not local yet).
        referenced.insert("not_on_disk".to_string());

        let removed = reclaim_orphans(&blobs, &referenced).unwrap();
        assert_eq!(removed, 2);
        assert!(blobs.join("keep_me").exists());
        assert!(!blobs.join("orphan_1").exists());
        assert!(!blobs.join("orphan_2").exists());

        std::fs::remove_dir_all(&dir).ok();
    }

    #[test]
    fn reclaim_on_missing_dir_is_noop() {
        let missing = std::env::temp_dir().join("goingson-blobgc-does-not-exist-xyz");
        let removed = reclaim_orphans(&missing, &HashSet::new()).unwrap();
        assert_eq!(removed, 0);
    }
}