rpo 0.1.0-beta.4

//! Blame and blame-over-time reports.
//!
//! Pure transforms on the library's `blame` and `blame_over_time`
//! frames. No file-selection filtering — those frames don't carry
//! `is_generated`/`is_vendored` columns.

use polars::prelude::*;

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

/// Per-contributor blame report from a single (HEAD) blame frame.
///
/// Columns: `canonical_<aggregate>_<identify>`, `lines`, `files`.
pub fn blame_report(
    blame: &DataFrame,
    agg: Aggregation,
    activity: ActivityOptions,
) -> Result<DataFrame, RpoError> {
    let group = agg.group_col();
    let df = filter_bots(blame.clone().lazy(), agg, activity)
        .group_by([col(&group)])
        .agg([
            col("line_count").sum().alias("lines"),
            col("path").n_unique().alias("files"),
        ])
        .collect()?;
    Ok(df)
}

/// Per-(snapshot, contributor) blame line counts from a `blame_over_time`
/// frame.
///
/// Columns: `snapshot_time`, `canonical_<aggregate>_<identify>`,
/// `lines`. Sorted by `snapshot_time` asc, then group asc. Suitable for
/// feeding directly into a stacked-area chart with one series per group.
pub fn blame_timeline_report(
    blame_over_time: &DataFrame,
    agg: Aggregation,
    activity: ActivityOptions,
) -> Result<DataFrame, RpoError> {
    let group = agg.group_col();
    let df = filter_bots(blame_over_time.clone().lazy(), agg, activity)
        .group_by([col("snapshot_time"), col(&group)])
        .agg([col("line_count").sum().alias("lines")])
        .sort(["snapshot_time", &group], SortMultipleOptions::default())
        .collect()?;
    Ok(df)
}

/// Like [`blame_timeline_report`], but also carries a `snapshot` column
/// that shows the tag label when present, falling back to a short SHA.
///
/// Columns: `snapshot_time`, `snapshot`, `canonical_<aggregate>_<identify>`, `lines`.
/// Used for the `Tags` strategy so the rendered table identifies each
/// snapshot by its tag.
pub fn blame_timeline_report_with_label(
    blame_over_time: &DataFrame,
    agg: Aggregation,
    activity: ActivityOptions,
) -> Result<DataFrame, RpoError> {
    let group = agg.group_col();
    let grouped = filter_bots(blame_over_time.clone().lazy(), agg, activity)
        .group_by([
            col("snapshot_time"),
            col("snapshot_sha"),
            col("snapshot_label"),
            col(&group),
        ])
        .agg([col("line_count").sum().alias("lines")])
        .sort(["snapshot_time", &group], SortMultipleOptions::default())
        .collect()?;

    // Build the `snapshot` column eagerly: prefer snapshot_label, else short SHA.
    let label_series = grouped.column("snapshot_label")?.str()?;
    let sha_series = grouped.column("snapshot_sha")?.str()?;
    let snapshot: StringChunked = label_series
        .iter()
        .zip(sha_series.iter())
        .map(|(label, sha)| match (label, sha) {
            (Some(l), _) if !l.is_empty() => Some(l.to_string()),
            (_, Some(s)) => Some(short_sha(s).to_string()),
            _ => None,
        })
        .collect();
    let snapshot = Column::new("snapshot".into(), snapshot.into_series());

    let mut out = grouped;
    out.with_column(snapshot)?;
    let out = out.select(["snapshot_time", "snapshot", &group, "lines"])?;
    Ok(out)
}

fn short_sha(sha: &str) -> &str {
    let end = sha.len().min(7);
    &sha[..end]
}

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

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

    /// Build a `blame` frame matching the library's schema for the columns
    /// the report pipeline reads (`path`, `line_count`, `canonical_author_name`).
    fn fixture_blame() -> DataFrame {
        df! {
            "path" => ["a.rs", "a.rs", "b.rs", "c.rs", "c.rs"],
            "line_count" => [10u32, 20, 5, 8, 3],
            "canonical_author_name" => ["alice", "bob", "alice", "bob", "alice"],
        }
        .unwrap()
    }

    #[test]
    fn blame_report_groups_by_canonical_author_name() {
        let blame = fixture_blame();
        let df = blame_report(&blame, agg_author_name(), ActivityOptions::default())
            .expect("blame_report failed");

        assert_eq!(df.height(), 2);

        for expected in ["canonical_author_name", "lines", "files"] {
            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 lines = df.column("lines").unwrap().u32().unwrap();
        assert_eq!(lines.get(alice_idx).unwrap(), 18);

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

    /// Blame-over-time fixture mirroring the library's schema for the
    /// columns the timeline report reads: `snapshot_time` (datetime ms UTC),
    /// `line_count` (u32), and the canonical identity columns.
    fn fixture_blame_over_time() -> DataFrame {
        df! {
            "snapshot_time" => [
                1_700_000_000_000i64, 1_700_000_000_000,
                1_700_000_100_000,   1_700_000_100_000,
                1_700_000_200_000,
            ],
            "path" => ["a.rs", "b.rs", "a.rs", "b.rs", "a.rs"],
            "line_count" => [10u32, 5, 12, 7, 20],
            "canonical_author_name" => ["alice", "bob", "alice", "bob", "alice"],
        }
        .unwrap()
        .lazy()
        .with_column(col("snapshot_time").cast(DataType::Datetime(
            TimeUnit::Milliseconds,
            Some(TimeZone::UTC),
        )))
        .collect()
        .unwrap()
    }

    #[test]
    fn blame_timeline_report_groups_by_snapshot_and_author() {
        let frame = fixture_blame_over_time();
        let df = blame_timeline_report(&frame, agg_author_name(), ActivityOptions::default())
            .expect("blame_timeline_report failed");

        assert_eq!(df.height(), 5);

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

        let lines = df.column("lines").unwrap();
        assert!(
            matches!(
                lines.dtype(),
                DataType::UInt32 | DataType::UInt64 | DataType::Int64
            ),
            "unexpected lines dtype: {:?}",
            lines.dtype()
        );

        let name_col = df.column("canonical_author_name").unwrap().str().unwrap();
        let last_alice_idx = (0..df.height())
            .rev()
            .find(|&i| name_col.get(i).unwrap() == "alice")
            .unwrap();
        let lines_u64 = df.column("lines").unwrap().cast(&DataType::UInt64).unwrap();
        let lines_u64 = lines_u64.u64().unwrap();
        assert_eq!(lines_u64.get(last_alice_idx).unwrap(), 20);
    }

    fn fixture_blame_over_time_with_labels() -> DataFrame {
        df! {
            "snapshot_time" => [
                1_700_000_000_000i64, 1_700_000_000_000,
                1_700_000_100_000,    1_700_000_100_000,
            ],
            "snapshot_sha" => [
                "aaaaaaa1111111111111111111111111111111aa",
                "aaaaaaa1111111111111111111111111111111aa",
                "bbbbbbbcccccccccccccccccccccccccccccccbb",
                "bbbbbbbcccccccccccccccccccccccccccccccbb",
            ],
            "snapshot_label" => [Some("v1.0"), Some("v1.0"), None, None],
            "path" => ["a.rs", "b.rs", "a.rs", "b.rs"],
            "line_count" => [10u32, 5, 12, 7],
            "canonical_author_name" => ["alice", "bob", "alice", "bob"],
        }
        .unwrap()
        .lazy()
        .with_column(col("snapshot_time").cast(DataType::Datetime(
            TimeUnit::Milliseconds,
            Some(TimeZone::UTC),
        )))
        .collect()
        .unwrap()
    }

    #[test]
    fn blame_timeline_report_with_label_preserves_tag_and_falls_back_to_short_sha() {
        let frame = fixture_blame_over_time_with_labels();
        let df =
            blame_timeline_report_with_label(&frame, agg_author_name(), ActivityOptions::default())
                .expect("blame_timeline_report_with_label failed");

        assert_eq!(df.height(), 4);

        let cols = df.get_column_names();
        let names: Vec<&str> = cols.iter().map(|s| s.as_str()).collect();
        assert_eq!(
            names,
            vec![
                "snapshot_time",
                "snapshot",
                "canonical_author_name",
                "lines"
            ]
        );

        let snapshot_col = df.column("snapshot").unwrap().str().unwrap();
        let snapshots: Vec<&str> = snapshot_col.iter().map(|v| v.unwrap()).collect();
        assert!(snapshots.contains(&"v1.0"));
        assert!(
            snapshots.contains(&"bbbbbbb"),
            "expected short-SHA fallback 'bbbbbbb' in {snapshots:?}"
        );
    }

    #[test]
    fn blame_report_drops_bot_rows_when_ignore_bots_is_set() {
        let blame = df! {
            "path" => ["a.rs", "a.rs", "b.rs"],
            "line_count" => [10u32, 20, 5],
            "canonical_author_name" => ["alice", "dependabot[bot]", "bob"],
        }
        .unwrap();

        let activity_off = blame_report(&blame, agg_author_name(), ActivityOptions::default())
            .expect("blame_report (default)");
        assert_eq!(activity_off.height(), 3);

        let activity_on = blame_report(
            &blame,
            agg_author_name(),
            ActivityOptions {
                ignore_bots: true,
                ..ActivityOptions::default()
            },
        )
        .expect("blame_report (ignore_bots)");
        // dependabot dropped → only alice and bob
        assert_eq!(activity_on.height(), 2);
        let names: Vec<&str> = activity_on
            .column("canonical_author_name")
            .unwrap()
            .str()
            .unwrap()
            .iter()
            .map(|v| v.unwrap())
            .collect();
        assert!(names.contains(&"alice"));
        assert!(names.contains(&"bob"));
        assert!(!names.contains(&"dependabot[bot]"));
    }
}