chronix 0.4.0

//! Time buckets — the one answer to "what is a day?".
//!
//! Everything that divides a stream of instants into buckets goes through
//! [`TimeBucket`]: the `time_bucket()` SQL function and every rollup tier.
//!
//! ## What a bucket width means
//!
//! The **unit** decides, not the magnitude. That is the whole rule:
//!
//! - **Sub-day units** (`ns`, `us`, `ms`, `s`, `m`/`min`, `h`) are
//!   [`BucketWidth::Fixed`]: a uniform span that never varies. An hour bucket
//!   is always an hour. In a zone they align to the zone's **standard**
//!   (non-summer) offset, so boundaries stay evenly spaced across a
//!   daylight-saving transition — truncating the *local* clock instead would
//!   merge the repeated hour of an autumn fall-back into one bucket.
//! - **Super-day units** (`d`, `w`, `mo`, `y`) follow the **local calendar**:
//!   local midnight to local midnight, which is 23 or 25 hours on a
//!   transition day, and a month of 28, 29, 30 or 31 days. A fixed span
//!   cannot be either.
//!
//! This is the rule QuestDB's `SAMPLE BY … ALIGN TO CALENDAR TIME ZONE`
//! settles on, and TimescaleDB's `time_bucket` takes the same timezone
//! argument.
//!
//! ## Why a type and not an `i64`
//!
//! `start + width` is not the next bucket when the width is a month, nor on a
//! transition day when it is a day. Callers ask [`TimeBucket::next`] instead,
//! and cannot reach a width to add — which is the point: that arithmetic is
//! correct for every value an `i64` can hold and wrong for the quantity.
//!
//! ## Anchors
//!
//! A width of one needs none: every local midnight is a day boundary and
//! every local 1st a month boundary. Wider ones are anchored on **Monday
//! 5 January 1970** (so `1w` starts on a Monday, not on the Thursday the
//! epoch happens to be) and on **January 1970** (so `3mo` is a calendar
//! quarter and `1y` a calendar year).

use std::fmt;
use std::str::FromStr;

use chrono::{DateTime, Datelike, NaiveDate, TimeZone};
use chrono_tz::{OffsetComponents, Tz};
use serde::{Deserialize, Serialize};

const NS: i64 = 1;
const US: i64 = 1_000;
const MS: i64 = 1_000_000;
const SEC: i64 = 1_000_000_000;
const MIN: i64 = 60 * SEC;
const HOUR: i64 = 60 * MIN;
const MINUTE_NS: i64 = MIN;

/// Days from 1970-01-01 to Monday 1970-01-05.
const DAY_ANCHOR: i64 = 4;

/// A daylight-saving gap is at most a couple of hours anywhere in the tz
/// database; three hours of one-minute probes covers every transition in it.
const MAX_GAP_PROBES: i64 = 180;

/// How wide a bucket is, and therefore whether it varies.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub enum BucketWidth {
    /// A fixed span of nanoseconds. Never varies.
    Fixed(i64),
    /// A whole number of calendar days. Varies across a daylight-saving
    /// transition when the bucket is zoned.
    Days(u32),
    /// A whole number of calendar months. `3` is a quarter, `12` a year.
    Months(u32),
}

/// Error parsing a bucket specification.
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
#[error("{0}")]
pub struct BucketParseError(pub String);

impl BucketWidth {
    /// Roughly how long this bucket is, for a caller that needs one number.
    ///
    /// **An estimate for calendar widths**, and only ever used where an
    /// estimate is correct — sizing a scan, ordering two tiers by coarseness.
    /// A month is 30.44 days here and never that in reality, so anything that
    /// must be *right* uses [`TimeBucket::next`].
    #[must_use]
    pub fn nominal_ns(self) -> i64 {
        match self {
            Self::Fixed(n) => n,
            Self::Days(n) => i64::from(n).saturating_mul(24 * HOUR),
            // The mean Gregorian month: 146 097 days per 400 years / 4 800
            // months, to the nanosecond.
            Self::Months(n) => i64::from(n).saturating_mul(2_629_746 * SEC),
        }
    }

    /// Whether this width follows a calendar rather than a fixed span.
    #[must_use]
    pub fn is_calendar(self) -> bool {
        !matches!(self, Self::Fixed(_))
    }
}

impl fmt::Display for BucketWidth {
    /// The canonical spelling, which parses back to the same width.
    ///
    /// Round-tripping matters: a rollup reports the width it was created
    /// with, and an operator has to be able to type that back in.
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match *self {
            Self::Fixed(n) => {
                for (unit, suffix) in [(HOUR, "h"), (MIN, "m"), (SEC, "s"), (MS, "ms"), (US, "us")]
                {
                    if n % unit == 0 {
                        return write!(f, "{}{suffix}", n / unit);
                    }
                }
                write!(f, "{n}ns")
            }
            Self::Days(n) if n % 7 == 0 => write!(f, "{}w", n / 7),
            Self::Days(n) => write!(f, "{n}d"),
            Self::Months(n) if n % 12 == 0 => write!(f, "{}y", n / 12),
            Self::Months(n) => write!(f, "{n}mo"),
        }
    }
}

impl FromStr for BucketWidth {
    type Err = BucketParseError;

    /// Parse `15m`, `1h`, `1d`, `1w`, `1mo`, `1y`.
    ///
    /// `mo` is the month, never `M`: a case-sensitive distinction between
    /// minute and month is the kind that is wrong once and then wrong for
    /// three years of stored data. InfluxDB spells it `mo` for the same
    /// reason.
    fn from_str(s: &str) -> Result<Self, Self::Err> {
        let s = s.trim();
        if s.is_empty() {
            return Err(BucketParseError("empty bucket width".into()));
        }
        let split = s.find(|c: char| !c.is_ascii_digit()).unwrap_or(s.len());
        let (num_str, suffix) = s.split_at(split);
        let n: i64 = num_str.parse().map_err(|_| {
            BucketParseError(format!(
                "'{s}' does not start with a number — write a width like 15m, 1h, 1d, 1mo"
            ))
        })?;
        if n <= 0 {
            return Err(BucketParseError(format!(
                "a bucket width must be positive, got '{s}'"
            )));
        }
        let suffix = suffix.trim();
        let mul = |unit: i64| -> Result<Self, BucketParseError> {
            n.checked_mul(unit).map(Self::Fixed).ok_or_else(|| {
                BucketParseError(format!("bucket width '{s}' overflows the nanosecond range"))
            })
        };
        let count = |per: i64| -> Result<u32, BucketParseError> {
            u32::try_from(n.saturating_mul(per))
                .map_err(|_| BucketParseError(format!("bucket width '{s}' is too large")))
        };
        // The long spellings are accepted beside the short ones, because
        // `time_bucket('1 hour', _time)` is what somebody coming from
        // TimescaleDB or PostgreSQL writes, and the parser already tolerates
        // the space. What is deliberately *not* accepted is a bare `M` for
        // the month: `m` is the minute here, and a unit that means one thing
        // in one case and 43 200 times that in the other is a trap. The
        // month is `mo` or `month`, always.
        match suffix {
            "ns" | "nanosecond" | "nanoseconds" => mul(NS),
            "us" | "µs" | "microsecond" | "microseconds" => mul(US),
            "ms" | "millisecond" | "milliseconds" => mul(MS),
            "s" | "sec" | "secs" | "second" | "seconds" => mul(SEC),
            "m" | "min" | "mins" | "minute" | "minutes" => mul(MIN),
            "h" | "hr" | "hrs" | "hour" | "hours" => mul(HOUR),
            "d" | "day" | "days" => count(1).map(Self::Days),
            "w" | "week" | "weeks" => count(7).map(Self::Days),
            "mo" | "month" | "months" => count(1).map(Self::Months),
            "y" | "yr" | "yrs" | "year" | "years" => count(12).map(Self::Months),
            "" => Err(BucketParseError(format!(
                "'{s}' has no unit — write a width like 15m, 1h, 1d, 1mo"
            ))),
            other => Err(BucketParseError(format!(
                "unknown bucket unit '{other}'. Use ns, us, ms, s, m/min, h, d, w, mo, y — \
                 or their long forms (second, minute, hour, day, week, month, year) \
                 (the month is 'mo'; 'm' is always the minute)"
            ))),
        }
    }
}

/// A bucket width together with the calendar it is read against.
///
/// See the [module documentation](self) for what each width means.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct TimeBucket {
    width: BucketWidth,
    tz: Option<Tz>,
}

/// The on-disk and on-the-wire form.
///
/// The zone travels as its IANA name rather than as whatever `chrono_tz`
/// happens to serialise a `Tz` as, so a rollup's definition survives a
/// dependency bump and can be read by anything.
///
/// **No `skip_serializing_if`, deliberately.** The rollup catalog is
/// `postcard`, which is not self-describing: a field the serialiser omits is
/// a field the deserialiser still expects, and every following byte is read
/// as the wrong thing. An `Option` costs one tag byte and is always written.
#[derive(Serialize, Deserialize)]
struct TimeBucketRepr {
    width: BucketWidth,
    timezone: Option<String>,
}

impl Serialize for TimeBucket {
    fn serialize<S: serde::Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
        TimeBucketRepr {
            width: self.width,
            timezone: self.tz.map(|tz| tz.name().to_owned()),
        }
        .serialize(s)
    }
}

impl<'de> Deserialize<'de> for TimeBucket {
    fn deserialize<D: serde::Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
        let repr = TimeBucketRepr::deserialize(d)?;
        let tz = match repr.timezone {
            None => None,
            Some(name) => Some(name.parse::<Tz>().map_err(serde::de::Error::custom)?),
        };
        Ok(Self {
            width: repr.width,
            tz,
        })
    }
}

impl TimeBucket {
    /// A bucket of `width`, read against UTC.
    #[must_use]
    pub const fn utc(width: BucketWidth) -> Self {
        Self { width, tz: None }
    }

    /// A fixed span of nanoseconds against UTC — what an `interval_ns` was.
    #[must_use]
    pub const fn fixed_ns(ns: i64) -> Self {
        Self::utc(BucketWidth::Fixed(ns))
    }

    /// Read this bucket against an IANA time zone.
    #[must_use]
    pub const fn in_zone(mut self, tz: Tz) -> Self {
        self.tz = Some(tz);
        self
    }

    /// Parse a width, and optionally a zone.
    ///
    /// # Errors
    /// [`BucketParseError`] if the width or the zone name is not understood.
    pub fn parse(width: &str, timezone: Option<&str>) -> Result<Self, BucketParseError> {
        let width: BucketWidth = width.parse()?;
        let tz = match timezone.map(str::trim).filter(|s| !s.is_empty()) {
            None => None,
            Some(name) => Some(name.parse::<Tz>().map_err(|_| {
                BucketParseError(format!(
                    "unknown time zone '{name}' — use an IANA name like Europe/Berlin"
                ))
            })?),
        };
        Ok(Self { width, tz })
    }

    /// This bucket's width.
    #[must_use]
    pub const fn width(&self) -> BucketWidth {
        self.width
    }

    /// The IANA zone this bucket is read against, if any.
    #[must_use]
    pub fn timezone(&self) -> Option<&'static str> {
        self.tz.map(chrono_tz::Tz::name)
    }

    /// Roughly how long a bucket is — see [`BucketWidth::nominal_ns`].
    #[must_use]
    pub fn nominal_ns(&self) -> i64 {
        self.width.nominal_ns()
    }

    /// Whether a bucket's length can vary between one bucket and the next.
    ///
    /// True for a calendar width in *any* zone (February is short in UTC
    /// too), and for a day only where a zone has transitions.
    #[must_use]
    pub fn is_calendar(&self) -> bool {
        self.width.is_calendar()
    }

    /// The start of the bucket containing `ts`.
    #[must_use]
    pub fn start_of(&self, ts: i64) -> i64 {
        match self.width {
            BucketWidth::Fixed(n) => {
                let n = n.max(1);
                // Aligned to the zone's *standard* offset, not to whatever
                // offset is in force at `ts`: an alignment that moved with
                // summer time would make one bucket an hour long and its
                // neighbour two, which is the defect this rule exists to
                // avoid. In a whole-hour zone this leaves hour buckets on
                // UTC hour boundaries; in Asia/Kolkata (+05:30) they land at
                // half past, which is the local clock's hour.
                let offset = self.standard_offset_ns(ts);
                ts.saturating_sub(ts.wrapping_sub(offset).rem_euclid(n))
            }
            BucketWidth::Days(n) => {
                let n = i64::from(n.max(1));
                let Some(date) = self.local_date(ts) else {
                    return ts;
                };
                let days = days_from_epoch(date);
                let start = (days - DAY_ANCHOR).div_euclid(n) * n + DAY_ANCHOR;
                self.local_midnight(date_from_epoch_days(start), ts)
            }
            BucketWidth::Months(n) => {
                let n = i64::from(n.max(1));
                let Some(date) = self.local_date(ts) else {
                    return ts;
                };
                let months = i64::from(date.year() - 1970) * 12 + i64::from(date.month0());
                let start = months.div_euclid(n) * n;
                self.local_midnight(first_of_month(start), ts)
            }
        }
    }

    /// The start of the bucket after the one starting at `bucket_start`.
    ///
    /// **Never `bucket_start + width`.** A calendar bucket's length depends
    /// on which bucket it is: February is three days shorter than March, and
    /// a local day is 23 or 25 hours on a transition. Code that adds is code
    /// that has assumed the calendar away.
    #[must_use]
    pub fn next(&self, bucket_start: i64) -> i64 {
        match self.width {
            BucketWidth::Fixed(n) => bucket_start.saturating_add(n.max(1)),
            BucketWidth::Days(n) => {
                let Some(date) = self.local_date(bucket_start) else {
                    return bucket_start.saturating_add(self.nominal_ns());
                };
                let next = days_from_epoch(date).saturating_add(i64::from(n.max(1)));
                self.local_midnight(date_from_epoch_days(next), bucket_start)
            }
            BucketWidth::Months(n) => {
                let Some(date) = self.local_date(bucket_start) else {
                    return bucket_start.saturating_add(self.nominal_ns());
                };
                let months = i64::from(date.year() - 1970) * 12 + i64::from(date.month0());
                self.local_midnight(
                    first_of_month(months.saturating_add(i64::from(n.max(1)))),
                    bucket_start,
                )
            }
        }
    }

    /// The local calendar date `ts` falls on.
    fn local_date(&self, ts: i64) -> Option<NaiveDate> {
        let utc = DateTime::from_timestamp_nanos(ts);
        Some(match self.tz {
            None => utc.date_naive(),
            Some(tz) => utc.with_timezone(&tz).date_naive(),
        })
    }

    /// The instant local midnight of `date` happens at.
    ///
    /// `fallback` is returned only where the date cannot be represented at
    /// all, which needs a timestamp outside the ±292 years an `i64` of
    /// nanoseconds can hold.
    fn local_midnight(&self, date: Option<NaiveDate>, fallback: i64) -> i64 {
        let Some(naive) = date.and_then(|d| d.and_hms_opt(0, 0, 0)) else {
            return fallback;
        };
        let Some(tz) = self.tz else {
            return naive.and_utc().timestamp_nanos_opt().unwrap_or(fallback);
        };
        match tz.from_local_datetime(&naive) {
            // Autumn fall-back: this local time happens twice. The earliest
            // is the start of the bucket, which is what a boundary means.
            chrono::LocalResult::Single(dt) => dt.timestamp_nanos_opt().unwrap_or(fallback),
            chrono::LocalResult::Ambiguous(earliest, _) => {
                earliest.timestamp_nanos_opt().unwrap_or(fallback)
            }
            // Spring forward *at midnight* — which several zones do, among
            // them America/Santiago and Asia/Beirut. The local time does not
            // exist, so the bucket starts at the instant the clock resumed.
            chrono::LocalResult::None => (1..=MAX_GAP_PROBES)
                .find_map(|step| {
                    let probe = naive.checked_add_signed(chrono::TimeDelta::nanoseconds(
                        step.saturating_mul(MINUTE_NS),
                    ))?;
                    match tz.from_local_datetime(&probe) {
                        chrono::LocalResult::Single(dt) => dt.timestamp_nanos_opt(),
                        chrono::LocalResult::Ambiguous(earliest, _) => {
                            earliest.timestamp_nanos_opt()
                        }
                        chrono::LocalResult::None => None,
                    }
                })
                .unwrap_or(fallback),
        }
    }

    /// The zone's standard (non-summer) UTC offset at `ts`, in nanoseconds.
    ///
    /// A zone's *base* offset does change, but on the scale of decades —
    /// Europe/Moscow moved twice — so this is read at `ts` rather than
    /// assumed constant.
    fn standard_offset_ns(&self, ts: i64) -> i64 {
        let Some(tz) = self.tz else { return 0 };
        let naive = DateTime::from_timestamp_nanos(ts).naive_utc();
        tz.offset_from_utc_datetime(&naive)
            .base_utc_offset()
            .num_nanoseconds()
            .unwrap_or(0)
    }
}

impl fmt::Display for TimeBucket {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self.tz {
            None => write!(f, "{}", self.width),
            Some(tz) => write!(f, "{} {}", self.width, tz.name()),
        }
    }
}

/// `num_days_from_ce()` of 1970-01-01, so that the epoch is day zero.
///
/// Pinned by a test rather than trusted: an off-by-one here is invisible for
/// every one-day bucket — the anchor cannot matter when every day is a
/// boundary — and moves `1w` off Monday, which is the only place it shows.
const EPOCH_DAYS_FROM_CE: i64 = 719_163;

/// Days from 1970-01-01 to `date`.
fn days_from_epoch(date: NaiveDate) -> i64 {
    i64::from(date.num_days_from_ce()) - EPOCH_DAYS_FROM_CE
}

/// The date `days` after 1970-01-01.
fn date_from_epoch_days(days: i64) -> Option<NaiveDate> {
    i32::try_from(days.checked_add(EPOCH_DAYS_FROM_CE)?)
        .ok()
        .and_then(NaiveDate::from_num_days_from_ce_opt)
}

/// The first of the month `months` after January 1970.
fn first_of_month(months: i64) -> Option<NaiveDate> {
    let year = i32::try_from(months.div_euclid(12))
        .ok()?
        .checked_add(1970)?;
    let month = u32::try_from(months.rem_euclid(12)).ok()?.checked_add(1)?;
    NaiveDate::from_ymd_opt(year, month, 1)
}

// ── Tests ──────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    #![allow(clippy::unwrap_used, clippy::expect_used)]

    use super::*;

    fn ts(s: &str) -> i64 {
        chrono::DateTime::parse_from_rfc3339(s)
            .unwrap()
            .timestamp_nanos_opt()
            .unwrap()
    }

    fn berlin() -> Tz {
        "Europe/Berlin".parse().unwrap()
    }

    // ── Parsing ────────────────────────────────────────────────────────

    #[test]
    fn the_unit_decides_whether_a_width_is_a_calendar_one() {
        use BucketWidth::{Days, Fixed, Months};
        let cases = [
            ("30s", Fixed(30 * SEC)),
            ("15m", Fixed(15 * MIN)),
            ("15min", Fixed(15 * MIN)),
            // A sub-day unit stays fixed however large the number: 36 hours
            // is thirty-six hours, not a day and a half of calendar.
            ("36h", Fixed(36 * HOUR)),
            ("1d", Days(1)),
            ("1w", Days(7)),
            ("2w", Days(14)),
            ("1mo", Months(1)),
            ("3mo", Months(3)),
            ("1y", Months(12)),
        ];
        for (text, want) in cases {
            assert_eq!(text.parse::<BucketWidth>().unwrap(), want, "{text}");
        }
    }

    /// Every width round-trips through its own spelling.
    ///
    /// A rollup reports the width it was created with and an operator has to
    /// be able to type that back in (the invertible-name rule).
    #[test]
    fn a_widths_spelling_parses_back_to_the_same_width() {
        for text in [
            "500ns", "5us", "250ms", "30s", "15m", "6h", "1d", "1w", "3mo", "1y",
        ] {
            let width: BucketWidth = text.parse().unwrap();
            let spelled = width.to_string();
            assert_eq!(
                spelled.parse::<BucketWidth>().unwrap(),
                width,
                "{text} spelled as {spelled}"
            );
        }
    }

    #[test]
    fn the_month_is_mo_and_m_is_always_the_minute() {
        assert_eq!(
            "1m".parse::<BucketWidth>().unwrap(),
            BucketWidth::Fixed(MIN)
        );
        assert_eq!(
            "1mo".parse::<BucketWidth>().unwrap(),
            BucketWidth::Months(1)
        );
        // `M` is refused rather than guessed at: a case-sensitive difference
        // between a minute and a month is wrong once and then wrong for
        // three years of stored data.
        let err = "1M".parse::<BucketWidth>().unwrap_err();
        assert!(err.0.contains("'mo'"), "{err}");
    }

    #[test]
    fn the_long_unit_names_mean_the_same_as_the_short_ones() {
        // `time_bucket('1 hour', _time)` is what somebody coming from
        // TimescaleDB writes, and the parser already tolerated the space —
        // so the only thing standing between that query and an answer was
        // the vocabulary.
        for (long, short) in [
            ("1 second", "1s"),
            ("30 seconds", "30s"),
            ("15 minutes", "15m"),
            ("1 minute", "1m"),
            ("1 hour", "1h"),
            ("24 hours", "24h"),
            ("1 day", "1d"),
            ("2 weeks", "2w"),
            ("1 month", "1mo"),
            ("3 months", "3mo"),
            ("1 year", "1y"),
        ] {
            assert_eq!(
                long.parse::<BucketWidth>().unwrap(),
                short.parse::<BucketWidth>().unwrap(),
                "'{long}' must mean '{short}'"
            );
        }
        // And the long forms work without the space too.
        assert_eq!(
            "1hour".parse::<BucketWidth>().unwrap(),
            BucketWidth::Fixed(HOUR)
        );
    }

    #[test]
    fn no_long_form_reopens_the_minute_month_trap() {
        // `month` is unambiguous; `M`, `Min` and `MO` are not spellings this
        // parser accepts, because a case-sensitive difference between a
        // minute and a month is wrong once and then wrong for three years of
        // stored data.
        assert_eq!(
            "1month".parse::<BucketWidth>().unwrap(),
            BucketWidth::Months(1)
        );
        assert_eq!(
            "1minute".parse::<BucketWidth>().unwrap(),
            BucketWidth::Fixed(MIN)
        );
        for ambiguous in ["1M", "1MO", "1Month", "1Min"] {
            assert!(
                ambiguous.parse::<BucketWidth>().is_err(),
                "'{ambiguous}' must be refused rather than guessed at"
            );
        }
    }

    #[test]
    fn a_width_without_a_unit_or_with_a_bad_one_is_refused() {
        for bad in ["", "5", "0h", "-1h", "1fortnight", "h"] {
            assert!(
                bad.parse::<BucketWidth>().is_err(),
                "'{bad}' must be refused"
            );
        }
    }

    #[test]
    fn an_unknown_zone_is_refused_by_name() {
        let err = TimeBucket::parse("1d", Some("Europe/Atlantis")).unwrap_err();
        assert!(err.0.contains("Europe/Atlantis"), "{err}");
        assert!(err.0.contains("IANA"), "{err}");
    }

    // ── Fixed widths stay uniform ──────────────────────────────────────

    /// The defect this type exists for: two instants an hour apart cannot
    /// share a one-hour bucket, even where the local clock says they can.
    #[test]
    fn a_fixed_bucket_does_not_merge_the_repeated_hour() {
        let b = TimeBucket::utc(BucketWidth::Fixed(HOUR)).in_zone(berlin());
        // 2026-10-25T01:00:00Z is the transition: 03:00 CEST → 02:00 CET.
        // Both of these read 02:30 on the local clock.
        let first = b.start_of(ts("2026-10-25T00:30:00Z"));
        let second = b.start_of(ts("2026-10-25T01:30:00Z"));
        assert_eq!(second - first, HOUR, "hour buckets stay an hour apart");
        assert_eq!(first, ts("2026-10-25T00:00:00Z"));
        assert_eq!(second, ts("2026-10-25T01:00:00Z"));
    }

    /// …so a 25-hour local day holds 25 hourly buckets, evenly spaced.
    #[test]
    fn a_fall_back_day_holds_twenty_five_hourly_buckets() {
        let b = TimeBucket::utc(BucketWidth::Fixed(HOUR)).in_zone(berlin());
        let day_start = ts("2026-10-24T22:00:00Z"); // local midnight, CEST
        let buckets: std::collections::BTreeSet<i64> = (0..25 * 60)
            .map(|m| b.start_of(day_start + m * MIN))
            .collect();
        assert_eq!(buckets.len(), 25);
        let mut prev: Option<i64> = None;
        for start in buckets {
            if let Some(p) = prev {
                assert_eq!(start - p, HOUR, "evenly spaced");
            }
            prev = Some(start);
        }
    }

    /// A half-hour zone buckets on its own clock, which is what makes the
    /// standard-offset alignment observable at all.
    #[test]
    fn a_half_hour_zone_buckets_on_its_own_clock() {
        let kolkata: Tz = "Asia/Kolkata".parse().unwrap(); // UTC+05:30, no DST
        let b = TimeBucket::utc(BucketWidth::Fixed(HOUR)).in_zone(kolkata);
        // 00:10 UTC is 05:40 local; the hour bucket is 05:00 local = 23:30Z
        // the previous day.
        assert_eq!(
            b.start_of(ts("2024-03-15T00:10:00Z")),
            ts("2024-03-14T23:30:00Z")
        );
    }

    #[test]
    fn an_unzoned_fixed_bucket_truncates_utc() {
        let b = TimeBucket::fixed_ns(HOUR);
        assert_eq!(
            b.start_of(ts("2024-03-15T02:34:56Z")),
            ts("2024-03-15T02:00:00Z")
        );
        assert_eq!(
            b.next(ts("2024-03-15T02:00:00Z")),
            ts("2024-03-15T03:00:00Z")
        );
    }

    // ── Calendar widths follow the calendar ────────────────────────────

    #[test]
    fn a_zoned_day_is_local_midnight_to_local_midnight() {
        let ny: Tz = "America/New_York".parse().unwrap();
        let b = TimeBucket::utc(BucketWidth::Days(1)).in_zone(ny);
        // 02:00 UTC on the 15th is 22:00 on the *14th* in New York.
        assert_eq!(
            b.start_of(ts("2024-03-15T02:00:00Z")),
            ts("2024-03-14T04:00:00Z")
        );
    }

    /// A transition day is 23 or 25 hours, and `next` says so.
    #[test]
    fn a_transition_day_is_not_twenty_four_hours() {
        let ny: Tz = "America/New_York".parse().unwrap();
        let day = TimeBucket::utc(BucketWidth::Days(1)).in_zone(ny);

        // Spring forward: 2024-03-10 is 23 hours in New York.
        let start = day.start_of(ts("2024-03-10T20:00:00Z"));
        assert_eq!(start, ts("2024-03-10T05:00:00Z"));
        assert_eq!(day.next(start) - start, 23 * HOUR);

        // Fall back: 2024-11-03 is 25 hours.
        let start = day.start_of(ts("2024-11-03T20:00:00Z"));
        assert_eq!(start, ts("2024-11-03T04:00:00Z"));
        assert_eq!(day.next(start) - start, 25 * HOUR);
    }

    /// A month is a month — and `start + width` would never have been one.
    #[test]
    fn a_month_bucket_is_a_calendar_month() {
        let b = TimeBucket::utc(BucketWidth::Months(1)).in_zone(berlin());
        let feb = b.start_of(ts("2024-02-17T12:00:00Z"));
        assert_eq!(feb, ts("2024-01-31T23:00:00Z"), "1 Feb 2024, Berlin (CET)");
        // 2024 is a leap year: February is 29 days.
        assert_eq!(b.next(feb) - feb, 29 * 24 * HOUR);
        // …and 2023's was 28.
        let feb23 = b.start_of(ts("2023-02-17T12:00:00Z"));
        assert_eq!(b.next(feb23) - feb23, 28 * 24 * HOUR);
        // March crosses the spring transition, so it is an hour short of 31
        // days.
        let mar = b.start_of(ts("2024-03-17T12:00:00Z"));
        assert_eq!(b.next(mar) - mar, 31 * 24 * HOUR - HOUR);
    }

    #[test]
    fn a_year_bucket_is_a_calendar_year() {
        let b = TimeBucket::utc(BucketWidth::Months(12));
        let y = b.start_of(ts("2024-07-04T00:00:00Z"));
        assert_eq!(y, ts("2024-01-01T00:00:00Z"));
        assert_eq!(b.next(y), ts("2025-01-01T00:00:00Z"));
    }

    #[test]
    fn a_quarter_is_three_calendar_months_from_january() {
        let b = TimeBucket::utc(BucketWidth::Months(3));
        assert_eq!(
            b.start_of(ts("2024-05-17T00:00:00Z")),
            ts("2024-04-01T00:00:00Z")
        );
        assert_eq!(
            b.start_of(ts("2024-12-31T23:59:59Z")),
            ts("2024-10-01T00:00:00Z")
        );
    }

    /// The epoch is day zero, and it is a Thursday.
    ///
    /// Pinned because an off-by-one is invisible everywhere except in the
    /// weekday a `1w` bucket starts on.
    #[test]
    fn the_epoch_is_day_zero_and_a_thursday() {
        let epoch = NaiveDate::from_ymd_opt(1970, 1, 1).unwrap();
        assert_eq!(days_from_epoch(epoch), 0);
        assert_eq!(epoch.weekday(), chrono::Weekday::Thu);
        assert_eq!(
            date_from_epoch_days(DAY_ANCHOR).unwrap().weekday(),
            chrono::Weekday::Mon,
            "the multi-day anchor is a Monday"
        );
    }

    /// A week starts on a Monday, not on the Thursday the epoch happens to
    /// be.
    #[test]
    fn a_week_starts_on_monday() {
        let b = TimeBucket::utc(BucketWidth::Days(7));
        // 2024-03-15 is a Friday; its week began Monday 2024-03-11.
        let start = b.start_of(ts("2024-03-15T12:00:00Z"));
        assert_eq!(start, ts("2024-03-11T00:00:00Z"));
        assert_eq!(
            DateTime::from_timestamp_nanos(start).date_naive().weekday(),
            chrono::Weekday::Mon
        );
    }

    /// Every bucket is half-open and they tile the line with no gap and no
    /// overlap — the property every consumer of `start_of`/`next` assumes.
    #[test]
    fn buckets_tile_the_line() {
        let specs = [
            TimeBucket::fixed_ns(HOUR),
            TimeBucket::utc(BucketWidth::Fixed(15 * MIN)).in_zone(berlin()),
            TimeBucket::utc(BucketWidth::Days(1)).in_zone(berlin()),
            TimeBucket::utc(BucketWidth::Months(1)).in_zone(berlin()),
            TimeBucket::utc(BucketWidth::Days(7)),
        ];
        // A span that crosses both of Berlin's 2026 transitions and a
        // February.
        let from = ts("2026-01-15T00:00:00Z");
        let to = ts("2027-01-15T00:00:00Z");
        for b in specs {
            let mut cursor = b.start_of(from);
            let mut steps = 0;
            while cursor < to {
                let next = b.next(cursor);
                assert!(next > cursor, "{b} did not advance at {cursor}");
                // Every instant inside [cursor, next) belongs to this bucket.
                for probe in [cursor, cursor + (next - cursor) / 2, next - 1] {
                    assert_eq!(b.start_of(probe), cursor, "{b} at {probe}");
                }
                // …and the first instant of the next one does not.
                assert_eq!(b.start_of(next), next, "{b} boundary at {next}");
                cursor = next;
                steps += 1;
                assert!(steps < 40_000, "{b} is not advancing");
            }
        }
    }

    // ── Serialisation ──────────────────────────────────────────────────

    /// A rollup's definition survives a restart, zone and all — through the
    /// format the catalog actually uses.
    ///
    /// `postcard` is not self-describing, so a `skip_serializing_if` on the
    /// zone made every rollup in the catalog unreadable after the first one
    /// without a zone: the omitted field was still expected, and the bytes
    /// after it were read as the wrong thing. JSON hid it completely.
    #[test]
    fn a_bucket_round_trips_through_the_catalog_format() {
        for b in [
            TimeBucket::fixed_ns(15 * MIN),
            TimeBucket::utc(BucketWidth::Days(1)),
            TimeBucket::utc(BucketWidth::Days(1)).in_zone(berlin()),
            TimeBucket::utc(BucketWidth::Months(3)).in_zone(berlin()),
        ] {
            let bytes = postcard::to_allocvec(&b).unwrap();
            assert_eq!(postcard::from_bytes::<TimeBucket>(&bytes).unwrap(), b);
        }
        // …and a sequence of them, which is what the catalog stores: a
        // truncated field only shows up once something follows it.
        let seq = vec![
            TimeBucket::fixed_ns(15 * MIN),
            TimeBucket::utc(BucketWidth::Months(1)).in_zone(berlin()),
            TimeBucket::utc(BucketWidth::Days(7)),
        ];
        let bytes = postcard::to_allocvec(&seq).unwrap();
        assert_eq!(
            postcard::from_bytes::<Vec<TimeBucket>>(&bytes).unwrap(),
            seq
        );
    }

    /// A rollup's definition survives a restart, zone and all.
    #[test]
    fn a_bucket_round_trips_through_json_by_zone_name() {
        for b in [
            TimeBucket::fixed_ns(15 * MIN),
            TimeBucket::utc(BucketWidth::Days(1)).in_zone(berlin()),
            TimeBucket::utc(BucketWidth::Months(3)).in_zone(berlin()),
        ] {
            let json = serde_json::to_string(&b).unwrap();
            assert_eq!(
                serde_json::from_str::<TimeBucket>(&json).unwrap(),
                b,
                "{json}"
            );
            if let Some(name) = b.timezone() {
                assert!(json.contains(name), "the zone travels by name: {json}");
            }
        }
    }

    #[test]
    fn display_round_trips_through_parse() {
        for (width, zone) in [("15m", None), ("1d", Some("Europe/Berlin")), ("1mo", None)] {
            let b = TimeBucket::parse(width, zone).unwrap();
            assert_eq!(b.width().to_string(), width);
            assert_eq!(b.timezone(), zone);
        }
    }
}