monetize-product 0.1.1

//! **Reading a DAILY COUNTER out of a series of readings** — the arithmetic
//! behind the console's SILENT and FRICTION lists.
//!
//! Pure. No store, no clock, no product — [`crate::trend`]'s shape, and for the
//! same reason: the fold has to be assertable without a wire.
//!
//! # The problem this exists to solve
//!
//! gunnar's flow meters are `*_today` counters. `refused_today` is *how many
//! transfers this namespace has had refused since midnight UTC*: it starts at 0
//! every day, only ever grows within a day, and drops back at the boundary. The
//! poller reads it every fifteen minutes and `UsageStore` keeps every reading
//! for 400 days.
//!
//! So the question *how many refusals last week* has an obvious wrong answer and
//! a correct one:
//!
//! * **Wrong: sum the readings.** A tenant refused once at 09:00 has
//!   `refused_today = 1` in every one of the day's remaining sixty readings, and
//!   summing them reports sixty refusals. The busier the poller, the bigger the
//!   lie — a monitoring interval would be silently multiplying a customer's
//!   friction.
//! * **Right: the MAXIMUM reading of each UTC day is that day's total**, because
//!   the counter is monotonic within the day; then add the days. That is
//!   [`daily_totals`].
//!
//! # Why the maximum and not the last reading of the day
//!
//! The last reading of a day is the natural choice and it is wrong at exactly
//! one moment: the poll that lands a few milliseconds after midnight, whose
//! reading belongs to the NEW day but whose `measured_at` may still be
//! attributed to the old one by a clock that disagrees by a hair. That reading
//! is a small number where the day's real total is large, and taking it would
//! erase the day. The maximum cannot be wrong in that direction — a counter
//! that only grows within a day has its total at its peak — and its failure
//! mode if a reading is ever attributed to the wrong day is to keep the larger
//! of two true numbers rather than to invent one.
//!
//! # What is NOT here
//!
//! This does not fold `last_push_unix_ms`. That meter is not a daily counter —
//! it is an instant, and gunnar already looks back over its whole kept window to
//! produce it — so the answer to *when was this tenant last seen* is the latest
//! reading's value, read straight off, and folding it would be arithmetic on a
//! timestamp. [`quiet_for_ms`] is the whole of what the SILENT list needs.

use std::collections::BTreeMap;

use crate::Usage;

/// Milliseconds in a day. UTC, no leap seconds: the wire carries Unix
/// milliseconds and a Unix millisecond has never counted one.
pub const DAY_MS: u64 = 86_400_000;

/// The UTC day a reading falls in, as whole days since the epoch.
pub fn day_of(unix_ms: u64) -> u64 {
    unix_ms / DAY_MS
}

/// **Per UTC day, the total of one `*_today` counter** — the day's maximum
/// reading. See the module doc for why the maximum.
///
/// Readings that do not carry the meter are skipped, not counted as zero: a
/// product that stopped reporting a meter and a product reporting zero are
/// different facts, and only the second is a day with no refusals.
///
/// A day with no reading at all is ABSENT from the map. The caller knows which
/// days it asked about, and a zero for a day nobody measured is the one number
/// that would make an unreachable product look like a quiet week.
pub fn daily_totals(readings: &[Usage], meter: &str) -> BTreeMap<u64, u64> {
    let mut days: BTreeMap<u64, u64> = BTreeMap::new();
    for reading in readings {
        let Some(value) = reading.meters.get(meter).copied() else {
            continue;
        };
        let day = days.entry(day_of(reading.measured_at_unix_ms)).or_insert(0);
        *day = (*day).max(value);
    }
    days
}

/// **The total of one `*_today` counter over the readings whose day falls in
/// `[since_unix_ms, until_unix_ms]`.**
///
/// The window is resolved to whole UTC days at both ends, because the counter
/// is a per-day quantity and half a day of one is not a number that exists.
pub fn window_total(readings: &[Usage], meter: &str, since_unix_ms: u64, until_unix_ms: u64) -> u64 {
    let (from, to) = (day_of(since_unix_ms), day_of(until_unix_ms));
    daily_totals(readings, meter)
        .range(from..=to)
        .map(|(_, total)| *total)
        .fold(0u64, |acc, n| acc.saturating_add(n))
}

/// **How long a tenant has been quiet**, or `None` when it has not been.
///
/// `last_seen_unix_ms` is the later of the two `last_*_unix_ms` meters, which
/// gunnar computes over its WHOLE kept window rather than over today — so a
/// tenant with a today of zeroes and a push six weeks ago is six weeks quiet
/// and not silent forever.
///
/// **`0` means NEVER, not 1970.** proto3 cannot tell an unset timestamp from a
/// zero one, and gunnar spells "this namespace has never been touched inside
/// the kept window" as 0. Rendering that as an epoch date would put 1970 on a
/// console beside real dates, which reads as a very old customer rather than as
/// one nobody has ever used — so `None` for `last_seen == 0` is refused here
/// and [`Quiet::Never`] says it out loud.
pub fn quiet_for(last_seen_unix_ms: Option<u64>, now_unix_ms: u64) -> Quiet {
    // **NOTHING WAS MEASURED.** Not a silence — an absence of the meters that
    // would have measured one. See [`last_seen`] and [`Quiet::NoData`].
    let Some(last_seen_unix_ms) = last_seen_unix_ms else {
        return Quiet::NoData;
    };
    if last_seen_unix_ms == 0 {
        return Quiet::Never;
    }
    match now_unix_ms.checked_sub(last_seen_unix_ms) {
        // A reading from the future is a clock disagreement, not a visit that
        // has not happened yet. Treated as "just now", which is the closest
        // true statement and never puts a working tenant on the SILENT list.
        None => Quiet::For(0),
        Some(ms) => Quiet::For(ms),
    }
}

/// [`quiet_for`], as bare milliseconds — the ordering key the SILENT list sorts
/// by, descending.
///
/// `Never` is `u64::MAX`, so the tenant nobody has ever used sorts above the one
/// last seen a year ago. [`Quiet::NoData`] is `0` and therefore sorts LAST,
/// which is defensive only: it is not a member of that list at all
/// ([`Quiet::at_least_days`] is false for it), and if a caller ever puts one
/// there anyway it must not displace a real finding from the top of the page.
pub fn quiet_for_ms(last_seen_unix_ms: Option<u64>, now_unix_ms: u64) -> u64 {
    match quiet_for(last_seen_unix_ms, now_unix_ms) {
        Quiet::NoData => 0,
        Quiet::Never => u64::MAX,
        Quiet::For(ms) => ms,
    }
}

/// **How long since a tenant was last seen** — three arms, because there are
/// three facts and only two of them are about the customer.
#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Debug)]
pub enum Quiet {
    /// **The product did not report the flow meters at all**, so nothing is
    /// known about whether this tenant is being used.
    ///
    /// Ordered first so it can never outrank a measured silence.
    ///
    /// This is NOT [`Quiet::Never`] and the difference is the whole point:
    /// `Never` is a MEASUREMENT — the product counts flow, and it counted none
    /// — while `NoData` is a GAP, in a plugin or a poller or an older product
    /// build. Rendering a gap as `never seen` publishes a finding about a
    /// customer that nobody measured, which is exactly the sentence an operator
    /// would act on by ringing them up.
    NoData,
    /// Nothing has ever moved through this tenant inside the product's kept
    /// window. **Not the same as "a very long time ago"**, and a console must
    /// not render it as a date.
    Never,
    /// Milliseconds since the last push or fetch.
    For(u64),
}

impl Quiet {
    /// Whole days, or `None` for the two arms that have no number of days.
    ///
    /// **`None` is ambiguous on purpose and must not be the only thing a
    /// renderer gets**: both [`Quiet::NoData`] and [`Quiet::Never`] answer
    /// `None` here, and they are different facts. Pass the `Quiet` itself to
    /// anything that puts words on a screen; this is for arithmetic.
    pub fn days(self) -> Option<u64> {
        match self {
            Quiet::NoData | Quiet::Never => None,
            Quiet::For(ms) => Some(ms / DAY_MS),
        }
    }

    /// Has this tenant been quiet for at least `days`?
    ///
    /// `Never` always has. **`NoData` never has** — the SILENT list is a list of
    /// MEASURED silences, and a tenant whose product does not count flow has not
    /// been shown to be quiet, only to be unmeasured.
    pub fn at_least_days(self, days: u64) -> bool {
        match self {
            Quiet::NoData => false,
            Quiet::Never => true,
            Quiet::For(ms) => ms >= days.saturating_mul(DAY_MS),
        }
    }

    /// Is this an absence of measurement rather than a measurement?
    pub fn is_no_data(self) -> bool {
        matches!(self, Quiet::NoData)
    }
}

/// **The later of a tenant's two last-seen timestamps**, from a usage reading —
/// or `None` when the reading carries NEITHER of them.
///
/// One writer, because "last seen" is a word two lists and one panel use and a
/// second `max` somewhere else is how they would come to disagree about whether
/// a fetch counts as being seen. It does: a customer who clones every morning
/// and never pushes is using the product.
///
/// # Why this is an `Option` and not a `u64`
///
/// It returned `u64` until 2026-09-10, with `.unwrap_or(0)` on each meter —
/// which folded a meter that was **absent** into a meter that **read zero**,
/// and those are the two facts this module exists to keep apart. [`daily_totals`]
/// three functions up already refuses exactly that fold, in those words: *"a
/// product that stopped reporting a meter and a product reporting zero are
/// different facts"*. This function contradicted it.
///
/// The consequence was on the screen. A tenant whose product does not count
/// flow at all — an older gunnar, the `Admin.StorageInfo` fallback path, or one
/// the poller has never read — carries neither meter, folded to `0`, and came
/// out of [`quiet_for`] as `Quiet::Never`: **`never seen`, at the top of the
/// fleet page's SILENT list**, sorted above every real finding by
/// [`quiet_for_ms`]'s `u64::MAX`. That is a claim about a customer manufactured
/// from a gap in a plugin, and an operator acts on it by ringing them up.
///
/// The tenant page had already settled this the other way — it draws *"this
/// product does not report flow"* when neither key is present, and its comment
/// says why: *"`never seen` about a product that does not count flow would be a
/// claim about a customer made from a gap in a plugin"*. Two views of one fact
/// disagreed; this makes the type carry the distinction so they cannot.
///
/// `Some(0)` is still meaningful and still means NEVER: the product reported the
/// meters and they read zero. Only `None` is *nobody measured*.
pub fn last_seen(usage: &Usage) -> Option<u64> {
    let at = |meter: &str| usage.meters.get(meter).copied();
    match (at("last_push_unix_ms"), at("last_fetch_unix_ms")) {
        // Neither meter reported: this product does not count flow.
        (None, None) => None,
        // At least one reported. A missing twin is `0` for the `max` only —
        // a product may report a push meter and no fetch meter.
        (push, fetch) => Some(push.unwrap_or(0).max(fetch.unwrap_or(0))),
    }
}

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

    fn reading(at: u64, meters: &[(&str, u64)]) -> Usage {
        Usage {
            meters: meters.iter().map(|(k, v)| ((*k).to_string(), *v)).collect(),
            measured_at_unix_ms: at,
        }
    }

    const NOON: u64 = 1_789_041_600_000;

    /// ★ **Sixty readings of one refusal are one refusal, not sixty.**
    ///
    /// The defect this module exists to prevent, and the one a naive `sum` has:
    /// a `*_today` counter is repeated in every reading of the day it belongs
    /// to, so summing readings reports a number proportional to the POLL RATE
    /// rather than to what happened.
    #[test]
    fn a_daily_counter_is_read_once_per_day_however_often_it_is_polled() {
        let midnight = day_of(NOON) * DAY_MS;
        // One refusal at 09:00, then the counter sits at 1 for the rest of the
        // day across sixty more polls.
        let mut readings = vec![reading(midnight + 8 * 3_600_000, &[("refused_today", 0)])];
        for tick in 0..60u64 {
            readings.push(reading(
                midnight + 9 * 3_600_000 + tick * 900_000,
                &[("refused_today", 1)],
            ));
        }
        let totals = daily_totals(&readings, "refused_today");
        assert_eq!(totals.len(), 1, "one day: {totals:?}");
        assert_eq!(
            totals[&day_of(NOON)],
            1,
            "summing the readings would say 60; the counter is per day, not per poll"
        );
    }

    /// The maximum, not the last reading — see the module doc for the midnight
    /// poll that makes the last reading wrong.
    #[test]
    fn the_days_total_is_its_peak_and_not_its_last_reading() {
        let midnight = day_of(NOON) * DAY_MS;
        let readings = vec![
            reading(midnight + 3_600_000, &[("refused_today", 4)]),
            reading(midnight + 7_200_000, &[("refused_today", 9)]),
            // A reading attributed to this day whose counter has already
            // rolled: taking the last would erase eight of the nine.
            reading(midnight + DAY_MS - 1, &[("refused_today", 1)]),
        ];
        assert_eq!(daily_totals(&readings, "refused_today")[&day_of(NOON)], 9);
    }

    /// ★ **A window adds whole days, and a day with no reading is absent
    /// rather than zero.**
    #[test]
    fn a_window_adds_the_days_it_names_and_invents_none() {
        let day = |n: u64| day_of(NOON) * DAY_MS + n * DAY_MS + 3_600_000;
        let readings = vec![
            reading(day(0), &[("refused_today", 2)]),
            reading(day(1), &[("refused_today", 5)]),
            // day 2: the product was unreachable and the poller stored nothing.
            reading(day(3), &[("refused_today", 1)]),
        ];
        let totals = daily_totals(&readings, "refused_today");
        assert_eq!(totals.len(), 3, "three days were measured: {totals:?}");
        assert!(
            !totals.contains_key(&(day_of(NOON) + 2)),
            "a day nobody measured must not appear as a day with no refusals"
        );
        assert_eq!(window_total(&readings, "refused_today", day(0), day(3)), 8);
        assert_eq!(window_total(&readings, "refused_today", day(1), day(3)), 6, "the window is honoured");
        assert_eq!(window_total(&readings, "refused_today", day(4), day(9)), 0, "a window with nothing in it");
    }

    /// A meter the readings do not carry is skipped, never counted as zero: a
    /// product that stopped reporting and a product reporting nothing are
    /// different facts.
    #[test]
    fn a_meter_that_is_not_reported_is_not_a_zero() {
        let readings = vec![reading(NOON, &[("bytes_in_today", 4_096)])];
        assert!(
            daily_totals(&readings, "refused_today").is_empty(),
            "a meter nobody reported must not produce a day of zero refusals"
        );
        assert_eq!(window_total(&readings, "refused_today", 0, NOON), 0);
    }

    /// ★ **NEVER is its own answer, and it sorts above every date.**
    ///
    /// `0` on the wire means "nothing has ever moved through this tenant". Read
    /// as a timestamp it is 1970, which on a console beside real dates reads as
    /// a very old customer rather than as one nobody has ever used.
    #[test]
    fn never_is_not_nineteen_seventy() {
        assert_eq!(quiet_for(Some(0), NOON), Quiet::Never);
        assert_eq!(quiet_for(Some(0), NOON).days(), None, "never has no number of days");
        assert!(quiet_for(Some(0), NOON).at_least_days(14));
        assert_eq!(
            quiet_for_ms(Some(0), NOON),
            u64::MAX,
            "the tenant nobody has ever used must sort above one last seen a year ago"
        );

        let three_days = NOON - 3 * DAY_MS;
        assert_eq!(quiet_for(Some(three_days), NOON), Quiet::For(3 * DAY_MS));
        assert_eq!(quiet_for(Some(three_days), NOON).days(), Some(3));
        assert!(!quiet_for(Some(three_days), NOON).at_least_days(14));
        assert!(quiet_for(Some(NOON - 14 * DAY_MS), NOON).at_least_days(14), "exactly at the threshold is quiet");
    }

    /// ★ **A tenant nobody MEASURED is NO DATA, and a tenant measured at zero is
    /// NEVER SEEN. They are different facts and nothing may fold them.**
    ///
    /// The distinction this whole module turns on, at the one function that used
    /// to destroy it. Until 2026-09-10 [`last_seen`] read each meter with
    /// `.unwrap_or(0)`, so a product that does not count flow at all — an older
    /// gunnar, the `Admin.StorageInfo` fallback, a tenant the poller has never
    /// read — was indistinguishable from one whose meters were reported and read
    /// zero. Both came out `Quiet::Never`.
    ///
    /// What that put on the screen: `never seen`, on the fleet page's SILENT
    /// list, sorted to the TOP by `quiet_for_ms`'s `u64::MAX` — above every real
    /// finding. A claim about a customer, manufactured from a gap in a plugin,
    /// ranked as the most urgent row on the page. The tenant page had already
    /// refused to make that claim (*"this product does not report flow"*); the
    /// fleet page made it.
    ///
    /// **Measure, never assert**: every arm below has its twin, so a fold in
    /// either direction goes red here.
    #[test]
    fn a_product_that_reports_no_flow_is_no_data_and_never_a_zero() {
        // MEASURED, and the measurement is zero: the meters are present.
        let measured_zero = reading(NOON, &[("last_push_unix_ms", 0), ("last_fetch_unix_ms", 0)]);
        // NOT MEASURED: this product reports a stock meter and no flow meter.
        let unmeasured = reading(NOON, &[("pack_bytes", 4_096)]);

        assert_eq!(last_seen(&measured_zero), Some(0), "reported, and it reads zero");
        assert_eq!(last_seen(&unmeasured), None, "no flow meter was reported at all");
        assert_ne!(
            last_seen(&measured_zero),
            last_seen(&unmeasured),
            "an absent meter and a meter reading zero must not be the same value"
        );

        assert_eq!(quiet_for(last_seen(&measured_zero), NOON), Quiet::Never);
        assert_eq!(quiet_for(last_seen(&unmeasured), NOON), Quiet::NoData);
        assert!(!quiet_for(last_seen(&measured_zero), NOON).is_no_data());
        assert!(quiet_for(last_seen(&unmeasured), NOON).is_no_data());

        // THE CONSEQUENCE, and the reason the arm exists: the SILENT list is a
        // list of MEASURED silences. A gap is not a finding about a customer.
        assert!(
            quiet_for(last_seen(&measured_zero), NOON).at_least_days(14),
            "a tenant measured as never used belongs on the SILENT list"
        );
        assert!(
            !quiet_for(last_seen(&unmeasured), NOON).at_least_days(14),
            "a tenant nobody measured must not be published as a silence"
        );

        // And it can never outrank a real finding if a caller lists it anyway.
        assert_eq!(quiet_for_ms(last_seen(&unmeasured), NOON), 0);
        assert!(quiet_for_ms(last_seen(&unmeasured), NOON) < quiet_for_ms(last_seen(&measured_zero), NOON));

        // `days()` answers `None` for BOTH, which is why it is not enough on its
        // own — a renderer that reads only this cannot tell them apart, and the
        // one that matters is `Quiet` itself.
        assert_eq!(quiet_for(last_seen(&measured_zero), NOON).days(), None);
        assert_eq!(quiet_for(last_seen(&unmeasured), NOON).days(), None);
    }

    /// A reading from the future is a clock disagreement and reads as "just
    /// now" — never as a negative age, and never as a tenant to put on the
    /// SILENT list.
    #[test]
    fn a_clock_that_runs_ahead_does_not_make_a_busy_tenant_silent() {
        assert_eq!(quiet_for(Some(NOON + DAY_MS), NOON), Quiet::For(0));
        assert!(!quiet_for(Some(NOON + DAY_MS), NOON).at_least_days(1));
    }

    /// **A fetch counts as being seen.** A customer who clones every morning
    /// and never pushes is using the product, and one writer of that decision
    /// is what stops two lists disagreeing about it.
    #[test]
    fn last_seen_is_the_later_of_the_two_and_a_fetch_counts() {
        let only_fetch = reading(NOON, &[("last_push_unix_ms", 0), ("last_fetch_unix_ms", NOON)]);
        assert_eq!(last_seen(&only_fetch), Some(NOON));
        let older_fetch = reading(
            NOON,
            &[("last_push_unix_ms", NOON), ("last_fetch_unix_ms", NOON - DAY_MS)],
        );
        assert_eq!(last_seen(&older_fetch), Some(NOON), "the later of the two");

        // ONE of the two reported is still a measurement — a product may count
        // pushes and not fetches. Only NEITHER is an absence.
        let push_only = reading(NOON, &[("last_push_unix_ms", NOON - DAY_MS)]);
        assert_eq!(last_seen(&push_only), Some(NOON - DAY_MS), "one reported meter is a measurement");

        let neither = reading(NOON, &[("pack_bytes", 1)]);
        assert_eq!(
            last_seen(&neither),
            None,
            "a product that reports neither has not been MEASURED — it has not `never been seen`"
        );
    }
}