rpo 0.1.0-beta.4

//! Bus-factor primitives.
//!
//! Bus factor is an estimate of how concentrated contributor knowledge is
//! for a scope (a file, a directory, or the whole repository). The number
//! answers: "how many contributors must disappear before the project is in
//! trouble?" Smaller = more fragile.
//!
//! This module contains *primitives only*: pure polars transforms on the
//! library's canonical frames. The reducers that turn these primitives
//! into a "bus factor" number (Threshold, ABF, JBF) live in the binary
//! crate (`src/bus_factor.rs`). That split keeps this module small, pure,
//! and reusable by downstream consumers (including hypothetical Python
//! bindings) who may want to apply different reducers.
//!
//! # Methods
//!
//! Three methods share this module's primitives:
//!
//! ## Threshold
//!
//! Sort authors by their share of blame lines in the scope (descending).
//! The bus factor is the smallest `k` such that the top-`k` authors together
//! hold `>= threshold` of the lines. Default `threshold = 0.5` (50%).
//! Uses the output of [`file_ownership`].
//!
//! ## ABF (Authorship-Based Factor)
//!
//! From Ricca & Avelino (2015). An author is a "primary owner" of a file
//! if their share exceeds `ownership_threshold` (default 50%). The
//! algorithm iteratively removes the author with the highest Degree of
//! Authorship — the fraction of in-scope files they primarily own — until
//! fewer than 50% of files have at least one remaining primary owner.
//! The bus factor is the number of authors removed. Uses the output of
//! [`file_ownership`].
//!
//! ## JBF (Jabrayilzade Bus Factor) — time-weighted
//!
//! From Jabrayilzade et al. Same reducer as Threshold, but the
//! per-author share is time-decayed using an exponential half-life: a
//! touch at `now - half_life_months` counts for half as much as a touch
//! today. v1 decays per commit (see [`weighted_touches`]); a future
//! extension will decay per blame snapshot for a richer "who knows it
//! now" view.
//!
//! # Frame contract
//!
//! | Function           | Input frame(s)    | Reads columns                                    |
//! | ------------------ | ----------------- | ------------------------------------------------ |
//! | [`file_ownership`] | `blame`           | `path`, `line_count`, `canonical_<agg>_<id>`     |
//! | [`weighted_touches`] | `file_changes`  | `path`, `commit_time`, `canonical_<agg>_<id>`    |
//! | [`prefix_paths`]   | any               | the caller-chosen `path` column                  |
//!
//! All three return frames suitable for the reducers in the binary crate.

use polars::prelude::*;

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

/// Explode each row into `(row × ancestor_dir)` combinations, replacing the
/// `path` column with a new `dir` column. Ancestors include the empty-string
/// root and every intermediate directory; the leaf filename is excluded.
///
/// # Example
///
/// A single input row with `path = "rpo/src/frames/blame.rs"` produces four
/// output rows with `dir` values `""`, `"rpo"`, `"rpo/src"`,
/// `"rpo/src/frames"`.
///
/// All other columns are preserved unchanged on every emitted row.
///
/// # Errors
///
/// Returns `RpoError::Polars` if `path_col` is missing or non-string, or if
/// the intermediate polars operations fail.
pub fn prefix_paths(df: &DataFrame, path_col: &str) -> Result<DataFrame, RpoError> {
    // Build a Vec<Series> of ancestor lists (one per row), then construct
    // a List-dtype Series, attach it as a new column, explode, and drop
    // the original path column.
    let path_series = df.column(path_col)?.str()?;

    let ancestor_series_list: Vec<Series> = path_series
        .iter()
        .map(|opt| {
            let p = opt.unwrap_or("");
            let ancestors = ancestors_of(p);
            Series::new("dir".into(), ancestors)
        })
        .collect();

    let list_col = Column::new("dir".into(), ancestor_series_list);

    let mut out = df.clone();
    out.with_column(list_col)?;

    let out = out
        .lazy()
        .explode(
            cols(["dir"]),
            ExplodeOptions {
                empty_as_null: false,
                keep_nulls: false,
            },
        )
        .drop(cols([path_col]))
        .collect()?;

    Ok(out)
}

/// Compute every ancestor directory of `path`, including the empty-string
/// root. The leaf is excluded (since we're after *directories*, not files).
///
/// `"a/b/c.rs"` → `["", "a", "a/b"]`.
/// `"README.md"` → `[""]`.
fn ancestors_of(path: &str) -> Vec<String> {
    let mut parts: Vec<&str> = path.split('/').collect();
    parts.pop(); // drop the leaf
    let mut out = Vec::with_capacity(parts.len() + 1);
    out.push(String::new()); // repo root
    let mut acc = String::new();
    for p in parts {
        if !acc.is_empty() {
            acc.push('/');
        }
        acc.push_str(p);
        out.push(acc.clone());
    }
    out
}

/// Per-(path, author) blame-line share.
///
/// Input: a blame frame with columns `path`, `line_count`, and the canonical
/// identity column named by `(agg, id)`.
///
/// Output columns:
/// - `path`
/// - `<group>` — the canonical identity column chosen by `(agg, id)`.
/// - `lines` — sum of `line_count` for this author on this file.
/// - `file_lines` — total lines of the file.
/// - `share` — `lines / file_lines` (0..=1, f64).
/// - `is_primary_owner` — `share > ownership_threshold` (bool).
///
/// # Parameters
///
/// * `ownership_threshold` — primary-owner cutoff as a fraction in `0.0..1.0`.
///   A share *strictly greater than* this value marks the author as a primary
///   owner. For the ABF reducer's default of 50%, pass `0.5`.
///
/// # Errors
///
/// Returns `RpoError::Polars` if required columns are missing.
pub fn file_ownership(
    blame: &DataFrame,
    agg: Aggregation,
    activity: ActivityOptions,
    ownership_threshold: f64,
) -> Result<DataFrame, RpoError> {
    let group = agg.group_col();

    let per_pair = filter_bots(blame.clone().lazy(), agg, activity)
        .group_by([col("path"), col(&group)])
        .agg([col("line_count")
            .cast(DataType::UInt64)
            .sum()
            .alias("lines")])
        .collect()?;

    let file_totals = per_pair
        .clone()
        .lazy()
        .group_by([col("path")])
        .agg([col("lines").sum().alias("file_lines")])
        .collect()?;

    let joined = per_pair
        .lazy()
        .join(
            file_totals.lazy(),
            [col("path")],
            [col("path")],
            JoinArgs::new(JoinType::Left),
        )
        .with_columns([(col("lines").cast(DataType::Float64)
            / col("file_lines").cast(DataType::Float64))
        .alias("share")])
        .with_columns([col("share")
            .gt(lit(ownership_threshold))
            .alias("is_primary_owner")])
        .collect()?;

    Ok(joined)
}

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

    /// Blame fixture:
    ///   a.rs: alice 10 lines, bob 20 lines  -> alice 33%, bob 67%
    ///   b.rs: alice 5 lines                 -> alice 100%
    ///   c.rs: alice 3, bob 3, carol 4       -> alice 30%, bob 30%, carol 40%
    fn fixture() -> DataFrame {
        df! {
            "path" => ["a.rs", "a.rs", "b.rs", "c.rs", "c.rs", "c.rs"],
            "line_count" => [10u32, 20, 5, 3, 3, 4],
            "canonical_author_name" => ["alice", "bob", "alice", "alice", "bob", "carol"],
        }
        .unwrap()
    }

    #[test]
    fn shares_sum_to_one_per_file() {
        let df = file_ownership(
            &fixture(),
            crate::options::Aggregation {
                aggregate: crate::options::Aggregate::Author,
                identify: crate::options::Identify::Name,
            },
            ActivityOptions::default(),
            0.5,
        )
        .expect("file_ownership");

        let path_col = df.column("path").unwrap().str().unwrap();
        let share_col = df.column("share").unwrap().f64().unwrap();

        let mut sums: std::collections::HashMap<String, f64> = Default::default();
        for i in 0..df.height() {
            let p = path_col.get(i).unwrap().to_string();
            let s = share_col.get(i).unwrap();
            *sums.entry(p).or_insert(0.0) += s;
        }
        for (p, total) in &sums {
            assert!(
                (total - 1.0).abs() < 1e-9,
                "shares for {p} should sum to 1.0; got {total}"
            );
        }
    }

    #[test]
    fn primary_owner_flag_reflects_threshold() {
        let df = file_ownership(
            &fixture(),
            crate::options::Aggregation {
                aggregate: crate::options::Aggregate::Author,
                identify: crate::options::Identify::Name,
            },
            ActivityOptions::default(),
            0.5,
        )
        .expect("file_ownership");

        let path_col = df.column("path").unwrap().str().unwrap();
        let name_col = df.column("canonical_author_name").unwrap().str().unwrap();
        let owner_col = df.column("is_primary_owner").unwrap().bool().unwrap();

        // At threshold=0.5:
        //   a.rs: bob (67%) is primary, alice (33%) is not.
        //   b.rs: alice (100%) is primary.
        //   c.rs: no-one >50% -> no primary owner.
        for i in 0..df.height() {
            let p = path_col.get(i).unwrap();
            let n = name_col.get(i).unwrap();
            let is_owner = owner_col.get(i).unwrap();
            let expected = matches!((p, n), ("a.rs", "bob") | ("b.rs", "alice"));
            assert_eq!(
                is_owner, expected,
                "primary-owner mismatch for ({p}, {n}): got {is_owner}, expected {expected}"
            );
        }
    }

    #[test]
    fn custom_threshold_shifts_primary_owner_boundary() {
        // At threshold=0.25:
        //   a.rs: both alice and bob primary (both > 25%).
        //   b.rs: alice primary.
        //   c.rs: all three primary (all > 25%).
        let low = file_ownership(
            &fixture(),
            crate::options::Aggregation {
                aggregate: crate::options::Aggregate::Author,
                identify: crate::options::Identify::Name,
            },
            ActivityOptions::default(),
            0.25,
        )
        .expect("file_ownership");
        let owner_col = low.column("is_primary_owner").unwrap().bool().unwrap();
        let primary_count = (0..low.height())
            .filter(|&i| owner_col.get(i).unwrap())
            .count();
        // 2 + 1 + 3 = 6 primary owners at threshold=0.25.
        assert_eq!(primary_count, 6);
    }

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

        let off = file_ownership(
            &blame,
            crate::options::Aggregation::default(),
            ActivityOptions::default(),
            0.5,
        )
        .expect("file_ownership (default)");
        // alice 33%, dependabot 67% → 2 rows
        assert_eq!(off.height(), 2);

        let on = file_ownership(
            &blame,
            crate::options::Aggregation::default(),
            ActivityOptions {
                ignore_bots: true,
                ..ActivityOptions::default()
            },
            0.5,
        )
        .expect("file_ownership (ignore_bots)");
        // dependabot dropped → alice 100% → 1 row
        assert_eq!(on.height(), 1);
        let share = on.column("share").unwrap().f64().unwrap().get(0).unwrap();
        assert!(
            (share - 1.0).abs() < 1e-9,
            "alice should own 100% after bot filter; got {share}"
        );
    }
}

/// Average month in milliseconds (30.44 days × 86400 s × 1000 ms).
/// Used for age-in-months computation in the decay formula.
const AVG_MONTH_MS: f64 = 30.44 * 86_400.0 * 1000.0;

/// Per-(path, author) time-decayed touch weight and normalized share.
///
/// Each row in `file_changes` represents one touch of `path` by the
/// canonical author at the commit's `commit_time`. Each touch contributes:
///
/// ```text
/// weight = exp(-ln(2) * max(0, (now - commit_time_ms)) / (half_life_months * AVG_MONTH_MS))
/// ```
///
/// Touches at `now` get weight 1.0; touches at `now - half_life_months` get
/// weight 0.5; future-dated touches (clock skew) clamp to weight 1.0.
///
/// Output columns:
/// - `path`
/// - `<group>` — canonical identity column chosen by `(agg, id)`.
/// - `weight` — sum of per-touch weights for this author on this file (f64).
/// - `share` — `weight / sum(weight) over path` (f64, 0..=1).
///
/// # Errors
///
/// Returns `RpoError::Polars` if required columns (`path`, `commit_time`,
/// the canonical identity column) are missing.
pub fn weighted_touches(
    file_changes: &DataFrame,
    agg: Aggregation,
    activity: ActivityOptions,
    half_life_months: u32,
    now_ms: i64,
) -> Result<DataFrame, RpoError> {
    let group = agg.group_col();
    let half_life_ms = half_life_months as f64 * AVG_MONTH_MS;

    let per_touch = filter_bots(file_changes.clone().lazy(), agg, activity)
        .with_columns([col("commit_time")
            .cast(DataType::Int64)
            .alias("commit_time_ms")])
        .with_columns([when((lit(now_ms) - col("commit_time_ms")).lt(lit(0i64)))
            .then(lit(0i64))
            .otherwise(lit(now_ms) - col("commit_time_ms"))
            .cast(DataType::Float64)
            .alias("age_ms")])
        .with_columns([
            // weight = 2^(-age / half_life_ms)  ≡  exp(-ln2 * age / half_life_ms)
            lit(2.0f64)
                .pow(col("age_ms") * lit(-1.0f64 / half_life_ms))
                .alias("weight"),
        ])
        .group_by([col("path"), col(&group)])
        .agg([col("weight").sum().alias("weight")])
        .collect()?;

    let path_totals = per_touch
        .clone()
        .lazy()
        .group_by([col("path")])
        .agg([col("weight").sum().alias("path_weight")])
        .collect()?;

    let joined = per_touch
        .lazy()
        .join(
            path_totals.lazy(),
            [col("path")],
            [col("path")],
            JoinArgs::new(JoinType::Left),
        )
        .with_columns([(col("weight") / col("path_weight")).alias("share")])
        .drop(cols(["path_weight"]))
        .collect()?;

    Ok(joined)
}

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

    fn fixture(now_ms: i64, half_life_months: i64) -> DataFrame {
        let month_ms: i64 = (30.44 * 86400.0 * 1000.0) as i64;
        let old_ms = now_ms - half_life_months * month_ms;
        let ancient_ms = now_ms - 2 * half_life_months * month_ms;

        df! {
            "sha" => ["c1", "c2", "c3"],
            "commit_time" => [now_ms, old_ms, ancient_ms],
            "canonical_author_name" => ["alice", "alice", "bob"],
            "canonical_author_email" => ["a@x", "a@x", "b@x"],
            "canonical_committer_name" => ["alice", "alice", "bob"],
            "canonical_committer_email" => ["a@x", "a@x", "b@x"],
            "path" => ["a.rs", "a.rs", "a.rs"],
            "insertions" => [10u64, 5, 3],
            "deletions" => [0u64, 0, 0],
            "is_generated" => [false, false, false],
            "is_vendored" => [false, false, false],
        }
        .unwrap()
        .lazy()
        .with_column(col("commit_time").cast(DataType::Datetime(
            TimeUnit::Milliseconds,
            Some(TimeZone::UTC),
        )))
        .collect()
        .unwrap()
    }

    #[test]
    fn weights_match_half_life_decay() {
        let now_ms: i64 = 1_800_000_000_000;
        let half_life: i64 = 12;
        let df = weighted_touches(
            &fixture(now_ms, half_life),
            crate::options::Aggregation::default(),
            ActivityOptions::default(),
            half_life as u32,
            now_ms,
        )
        .expect("weighted_touches");

        let name_col = df.column("canonical_author_name").unwrap().str().unwrap();
        let weight_col = df.column("weight").unwrap().f64().unwrap();

        let mut weights = std::collections::HashMap::<String, f64>::new();
        for i in 0..df.height() {
            weights.insert(
                name_col.get(i).unwrap().to_string(),
                weight_col.get(i).unwrap(),
            );
        }

        // alice has two touches: 1.0 + 0.5 = 1.5
        assert!(
            (weights["alice"] - 1.5).abs() < 1e-6,
            "alice weight: {}",
            weights["alice"]
        );
        // bob has one ancient touch: 0.25
        assert!(
            (weights["bob"] - 0.25).abs() < 1e-6,
            "bob weight: {}",
            weights["bob"]
        );
    }

    #[test]
    fn shares_normalize_to_one_per_file() {
        let now_ms: i64 = 1_800_000_000_000;
        let df = weighted_touches(
            &fixture(now_ms, 12),
            crate::options::Aggregation::default(),
            ActivityOptions::default(),
            12,
            now_ms,
        )
        .expect("weighted_touches");

        let path_col = df.column("path").unwrap().str().unwrap();
        let share_col = df.column("share").unwrap().f64().unwrap();

        let mut sums = std::collections::HashMap::<String, f64>::new();
        for i in 0..df.height() {
            *sums
                .entry(path_col.get(i).unwrap().to_string())
                .or_insert(0.0) += share_col.get(i).unwrap();
        }
        for (p, s) in &sums {
            assert!(
                (s - 1.0).abs() < 1e-9,
                "shares for {p} should sum to 1.0; got {s}"
            );
        }
    }

    #[test]
    fn future_commit_clamped_to_zero_age() {
        let now_ms: i64 = 1_800_000_000_000;
        let future_ms = now_ms + 86_400_000;
        let df = df! {
            "sha" => ["c1"],
            "commit_time" => [future_ms],
            "canonical_author_name" => ["alice"],
            "canonical_author_email" => ["a@x"],
            "canonical_committer_name" => ["alice"],
            "canonical_committer_email" => ["a@x"],
            "path" => ["a.rs"],
            "insertions" => [10u64],
            "deletions" => [0u64],
            "is_generated" => [false],
            "is_vendored" => [false],
        }
        .unwrap()
        .lazy()
        .with_column(col("commit_time").cast(DataType::Datetime(
            TimeUnit::Milliseconds,
            Some(TimeZone::UTC),
        )))
        .collect()
        .unwrap();

        let out = weighted_touches(
            &df,
            crate::options::Aggregation::default(),
            ActivityOptions::default(),
            12,
            now_ms,
        )
        .expect("weighted_touches");
        let w = out.column("weight").unwrap().f64().unwrap().get(0).unwrap();
        assert!(
            (w - 1.0).abs() < 1e-9,
            "future commit should have weight 1.0; got {w}"
        );
    }
}

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

    fn fixture() -> DataFrame {
        // Three paths: one deep, one shallow, one at repo root.
        df! {
            "path" => ["rpo/src/frames/blame.rs", "src/main.rs", "README.md"],
            "v" => [1u32, 2, 3],
        }
        .unwrap()
    }

    #[test]
    fn explodes_into_every_ancestor_including_root() {
        let df = prefix_paths(&fixture(), "path").expect("prefix_paths");

        // Expected ancestors (excluding the leaf):
        //   rpo/src/frames/blame.rs -> "", "rpo", "rpo/src", "rpo/src/frames"
        //   src/main.rs             -> "", "src"
        //   README.md               -> ""
        // Total rows: 4 + 2 + 1 = 7.
        assert_eq!(df.height(), 7);

        let dirs: Vec<&str> = df
            .column("dir")
            .unwrap()
            .str()
            .unwrap()
            .iter()
            .map(|o| o.unwrap())
            .collect();

        let mut counts = std::collections::HashMap::<&str, usize>::new();
        for d in &dirs {
            *counts.entry(d).or_insert(0) += 1;
        }
        // Root appears once per original row.
        assert_eq!(counts[""], 3);
        // Each intermediate directory appears exactly once.
        assert_eq!(counts["rpo"], 1);
        assert_eq!(counts["rpo/src"], 1);
        assert_eq!(counts["rpo/src/frames"], 1);
        assert_eq!(counts["src"], 1);

        // Leaf filename must NOT appear.
        assert!(!dirs.contains(&"rpo/src/frames/blame.rs"));
        assert!(!dirs.contains(&"README.md"));
        assert!(!dirs.contains(&"src/main.rs"));

        // `v` must be preserved per original row.
        let v: Vec<u32> = df
            .column("v")
            .unwrap()
            .u32()
            .unwrap()
            .iter()
            .map(|o| o.unwrap())
            .collect();
        assert_eq!(v.len(), 7);
    }
}