rpo 0.1.0-beta.4

//! Public entrypoint. See spec §5.

use std::collections::HashMap;
use std::path::Path;
use std::sync::Arc;

use polars::prelude::DataFrame;

use crate::{
    ActivityOptions, Analysis, Progress, RpoError, SnapshotSelector, StreamStats,
    backend::{DefaultBackend, GitBackend},
    blame::{self, CommitMetaBlame},
    filters::{FilterSet, LinguistAttrs},
    frames,
    identity::{Canonicalizer, IdentityMap},
    sinks::FrameSink,
    walk,
};

/// Entry point: [`RepoAnalyzer::open`] returns a [`Builder`].
pub struct RepoAnalyzer<B: GitBackend = DefaultBackend> {
    _phantom: std::marker::PhantomData<B>,
}

impl RepoAnalyzer<DefaultBackend> {
    /// Open a repository and start configuring an analysis.
    ///
    /// `path` may be a work tree or a `.git` directory.
    ///
    /// # Errors
    ///
    /// [`RpoError::NotARepo`] if `path` is not a git repository.
    ///
    /// ```no_run
    /// # use rpo::RepoAnalyzer;
    /// let commits = RepoAnalyzer::open(".")?.commits()?;
    /// # Ok::<(), rpo::RpoError>(())
    /// ```
    pub fn open(path: impl AsRef<Path>) -> Result<Builder<DefaultBackend>, RpoError> {
        let backend = DefaultBackend::open(path.as_ref())?;
        Ok(Builder::new(backend))
    }
}

type ProgressCb = Arc<dyn Fn(Progress) + Send + Sync>;

/// A configured analysis, awaiting a terminal.
///
/// Built by [`RepoAnalyzer::open`]. Set walk options with the `with_*`
/// methods, then call [`commits`](Self::commits),
/// [`file_changes`](Self::file_changes), [`blame`](Self::blame),
/// [`blame_over_time`](Self::blame_over_time), or [`all`](Self::all).
pub struct Builder<B: GitBackend> {
    backend: Arc<B>,
    mailmap: bool,
    overrides: IdentityMap,
    include_globs: Vec<String>,
    exclude_globs: Vec<String>,
    respect_gitattributes: bool,
    activity: ActivityOptions,
    snapshots: SnapshotSelector,
    thread_count: Option<usize>,
    progress: Option<ProgressCb>,
}

impl<B: GitBackend> Clone for Builder<B> {
    fn clone(&self) -> Self {
        Self {
            backend: Arc::clone(&self.backend),
            mailmap: self.mailmap,
            overrides: self.overrides.clone(),
            include_globs: self.include_globs.clone(),
            exclude_globs: self.exclude_globs.clone(),
            respect_gitattributes: self.respect_gitattributes,
            activity: self.activity,
            snapshots: self.snapshots.clone(),
            thread_count: self.thread_count,
            progress: self.progress.clone(),
        }
    }
}

impl<B: GitBackend + 'static> Builder<B> {
    fn new(backend: B) -> Self {
        Self {
            backend: Arc::new(backend),
            mailmap: true,
            overrides: IdentityMap::new(),
            include_globs: Vec::new(),
            exclude_globs: Vec::new(),
            respect_gitattributes: true,
            activity: ActivityOptions::default(),
            snapshots: SnapshotSelector::Head,
            thread_count: None,
            progress: None,
        }
    }

    /// Apply the repository's `.mailmap` when canonicalizing identities.
    ///
    /// On by default. Turning this off makes the `canonical_*` columns
    /// mirror the raw author and committer fields.
    pub fn with_mailmap(mut self, enabled: bool) -> Self {
        self.mailmap = enabled;
        self
    }
    /// Add identity aliases on top of `.mailmap`.
    ///
    /// Useful for collapsing addresses the repository's mailmap does not
    /// cover, without rewriting history.
    pub fn with_identity_overrides(mut self, map: IdentityMap) -> Self {
        self.overrides = map;
        self
    }
    /// Restrict the analysis to paths matching any of these globs.
    ///
    /// Empty (the default) includes every path. Applied during the walk,
    /// so excluded paths never enter any frame. See
    /// [`with_exclude_globs`](Self::with_exclude_globs) for the
    /// precedence rules.
    pub fn with_include_globs(mut self, globs: Vec<String>) -> Self {
        self.include_globs = globs;
        self
    }
    /// Skip paths matching any of these globs.
    ///
    /// Exclusion wins: a path matching both an include and an exclude
    /// glob is dropped. Patterns are [globset](https://docs.rs/globset)
    /// syntax matched against the full repo-relative path, and
    /// separators are not special — `*.rs` matches `src/web/pages.rs`.
    ///
    /// An invalid pattern is an error from the terminal, not a silent
    /// non-match.
    pub fn with_exclude_globs(mut self, globs: Vec<String>) -> Self {
        self.exclude_globs = globs;
        self
    }
    /// Honour `linguist-generated` and `linguist-vendored` in
    /// `.gitattributes`.
    ///
    /// On by default. Drives the `is_generated` and `is_vendored`
    /// columns, which [`FileSelection`](crate::options::FileSelection) then filters
    /// on.
    pub fn with_respect_gitattributes(mut self, enabled: bool) -> Self {
        self.respect_gitattributes = enabled;
        self
    }
    /// Set which commits enter the walk — merges, first-parent-only,
    /// and bot filtering. See [`ActivityOptions`].
    pub fn with_activity(mut self, activity: ActivityOptions) -> Self {
        self.activity = activity;
        self
    }
    /// Choose the revisions [`blame_over_time`](Self::blame_over_time)
    /// samples.
    ///
    /// Defaults to [`SnapshotSelector::Head`]. Has no effect on
    /// [`blame`](Self::blame), which is always HEAD.
    pub fn with_blame_snapshots(mut self, selector: SnapshotSelector) -> Self {
        self.snapshots = selector;
        self
    }
    /// Cap the threads used for blame.
    ///
    /// Defaults to rayon's choice, normally one per core.
    pub fn with_thread_count(mut self, n: usize) -> Self {
        self.thread_count = Some(n);
        self
    }
    /// Receive [`Progress`] updates during the walk and blame phases.
    ///
    /// The callback runs on worker threads and must be `Send + Sync`.
    pub fn with_progress<F: Fn(Progress) + Send + Sync + 'static>(mut self, cb: F) -> Self {
        self.progress = Some(Arc::new(cb));
        self
    }

    // Terminals.

    /// Walk history and return one row per commit.
    ///
    /// Cheaper than [`all`](Self::all) when the per-file detail is not
    /// needed; no blame is computed.
    pub fn commits(self) -> Result<DataFrame, RpoError> {
        let Prepared {
            canonicalizer,
            filters,
            ..
        } = self.prepare()?;
        let out = walk::run(
            self.backend.as_ref(),
            &canonicalizer,
            &filters,
            &self.activity,
            self.progress.as_deref(),
        )?;
        frames::commits::build(out.commits)
    }

    /// Walk history and return one row per (commit, file) touched.
    ///
    /// Path globs and `.gitattributes` classification apply here.
    pub fn file_changes(self) -> Result<DataFrame, RpoError> {
        let Prepared {
            canonicalizer,
            filters,
            ..
        } = self.prepare()?;
        let out = walk::run(
            self.backend.as_ref(),
            &canonicalizer,
            &filters,
            &self.activity,
            self.progress.as_deref(),
        )?;
        frames::file_changes::build(out.file_changes)
    }

    /// Blame at HEAD only. Any selector previously set via
    /// [`Builder::with_blame_snapshots`] is ignored — use
    /// [`Builder::blame_over_time`] for multi-snapshot blame.
    pub fn blame(self) -> Result<DataFrame, RpoError> {
        let mut me = self;
        me.snapshots = SnapshotSelector::Head;
        me.run_blame_in_memory()
    }

    /// Blame at every revision chosen by
    /// [`with_blame_snapshots`](Self::with_blame_snapshots).
    ///
    /// The result holds every snapshot at once. For histories too large
    /// to fit in memory, use
    /// [`blame_over_time_streaming`](Self::blame_over_time_streaming).
    pub fn blame_over_time(self) -> Result<DataFrame, RpoError> {
        self.run_blame_in_memory()
    }

    /// Blame over time, handing each snapshot to `sink` as it is
    /// produced rather than accumulating them.
    ///
    /// Keeps peak memory to a single snapshot. See [`ParquetSink`],
    /// [`ParquetDirSink`], and `DuckDbSink` (the last behind the
    /// `sink-duckdb` feature).
    ///
    /// [`ParquetSink`]: crate::ParquetSink
    /// [`ParquetDirSink`]: crate::ParquetDirSink
    pub fn blame_over_time_streaming<S: FrameSink + ?Sized>(
        self,
        sink: &mut S,
    ) -> Result<StreamStats, RpoError> {
        let Prepared {
            filters,
            commit_meta,
            ..
        } = self.prepare()?;
        blame::run_streaming(
            self.backend.as_ref(),
            &filters,
            &self.snapshots,
            &self.activity,
            self.thread_count,
            self.progress.as_deref(),
            &commit_meta,
            sink,
        )
    }

    /// Run every terminal from a single walk.
    ///
    /// Preferable to calling the terminals separately when more than one
    /// frame is needed — history is traversed once.
    ///
    /// `blame_over_time` is populated only when
    /// [`with_blame_snapshots`](Self::with_blame_snapshots) selects
    /// something other than [`SnapshotSelector::Head`].
    pub fn all(self) -> Result<Analysis, RpoError> {
        let Prepared {
            canonicalizer,
            filters,
            commit_meta,
        } = self.prepare()?;
        let walk_out = walk::run(
            self.backend.as_ref(),
            &canonicalizer,
            &filters,
            &self.activity,
            self.progress.as_deref(),
        )?;

        let (blame, blame_over_time, skipped) = match &self.snapshots {
            SnapshotSelector::Head => {
                let r = blame::run_in_memory(
                    self.backend.as_ref(),
                    &filters,
                    &SnapshotSelector::Head,
                    &self.activity,
                    self.thread_count,
                    self.progress.as_deref(),
                    &commit_meta,
                )?;
                let frame = frames::blame::build(r.records)?;
                (Some(frame), None, r.skipped)
            }
            other => {
                // The timeline pass always includes a HEAD snapshot (see
                // resolve_snapshots), so we can derive the HEAD-only blame
                // frame by filtering without a second blame walk.
                let r = blame::run_in_memory(
                    self.backend.as_ref(),
                    &filters,
                    other,
                    &self.activity,
                    self.thread_count,
                    self.progress.as_deref(),
                    &commit_meta,
                )?;
                let timeline = frames::blame::build(r.records)?;
                let head_sha = self.backend.head_commit()?.to_hex();
                let head_only = filter_blame_to_head(&timeline, &head_sha)?;
                (Some(head_only), Some(timeline), r.skipped)
            }
        };

        Ok(Analysis {
            commits: frames::commits::build(walk_out.commits)?,
            file_changes: frames::file_changes::build(walk_out.file_changes)?,
            blame,
            blame_over_time,
            skipped_files: skipped,
        })
    }

    // Internals.

    fn run_blame_in_memory(self) -> Result<DataFrame, RpoError> {
        let Prepared {
            filters,
            commit_meta,
            ..
        } = self.prepare()?;
        let r = blame::run_in_memory(
            self.backend.as_ref(),
            &filters,
            &self.snapshots,
            &self.activity,
            self.thread_count,
            self.progress.as_deref(),
            &commit_meta,
        )?;
        frames::blame::build(r.records)
    }

    fn prepare(&self) -> Result<Prepared, RpoError> {
        let mailmap_bytes = if self.mailmap {
            self.backend.mailmap_bytes()?
        } else {
            None
        };
        let canonicalizer = Canonicalizer::new(mailmap_bytes.as_deref(), self.overrides.clone())?;

        let attrs = if self.respect_gitattributes {
            self.backend
                .gitattributes_bytes()?
                .as_deref()
                .map(LinguistAttrs::parse)
        } else {
            None
        };
        let filters = FilterSet::build(
            &self.include_globs,
            &self.exclude_globs,
            attrs,
            self.respect_gitattributes,
        )?;

        // Build a commit-meta lookup covering every commit we might reference
        // from a blame hunk. We walk ALL commits (regardless of merge setting)
        // so blame joins don't lose rows.
        let mut commit_meta: HashMap<String, CommitMetaBlame> = HashMap::new();
        for c in self.backend.iter_commits(crate::backend::WalkOptions {
            first_parent_only: false,
            include_merges: true,
        }) {
            let c = c?;
            let (an, ae) = canonicalizer.canonicalize(&c.author);
            let (cn, ce) = canonicalizer.canonicalize(&c.committer);
            commit_meta.insert(
                c.id.to_hex(),
                CommitMetaBlame {
                    canonical_author_name: an,
                    canonical_author_email: ae,
                    canonical_committer_name: cn,
                    canonical_committer_email: ce,
                    commit_time_ms: c.committer.time_ms,
                },
            );
        }

        Ok(Prepared {
            canonicalizer,
            filters,
            commit_meta,
        })
    }
}

struct Prepared {
    canonicalizer: Canonicalizer,
    filters: FilterSet,
    commit_meta: HashMap<String, CommitMetaBlame>,
}

/// Derive a HEAD-only blame frame from a multi-snapshot timeline frame by
/// filtering on `snapshot_sha`. Used by `Builder::all()` to avoid a second
/// blame walk when the user requested a non-HEAD snapshot cadence.
fn filter_blame_to_head(timeline: &DataFrame, head_sha: &str) -> Result<DataFrame, RpoError> {
    use polars::prelude::*;
    let out = timeline
        .clone()
        .lazy()
        .filter(col("snapshot_sha").eq(lit(head_sha)))
        .collect()?;
    Ok(out)
}

#[cfg(all(test, feature = "backend-gix"))]
mod tests {
    use std::process::Command;

    use super::*;

    fn git(repo: &std::path::Path, args: &[&str]) {
        let status = Command::new("git")
            .arg("-C")
            .arg(repo)
            .args(args)
            .status()
            .expect("git command should run");
        assert!(status.success(), "git {args:?} failed");
    }

    #[test]
    fn root_commit_file_changes_include_added_line_counts() {
        let dir = tempfile::tempdir().expect("tempdir");
        let repo = dir.path();

        Command::new("git")
            .arg("init")
            .arg("-q")
            .arg(repo)
            .status()
            .expect("git init should run");
        git(repo, &["config", "user.name", "Ada"]);
        git(repo, &["config", "user.email", "ada@example.com"]);

        std::fs::write(repo.join("README.md"), "one\ntwo\nthree\n").expect("write readme");
        std::fs::write(repo.join("binary.dat"), b"one\0two\n").expect("write binary");
        git(repo, &["add", "."]);
        git(repo, &["commit", "-q", "-m", "initial"]);

        let df = RepoAnalyzer::open(repo)
            .and_then(|b| b.file_changes())
            .expect("file changes");

        let path_col = df.column("path").unwrap().str().unwrap();
        let insertions_col = df.column("insertions").unwrap().u64().unwrap();
        let readme_idx = (0..df.height())
            .find(|&i| path_col.get(i) == Some("README.md"))
            .expect("README.md row");
        let binary_idx = (0..df.height())
            .find(|&i| path_col.get(i) == Some("binary.dat"))
            .expect("binary.dat row");

        assert_eq!(insertions_col.get(readme_idx), Some(3));
        assert_eq!(insertions_col.get(binary_idx), Some(0));
    }
}