//! Per-file source extraction: walker, parallel parse, and hash persistence.
//!
//! `collect_candidates` walks a repo tree respecting `.gitignore` and
//! hardcoded build-output excludes, then `run_parallel_extract` fans out
//! over them with rayon and aggregates results for the serial DB flush.

use std::path::{Path, PathBuf};

use anyhow::{Context, Result};
use ckg_storage::Storage;
use cozo::DataValue;
use rayon::prelude::*;

const SOURCE_FILE_LIMIT_BYTES: u64 = 1_000_000;

/// Per-file extraction outcome, produced by the rayon parallel pass.
pub(super) enum ExtractResult {
    Done {
        rel: String,
        lang: String,
        hash: String,
        symbols: Vec<ckg_core::Symbol>,
        edges: Vec<ckg_core::Edge>,
    },
    Skipped,
    Failed,
}

/// Walk `repo_root` with `.gitignore` + hardcoded build-output exclusions.
/// Returns `(absolute_path, language_str)` for every parseable source file
/// under the size limit.
///
/// Walker errors are surfaced as warnings (CR-I5) rather than hard failures.
pub(super) fn collect_candidates(repo_root: &Path) -> Result<Vec<(PathBuf, &'static str)>> {
    let mut candidates: Vec<(PathBuf, &'static str)> = Vec::new();
    let mut overrides = ignore::overrides::OverrideBuilder::new(repo_root);
    // Phase-3 refactor: build-output dir excludes come from the active
    // adapter set (target/ for Rust, node_modules/ for JS, __pycache__/
    // for Python, etc.). Fallback dirs are added unconditionally so a
    // repo with no fingerprint files still skips the obvious cases.
    let registry = ckg_lang::AdapterRegistry::detect(repo_root);
    let mut excludes: Vec<String> = registry
        .union_build_dirs()
        .into_iter()
        .map(|d| format!("!{d}"))
        .collect();
    for fallback in &["!target", "!node_modules", "!dist", "!build", "!__pycache__"] {
        if !excludes.iter().any(|e| e == fallback) {
            excludes.push((*fallback).into());
        }
    }
    for excl in &excludes {
        let _ = overrides.add(excl);
    }
    let walker = ignore::WalkBuilder::new(repo_root)
        .git_ignore(true)
        .git_global(true)
        .git_exclude(true)
        .hidden(true)
        .follow_links(false)
        .overrides(overrides.build().with_context(|| "build override matcher")?)
        .build();
    for entry in walker {
        let entry = match entry {
            Ok(e) => e,
            Err(e) => {
                // CR-I5: surface walker errors instead of silent skip.
                tracing::warn!("walker: {e}");
                continue;
            }
        };
        if !entry.file_type().map(|t| t.is_file()).unwrap_or(false) {
            continue;
        }
        let path = entry.path();
        if path
            .metadata()
            .map(|m| m.len() > SOURCE_FILE_LIMIT_BYTES)
            .unwrap_or(false)
        {
            continue;
        }
        let Some(lang) = ckg_lang::detect_language(path) else {
            continue;
        };
        candidates.push((path.to_path_buf(), lang));
    }
    Ok(candidates)
}

/// Aggregate produced by `run_parallel_extract`, ready for the serial DB flush.
pub(super) struct ExtractBatch {
    pub files_scanned: usize,
    pub outputs: Vec<ExtractResult>,
}

/// Fan out extraction over `candidates` in parallel (rayon). Each file is
/// hashed with blake3; files whose hash matches `known_hashes` are skipped
/// (incremental mode). Returns the raw per-file results for the caller to
/// aggregate and flush serially.
pub(super) fn run_parallel_extract(
    repo_id: &str,
    repo_root: &Path,
    candidates: Vec<(PathBuf, &'static str)>,
    known_hashes: &std::collections::HashMap<String, String>,
) -> ExtractBatch {
    let files_scanned = candidates.len();
    let outputs: Vec<ExtractResult> = candidates
        .par_iter()
        .filter_map(|(path, lang)| {
            let source = std::fs::read_to_string(path).ok()?;
            // L7: truncate to 24 hex chars (96 bits) for parity with
            // RepoId's 24-hex representation. 16 hex (64 bits) was
            // marginally collision-prone for very large monorepos.
            let hash = blake3::hash(source.as_bytes()).to_hex()[..24].to_string();
            let rel = path
                .strip_prefix(repo_root)
                .unwrap_or(path)
                .to_string_lossy()
                .into_owned();
            if known_hashes.get(&rel) == Some(&hash) {
                return Some(ExtractResult::Skipped);
            }
            match ckg_lang::extract(repo_id, lang, Path::new(&rel), &source) {
                Ok(mut out) => Some(ExtractResult::Done {
                    rel,
                    lang: (*lang).to_string(),
                    hash,
                    symbols: std::mem::take(&mut out.symbols),
                    edges: std::mem::take(&mut out.edges),
                }),
                Err(e) => {
                    tracing::warn!("extract failed for {}: {e}", path.display());
                    Some(ExtractResult::Failed)
                }
            }
        })
        .collect();
    ExtractBatch {
        files_scanned,
        outputs,
    }
}

/// Load per-file blake3 hashes from the `File` relation for incremental skip.
/// Returns an empty map when the relation is empty or on any read error.
pub(super) fn load_file_hashes(storage: &Storage) -> std::collections::HashMap<String, String> {
    let mut out = std::collections::HashMap::new();
    if let Ok(rows) = storage.run_immutable("?[path, hash] := *File{path, hash}") {
        for row in rows.rows {
            if let (DataValue::Str(p), DataValue::Str(h)) = (&row[0], &row[1]) {
                out.insert(p.to_string(), h.to_string());
            }
        }
    }
    out
}

/// Persist `(path, language, hash)` rows into the `File` relation.
pub(super) fn persist_file_hashes(
    storage: &Storage,
    files: &[(String, String, String)],
) -> Result<()> {
    if files.is_empty() {
        return Ok(());
    }
    let rows: Vec<DataValue> = files
        .iter()
        .map(|(p, l, h)| {
            DataValue::List(vec![
                DataValue::from(p.as_str()),
                DataValue::from(l.as_str()),
                DataValue::from(h.as_str()),
            ])
        })
        .collect();
    let mut params = std::collections::BTreeMap::new();
    params.insert("rows".into(), DataValue::List(rows));
    storage.run_with(
        "?[path, language, hash] <- $rows :put File {path => language, hash}",
        params,
    )?;
    Ok(())
}
