server/lib/src/audit/aggregate.rs
//! **Count the audit log** — records and bytes per repository, action, day,
//! actor, client or status (lane HOL-AUDIT, 2026-10-30).
//!
//! An aggregate is a fold of the Arrow index ([`super::arrow_ipc`]) under an
//! [`AuditFilter`]: one row per distinct value of the chosen keys, with the record
//! count, the bytes summed and the first and last moment. Keys left out of
//! `group_by` are empty in every row. The day is the UTC calendar day
//! (`YYYY-MM-DD`); the client is the recorded address without its port
//! ([`super::client_host`]): an address is not an identity (NAT puts a whole office
//! behind one), so it says the shape of who pulls, never who is allowed.
//!
//! The flow picture's fetches, bytes sent and pullers are aggregates
//! (`crate::flow::AuditTally::from_table`).
use std::collections::BTreeMap;
use serde::Serialize;
use super::arrow_ipc::AuditTable;
use super::query::AuditFilter;
use super::{AuditError, QuerySource};
/// What an aggregate groups on.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub enum GroupKey {
Repo,
Action,
Day,
Ident,
Client,
Status,
}
impl GroupKey {
/// From the wire word (`repo`, `action`, `day`, `ident`, `client`, `status`).
pub fn from_wire(s: &str) -> Result<Self, AuditError> {
Ok(match s.trim() {
"repo" | "repository" => GroupKey::Repo,
"action" => GroupKey::Action,
"day" => GroupKey::Day,
"ident" | "actor" => GroupKey::Ident,
"client" | "source_ip" => GroupKey::Client,
"status" => GroupKey::Status,
other => {
return Err(AuditError::Other(format!(
"`{other}` is no aggregate key (repo, action, day, ident, client, status)"
)))
}
})
}
/// The wire word.
pub fn as_str(&self) -> &'static str {
match self {
GroupKey::Repo => "repo",
GroupKey::Action => "action",
GroupKey::Day => "day",
GroupKey::Ident => "ident",
GroupKey::Client => "client",
GroupKey::Status => "status",
}
}
}
/// The keys when the caller names none: per repository, action and day.
pub const DEFAULT_GROUP_BY: [GroupKey; 3] = [GroupKey::Repo, GroupKey::Action, GroupKey::Day];
/// The most groups one answer carries.
pub const MAX_GROUPS: usize = 10_000;
/// One aggregate.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct AggregateQuery {
/// The keys; empty = [`DEFAULT_GROUP_BY`].
pub group_by: Vec<GroupKey>,
pub filter: AuditFilter,
/// At most this many groups (the first in key order); 0 = [`MAX_GROUPS`].
pub max_groups: usize,
}
/// One group.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
pub struct AggregateRow {
pub repo: String,
pub action: String,
/// `YYYY-MM-DD`, UTC.
pub day: String,
pub ident: String,
pub client: String,
/// 0 unless grouped by status.
pub status: u16,
pub count: u64,
pub bytes: u64,
pub first_ts_nanos: i64,
pub last_ts_nanos: i64,
}
/// The answer, in key order.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
pub struct Aggregates {
pub group_by: Vec<&'static str>,
pub rows: Vec<AggregateRow>,
/// More groups matched than [`AggregateQuery::max_groups`]; the rows kept are
/// the first in key order.
pub groups_truncated: bool,
/// Records counted (the filter's matches).
pub rows_counted: u64,
/// Records read.
pub scanned: u64,
}
/// The UTC calendar day of `ts_nanos` (Hinnant's civil-from-days).
pub fn day_of(ts_nanos: i64) -> String {
let days = ts_nanos.div_euclid(86_400 * 1_000_000_000);
let z = days + 719_468;
let era = z.div_euclid(146_097);
let doe = z - era * 146_097;
let yoe = (doe - doe / 1_460 + doe / 36_524 - doe / 146_096) / 365;
let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
let mp = (5 * doy + 2) / 153;
let d = doy - (153 * mp + 2) / 5 + 1;
let m = if mp < 10 { mp + 3 } else { mp - 9 };
let y = yoe + era * 400 + i64::from(m <= 2);
format!("{y:04}-{m:02}-{d:02}")
}
type Key = (String, String, String, String, String, u16);
/// **Aggregate `source`.**
pub fn aggregate(source: &QuerySource, q: &AggregateQuery) -> Result<Aggregates, AuditError> {
aggregate_table(&source.table()?, q)
}
/// **Aggregate a table already read** (several aggregates over one read).
pub fn aggregate_table(table: &AuditTable, q: &AggregateQuery) -> Result<Aggregates, AuditError> {
q.filter.check()?;
let keys: Vec<GroupKey> = if q.group_by.is_empty() {
DEFAULT_GROUP_BY.to_vec()
} else {
q.group_by.clone()
};
let has = |k: GroupKey| keys.contains(&k);
let (by_repo, by_action, by_day, by_ident, by_client, by_status) = (
has(GroupKey::Repo),
has(GroupKey::Action),
has(GroupKey::Day),
has(GroupKey::Ident),
has(GroupKey::Client),
has(GroupKey::Status),
);
let mut groups: BTreeMap<Key, AggregateRow> = BTreeMap::new();
let (mut scanned, mut counted) = (0u64, 0u64);
table.for_each_row(|r| {
scanned += 1;
if !q.filter.matches(&r) {
return;
}
counted += 1;
let key: Key = (
if by_repo {
r.repo.to_string()
} else {
String::new()
},
if by_action {
r.action.to_string()
} else {
String::new()
},
if by_day {
day_of(r.ts_nanos)
} else {
String::new()
},
if by_ident {
r.ident.to_string()
} else {
String::new()
},
if by_client {
super::client_host(r.source_ip)
} else {
String::new()
},
if by_status { r.status } else { 0 },
);
let g = groups.entry(key).or_insert_with_key(|k| AggregateRow {
repo: k.0.clone(),
action: k.1.clone(),
day: k.2.clone(),
ident: k.3.clone(),
client: k.4.clone(),
status: k.5,
first_ts_nanos: i64::MAX,
last_ts_nanos: i64::MIN,
..Default::default()
});
g.count += 1;
g.bytes = g.bytes.saturating_add(r.bytes);
g.first_ts_nanos = g.first_ts_nanos.min(r.ts_nanos);
g.last_ts_nanos = g.last_ts_nanos.max(r.ts_nanos);
})?;
let cap = match q.max_groups {
0 => MAX_GROUPS,
n => n.min(MAX_GROUPS),
};
let truncated = groups.len() > cap;
Ok(Aggregates {
group_by: keys.iter().map(GroupKey::as_str).collect(),
rows: groups.into_values().take(cap).collect(),
groups_truncated: truncated,
rows_counted: counted,
scanned,
})
}
#[cfg(test)]
mod tests {
use super::super::{AuditAction, AuditEvent, AuditLog, FileAuditLog, MemoryAuditLog};
use super::*;
const DAY: i64 = 86_400 * 1_000_000_000;
/// 2026-10-30T00:00:00Z.
const D30: i64 = 1_793_318_400 * 1_000_000_000;
fn at(ev: AuditEvent, ts: i64) -> AuditEvent {
AuditEvent { ts_nanos: ts, ..ev }
}
fn sample(log: &dyn AuditLog) {
let rows = [
(
"c",
AuditAction::Download,
"crates",
"a-1",
"10.0.0.9:1",
200,
10,
D30 + 5,
),
(
"c",
AuditAction::Download,
"crates",
"a-1",
"10.0.0.9:2",
200,
10,
D30 + 6,
),
(
"d",
AuditAction::Download,
"crates",
"b-1",
"10.0.0.7:1",
404,
0,
D30 + 7,
),
(
"d",
AuditAction::Upload,
"crates",
"b-1",
"10.0.0.7:1",
200,
99,
D30 + DAY + 1,
),
(
"c",
AuditAction::Download,
"pypi",
"w.whl",
"10.0.0.9:3",
200,
5,
D30 + DAY + 2,
),
];
for (who, a, repo, art, ip, st, b, ts) in rows {
log.record(at(AuditEvent::new(who, a, repo, art, ip, st, b), ts))
.unwrap();
}
}
#[test]
fn the_day_is_the_utc_calendar_day() {
assert_eq!(day_of(D30), "2026-10-30");
assert_eq!(day_of(D30 + DAY - 1), "2026-10-30");
assert_eq!(day_of(D30 + DAY), "2026-10-31");
assert_eq!(day_of(0), "1970-01-01");
assert_eq!(day_of(951_782_400 * 1_000_000_000), "2000-02-29");
}
/// ★ Counts per repository, action and day; per client; under a filter.
#[test]
fn counts_per_repo_action_and_day_and_per_client() {
let log = MemoryAuditLog::new();
sample(&log);
let src = log.query_source().unwrap();
let a = aggregate(&src, &AggregateQuery::default()).unwrap();
let got: Vec<(&str, &str, &str, u64, u64)> = a
.rows
.iter()
.map(|r| {
(
r.repo.as_str(),
r.action.as_str(),
r.day.as_str(),
r.count,
r.bytes,
)
})
.collect();
assert_eq!(
got,
[
("crates", "download", "2026-10-30", 3, 20),
("crates", "upload", "2026-10-31", 1, 99),
("pypi", "download", "2026-10-31", 1, 5),
]
);
assert_eq!(
(a.rows_counted, a.scanned, a.groups_truncated),
(5, 5, false)
);
let ok = AuditFilter {
action: Some("download".into()),
status_class: Some(2),
..Default::default()
};
let by_client = aggregate(
&src,
&AggregateQuery {
group_by: vec![GroupKey::Client],
filter: ok,
max_groups: 0,
},
)
.unwrap();
assert_eq!(
by_client.rows.len(),
1,
"the 404 is not a fetch; ports collapse"
);
assert_eq!(
(
by_client.rows[0].client.as_str(),
by_client.rows[0].count,
by_client.rows[0].bytes
),
("10.0.0.9", 3, 25)
);
assert_eq!(
(
by_client.rows[0].first_ts_nanos,
by_client.rows[0].last_ts_nanos
),
(D30 + 5, D30 + DAY + 2)
);
let capped = aggregate(
&src,
&AggregateQuery {
group_by: vec![GroupKey::Ident, GroupKey::Status],
max_groups: 1,
..Default::default()
},
)
.unwrap();
assert!(capped.groups_truncated);
assert_eq!(capped.rows.len(), 1);
assert!(GroupKey::from_wire("colour").is_err());
}
/// ★ The file sink's index answers exactly what the in-memory snapshot does.
#[test]
fn the_index_answers_like_the_snapshot() {
let dir = std::env::temp_dir().join(format!(
"holger-agg-{}-{}",
std::process::id(),
super::super::now_nanos()
));
let file = FileAuditLog::new(&dir).unwrap().with_checkpoint_every(2);
let mem = MemoryAuditLog::new();
sample(&file);
sample(&mem);
let q = AggregateQuery {
group_by: vec![GroupKey::Repo, GroupKey::Ident, GroupKey::Client],
..Default::default()
};
assert_eq!(
aggregate(&file.query_source().unwrap(), &q).unwrap(),
aggregate(&mem.query_source().unwrap(), &q).unwrap()
);
let _ = std::fs::remove_dir_all(&dir);
}
}