rpo 0.1.0-beta.4

//! Activity reports: pure transforms on `commits` and `file_changes`
//! frames.
//!
//! Each function reads only the columns it needs and applies
//! [`crate::reports::filter_files`] internally so callers don't have
//! to remember to pre-filter.

use polars::prelude::*;

use crate::RpoError;
use crate::options::{ActivityOptions, Aggregation, FileSelection};
use crate::reports::{filter_bots, filter_files};

/// Per-contributor activity report from cloned commits + file_changes frames.
///
/// Columns: `canonical_<aggregate>_<identify>`, `commits`, `first_commit`,
/// `last_commit`, `files`, `insertions`, `deletions`.
pub fn author_report(
    commits: &DataFrame,
    file_changes: &DataFrame,
    agg: Aggregation,
    sel: FileSelection,
    activity: ActivityOptions,
) -> Result<DataFrame, RpoError> {
    let group = agg.group_col();

    let files_lf = filter_bots(
        filter_files(file_changes.clone().lazy(), sel),
        agg,
        activity,
    );
    let included_shas = files_lf
        .clone()
        .select([col("sha")])
        .unique(None, UniqueKeepStrategy::Any);

    let commits_by = filter_bots(commits.clone().lazy(), agg, activity)
        .join(
            included_shas,
            [col("sha")],
            [col("sha")],
            JoinArgs::new(JoinType::Inner),
        )
        .group_by([col(&group)])
        .agg([
            col("sha").n_unique().alias("commits"),
            col("author_time").min().alias("first_commit"),
            col("author_time").max().alias("last_commit"),
        ]);

    let changes_by = files_lf.group_by([col(&group)]).agg([
        col("path").n_unique().alias("files"),
        col("insertions").sum().alias("insertions"),
        col("deletions").sum().alias("deletions"),
    ]);

    let df = commits_by
        .join(
            changes_by,
            [col(&group)],
            [col(&group)],
            JoinArgs::new(JoinType::Left),
        )
        .collect()?;
    Ok(df)
}

/// Single-row summary of repository totals.
///
/// Columns: `contributors`, `commits`, `first_commit`, `last_commit`, `files`.
pub fn summary(
    commits: &DataFrame,
    file_changes: &DataFrame,
    agg: Aggregation,
    sel: FileSelection,
    activity: ActivityOptions,
) -> Result<DataFrame, RpoError> {
    let group = agg.group_col();

    let files_lf = filter_bots(
        filter_files(file_changes.clone().lazy(), sel),
        agg,
        activity,
    );
    let included_shas = files_lf
        .clone()
        .select([col("sha")])
        .unique(None, UniqueKeepStrategy::Any);

    let commits_scoped = filter_bots(commits.clone().lazy(), agg, activity).join(
        included_shas,
        [col("sha")],
        [col("sha")],
        JoinArgs::new(JoinType::Inner),
    );

    let mut df = commits_scoped
        .select([
            col(&group).n_unique().alias("contributors"),
            col("sha").n_unique().alias("commits"),
            col("author_time").min().alias("first_commit"),
            col("author_time").max().alias("last_commit"),
        ])
        .collect()?;

    let files = files_lf
        .select([col("path").n_unique().alias("files")])
        .collect()?;

    for c in files.columns() {
        df.with_column(c.clone())?;
    }

    Ok(df)
}

/// Per-(path, author) commit matrix. Used by the /file-report heatmap.
///
/// Columns: `path`, `canonical_<aggregate>_<identify>`, `commits`.
pub fn file_author_matrix(
    file_changes: &DataFrame,
    agg: Aggregation,
    sel: FileSelection,
    activity: ActivityOptions,
) -> Result<DataFrame, RpoError> {
    let group = agg.group_col();

    let files_lf = filter_bots(
        filter_files(file_changes.clone().lazy(), sel),
        agg,
        activity,
    );

    let df = files_lf
        .group_by([col("path"), col(&group)])
        .agg([col("sha").n_unique().alias("commits")])
        .collect()?;

    Ok(df)
}

/// Per-file activity report.
///
/// Columns: `path`, `commits`, `contributors`, `insertions`, `deletions`, `net`.
pub fn file_report(
    commits: &DataFrame,
    file_changes: &DataFrame,
    agg: Aggregation,
    sel: FileSelection,
    activity: ActivityOptions,
) -> Result<DataFrame, RpoError> {
    let _ = commits;
    let group = agg.group_col();

    let files_lf = filter_bots(
        filter_files(file_changes.clone().lazy(), sel),
        agg,
        activity,
    );

    let df = files_lf
        .group_by([col("path")])
        .agg([
            col("sha").n_unique().alias("commits"),
            col(&group).n_unique().alias("contributors"),
            col("insertions").sum().alias("insertions"),
            col("deletions").sum().alias("deletions"),
            (col("insertions").cast(DataType::Int64).sum()
                - col("deletions").cast(DataType::Int64).sum())
            .alias("net"),
        ])
        .collect()?;

    Ok(df)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::options::{ActivityOptions, Aggregate, Aggregation, FileSelection, Identify};

    fn agg_author_name() -> Aggregation {
        Aggregation {
            aggregate: Aggregate::Author,
            identify: Identify::Name,
        }
    }

    fn fixture_commits() -> DataFrame {
        let df = df! {
            "sha" => ["c1", "c2", "c3"],
            "canonical_author_name" => ["alice", "bob", "alice"],
            "author_time" => [1_700_000_000_000i64, 1_700_000_100_000, 1_700_000_200_000],
        }
        .unwrap();
        df.lazy()
            .with_column(col("author_time").cast(DataType::Datetime(
                TimeUnit::Milliseconds,
                Some(TimeZone::UTC),
            )))
            .collect()
            .unwrap()
    }

    fn fixture_file_changes() -> DataFrame {
        df! {
            "sha" => ["c1", "c1", "c2", "c2", "c3", "c3"],
            "canonical_author_name" => ["alice", "alice", "bob", "bob", "alice", "alice"],
            "path" => ["a.rs", "b.rs", "a.rs", "c.rs", "a.rs", "d.rs"],
            "insertions" => [10u32, 20, 5, 8, 3, 4],
            "deletions" => [1u32, 2, 0, 1, 0, 0],
            "is_generated" => [false, false, false, false, false, false],
            "is_vendored" => [false, false, false, false, false, false],
        }
        .unwrap()
    }

    #[test]
    fn author_report_groups_by_canonical_author_name() {
        let commits = fixture_commits();
        let file_changes = fixture_file_changes();
        let df = author_report(
            &commits,
            &file_changes,
            agg_author_name(),
            FileSelection::default(),
            ActivityOptions::default(),
        )
        .expect("author_report failed");

        assert_eq!(df.height(), 2, "expected one row per author");

        let mut names: Vec<String> = df
            .column("canonical_author_name")
            .unwrap()
            .str()
            .unwrap()
            .iter()
            .map(|s| s.unwrap().to_string())
            .collect();
        names.sort();
        assert_eq!(names, vec!["alice".to_string(), "bob".to_string()]);

        for expected in [
            "canonical_author_name",
            "commits",
            "first_commit",
            "last_commit",
            "files",
            "insertions",
            "deletions",
        ] {
            assert!(
                df.get_column_names().iter().any(|n| n.as_str() == expected),
                "missing column {expected:?}; got: {:?}",
                df.get_column_names()
            );
        }

        let name_col = df.column("canonical_author_name").unwrap().str().unwrap();
        let alice_idx = (0..df.height())
            .find(|&i| name_col.get(i).unwrap() == "alice")
            .unwrap();

        let commits_col = df.column("commits").unwrap().u32().unwrap();
        assert_eq!(commits_col.get(alice_idx).unwrap(), 2);

        let files_col = df.column("files").unwrap().u32().unwrap();
        assert_eq!(files_col.get(alice_idx).unwrap(), 3);

        let ins_col = df.column("insertions").unwrap().u32().unwrap();
        assert_eq!(ins_col.get(alice_idx).unwrap(), 37);
    }

    #[test]
    fn summary_returns_single_row_with_aggregate_counts() {
        let commits = fixture_commits();
        let file_changes = fixture_file_changes();
        let df = summary(
            &commits,
            &file_changes,
            agg_author_name(),
            FileSelection::default(),
            ActivityOptions::default(),
        )
        .expect("summary failed");

        assert_eq!(df.height(), 1, "summary should produce one row");

        for expected in [
            "contributors",
            "commits",
            "first_commit",
            "last_commit",
            "files",
        ] {
            assert!(
                df.get_column_names().iter().any(|n| n.as_str() == expected),
                "missing column {expected:?}; got: {:?}",
                df.get_column_names()
            );
        }

        let contributors = df.column("contributors").unwrap().u32().unwrap();
        assert_eq!(contributors.get(0).unwrap(), 2);

        let commits_col = df.column("commits").unwrap().u32().unwrap();
        assert_eq!(commits_col.get(0).unwrap(), 3);

        let files = df.column("files").unwrap().u32().unwrap();
        assert_eq!(files.get(0).unwrap(), 4);
    }

    #[test]
    fn file_report_groups_by_path() {
        let commits = fixture_commits();
        let file_changes = fixture_file_changes();
        let df = file_report(
            &commits,
            &file_changes,
            agg_author_name(),
            FileSelection::default(),
            ActivityOptions::default(),
        )
        .expect("file_report failed");

        assert_eq!(df.height(), 4, "expected one row per path");

        for expected in [
            "path",
            "commits",
            "contributors",
            "insertions",
            "deletions",
            "net",
        ] {
            assert!(
                df.get_column_names().iter().any(|n| n.as_str() == expected),
                "missing column {expected:?}; got: {:?}",
                df.get_column_names()
            );
        }

        let path_col = df.column("path").unwrap().str().unwrap();
        let a_rs_idx = (0..df.height())
            .find(|&i| path_col.get(i).unwrap() == "a.rs")
            .unwrap();

        let commits_col = df.column("commits").unwrap().u32().unwrap();
        assert_eq!(commits_col.get(a_rs_idx).unwrap(), 3);

        let contributors = df.column("contributors").unwrap().u32().unwrap();
        assert_eq!(contributors.get(a_rs_idx).unwrap(), 2);

        let net = df.column("net").unwrap().i64().unwrap();
        assert_eq!(net.get(a_rs_idx).unwrap(), 17);
    }

    #[test]
    fn file_author_matrix_groups_by_path_and_author() {
        let file_changes = fixture_file_changes();
        let df = file_author_matrix(
            &file_changes,
            agg_author_name(),
            FileSelection::default(),
            ActivityOptions::default(),
        )
        .expect("file_author_matrix failed");

        // fixture has 6 rows across 4 distinct paths and 2 authors.
        // Pairs: (a.rs,alice) from c1,c3 → 2; (b.rs,alice) from c1 → 1;
        // (a.rs,bob) from c2 → 1; (c.rs,bob) from c2 → 1;
        // (d.rs,alice) from c3 → 1. Total 5 distinct (path, author) pairs.
        assert_eq!(df.height(), 5);

        for expected in ["path", "canonical_author_name", "commits"] {
            assert!(
                df.get_column_names().iter().any(|n| n.as_str() == expected),
                "missing column {expected:?}; got: {:?}",
                df.get_column_names()
            );
        }

        // Check (a.rs, alice) → 2 commits.
        let path_col = df.column("path").unwrap().str().unwrap();
        let name_col = df.column("canonical_author_name").unwrap().str().unwrap();
        let idx = (0..df.height())
            .find(|&i| path_col.get(i).unwrap() == "a.rs" && name_col.get(i).unwrap() == "alice")
            .unwrap();
        let commits_col = df.column("commits").unwrap().u32().unwrap();
        assert_eq!(commits_col.get(idx).unwrap(), 2);
    }

    #[test]
    fn author_report_drops_bot_rows_when_ignore_bots_is_set() {
        // Reuse fixture_commits/fixture_file_changes plus a bot row.
        let commits = df! {
            "sha" => ["c1", "c2", "c3"],
            "canonical_author_name" => ["alice", "dependabot[bot]", "alice"],
            "author_time" => [1_700_000_000_000i64, 1_700_000_100_000, 1_700_000_200_000],
        }
        .unwrap()
        .lazy()
        .with_column(col("author_time").cast(DataType::Datetime(
            TimeUnit::Milliseconds,
            Some(TimeZone::UTC),
        )))
        .collect()
        .unwrap();

        let file_changes = df! {
            "sha" => ["c1", "c2", "c3"],
            "canonical_author_name" => ["alice", "dependabot[bot]", "alice"],
            "path" => ["a.rs", "b.rs", "a.rs"],
            "insertions" => [10u32, 20, 5],
            "deletions" => [0u32, 0, 0],
            "is_generated" => [false, false, false],
            "is_vendored" => [false, false, false],
        }
        .unwrap();

        let activity_off = author_report(
            &commits,
            &file_changes,
            agg_author_name(),
            FileSelection::default(),
            ActivityOptions::default(),
        )
        .expect("author_report (default) failed");
        // alice + dependabot[bot] = 2 rows
        assert_eq!(activity_off.height(), 2);

        let activity_on = author_report(
            &commits,
            &file_changes,
            agg_author_name(),
            FileSelection::default(),
            ActivityOptions {
                ignore_bots: true,
                ..ActivityOptions::default()
            },
        )
        .expect("author_report (ignore_bots) failed");
        // dependabot dropped → only alice
        assert_eq!(activity_on.height(), 1);
        let names: Vec<&str> = activity_on
            .column("canonical_author_name")
            .unwrap()
            .str()
            .unwrap()
            .iter()
            .map(|v| v.unwrap())
            .collect();
        assert_eq!(names, vec!["alice"]);
    }
}