//! One rendering, however many logs.
//!
//! # Why this is generic
//!
//! `atgc logs oauth` and `atgc logs pds` read different events out of
//! different files, and a person reading them is doing the same thing both
//! times: narrowing to a window, grouping by the run that produced the lines,
//! following the file while a command runs. Two readers would have meant two
//! `--since` parsers, two ideas of what `-n` counts, two renderings of a
//! timestamp — and the moment they disagreed, a reader lining one log up
//! against the other would be wrong without being told.
//!
//! So everything that is about *reading a log* lives here once, over a
//! [`Kind`]: the filters, the palette, the grouping, the follow loop, the
//! rotation handling. What each log implements is only the part that is
//! genuinely about its own events — what they are called, how a row of one
//! reads, and which of them are failures.
//!
//! # What a log still owns
//!
//! [`Kind`] is deliberately small and every method on it answers a question
//! the rendering asks. `detail` is where a log spends its character budget
//! and is the only place either module writes a sentence; everything else is
//! classification. A log that wants a view this file does not have — the
//! OAuth log's `--incident` — passes it in as an analysis of the entries it
//! already parsed, rather than growing a flag here that one log answers and
//! the other refuses.

use anyhow::{Context, Result, bail};
use chrono::{DateTime, Local, NaiveDate, NaiveTime, TimeZone, Utc};
use serde::Deserialize;
use std::collections::{BTreeMap, BTreeSet};
use std::io::{Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};

/// How long to wait between polls in `--follow`.
///
/// A quarter second is under the threshold at which a person tailing a log
/// notices lag, and 4 `stat`/`read` pairs a second on one file is free. inotify
/// would be tighter and would mean a dependency and a second code path for
/// the rotation case, which is the case that actually matters here.
const POLL: std::time::Duration = std::time::Duration::from_millis(250);

/// How many events to show when nothing says otherwise.
///
/// A screen holds forty or so; two hundred is a few screens of scrollback,
/// which is the amount a person scrolls up through looking for the thing they
/// just did. `--all` and `-n` are one keystroke away when it is the wrong
/// amount.
const DEFAULT_LINES: usize = 200;

/// What one log's events have to be able to say about themselves.
///
/// Implemented by [`crate::logging::oauth::Event`] and [`crate::logging::pds::Event`],
/// beside the writers that produce them — the reader deserializes through the
/// writer's own enum, so a variant added to a writer is a compile error in
/// its `impl Kind` rather than a line that renders as nothing.
pub(super) trait Kind: serde::de::DeserializeOwned {
    /// Every event name this log can write, for `--event` to validate
    /// against. The order is the writer's, which is roughly causal, and it is
    /// what an unknown-kind error prints back.
    fn kinds() -> &'static [&'static str];

    /// This event's own name — the `event` tag it was written under.
    fn name(&self) -> &'static str;

    /// The payload, rendered for one row. The only place a log writes prose.
    fn detail(&self, p: &Palette) -> String;

    /// Whether `--failures` should keep this event.
    ///
    /// Wider than "an error happened" on purpose in both logs: an event can
    /// be a successful operation and still be the line an investigation is
    /// looking for.
    fn is_failure(&self) -> bool;

    /// Every DID this event names, for `--did`.
    ///
    /// A `Vec` because one event can name several — `logout --all` takes out
    /// four accounts at once.
    fn dids(&self) -> Vec<&str>;

    /// How to colour the event's name.
    fn severity(&self) -> Severity;

    /// Whether this is the invocation's head line, which renders as a group
    /// heading instead of as a row.
    fn is_head(&self) -> bool;
}

/// What colour an event's name is printed in.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum Severity {
    /// Ordinary.
    Plain,
    /// Structural rather than eventful: the invocation head.
    Note,
    /// It worked, and the fact that it worked is the news.
    Good,
    /// It failed, or it is the line an investigation is looking for.
    Bad,
}

/// Which log a reader is pointed at, and what to say when it is not there.
pub(crate) struct Source {
    /// The writer's own handle, which owns the path and the switch.
    pub log: &'static crate::logging::file::Log,
    /// What this log is called in a sentence: "OAuth", "PDS write".
    pub noun: &'static str,
    /// The cheapest command that would produce a line, for an empty log.
    pub how_to_fill: &'static str,
}

/// The flags every log reader takes, resolved from the CLI.
///
/// `--incident` is absent because it belongs to one log; a view only one
/// reader can answer is passed to [`run`] as an analysis, not carried here
/// as a flag the other would have to refuse.
/// The flags every `logs` subcommand takes.
///
/// One definition flattened into each, rather than a copy per command that
/// somebody has to remember to edit. The readers share this rendering, and
/// sharing the flag definitions is what keeps that promise true at the
/// command line as well as on screen: a flag added here reaches every log,
/// and one that reached only some of them would not compile.
#[derive(clap::Args, Debug)]
pub(crate) struct Args {
    /// Stream new events as they are appended (implies --interleave)
    #[arg(short, long)]
    pub follow: bool,
    /// Show at most this many of the most recent events (0 for none)
    ///
    /// `--limit` is accepted for it too. `-n`/`--lines` is `tail(1)`'s
    /// spelling and is the right one for a log, but `--limit` is what every
    /// listing in the tool takes and what a caller arrives here having
    /// learned, and `atgc logs oauth --limit 20` used to be "unexpected
    /// argument '--limit' found". Unlike the listings, 0 is meaningful
    /// here — the head lines and the summary with no events between them —
    /// so there is no floor on it.
    #[arg(short = 'n', long, visible_alias = "limit", value_name = "N")]
    pub lines: Option<usize>,
    /// Show every event, however many there are
    #[arg(long, conflicts_with = "lines")]
    pub all: bool,
    /// Only this invocation (any substring of its id, e.g. the short tag)
    #[arg(long, value_name = "ID")]
    pub inv: Option<String>,
    /// Only these event kinds, comma-separated; a trailing * takes a family
    #[arg(long, value_name = "KIND")]
    pub event: Vec<String>,
    /// Only events naming this account (DID, or part of one)
    #[arg(long, value_name = "DID")]
    pub did: Option<String>,
    /// Only events at or after this time (30m, 08:55, 2026-08-06, RFC 3339)
    #[arg(long, value_name = "WHEN")]
    pub since: Option<String>,
    /// Only events at or before this time
    #[arg(long, value_name = "WHEN")]
    pub until: Option<String>,
    /// Only refusals, errors, session deletions and dropped records
    #[arg(long)]
    pub failures: bool,
    /// Print the log records themselves, one JSON object per line, for jq
    #[arg(long)]
    pub json: bool,
    /// Strict time order across invocations instead of grouping by one
    #[arg(long)]
    pub interleave: bool,
    /// Read this file instead of the configured log
    #[arg(long, value_name = "PATH")]
    pub file: Option<PathBuf>,
}

// ===========================================================================
// The record, as read
// ===========================================================================

/// One line of the log, parsed.
///
/// `raw` is kept beside the parsed form so `--json` can emit the bytes that
/// were actually on disk rather than a re-serialization of them. Those two
/// differ in key order and in anything a newer writer added that this build's
/// event enum does not know about, and the whole promise of `--json` is that
/// the `jq` pipelines people already have keep working. See
/// [`write_json_line`].
#[derive(Debug)]
pub(crate) struct Entry<E> {
    pub ts: DateTime<Utc>,
    pub inv: String,
    pub seq: u64,
    pub event: E,
    pub raw: String,
}

/// The envelope, mirroring the writer's private `Record`.
///
/// Not shared with the writer because the writer's version borrows (`inv:
/// &'a str`, `event: &'a Event`) so that emitting an event allocates as
/// little as possible on a hot path that must finish inside one `write(2)`.
/// Reading has the opposite priority and wants owned data.
#[derive(Debug, Deserialize)]
struct Wire<E> {
    ts: String,
    inv: String,
    seq: u64,
    #[serde(flatten)]
    event: E,
}

/// Parse one line, or say why it could not be parsed.
///
/// The error is a `String` rather than an `anyhow::Error` because every
/// caller reports it in-band and continues; there is no path here where a bad
/// line propagates upward, and typing it as a hard error would invite one.
pub(super) fn parse_line<E: Kind>(raw: &str) -> Result<Entry<E>, String> {
    let wire: Wire<E> = serde_json::from_str(raw).map_err(|e| e.to_string())?;
    let ts = DateTime::parse_from_rfc3339(&wire.ts)
        .map_err(|e| format!("bad timestamp {:?}: {e}", wire.ts))?
        .with_timezone(&Utc);
    Ok(Entry {
        ts,
        inv: wire.inv,
        seq: wire.seq,
        event: wire.event,
        raw: raw.to_string(),
    })
}

// Each log names its own events, in `Kind::name`. It is written out there
// rather than obtained by serializing to a `serde_json::Value` and reading
// the tag back, which is what the writers do in their oversize path: that
// costs an allocation and a full serialization per line, and this runs on
// every line of a file that can hold tens of thousands.

// ===========================================================================
// Filtering
// ===========================================================================

/// A compiled set of `--event` selectors.
///
/// A selector is a kind name, or a prefix ending in `*`. The glob exists
/// because the kinds fall into obvious families — `token_*`, `session_*`,
/// `auth_state_*` — and asking for one family is the common case; without it
/// the equivalent is four comma-separated names typed exactly right.
#[derive(Debug, Default)]
pub(super) struct KindFilter {
    exact: BTreeSet<String>,
    prefixes: Vec<String>,
}

impl KindFilter {
    pub(super) fn parse(specs: &[String], kinds: &'static [&'static str]) -> Result<Self> {
        let mut filter = KindFilter::default();
        for spec in specs.iter().flat_map(|s| s.split(',')) {
            let spec = spec.trim();
            if spec.is_empty() {
                continue;
            }
            match spec.strip_suffix('*') {
                Some(prefix) => {
                    if !kinds.iter().any(|k| k.starts_with(prefix)) {
                        bail!("no event kind starts with {prefix:?}\n{}", kind_help(kinds));
                    }
                    filter.prefixes.push(prefix.to_string());
                }
                None => {
                    if !kinds.contains(&spec) {
                        bail!("unknown event kind {spec:?}\n{}", kind_help(kinds));
                    }
                    filter.exact.insert(spec.to_string());
                }
            }
        }
        Ok(filter)
    }

    fn is_empty(&self) -> bool {
        self.exact.is_empty() && self.prefixes.is_empty()
    }

    pub(super) fn matches(&self, kind: &str) -> bool {
        self.is_empty()
            || self.exact.contains(kind)
            || self.prefixes.iter().any(|p| kind.starts_with(p.as_str()))
    }
}

fn kind_help(kinds: &[&str]) -> String {
    format!(
        "known kinds: {}\n(a trailing * matches a family, e.g. token_*)",
        kinds.join(", ")
    )
}

/// Everything the row filters need, resolved once.
#[derive(Debug)]
pub(super) struct Filters {
    inv: Option<String>,
    kinds: KindFilter,
    did: Option<String>,
    since: Option<DateTime<Utc>>,
    until: Option<DateTime<Utc>>,
    failures: bool,
}

impl Filters {
    pub(super) fn from_args<E: Kind>(args: &Args, now: DateTime<Utc>) -> Result<Self> {
        Ok(Filters {
            inv: args.inv.clone(),
            kinds: KindFilter::parse(&args.event, E::kinds())?,
            did: args.did.clone(),
            since: args
                .since
                .as_deref()
                .map(|s| parse_time(s, now))
                .transpose()
                .context("--since")?,
            until: args
                .until
                .as_deref()
                .map(|s| parse_time(s, now))
                .transpose()
                .context("--until")?,
            failures: args.failures,
        })
    }

    /// Applied uniformly to every row, `invocation` included.
    ///
    /// No event is exempt, because an exemption is the kind of rule that
    /// makes a reader distrust the output: if `--event token_refused` printed
    /// invocation headers "for context", a count of the lines on screen would
    /// stop meaning anything. Grouped rendering gets its headers from the
    /// unfiltered stream instead, so context is restored where it belongs —
    /// as a heading, not as a row that pretends to have matched.
    pub(super) fn keep<E: Kind>(&self, entry: &Entry<E>) -> bool {
        if let Some(inv) = &self.inv
            && !entry.inv.contains(inv.as_str())
        {
            return false;
        }
        if !self.kinds.matches(entry.event.name()) {
            return false;
        }
        if let Some(did) = &self.did
            && !entry.event.dids().iter().any(|d| d.contains(did.as_str()))
        {
            return false;
        }
        if self.since.is_some_and(|t| entry.ts < t) || self.until.is_some_and(|t| entry.ts > t) {
            return false;
        }
        if self.failures && !entry.event.is_failure() {
            return false;
        }
        true
    }
}

/// Parse a `--since`/`--until` value.
///
/// Four spellings, in the order people reach for them: a bare duration ago
/// (`30m`, `2h`, `3d`) because the question is nearly always "what happened
/// while I was doing that"; a wall-clock time today (`08:55`) because that is
/// what a person reads off a screenshot; a date; and full RFC 3339, which is
/// what the log itself prints and so what a copy-paste out of it produces.
///
/// The middle two are interpreted in the *local* zone even though the log
/// stores UTC, because a bare `08:55` typed by a person means their clock.
/// The log's own timestamps carry `Z`, so pasting one back in is unambiguous
/// and unaffected.
///
/// `pub(crate)` rather than `pub(super)` for [`crate::cmd::search`], whose
/// `--since`/`--until` bound a Bobbin query rather than a log. Copying it
/// there would have been the worse trade twice over: four spellings is a
/// contract worth stating once, and a `--since 2h` that means one thing in
/// `atgc logs` and another in `atgc search` is the kind of difference nobody
/// reads the help to discover.
pub(crate) fn parse_time(spec: &str, now: DateTime<Utc>) -> Result<DateTime<Utc>> {
    let spec = spec.trim();
    if spec.is_empty() {
        bail!("empty time");
    }

    // A duration ago. Checked first because `3d` is not a prefix of anything
    // else here and the check is cheap.
    //
    // The unit is the last *character*, not the last byte: a byte index lands
    // inside the code point on a value like `2д` and aborts the process in
    // `str::split_at` before anything here can refuse it.
    let mut rest = spec.chars();
    let unit = rest.next_back();
    let digits = rest.as_str();
    if !digits.is_empty()
        && digits.bytes().all(|b| b.is_ascii_digit())
        && let Some(seconds) = match unit {
            Some('s') => Some(1i64),
            Some('m') => Some(60),
            Some('h') => Some(3600),
            Some('d') => Some(86_400),
            Some('w') => Some(604_800),
            _ => None,
        }
    {
        // `checked_mul` and not `n * seconds`: `n` is bounded only by `i64`
        // and `seconds` is up to a week's worth, so the product overflows
        // for a value `parse::<i64>` accepts happily. That is not a
        // theoretical edge — it is `--since 99999999999999999d`, one key
        // held down. Unchecked it panics in a build with overflow checks and
        // wraps in one without, and a wrapped negative delta reads as an
        // instant in the *future*, which filters every event out and reports
        // "no events matched" over a log that has them.
        let delta = digits
            .parse::<i64>()
            .ok()
            .and_then(|n| n.checked_mul(seconds))
            .and_then(chrono::TimeDelta::try_seconds)
            .ok_or_else(|| {
                crate::exit::fail(
                    crate::exit::Exit::Usage,
                    format!("{spec:?} is longer than any time this can name"),
                )
            })?;
        return now.checked_sub_signed(delta).ok_or_else(|| {
            crate::exit::fail(
                crate::exit::Exit::Usage,
                format!("{spec:?} is longer than any time this can name"),
            )
        });
    }

    if let Ok(t) = DateTime::parse_from_rfc3339(spec) {
        return Ok(t.with_timezone(&Utc));
    }

    if let Ok(date) = NaiveDate::parse_from_str(spec, "%Y-%m-%d") {
        return local_to_utc(date.and_time(NaiveTime::MIN), spec);
    }

    for format in ["%H:%M:%S%.f", "%H:%M"] {
        if let Ok(time) = NaiveTime::parse_from_str(spec, format) {
            let today = now.with_timezone(&Local).date_naive();
            return local_to_utc(today.and_time(time), spec);
        }
    }

    // `Usage`, not `Failure`: nothing was read and nothing was missing — the
    // spec on the command line is the problem, and retrying it unchanged
    // fails the same way.
    Err(crate::exit::fail(
        crate::exit::Exit::Usage,
        format!(
            "cannot read {spec:?} as a time. Try a duration ago (30m, 2h, 3d), \
             a clock time today (08:55), a date (2026-08-06), or a full \
             timestamp (2026-08-06T08:55:01Z)"
        ),
    ))
}

/// Interpret a naive local wall-clock instant, tolerating the two days a
/// year on which it is ambiguous or does not exist.
fn local_to_utc(naive: chrono::NaiveDateTime, spec: &str) -> Result<DateTime<Utc>> {
    // A time inside a DST spring-forward gap has no local instant, and one
    // inside an autumn fall-back has two. Taking the earliest of whatever
    // exists is right for a log filter: `--since` wants the widest window
    // that could be meant, and being an hour early costs nothing but a few
    // extra lines.
    Local
        .from_local_datetime(&naive)
        .earliest()
        .map(|t| t.with_timezone(&Utc))
        .with_context(|| format!("{spec:?} is not a real local time (daylight saving?)"))
}

// ===========================================================================
// Colour
// ===========================================================================

/// Colour, gated exactly as `about` gates it: off when stdout is not a
/// terminal, off when `NO_COLOR` is set to anything non-empty, off when
/// `TERM=dumb` says the terminal cannot render escapes at all. Matching that
/// precedent — [`crate::term::hyperlink::stdout_escapes_wanted`] — rather than
/// inventing a second rule is the whole design here.
#[derive(Debug, Clone, Copy)]
pub(super) struct Palette {
    on: bool,
}

impl Palette {
    pub(super) fn detect() -> Self {
        Palette {
            on: crate::term::hyperlink::stdout_escapes_wanted(),
        }
    }

    fn paint(&self, style: anstyle::Style, text: &str) -> String {
        crate::term::style::paint(self.on, style, text)
    }

    /// A palette that paints nothing, for the tests that assert on the text
    /// of a row. Escape sequences in an assertion make a failure unreadable
    /// and the assertion itself a claim about colour rather than content.
    #[cfg(test)]
    pub(in crate::cmd) fn none() -> Self {
        Palette { on: false }
    }

    /// A palette that always paints, for the one test that asserts a
    /// severity reaches the escape sequence rather than only the text.
    #[cfg(test)]
    pub(super) fn colored() -> Self {
        Palette { on: true }
    }

    fn paint_dim(&self, text: &str) -> String {
        self.paint(crate::term::style::DIM, text)
    }

    pub(super) fn dim(&self, text: &str) -> String {
        self.paint_dim(text)
    }
    pub(super) fn bad(&self, text: &str) -> String {
        self.paint(crate::term::style::BAD, text)
    }
    pub(super) fn good(&self, text: &str) -> String {
        self.paint(crate::term::style::GOOD, text)
    }
    pub(super) fn warn(&self, text: &str) -> String {
        self.paint(crate::term::style::WARN, text)
    }
    pub(super) fn note(&self, text: &str) -> String {
        self.paint(crate::term::style::NOTE, text)
    }
    pub(super) fn bold(&self, text: &str) -> String {
        self.paint(crate::term::style::BOLD, text)
    }
}

// ===========================================================================
// Rendering
// ===========================================================================

/// The short label an invocation gets on every line.
///
/// The full id is `<millis-hex>-<pid>-<random>` and runs to about 26
/// characters, which is too much to repeat on every line and too much for the
/// eye to compare two of. The trailing random field is 32 bits seeded from
/// the OS per process — the part that exists precisely so two invocations
/// starting in the same millisecond are distinguishable — so it is the right
/// eight characters to keep. The full id is still printed in the group
/// header, and `--inv` matches on any substring, so a short tag copied off a
/// line is a usable selector.
pub(super) fn short_inv(inv: &str) -> &str {
    match inv.rsplit_once('-') {
        Some((_, tail)) if !tail.is_empty() => tail,
        _ => inv,
    }
}

/// `HH:MM:SS.ffffff`, in UTC as stored.
///
/// Microseconds are not decoration: the question this log has to answer is
/// whether two processes overlapped, and at second resolution they always
/// appear to. The date is deliberately absent from the row and shown in the
/// group header instead, where it costs one line rather than one field per
/// line.
pub(super) fn clock(ts: &DateTime<Utc>) -> String {
    ts.format("%H:%M:%S%.6f").to_string()
}

/// A `client_id` reduced to the two things that decide hypothesis 1.
/// A path with `$HOME` written as `~`.
///
/// Two of these appear on the invocation line of every group and they are
/// long enough to push everything after them off an 80-column terminal.
/// Nothing else is abbreviated, and a path outside `$HOME` — which is where
/// a redirected log lives — passes the path through untouched.
pub(super) fn tilde(path: &str) -> String {
    match std::env::var("HOME") {
        Ok(home) if !home.is_empty() && path.starts_with(&home) => {
            format!("~{}", &path[home.len()..])
        }
        _ => path.to_string(),
    }
}

/// A bounded rendering of a list of removed keys.
pub(super) fn some_of(items: &[String], keep: usize) -> String {
    if items.is_empty() {
        return "none".into();
    }
    if items.len() <= keep {
        return items.join(", ");
    }
    format!("{}, +{} more", items[..keep].join(", "), items.len() - keep)
}

pub(super) fn opt(label: &str, value: &Option<String>) -> String {
    match value {
        Some(v) if !v.is_empty() => format!(" {label}={v}"),
        _ => String::new(),
    }
}

// The payload half of a row — everything after the event's name — is
// `Kind::detail`, one arm per variant in each log. That exhaustiveness is
// the point of deserializing into the real enum: a variant added to a writer
// stops its reader compiling until someone decides how it should read.

/// Colour for the event's own name, by the severity the log assigns it.
fn kind_colored<E: Kind>(event: &E, p: &Palette) -> String {
    let padded = crate::term::column::pad_to(event.name(), 24);
    match event.severity() {
        Severity::Bad => p.bad(&padded),
        Severity::Good => p.good(&padded),
        Severity::Note => p.note(&padded),
        Severity::Plain => padded,
    }
}

/// One row: time, invocation, sequence, kind, payload.
///
/// The invocation tag is on every single row on purpose. Two processes
/// interleaving is one of the two hypotheses this file exists to separate,
/// and a rendering that only names the invocation in a heading makes that
/// interleaving invisible — which is precisely the failure mode `jq` already
/// had.
pub(super) fn row<E: Kind>(entry: &Entry<E>, p: &Palette) -> String {
    format!(
        "{} {} {}{}",
        p.dim(&clock(&entry.ts)),
        p.note(&format!(
            "{}#{:<3}",
            crate::term::column::pad_start(short_inv(&entry.inv), 8),
            entry.seq
        )),
        kind_colored(&entry.event, p),
        {
            let d = entry.event.detail(p);
            // Only on the `invocation` record: one mark per run is
            // orientation, one on every row would be noise.
            let mine = if entry.event.is_head() {
                self_mark(&entry.inv, p)
            } else {
                String::new()
            };
            if d.is_empty() {
                mine
            } else {
                format!(" {d}{mine}")
            }
        }
    )
}

/// `" (this command)"` when the id is this very process's.
///
/// Each writer's `init` puts an `invocation` record down before any command
/// runs, a `logs` command included, so a reader always finds itself as the last
/// block in its own output. Saying so is cheaper than explaining it and
/// far better than filtering it out, which would make the rendering disagree
/// with the file.
fn self_mark(inv: &str, p: &Palette) -> String {
    if crate::logging::file::invocation().id == inv {
        p.dim(" (this command)")
    } else {
        String::new()
    }
}

/// The heading for an invocation's block in grouped mode.
fn header<E: Kind>(inv: &str, head: Option<&Entry<E>>, p: &Palette) -> String {
    match head {
        Some(entry) => format!(
            "{} {}{} {}",
            p.note("──"),
            p.bold(inv),
            self_mark(inv, p),
            entry.event.detail(p)
        ),
        // The `invocation` record is the first line an invocation writes and
        // rotation only ever happens before one, so a missing head means the
        // block is a tail left over on the far side of a rotation — worth
        // saying, because the pid, subcommand and store path are all in that
        // record and their absence is otherwise unexplained.
        None => format!(
            "{} {}{} {}",
            p.note("──"),
            p.bold(inv),
            self_mark(inv, p),
            p.dim("(its invocation record is not in this file: rotated out?)")
        ),
    }
}

/// A dated separator, printed when the day changes.
fn day_line(day: &str, p: &Palette) -> String {
    p.dim(&format!("     ── {day} (times are UTC) ──"))
}

// ===========================================================================
// Reading the files
// ===========================================================================

/// Where the log lives, newest generation last.
///
/// Both generations are read, always, and in the order they were written.
/// The alternative — reading `oauth.jsonl.1` only when a filter seems to
/// reach back far enough — needs a guess about what "far enough" means before
/// any timestamps have been read, and gets it wrong exactly when the log has
/// just rotated, which is when the interesting history is *entirely* in the
/// older file. Both files together are bounded at about 16 MiB by the
/// writer's rotation, so the cost of always reading them is bounded too.
pub(super) fn generations(live: &Path) -> Vec<PathBuf> {
    let mut previous = live.as_os_str().to_owned();
    previous.push(".1");
    let previous = PathBuf::from(previous);
    let mut paths = Vec::new();
    if previous.exists() {
        paths.push(previous);
    }
    paths.push(live.to_path_buf());
    paths
}

/// A line that could not be parsed, kept in file order with the good ones.
#[derive(Debug)]
struct Bad {
    path: PathBuf,
    line_no: usize,
    problem: String,
    excerpt: String,
}

/// Read every generation, parsing what parses and noting what does not.
///
/// A malformed line never stops the read. The writer makes a torn line
/// impossible for records under 4 KiB written to a local filesystem, but that
/// guarantee has documented edges (NFS, a writer killed mid-`write`) and none
/// at all against a person opening the file in an editor, so the reader
/// treats a bad line as data about the log rather than as an error.
fn read_all<E: Kind>(paths: &[PathBuf]) -> (Vec<Entry<E>>, Vec<Bad>, u64) {
    let mut entries = Vec::new();
    let mut bad = Vec::new();
    let mut live_len = 0;
    for (n, path) in paths.iter().enumerate() {
        // Bytes and a lossy decode, not `read_to_string`, which is
        // all-or-nothing on encoding. A writer killed mid-`write` can leave
        // one line truncated inside a multi-byte character — which is one of
        // the two cases the doc above promises to survive — and a strict
        // read turns that single bad byte into "this whole generation is
        // unreadable", losing every well-formed line behind it and printing
        // "nothing has been recorded yet" over a log full of events.
        // Decoding leniently leaves the damage in the one line it belongs
        // to, where `parse_line` reports it as the `Bad` entry it is.
        let raw = match std::fs::read(path) {
            Ok(raw) => raw,
            Err(e) => {
                bad.push(Bad {
                    path: path.clone(),
                    line_no: 0,
                    problem: e.to_string(),
                    excerpt: String::new(),
                });
                continue;
            }
        };
        // The offset `--follow` picks up from is the byte length of the live
        // file as it was when we read it, not a fresh `stat` afterwards:
        // anything appended between the two would be skipped silently. Taken
        // off the bytes rather than off the decoded string, because it is a
        // file offset and a replacement character is not the width of what
        // it replaced.
        if n + 1 == paths.len() {
            live_len = raw.len() as u64;
        }
        let text = String::from_utf8_lossy(&raw);
        for (line_no, line) in text.lines().enumerate() {
            if line.trim().is_empty() {
                continue;
            }
            match parse_line::<E>(line) {
                Ok(entry) => entries.push(entry),
                Err(problem) => bad.push(Bad {
                    path: path.clone(),
                    line_no: line_no + 1,
                    problem,
                    excerpt: crate::logging::file::clip(line, 120),
                }),
            }
        }
    }
    (entries, bad, live_len)
}

/// Two invocations whose spans intersect in time.
///
/// Reported by the grouped rendering, and by the OAuth log's `--incident`
/// analysis, off one implementation rather than two that drift.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct Overlap {
    pub a: String,
    pub b: String,
    pub from: DateTime<Utc>,
    pub to: DateTime<Utc>,
}

/// Each invocation's first and last event, in start order.
pub(super) fn spans_of<'a, E: Kind + 'a>(
    entries: impl IntoIterator<Item = &'a Entry<E>>,
) -> Vec<(&'a str, DateTime<Utc>, DateTime<Utc>)> {
    let mut spans: BTreeMap<&str, (DateTime<Utc>, DateTime<Utc>)> = BTreeMap::new();
    for entry in entries {
        spans
            .entry(entry.inv.as_str())
            .and_modify(|(first, last)| {
                *first = (*first).min(entry.ts);
                *last = (*last).max(entry.ts);
            })
            .or_insert((entry.ts, entry.ts));
    }
    let mut spans: Vec<(&str, DateTime<Utc>, DateTime<Utc>)> = spans
        .into_iter()
        .map(|(inv, (first, last))| (inv, first, last))
        .collect();
    spans.sort_by_key(|(_, first, _)| *first);
    spans
}

/// Pairs of invocations whose spans intersect.
///
/// An invocation's span is bounded by the events it wrote, so this
/// under-reports rather than over-reports: a process that started, sat in a
/// browser consent page for a minute and then wrote its next line looks alive
/// for that whole minute (both endpoints are its own events), but a process
/// that wrote nothing after its `invocation` record looks instantaneous. That
/// direction is the right one for evidence — a reported overlap is real.
pub(super) fn overlaps_from(spans: &[(&str, DateTime<Utc>, DateTime<Utc>)]) -> Vec<Overlap> {
    let mut found = Vec::new();
    for (i, (a, a_first, a_last)) in spans.iter().enumerate() {
        for (b, b_first, b_last) in spans.iter().skip(i + 1) {
            let from = *a_first.max(b_first);
            let to = *a_last.min(b_last);
            if from <= to {
                found.push(Overlap {
                    a: (*a).to_string(),
                    b: (*b).to_string(),
                    from,
                    to,
                });
            }
        }
    }
    found
}

// ===========================================================================
// The command
// ===========================================================================

/// A view of a whole log that only one reader has.
///
/// Named because the signature is otherwise a mouthful in three places, and
/// because it says what the thing is: not a row renderer but something handed
/// every entry at once, which is what an analysis needs and a row never does.
pub(super) type Analysis<'a, E> = &'a dyn Fn(&[Entry<E>], &Palette) -> String;

/// Read one log back: the whole of `atgc logs <name>`.
///
/// `analyze` is the escape hatch for a view only one log has. It is handed
/// the parsed entries and renders the whole output itself, which is what
/// keeps `--incident` out of this file: the alternative was a flag here that
/// the PDS log would have to answer with a refusal.
pub(super) fn run<E: Kind>(
    args: Args,
    source: &Source,
    analyze: Option<Analysis<'_, E>>,
) -> Result<()> {
    let live = match &args.file {
        Some(path) => path.clone(),
        None => match source.log.resolve_path() {
            Some(path) => path,
            None => {
                // Setting the variable to 0 turns the *writer* off. Reading is
                // still a reasonable thing to want — the history from before
                // it was switched off is still on disk — so say what is
                // happening and read the default location anyway.
                let path = source.log.path_in(&crate::config::dir::config_dir()?);
                crate::term::say::note!(
                    Log,
                    "writing is disabled by {}; reading {} anyway",
                    source.log.env_var(),
                    tilde(&path.display().to_string())
                );
                path
            }
        },
    };

    // Said out loud even though this reader consults `args.json` directly:
    // it is what the shared printers read, and a `logs` stdout under `--json`
    // is as reserved as any other command's. See [`write_json_line`].
    crate::term::jsonout::init(args.json);

    let p = Palette::detect();
    let filters = Filters::from_args::<E>(&args, Utc::now())?;
    let limit = if args.all {
        usize::MAX
    } else {
        args.lines.unwrap_or(DEFAULT_LINES)
    };

    let paths = generations(&live);
    let existing: Vec<&PathBuf> = paths.iter().filter(|p| p.exists()).collect();
    if existing.is_empty() {
        return missing::<E>(&live, &args, source, &p);
    }

    let (entries, bad, live_len) = read_all::<E>(&paths);
    let mut out = std::io::stdout().lock();

    if let Some(analyze) = analyze {
        // Filtered, like every other view. An analysis draws conclusions
        // from the *population* it is given — which invocations overlapped,
        // whether more than one login is in the window, how often a refresh
        // token was seen — so handing it the whole file while the command
        // line said `--since 2h` answers a question nobody asked. The
        // incident text itself tells people to narrow with `--since`, and
        // doing so used to change nothing at all.
        //
        // `--lines` deliberately does not apply here: it is a display limit,
        // and a verdict that changed with how many rows fit on a screen
        // would be worse than one drawn from too much.
        let mut entries = entries;
        entries.retain(|e| filters.keep(e));
        write!(out, "{}", analyze(&entries, &p))?;
        return Ok(());
    }

    // Malformed lines are reported before the body rather than in place. They
    // have no timestamp — that is why they are malformed — so there is no
    // honest place to slot them into a time-ordered stream, and a note at the
    // top is what a reader needs to know before believing the rest.
    report_bad(&bad);

    let kept: Vec<&Entry<E>> = entries.iter().filter(|e| filters.keep(e)).collect();
    let shown = &kept[kept.len().saturating_sub(limit)..];

    if args.json {
        for entry in shown {
            write_json_line(entry, &mut out)?;
        }
    } else if shown.is_empty() {
        writeln!(
            out,
            "{}",
            p.dim(&if entries.is_empty() {
                format!(
                    "{} is empty: nothing has been recorded yet",
                    tilde(&live.display().to_string())
                )
            } else {
                format!(
                    "no events matched, out of {} in {} generation{}",
                    entries.len(),
                    existing.len(),
                    if existing.len() == 1 { "" } else { "s" }
                )
            })
        )?;
    } else if args.interleave || args.follow {
        write_interleaved(shown, &mut out, &p)?;
    } else {
        write_grouped(shown, &entries, &mut out, &p)?;
    }

    if args.follow {
        out.flush()?;
        follow::<E>(&live, live_len, &filters, &args, &mut out, &p)?;
    }
    out.flush()?;
    Ok(())
}

/// "No log file yet" is an ordinary state, not an error: it is what every
/// machine looks like until the first command that would write one runs.
fn missing<E: Kind>(live: &Path, args: &Args, source: &Source, p: &Palette) -> Result<()> {
    let shown = tilde(&live.display().to_string());
    if args.follow {
        // Waiting is progress, not an answer, on either side of `--json`:
        // what stdout is about to carry is the events, if any arrive.
        crate::term::say::step!(Log, "waiting for {shown} …");
        let mut out = std::io::stdout().lock();
        let filters = Filters::from_args::<E>(args, Utc::now())?;
        return follow::<E>(live, 0, &filters, args, &mut out, p);
    }
    // Nothing on stdout: the events are the answer, and there are none —
    // which under `--json` is the empty stream a `jq` pipeline expects and
    // which this used to print two sentences of prose into. Why there are
    // none is commentary, and reads the same on either side of the flag.
    crate::term::say::note!(
        Log,
        "no {} log at {shown} yet. {} Set {} to point it elsewhere, or to 0 to turn it off.",
        source.noun,
        source.how_to_fill,
        source.log.env_var()
    );
    Ok(())
}

/// A line of the log that would not parse, on stderr on every flag: under
/// `--json` stdout is a JSON Lines stream something downstream is parsing,
/// and everywhere else it is the rows this line is not one of.
fn report_bad(bad: &[Bad]) {
    for line in bad {
        crate::term::say::warning!(
            Log,
            "unreadable line: {}:{}: {}{}",
            tilde(&line.path.display().to_string()),
            line.line_no,
            line.problem,
            if line.excerpt.is_empty() {
                String::new()
            } else {
                format!("\n{}", line.excerpt)
            }
        );
    }
}

/// One event, as the bytes that were on disk, and a newline.
///
/// # Why this is JSON Lines and not one JSON value
///
/// `docs/output.md`'s second rule is that `--json` puts exactly one value on
/// stdout, and this is the one place in the tool that deliberately does not.
/// The rule is a promise to a caller that it can read to EOF and parse once,
/// and `--follow` cannot keep it: a follow is an unbounded stream that ends
/// when somebody interrupts it, so the closing `]` of an array is a byte that
/// never arrives. A caller that waited for it would hang on the case the flag
/// exists for. Emitting an array without `--follow` and lines with it would
/// be worse than either: the same flag on the same command would return two
/// shapes, and which one a script got would depend on a flag it did not pass.
///
/// So the exception is the whole command, on every flag, and it is written
/// down in `docs/output.md` beside the other four rather than left for a
/// caller to discover.
///
/// # Why the bytes rather than a re-serialization
///
/// [`Entry::raw`] is the line as the writer wrote it. Re-serializing through
/// [`Wire`] would drop every field this build's event enum does not know
/// about — which is precisely the fields a *newer* atgc added, in a log this
/// one is being pointed at to read. Dropping them would be silent, and it
/// would happen exactly during the incident where an old binary is reading a
/// new machine's log. Passing the bytes through cannot lose anything.
///
/// The consequence is that the log record shape is the `--json` schema:
/// [`crate::logging::file`]'s envelope and the three `Event` enums are a
/// public interface, and their doc comments say so.
fn write_json_line<E>(entry: &Entry<E>, out: &mut impl Write) -> Result<()> {
    writeln!(out, "{}", entry.raw)?;
    Ok(())
}

/// Strict file order, one row per event, each tagged with its invocation.
fn write_interleaved<E: Kind>(
    entries: &[&Entry<E>],
    out: &mut impl Write,
    p: &Palette,
) -> Result<()> {
    let mut day = String::new();
    for entry in entries {
        let this_day = entry.ts.format("%Y-%m-%d").to_string();
        if this_day != day {
            writeln!(out, "{}", day_line(&this_day, p))?;
            day = this_day;
        }
        writeln!(out, "{}", row(entry, p))?;
    }
    Ok(())
}

/// Grouped by invocation, in order of first appearance.
///
/// This is the default because the ordinary question is "what did that run
/// do", and a single `atgc` run's events are contiguous only when nothing
/// else was running. The headers come from `all` rather than from the
/// filtered rows, so `--did` or `--failures` still says which invocation each
/// surviving row belongs to.
fn write_grouped<E: Kind>(
    entries: &[&Entry<E>],
    all: &[Entry<E>],
    out: &mut impl Write,
    p: &Palette,
) -> Result<()> {
    let heads: BTreeMap<&str, &Entry<E>> = all
        .iter()
        .filter(|e| e.event.is_head())
        .map(|e| (e.inv.as_str(), e))
        .collect();

    // Groups in order of first appearance in the file, which is very nearly
    // start order and is exactly file order, so a reader comparing this
    // against the raw file is never surprised.
    let mut order: Vec<&str> = Vec::new();
    let mut groups: BTreeMap<&str, Vec<&Entry<E>>> = BTreeMap::new();
    for entry in entries {
        let inv = entry.inv.as_str();
        if !groups.contains_key(inv) {
            order.push(inv);
        }
        groups.entry(inv).or_default().push(entry);
    }

    let mut day = String::new();
    for (n, inv) in order.iter().enumerate() {
        let rows = &groups[inv];
        let this_day = rows[0].ts.format("%Y-%m-%d").to_string();
        if this_day != day {
            writeln!(out, "{}", day_line(&this_day, p))?;
            day = this_day;
        } else if n > 0 {
            writeln!(out)?;
        }
        writeln!(out, "{}", header(inv, heads.get(inv).copied(), p))?;
        // The `invocation` record renders as the heading, so printing it as a
        // row too would say everything twice.
        for entry in rows.iter().filter(|e| !e.event.is_head()) {
            writeln!(out, "{}", row(entry, p))?;
        }
    }

    // Grouping is the one view that can hide the interleaving, so it says so
    // when there is interleaving to hide. This is the cheap half of what
    // `--incident` does properly.
    let overlaps = overlaps_from(&spans_of(entries.iter().copied()));
    if !overlaps.is_empty() {
        writeln!(
            out,
            "\n{}",
            p.warn(&format!(
                "note: {} pair{} of invocations overlap in time: \
                 --interleave shows them in order",
                overlaps.len(),
                if overlaps.len() == 1 { "" } else { "s" }
            ))
        )?;
    }
    Ok(())
}

// ===========================================================================
// Follow
// ===========================================================================

/// Identity of an open file, for noticing that the name now points somewhere
/// else. `None` off Unix, where this reader simply does not detect rotation.
#[cfg(unix)]
fn identity(meta: &std::fs::Metadata) -> Option<(u64, u64)> {
    use std::os::unix::fs::MetadataExt;
    Some((meta.dev(), meta.ino()))
}

#[cfg(not(unix))]
fn identity(_meta: &std::fs::Metadata) -> Option<(u64, u64)> {
    None
}

/// Stream events as they are appended, `tail -f` style.
///
/// Three things can happen to the file underneath and all three are handled
/// rather than hoped against, because the writer rotates at 8 MiB *at process
/// start* — so the moment a long-running follow is most useful, a burst of
/// invocations, is exactly the moment a rotation can land:
///
/// - **Rotation.** The name is re-`stat`ed every poll and its (device, inode)
///   compared against the open handle's. When they differ the old handle is
///   drained one last time — the writer that rotated may have raced with a
///   writer still appending to the old inode — and then the new file is
///   followed from its start. Following the old inode forever is the failure
///   this exists to prevent, and it is silent, which is worse.
/// - **Truncation.** A file shorter than the offset we are reading from was
///   replaced or emptied; the read restarts at zero.
/// - **Absence.** Between the rename and the create there is a window with no
///   file at all, and `--follow` on a machine that has never run an OAuth
///   command starts in that state deliberately.
///
/// A partial final line is held in `pending` until its newline arrives, which
/// is what makes following a writer mid-`write` safe rather than a stream of
/// parse errors.
///
/// One case is detected only by luck, and it is worth naming rather than
/// leaving to be rediscovered: a truncation followed by enough new data to
/// pass the old offset, all inside one poll interval. The length test sees a
/// file longer than where it left off and reads on from there, which lands
/// in the middle of a line. Observed, and the outcome is one in-band
/// `unreadable line` report before the next newline resynchronises the
/// stream — the same thing `tail -f` does, and the reason the reader treats
/// a bad line as data rather than as an error. Closing it properly would
/// mean tracking file content rather than file length, which is a great deal
/// of machinery for a case only `: > oauth.jsonl` and its like produce: atgc
/// itself only ever appends, and only ever rotates by rename.
fn follow<E: Kind>(
    path: &Path,
    from: u64,
    filters: &Filters,
    args: &Args,
    out: &mut impl Write,
    p: &Palette,
) -> Result<()> {
    let mut file: Option<std::fs::File> = None;
    let mut ident: Option<(u64, u64)> = None;
    let mut pos = from;
    let mut pending = String::new();
    let mut announced_absent = false;

    loop {
        // Open, or re-open after a rotation or a period of absence.
        if file.is_none() {
            match std::fs::File::open(path) {
                Ok(mut f) => {
                    ident = f.metadata().ok().and_then(|m| identity(&m));
                    f.seek(SeekFrom::Start(pos))?;
                    if announced_absent {
                        writeln!(
                            out,
                            "{}",
                            p.note(&format!(
                                "── {} appeared",
                                tilde(&path.display().to_string())
                            ))
                        )?;
                        announced_absent = false;
                    }
                    file = Some(f);
                }
                Err(_) => {
                    if !announced_absent {
                        writeln!(
                            out,
                            "{}",
                            p.warn(&format!(
                                "── {} is not there; waiting",
                                tilde(&path.display().to_string())
                            ))
                        )?;
                        out.flush()?;
                        announced_absent = true;
                    }
                    std::thread::sleep(POLL);
                    continue;
                }
            }
        }

        if let Some(f) = file.as_mut() {
            pos += drain::<E>(f, &mut pending, filters, args, out, p)?;
        }
        out.flush()?;

        // Has the name moved to a different file, or been cut short?
        match std::fs::metadata(path) {
            Ok(meta) => {
                let now = identity(&meta);
                if now.is_some() && now != ident {
                    // Drain whatever the old inode still holds before letting
                    // go of it: another process may have appended to it after
                    // the rename and before we noticed.
                    if let Some(f) = file.as_mut() {
                        drain::<E>(f, &mut pending, filters, args, out, p)?;
                    }
                    if !pending.is_empty() {
                        writeln!(
                            out,
                            "{}",
                            p.warn(&format!(
                                "── a partial line was left unfinished in the rotated-out file: {}",
                                crate::logging::file::clip(&pending, 120)
                            ))
                        )?;
                        pending.clear();
                    }
                    writeln!(
                        out,
                        "{}",
                        p.note("── log rotated; following the new file from its start")
                    )?;
                    out.flush()?;
                    file = None;
                    pos = 0;
                } else if meta.len() < pos {
                    writeln!(
                        out,
                        "{}",
                        p.warn("── log got shorter; re-reading from the start")
                    )?;
                    out.flush()?;
                    pending.clear();
                    file = None;
                    pos = 0;
                }
            }
            Err(_) => {
                file = None;
                pos = 0;
                pending.clear();
            }
        }

        std::thread::sleep(POLL);
    }
}

/// Read whatever has been appended, render every whole line, and keep any
/// trailing fragment for next time. Returns the number of bytes consumed.
fn drain<E: Kind>(
    file: &mut std::fs::File,
    pending: &mut String,
    filters: &Filters,
    args: &Args,
    out: &mut impl Write,
    p: &Palette,
) -> Result<u64> {
    let mut buf = Vec::new();
    let read = file.read_to_end(&mut buf)? as u64;
    if read == 0 {
        return Ok(0);
    }
    // Lossy rather than strict: a torn multi-byte character at the read
    // boundary would otherwise abort a follow that is otherwise working, and
    // the line it belongs to will be reported as unreadable anyway.
    pending.push_str(&String::from_utf8_lossy(&buf));
    while let Some(at) = pending.find('\n') {
        let line: String = pending.drain(..=at).collect();
        let line = line.trim_end_matches(['\n', '\r']);
        if line.trim().is_empty() {
            continue;
        }
        match parse_line::<E>(line) {
            Ok(entry) => {
                if !filters.keep(&entry) {
                    continue;
                }
                if args.json {
                    write_json_line(&entry, out)?;
                } else {
                    writeln!(out, "{}", row(&entry, p))?;
                }
            }
            Err(problem) => {
                crate::term::say::warning!(
                    Log,
                    "unreadable line: {problem}\n{}",
                    crate::logging::file::clip(line, 120)
                );
            }
        }
    }
    Ok(read)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::exit::{Exit, classify};
    use chrono::TimeDelta;

    /// A fixed `now`, so what a duration resolves to is arithmetic rather
    /// than a race against the clock the test is running on.
    fn now() -> DateTime<Utc> {
        DateTime::parse_from_rfc3339("2026-08-06T08:55:01Z")
            .expect("the fixture is valid RFC 3339")
            .with_timezone(&Utc)
    }

    /// `atgc logs pds --since 2д` aborted the process. The unit was taken
    /// with a byte index, which lands inside the two bytes of `д` and panics
    /// in `str::split_at` before the value can be refused — every `logs`
    /// subcommand and `search` reach this on both `--since` and `--until`, so
    /// a mistyped keyboard layout crashed the CLI. A spec nobody can read is
    /// a usage error, whatever alphabet it is written in.
    #[test]
    fn a_multi_byte_final_character_is_refused_rather_than_fatal() {
        for spec in ["2д", "30мин", "3日", "2h🙂"] {
            match parse_time(spec, now()) {
                Ok(t) => panic!("{spec} parsed as {t}"),
                Err(err) => assert_eq!(classify(&err), Exit::Usage, "{spec}"),
            }
        }
    }

    /// One invalid byte does not take a whole generation with it.
    ///
    /// `read_to_string` is all-or-nothing on encoding, so a single bad byte
    /// — which is what a writer killed mid-`write` leaves when the cut lands
    /// inside a multi-byte character, one of the two cases `read_all`'s own
    /// doc promises to survive — used to collapse the entire file into one
    /// `Bad` entry. Every well-formed line in it vanished, and the command
    /// printed "nothing has been recorded yet" over a log full of events.
    ///
    /// Driven through a real file because `read_all` takes paths; that is
    /// also what makes the `live_len` half checkable, and `live_len` is the
    /// offset `--follow` resumes from, so a zero there reprints the file.
    #[test]
    fn one_invalid_byte_does_not_discard_the_lines_around_it() {
        let dir = tempfile::tempdir().expect("a scratch dir");
        let path = dir.path().join("git.jsonl");
        let good = br#"{"ts":"2026-08-06T08:55:01Z","inv":"abc","seq":1,"event":"invocation","pid":1,"version":"0","subcommand":"pr"}"#;
        let mut raw = Vec::new();
        raw.extend_from_slice(good);
        raw.push(b'\n');
        // A line cut inside a two-byte character: the lead byte with nothing
        // after it.
        raw.extend_from_slice(b"{\"ts\":\"2026-08-06T08:55:02Z\",\"inv\":\"ab\xd0");
        raw.push(b'\n');
        raw.extend_from_slice(good);
        raw.push(b'\n');
        std::fs::write(&path, &raw).expect("write the fixture");

        let (entries, bad, live_len) =
            read_all::<crate::logging::git::Event>(std::slice::from_ref(&path));
        assert_eq!(
            entries.len(),
            2,
            "the good lines were lost with the bad one"
        );
        assert_eq!(bad.len(), 1, "the damaged line should be reported, once");
        assert_eq!(
            live_len,
            raw.len() as u64,
            "the follow offset must be the file's own byte length"
        );
    }

    /// A duration too large to represent is refused, not multiplied.
    ///
    /// The same shape as the byte-split above, one arithmetic operation
    /// later: `n` is bounded only by `i64` and the unit multiplies it, so a
    /// value `parse::<i64>` accepts can still overflow the product. This
    /// build has overflow checks on and would panic; a release build has
    /// them off and wraps, and a wrapped negative delta subtracts to an
    /// instant in the *future*, so `--since` filters out every event and
    /// reports "no events matched" over a log that has them. Both are worse
    /// than the refusal every other unreadable spec gets.
    #[test]
    fn a_duration_too_large_to_represent_is_refused_rather_than_wrapped() {
        for spec in [
            "99999999999999999d",
            "9223372036854775807w",
            "10000000000000000h",
        ] {
            match parse_time(spec, now()) {
                Ok(t) => panic!("{spec} parsed as {t}"),
                Err(err) => assert_eq!(classify(&err), Exit::Usage, "{spec}"),
            }
        }
    }

    /// And the durations that do fit still work, in every unit.
    ///
    /// The guard above is a `checked_mul` in the middle of the one path that
    /// answers `--since 2h`, so it is worth proving it did not narrow what
    /// the flag accepts.
    #[test]
    fn the_durations_that_fit_are_unaffected() {
        let now = now();
        for (spec, secs) in [
            ("30s", 30),
            ("30m", 1800),
            ("2h", 7200),
            ("3d", 259_200),
            ("1w", 604_800),
        ] {
            let parsed = parse_time(spec, now).unwrap_or_else(|e| panic!("{spec}: {e}"));
            assert_eq!(
                now - parsed,
                TimeDelta::try_seconds(secs).unwrap(),
                "{spec}"
            );
        }
    }

    /// The same split with nothing in front of the unit: a single non-ASCII
    /// character is one byte short of a boundary all on its own, and `--since
    /// ✓` panicked where `--since x` had always refused.
    #[test]
    fn a_lone_non_ascii_character_is_refused() {
        for spec in ["✓", "д", "日", "🙂"] {
            match parse_time(spec, now()) {
                Ok(t) => panic!("{spec} parsed as {t}"),
                Err(err) => assert_eq!(classify(&err), Exit::Usage, "{spec}"),
            }
        }
    }

    /// What a refused spec is owed: the value back, so a shell that mangled
    /// it is visible, and the four spellings, so the next attempt is a
    /// correction rather than a guess. The non-ASCII path has to arrive here
    /// too — the fix is worth nothing if it trades a panic for a bare error.
    #[test]
    fn the_refusal_quotes_the_spec_and_names_the_alternatives() {
        for spec in ["garbage", "2д", "✓"] {
            let message = parse_time(spec, now()).expect_err("not a time").to_string();
            assert!(message.contains(&format!("{spec:?}")), "{message}");
            for hint in ["30m", "08:55", "2026-08-06", "2026-08-06T08:55:01Z"] {
                assert!(message.contains(hint), "{hint} missing from {message}");
            }
        }
    }

    /// The four spellings the doc comment and the refusal both promise. They
    /// are a contract two commands rely on — `logs` and `search` share this
    /// parser precisely so that `--since 2h` cannot mean two things — and
    /// nothing else in the tree pins them.
    #[test]
    fn the_four_documented_spellings_all_parse() {
        let now = now();

        // A duration ago, in each unit the parser accepts.
        for (spec, seconds) in [
            ("45s", 45),
            ("30m", 1_800),
            ("2h", 7_200),
            ("3d", 259_200),
            ("1w", 604_800),
        ] {
            let t = parse_time(spec, now).unwrap_or_else(|e| panic!("{spec}: {e}"));
            assert_eq!(now - t, TimeDelta::try_seconds(seconds).unwrap(), "{spec}");
        }

        // A wall-clock time today, which is what a person reads off a
        // screenshot, interpreted in their own zone rather than the log's.
        let t = parse_time("08:55", now).expect("a clock time today parses");
        let local = t.with_timezone(&Local);
        assert_eq!(local.date_naive(), now.with_timezone(&Local).date_naive());
        assert_eq!(local.time(), NaiveTime::from_hms_opt(8, 55, 0).unwrap());

        // A date, meaning local midnight that morning.
        let t = parse_time("2026-08-06", now).expect("a date parses");
        let local = t.with_timezone(&Local);
        assert_eq!(
            local.date_naive(),
            NaiveDate::from_ymd_opt(2026, 8, 6).unwrap()
        );
        assert_eq!(local.time(), NaiveTime::MIN);

        // Full RFC 3339, which is what the log prints and so what a
        // copy-paste out of one produces. `Z` is unambiguous: no local zone.
        assert_eq!(
            parse_time("2026-08-06T08:55:01Z", now).expect("a timestamp parses"),
            now
        );
    }
}