use std::collections::BTreeMap;
use std::path::Path;
use chrono::{Days, NaiveDate};
use serde::{Deserialize, Serialize};
use crate::datastore::{Datastore, schema};
use crate::firehose::Pushes;
/// the header's strip covers this many days, today last
pub const DAYS: u64 = 30;
/// how far the push counts read the firehose, and what they found per utc day
#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Counted {
offset: u64,
days: BTreeMap<NaiveDate, u32>,
}
impl Counted {
/// reads only the frames past the last offset
fn advance(self, output_root: &Path, today: NaiveDate) -> anyhow::Result<Self> {
let pushes = crate::firehose::pushes_after(output_root, self.offset)?;
Ok(self.add(pushes, today))
}
/// the log starting over starts the counts over
fn add(mut self, pushes: Pushes, today: NaiveDate) -> Self {
if pushes.restarted {
self.days.clear();
}
self.offset = pushes.end;
for day in pushes.days {
if let Ok(day) = day.parse::<NaiveDate>() {
*self.days.entry(day).or_default() += 1;
}
}
let first = today - Days::new(DAYS - 1);
self.days.retain(|day, _| *day >= first);
self
}
/// oldest first, empty without a firehose
fn strip(&self, today: NaiveDate) -> Vec<(NaiveDate, u32)> {
if self.offset == 0 {
return Vec::new();
}
(0..DAYS)
.rev()
.map(|back| {
let day = today - Days::new(back);
(day, self.days.get(&day).copied().unwrap_or(0))
})
.collect()
}
}
/// pushes per day for the header's strip. a firehose that can't be read leaves the strip out
pub(super) fn read(output_root: &Path, today: NaiveDate) -> Vec<(NaiveDate, u32)> {
let store = Datastore::at(output_root);
let record = store.record::<schema::IndexPulse>(&());
let before = record.read().ok().flatten().unwrap_or_default();
match before.clone().advance(output_root, today) {
Ok(counted) => {
if counted != before
&& let Err(error) = record.put(&counted)
{
tracing::warn!(%error, "keep the index page's push counts");
}
counted.strip(today)
}
Err(error) => {
tracing::warn!(%error, "count pushes for the index page");
Vec::new()
}
}
}
#[cfg(test)]
mod tests {
use super::*;
fn day(text: &str) -> NaiveDate {
text.parse().unwrap()
}
fn pushes(restarted: bool, end: u64, days: &[&str]) -> Pushes {
Pushes {
restarted,
end,
days: days.iter().map(|day| day.to_string()).collect(),
}
}
#[test]
fn counts_add_up_across_reads_and_drop_days_past_the_strip() {
let today = day("2026-11-10");
let counted = Counted::default()
.add(pushes(false, 10, &["2026-10-01", "2026-11-09"]), today)
.add(pushes(false, 25, &["2026-11-09", "2026-11-10"]), today);
assert_eq!(counted.offset, 25);
let strip = counted.strip(today);
assert_eq!(strip.len(), DAYS as usize);
assert_eq!(strip.first(), Some(&(day("2026-10-12"), 0)));
assert_eq!(
&strip[28..],
[(day("2026-11-09"), 2), (day("2026-11-10"), 1)]
);
assert!(!counted.days.contains_key(&day("2026-10-01")));
}
#[test]
fn a_log_that_starts_over_starts_the_counts_over() {
let today = day("2026-11-10");
let counted = Counted::default()
.add(pushes(false, 40, &["2026-11-08", "2026-11-08"]), today)
.add(pushes(true, 12, &["2026-11-10"]), today);
assert_eq!(counted.offset, 12);
assert_eq!(counted.days, BTreeMap::from([(day("2026-11-10"), 1)]));
}
#[test]
fn no_firehose_draws_no_strip() {
let root = tempfile::tempdir().unwrap();
assert_eq!(read(root.path(), day("2026-11-10")), Vec::new());
assert!(Counted::default().strip(day("2026-11-10")).is_empty());
}
}