pitchfork_cli/cli/

logs.rs

use crate::cli::json_output::{JsonLogEntry, print_json};
use crate::daemon_id::DaemonId;
use crate::log_store::sqlite::LOG_STORE;
use crate::log_store::{FieldFilter, LogEntry, LogQuery, LogStore, MessageFilter};
use crate::pitchfork_toml::PitchforkToml;
use crate::settings::settings;
use crate::state_file::StateFile;
use crate::ui::style::{edim, estyle, ndim};
use crate::{Result, env};
use chrono::{DateTime, Local, NaiveDateTime, NaiveTime, TimeZone};
use console;
use itertools::Itertools;
use miette::IntoDiagnostic;
use std::collections::BTreeSet;
use std::fmt::Write as _;
use std::io::{self, IsTerminal, Write};
use std::process::{Child, Command, Stdio};
use std::time::Duration;

/// Pager configuration for displaying logs
struct PagerConfig {
    command: String,
    args: Vec<String>,
}

impl PagerConfig {
    /// Select and configure the appropriate pager.
    /// Uses $PAGER environment variable if set, otherwise defaults to less.
    fn new(start_at_end: bool) -> Self {
        let command = std::env::var("PAGER").unwrap_or_else(|_| "less".to_string());
        let args = Self::build_args(&command, start_at_end);
        Self { command, args }
    }

    fn build_args(pager: &str, start_at_end: bool) -> Vec<String> {
        let mut args = vec![];
        if pager == "less" {
            args.push("-R".to_string());
            if start_at_end {
                args.push("+G".to_string());
            }
        }
        args
    }

    /// Spawn the pager with piped stdin
    fn spawn_piped(&self) -> io::Result<Child> {
        Command::new(&self.command)
            .args(&self.args)
            .stdin(Stdio::piped())
            .spawn()
    }
}

/// Fields already displayed in dedicated positions; excluded from the
/// extra-fields section of structured log output.
const KNOWN_FIELD_KEYS: &[&str] = &[
    "level",
    "severity",
    "lvl",
    "PRIORITY",
    "@level",
    "msg",
    "message",
    "event",
    "@message",
    "logger",
    "name",
    "component",
    "module",
    "timestamp",
    "ts",
    "time",
    "@timestamp",
];

/// Format a log level badge: `[ERR]`, `[WRN|`, etc.
///
/// Brackets use the same dim color as the logger; the level letters use the
/// level's accent color in bold.
fn level_badge(level: &str) -> String {
    let label = match level {
        "error" => console::style("ERR").red().bold(),
        "warn" => console::style("WRN").yellow().bold(),
        "info" => console::style("INF").cyan().bold(),
        "debug" => console::style("DBG").magenta().bold(),
        "trace" => console::style("TRC").dim().bold(),
        _ => return String::new(),
    };
    format!("{}{}{}", ndim("["), label, ndim("]"))
}

/// Format a single JSON value for display in the fields section.
///
/// Strings that look like booleans, null, numbers, or contain spaces are
/// quoted to avoid ambiguity. Numbers are green, booleans are
/// yellow/red, null is dim.
fn format_field_value(value: &serde_json::Value) -> String {
    match value {
        serde_json::Value::String(s) => {
            let needs_quotes = s.is_empty()
                || s.contains(' ')
                || s.contains('=')
                || s.contains('\t')
                || s == "true"
                || s == "false"
                || s == "null"
                || s.parse::<f64>().is_ok();
            if needs_quotes {
                format!("\"{s}\"")
            } else {
                s.clone()
            }
        }
        serde_json::Value::Number(n) => console::style(n).green().to_string(),
        serde_json::Value::Bool(true) => console::style("true").yellow().to_string(),
        serde_json::Value::Bool(false) => console::style("false").red().to_string(),
        serde_json::Value::Null => console::style("null").dim().to_string(),
        serde_json::Value::Object(_) | serde_json::Value::Array(_) => {
            serde_json::to_string(value).unwrap_or_else(|_| value.to_string())
        }
    }
}

/// Format structured fields as `key=value` pairs, excluding known keys.
fn format_fields(fields_json: &str) -> String {
    let Ok(serde_json::Value::Object(obj)) = serde_json::from_str(fields_json) else {
        return String::new();
    };
    let mut parts = Vec::new();
    for (key, value) in &obj {
        if KNOWN_FIELD_KEYS.contains(&key.as_str()) {
            continue;
        }
        let key_styled = console::style(key.as_str()).blue().to_string();
        let value_styled = format_field_value(value);
        parts.push(format!("{key_styled}={value_styled}"));
    }
    parts.join(" ")
}

/// Format and write a single log entry with structured field highlighting.
///
/// When `raw` is set, the original message is emitted verbatim (after ANSI
/// stripping if requested), ignoring any structured fields. This keeps `--raw`
/// honored even when `--jq` populates `fields_json` for filtering.
///
/// Otherwise, when the entry has structured fields (`fields_json` is `Some`),
/// renders as: `<timestamp> [LEVEL] logger: message > key=value key=value`
/// For warn/error levels, the timestamp and message use the level's accent color.
///
/// Falls back to plain display: `<timestamp> message`
#[allow(clippy::too_many_arguments)]
fn write_formatted_log(
    w: &mut dyn Write,
    entry: &LogEntry,
    date: &str,
    single_daemon: bool,
    strip_ansi: bool,
    show_timestamp: bool,
    raw: bool,
) -> io::Result<()> {
    // Always strip PTY control sequences (cursor moves, clear screen, etc.)
    // while preserving SGR color codes. PTY daemons emit these sequences which
    // can corrupt pager output (e.g. triggering page breaks in `less`).
    let clean_msg = strip_pty_controls(&entry.message);
    let raw_msg: std::borrow::Cow<'_, str> = if strip_ansi {
        console::strip_ansi_codes(&clean_msg)
    } else {
        std::borrow::Cow::Owned(clean_msg)
    };

    let mut out = String::with_capacity(256);

    if raw {
        // Raw mode: emit timestamp + original line verbatim. Structured
        // fields may still be populated (e.g. --jq needs them for
        // filtering) but must not alter the raw output format.
        if show_timestamp {
            out.push_str(&ndim(date).to_string());
            out.push(' ');
        }
        if !single_daemon {
            let colors_on = !strip_ansi && console::colors_enabled();
            out.push_str(&colored_id_label(&entry.daemon_id, colors_on));
            out.push(' ');
        }
        out.push_str(&raw_msg);
    } else if entry.fields_json.is_some() {
        // Structured display: [LEVEL] logger: message > key=value
        let level = entry.level.as_deref();
        let accent = match level {
            Some("error") => "error",
            Some("warn") => "warn",
            _ => "",
        };

        // Timestamp — accent-colored for warn/error
        if show_timestamp {
            let ts = match accent {
                "error" => console::style(date).red().to_string(),
                "warn" => console::style(date).yellow().to_string(),
                _ => ndim(date).to_string(),
            };
            out.push_str(&ts);
            out.push(' ');
        }

        // Daemon ID (multi-daemon mode)
        if !single_daemon {
            let colors_on = !strip_ansi && console::colors_enabled();
            out.push_str(&colored_id_label(&entry.daemon_id, colors_on));
            out.push(' ');
        }

        let mut need_sep = false;

        // Level badge: [ERR] with dim brackets + bold accent letters
        if let Some(lvl) = level {
            let badge = level_badge(lvl);
            if !badge.is_empty() {
                out.push_str(&badge);
                need_sep = true;
            }
        }

        // Separators (colon, arrow): always dim regardless of level
        let sep = ndim(":").to_string();
        let arrow = ndim(" > ").to_string();

        // Logger: italic + colon
        if let Some(logger) = &entry.logger {
            if need_sep {
                out.push(' ');
            }
            out.push_str(&console::style(logger).italic().dim().to_string());
            out.push_str(&sep);
            need_sep = true;
        }

        // Message (bold, accent-colored for warn/error) + fields
        let msg_cow = entry.msg.as_deref().filter(|s| !s.is_empty()).map(|s| {
            let cleaned = strip_pty_controls(s);
            if strip_ansi {
                std::borrow::Cow::<str>::Owned(console::strip_ansi_codes(&cleaned).to_string())
            } else {
                std::borrow::Cow::<str>::Owned(cleaned)
            }
        });
        let fields_str = entry
            .fields_json
            .as_deref()
            .map(format_fields)
            .filter(|s| !s.is_empty());

        if msg_cow.is_some() || fields_str.is_some() {
            if need_sep {
                out.push(' ');
            }
            if let Some(msg) = &msg_cow {
                let styled = match accent {
                    "error" => console::style(msg.as_ref()).red().bold().to_string(),
                    "warn" => console::style(msg.as_ref()).yellow().bold().to_string(),
                    _ => console::style(msg.as_ref()).bold().to_string(),
                };
                out.push_str(&styled);
                if fields_str.is_some() {
                    out.push_str(&arrow);
                }
            }
            if let Some(fields) = &fields_str {
                out.push_str(fields);
            }
        } else if !need_sep {
            // No level badge, logger, message, or fields — fall back to
            // the raw message so the line isn't empty.
            out.push_str(&raw_msg);
        }
    } else {
        // Unstructured log: timestamp + raw message
        if show_timestamp {
            out.push_str(&ndim(date).to_string());
            out.push(' ');
        }
        if !single_daemon {
            let colors_on = !strip_ansi && console::colors_enabled();
            out.push_str(&colored_id_label(&entry.daemon_id, colors_on));
            out.push(' ');
        }
        out.push_str(&raw_msg);
    }

    out.push('\n');
    w.write_all(out.as_bytes())
}

/// Return a colorized `[namespace/id]` label for display in progress jobs.
/// Uses brighter colors than `dimmed_id` and includes the square brackets.
pub fn colored_id_label(id: &str, colors_enabled: bool) -> String {
    if !colors_enabled {
        return format!("[{}]", id);
    }
    // Same palette as mise: Blue, Magenta, Cyan, Green
    // Excludes Red/Yellow to avoid confusion with errors/warnings.
    let colors: [u8; 4] = [34, 35, 36, 32]; // ANSI: Blue, Magenta, Cyan, Green
    let mut h: usize = 0x811C_9DC5; // FNV offset basis
    for b in id.bytes() {
        h = h.wrapping_mul(0x0100_0193).wrapping_add(b as usize);
    }
    let color = colors[h % colors.len()];
    format!("\x1b[{color}m[{id}]\x1b[0m")
}

/// Displays logs for daemon(s)
#[derive(Debug, usage_rs::Args)]
#[usage(
    verbatim_doc_comment,
    long_about = "\
Displays logs for daemon(s)

Shows logs from managed daemons. Logs are stored in the pitchfork logs directory
and include timestamps for filtering.

Examples:

    pitchfork logs api              Show all logs for 'api' (paged if needed)
    pitchfork logs api worker       Show logs for multiple daemons
    pitchfork logs                  Show logs for all daemons
    pitchfork logs api -n 50        Show last 50 lines
    pitchfork logs api --follow     Follow logs in real-time
    pitchfork logs api --since '2024-01-15 10:00:00'
                                    Show logs since a specific time (forward)
    pitchfork logs api --since '10:30:00'
                                    Show logs since 10:30:00 today
    pitchfork logs api --since '10:30' --until '12:00'
                                    Show logs since 10:30:00 until 12:00:00 today
    pitchfork logs api --since 5min Show logs from last 5 minutes
    pitchfork logs api --raw        Output raw log lines without formatting
    pitchfork logs api --raw -n 100 Output last 100 raw log lines
    pitchfork logs api --clear      Delete logs for 'api'
    pitchfork logs --clear          Delete logs for all daemons"
)]
pub struct Logs {
    /// Show only logs for the specified daemon(s)
    id: Vec<String>,

    /// Delete logs
    #[usage(short, long, effect = "destructive")]
    clear: bool,

    /// Show last N lines of logs
    ///
    /// With --since/--until, shows the last N matching lines in that time range.
    /// Without this option, all logs are shown.
    #[usage(short)]
    n: Option<usize>,

    /// Show logs in real-time
    #[usage(short = 't', short = 'f', long, visible_alias = "follow")]
    tail: bool,

    /// Show logs from this time
    ///
    /// Supports multiple formats:
    /// - Full datetime: "YYYY-MM-DD HH:MM:SS" or "YYYY-MM-DD HH:MM"
    /// - Time only: "HH:MM:SS" or "HH:MM" (uses today's date)
    /// - Relative time: "5min", "2h", "1d" (e.g., last 5 minutes)
    #[usage(short = 's', long)]
    since: Option<String>,

    /// Show logs until this time
    ///
    /// Supports multiple formats:
    /// - Full datetime: "YYYY-MM-DD HH:MM:SS" or "YYYY-MM-DD HH:MM"
    /// - Time only: "HH:MM:SS" or "HH:MM" (uses today's date)
    #[usage(short = 'u', long)]
    until: Option<String>,

    /// Disable pager even in interactive terminal
    #[usage(long)]
    no_pager: bool,

    /// Output raw log lines without color or formatting
    #[usage(long)]
    raw: bool,

    /// Output in JSON format
    #[usage(long, conflicts = "raw", conflicts = "tail")]
    json: bool,

    /// Filter logs by case-insensitive substring (can be repeated)
    ///
    /// Multiple --grep options are combined with OR.
    #[usage(long)]
    grep: Vec<String>,

    /// Filter logs by regular expression
    #[usage(long)]
    regex: Option<String>,

    /// Make --grep matching case-sensitive
    #[usage(long)]
    case_sensitive: bool,

    /// Filter by minimum log level (error, warn, info, debug, trace)
    ///
    /// Shows entries at or above the given severity.
    /// For example, `--level warn` shows warn and error.
    /// Only effective for daemons with log_format json or logfmt.
    #[usage(long)]
    level: Option<String>,

    /// Filter by structured field value (KEY=VALUE, can be repeated)
    ///
    /// Extracts the value from fields_json using json_extract($.KEY).
    /// Multiple --field options are combined with AND.
    #[usage(long, value_name = "KEY=VALUE")]
    field: Vec<String>,

    /// Filter log entries with a jq expression
    ///
    /// Each log entry is serialized as a JSON object with fields:
    /// timestamp, daemon_id, message, level, msg, logger, fields.
    /// Entries for which the expression produces a truthy value are shown.
    #[usage(long, value_name = "EXPR")]
    jq: Option<String>,

    /// Omit timestamps from log output
    #[usage(long)]
    no_timestamp: bool,
}

impl Logs {
    pub async fn run(&self) -> Result<()> {
        migrate_legacy_log_dirs();

        let resolved_ids: Vec<DaemonId> = if self.id.is_empty() {
            get_all_daemon_ids()?
        } else {
            PitchforkToml::resolve_ids(&self.id)?
        };

        if self.clear {
            LOG_STORE.clear(&resolved_ids)?;
            return Ok(());
        }

        let from = if let Some(since) = self.since.as_ref() {
            Some(parse_time_input(since, true)?)
        } else {
            None
        };
        let to = if let Some(until) = self.until.as_ref() {
            Some(parse_time_input(until, false)?)
        } else {
            None
        };

        let message_filters = self.build_message_filters()?;
        let field_filters = self.build_field_filters()?;

        // Compile jq filter early so parse errors surface before any query.
        let jq_filter = match self.jq.as_deref() {
            Some(expr) => Some(crate::log_jq::JqFilter::new(expr)?),
            None => None,
        };

        if self.json {
            return self.output_json(
                &resolved_ids,
                from,
                to,
                message_filters,
                field_filters,
                jq_filter.as_ref(),
            );
        }

        // Only suppress the daemon id label when the user explicitly asked
        // for a single daemon. When no daemon is named on the command line
        // (e.g. `pf logs`), ids are shown even if only one daemon has logs,
        // so the user can tell where each line came from.
        let single_daemon = resolved_ids.len() == 1 && !self.id.is_empty();
        let show_timestamp = settings().logs.timestamp && !self.no_timestamp && !self.raw;
        let has_time_filter = from.is_some() || to.is_some();

        self.query_and_output(
            &resolved_ids,
            from,
            to,
            message_filters.clone(),
            field_filters.clone(),
            jq_filter.as_ref(),
            single_daemon,
            has_time_filter,
            show_timestamp,
        )?;

        if self.tail {
            tail_logs(
                &resolved_ids,
                single_daemon,
                true,
                message_filters,
                field_filters,
                jq_filter.as_ref(),
                show_timestamp,
                self.raw,
            )
            .await?;
        }

        Ok(())
    }

    /// The row limit for the SQL query. `-n` is applied there only when
    /// nothing filters the rows afterwards: a time filter trims to the last N
    /// once the rows are in chronological order, and a jq filter runs after
    /// the query, so limiting first would show only the matches among the
    /// last N entries rather than the last N matches.
    fn query_limit(&self, has_time_filter: bool, has_jq_filter: bool) -> Option<usize> {
        if has_time_filter || has_jq_filter {
            None
        } else {
            self.n
        }
    }

    /// Run `opts` and apply the jq filter, if any: newest first without a
    /// time filter, chronological with one.
    ///
    /// With `--jq` and `-n` the last N matches are wanted, but jq runs after
    /// the query, so the query reads a window of the newest rows that grows
    /// until it holds N matches or covers them all (see [`last_matches`]).
    fn query_entries(
        &self,
        mut opts: LogQuery,
        jq_filter: Option<&crate::log_jq::JqFilter>,
        has_time_filter: bool,
    ) -> Result<Vec<LogEntry>> {
        let Some(jq) = jq_filter else {
            return LOG_STORE.query(&opts);
        };
        match self.n {
            Some(n) if !has_time_filter => last_matches(
                n,
                |window| {
                    opts.limit = Some(window);
                    LOG_STORE.query(&opts)
                },
                |rows| jq.filter(rows),
            ),
            _ => Ok(jq.filter(LOG_STORE.query(&opts)?)),
        }
    }

    fn build_message_filters(&self) -> Result<Vec<MessageFilter>> {
        if self.case_sensitive && self.grep.is_empty() {
            warn!("--case-sensitive has no effect without --grep");
        }
        let mut filters = Vec::new();
        for pattern in &self.grep {
            filters.push(MessageFilter::Contains {
                pattern: pattern.clone(),
                case_sensitive: self.case_sensitive,
            });
        }
        if let Some(pattern) = self.regex.as_ref() {
            // Validate the regex early so the user gets a clear CLI error
            // instead of a SQLite user-function failure at query time.
            let _ = regex::Regex::new(pattern)
                .into_diagnostic()
                .map_err(|e| miette::miette!("invalid regex pattern: {e}"))?;
            filters.push(MessageFilter::Regex {
                pattern: pattern.clone(),
            });
        }
        Ok(filters)
    }

    fn build_field_filters(&self) -> Result<Vec<FieldFilter>> {
        let mut filters = Vec::new();
        if let Some(level) = self.level.as_ref() {
            // Normalize the user-supplied level so "ERROR", "Error", "fatal"
            // all match the stored canonical form.
            let normalized = crate::log_parse::normalize_level_str(level).ok_or_else(|| {
                miette::miette!(
                    "invalid level '{level}'; expected one of: \
                         error, err, fatal, critical, panic, alert, emerg, \
                         warn, warning, info, inf, information, notice, \
                         debug, dbg, trace, trc"
                )
            })?;
            filters.push(FieldFilter::LevelMin(normalized));
        }
        for pair in &self.field {
            let (key, value) = pair
                .split_once('=')
                .ok_or_else(|| miette::miette!("--field expects KEY=VALUE, got: {pair}"))?;
            if key.is_empty() {
                miette::bail!("--field key cannot be empty: {pair}");
            }
            // Whitelist key characters to prevent SQL injection via
            // json_extract path interpolation.
            if !key
                .chars()
                .all(|c| c.is_alphanumeric() || c == '_' || c == '.')
            {
                miette::bail!(
                    "--field key may only contain alphanumeric, underscore, and dot; got: {key}"
                );
            }
            filters.push(FieldFilter::FieldEq {
                key: key.to_string(),
                value: value.to_string(),
            });
        }
        Ok(filters)
    }

    /// Query log entries and stream them directly to output, bypassing the
    /// intermediate `Vec<(String, String, String)>` layer. This eliminates
    /// 3 String allocations per row (timestamp format, daemon_id clone,
    /// message clone) by borrowing directly from `LogEntry` and reusing a
    /// single date buffer.
    #[allow(clippy::too_many_arguments)]
    fn query_and_output(
        &self,
        resolved_ids: &[DaemonId],
        from: Option<DateTime<Local>>,
        to: Option<DateTime<Local>>,
        message_filters: Vec<MessageFilter>,
        field_filters: Vec<FieldFilter>,
        jq_filter: Option<&crate::log_jq::JqFilter>,
        single_daemon: bool,
        has_time_filter: bool,
        show_timestamp: bool,
    ) -> Result<()> {
        let daemon_ids: Vec<String> = resolved_ids.iter().map(|id| id.qualified()).collect();

        let opts = LogQuery {
            daemon_ids,
            from,
            to,
            limit: self.query_limit(has_time_filter, jq_filter.is_some()),
            order_desc: !has_time_filter,
            after_id: None,
            before_id: None,
            message_filters,
            field_filters,
            include_structured: jq_filter.is_some() || !self.raw,
        };
        let mut entries = self.query_entries(opts, jq_filter, has_time_filter)?;

        // Apply time-filter take-last-N or reverse for chronological display.
        if has_time_filter {
            if let Some(n) = self.n {
                let len = entries.len();
                if len > n {
                    entries = entries.split_off(len - n);
                }
            }
        } else {
            entries.reverse();
        }

        if entries.is_empty() {
            return Ok(());
        }

        let strip_ansi = self.raw || !console::colors_enabled();
        let use_pager = !self.tail && !self.no_pager && should_use_pager(entries.len());
        let ts_format = &settings().logs.timestamp_format;

        // Reusable buffer for timestamp formatting — avoids N allocations.
        let mut date_buf = String::with_capacity(ts_format.len() + 6);

        let mut write_entries = |w: &mut dyn Write| -> io::Result<()> {
            for entry in &entries {
                date_buf.clear();
                write!(date_buf, "{}", entry.timestamp.format(ts_format))
                    .map_err(io::Error::other)?;
                write_formatted_log(
                    w,
                    entry,
                    &date_buf,
                    single_daemon,
                    strip_ansi,
                    show_timestamp,
                    self.raw,
                )?;
            }
            Ok(())
        };

        if use_pager {
            let pager_config = PagerConfig::new(!has_time_filter);
            match pager_config.spawn_piped() {
                Ok(mut child) => {
                    if let Some(stdin) = child.stdin.as_mut() {
                        if let Err(e) = write_entries(stdin) {
                            debug!("pager write error: {e}");
                        }
                        let _ = child.wait();
                    } else {
                        debug!("Failed to get pager stdin, falling back to direct output");
                        let stdout = io::stdout();
                        let mut buf = io::BufWriter::new(stdout.lock());
                        write_entries(&mut buf).into_diagnostic()?;
                    }
                }
                Err(e) => {
                    debug!("Failed to spawn pager: {e}, falling back to direct output");
                    let stdout = io::stdout();
                    let mut buf = io::BufWriter::new(stdout.lock());
                    write_entries(&mut buf).into_diagnostic()?;
                }
            }
        } else {
            let stdout = io::stdout();
            let mut buf = io::BufWriter::new(stdout.lock());
            write_entries(&mut buf).into_diagnostic()?;
        }

        Ok(())
    }

    fn output_json(
        &self,
        resolved_ids: &[DaemonId],
        from: Option<DateTime<Local>>,
        to: Option<DateTime<Local>>,
        message_filters: Vec<MessageFilter>,
        field_filters: Vec<FieldFilter>,
        jq_filter: Option<&crate::log_jq::JqFilter>,
    ) -> Result<()> {
        let daemon_ids: Vec<String> = resolved_ids.iter().map(|id| id.qualified()).collect();
        let has_time_filter = from.is_some() || to.is_some();

        let opts = LogQuery {
            daemon_ids,
            from,
            to,
            limit: self.query_limit(has_time_filter, jq_filter.is_some()),
            order_desc: !has_time_filter,
            after_id: None,
            before_id: None,
            message_filters,
            field_filters,
            include_structured: true,
        };
        let entries = self.query_entries(opts, jq_filter, has_time_filter)?;

        // Reverse only when the SQL query returned DESC (no time filter),
        // to produce chronological (oldest-first) output. With a time
        // filter, the query is ASC and already chronological.
        let mut entries: Vec<_> = if has_time_filter {
            entries
        } else {
            entries.into_iter().rev().collect()
        };

        // When a time filter is active, the SQL query returned all matching
        // entries without LIMIT. Trim to the last N here.
        if has_time_filter
            && let Some(n) = self.n
            && entries.len() > n
        {
            entries = entries.split_off(entries.len() - n);
        }

        let json_entries: Vec<JsonLogEntry> = entries.into_iter().map(Into::into).collect();

        print_json(&json_entries)
    }
}

/// The smallest window of newest rows [`last_matches`] reads.
const MIN_MATCH_WINDOW: usize = 1000;

/// The last `n` rows that `filter` keeps, newest first, reading no more of
/// the history than needed: `fetch(window)` returns the newest `window` rows,
/// and the window doubles until it holds `n` matches or all the rows.
///
/// Each read starts again from the newest row rather than paging on, so the
/// rows keep exactly the order the query gives them.
fn last_matches<T>(
    n: usize,
    mut fetch: impl FnMut(usize) -> Result<Vec<T>>,
    filter: impl Fn(Vec<T>) -> Vec<T>,
) -> Result<Vec<T>> {
    let mut window = n.max(MIN_MATCH_WINDOW);
    loop {
        let rows = fetch(window)?;
        let exhausted = rows.len() < window;
        let mut matches = filter(rows);
        if matches.len() >= n || exhausted {
            matches.truncate(n);
            return Ok(matches);
        }
        window = window.saturating_mul(2);
    }
}

fn should_use_pager(line_count: usize) -> bool {
    if !io::stdout().is_terminal() {
        return false;
    }

    let terminal_height = get_terminal_height().unwrap_or(24);
    line_count > terminal_height
}

fn get_terminal_height() -> Option<usize> {
    if let Ok(rows) = std::env::var("LINES")
        && let Ok(h) = rows.parse::<usize>()
    {
        return Some(h);
    }

    crossterm::terminal::size().ok().map(|(_, h)| h as usize)
}

/// Rename legacy log directories that predate namespace-qualified daemon IDs.
///
/// Old layout: `PITCHFORK_LOGS_DIR/<name>/<name>.log`
/// New layout: `PITCHFORK_LOGS_DIR/legacy--<name>/legacy--<name>.log`
///
/// Only directories that clearly match the old layout are migrated:
/// - directory name does not contain `"--"`
/// - directory contains `<name>.log`
/// - `<name>` is a valid daemon short name under current DaemonId rules
fn migrate_legacy_log_dirs() {
    let known_safe_paths = known_daemon_safe_paths();
    let dirs = match xx::file::ls(&*env::PITCHFORK_LOGS_DIR) {
        Ok(d) => d,
        Err(_) => return,
    };
    for dir in dirs {
        if dir.starts_with(".") || !dir.is_dir() {
            continue;
        }
        let name = match dir.file_name().map(|f| f.to_string_lossy().to_string()) {
            Some(n) => n,
            None => continue,
        };
        // Skip the supervisor's own log directory.
        if name == "pitchfork" {
            continue;
        }
        // New-format directories usually contain "--". For safety, only treat
        // them as new-format if they match a known daemon ID safe-path.
        if name.contains("--") {
            // If it parses as a valid safe-path, treat it as already migrated
            // and keep idempotent behavior silent.
            if DaemonId::from_safe_path(&name).is_ok() {
                continue;
            }
            // Keep noisy warnings only for invalid/ambiguous names that cannot
            // be interpreted as new-format IDs.
            if known_safe_paths.contains(&name) {
                continue;
            }
            warn!(
                "Skipping invalid legacy log directory '{name}': contains '--' but is not a valid daemon safe-path"
            );
            continue;
        }

        // Migrate only explicit old-layout directories to avoid renaming
        // unrelated folders under logs/.
        let old_log = dir.join(format!("{name}.log"));
        if !old_log.exists() {
            continue;
        }
        if DaemonId::try_new("legacy", &name).is_err() {
            warn!("Skipping invalid legacy log directory '{name}': not a valid daemon ID");
            continue;
        }

        let new_name = format!("legacy--{name}");
        let new_dir = env::PITCHFORK_LOGS_DIR.join(&new_name);
        // Skip if a target directory already exists to avoid clobbering data.
        if new_dir.exists() {
            continue;
        }
        if std::fs::rename(&dir, &new_dir).is_err() {
            continue;
        }
        // Also rename the log file inside the directory.
        let old_log = new_dir.join(format!("{name}.log"));
        let new_log = new_dir.join(format!("{new_name}.log"));
        if old_log.exists() {
            let _ = std::fs::rename(&old_log, &new_log);
        }
        debug!("Migrated legacy log dir '{name}' → '{new_name}'");
    }
}

fn known_daemon_safe_paths() -> BTreeSet<String> {
    let mut out = BTreeSet::new();

    match StateFile::read(&*env::PITCHFORK_STATE_FILE) {
        Ok(state) => {
            for id in state.daemons.keys() {
                out.insert(id.safe_path());
            }
        }
        Err(e) => {
            warn!("Failed to read state while checking known daemon IDs: {e}");
        }
    }

    match PitchforkToml::all_merged() {
        Ok(config) => {
            for id in config.daemons.keys() {
                out.insert(id.safe_path());
            }
        }
        Err(e) => {
            warn!("Failed to read config while checking known daemon IDs: {e}");
        }
    }

    out
}

fn get_all_daemon_ids() -> Result<Vec<DaemonId>> {
    let mut ids = BTreeSet::new();

    match StateFile::read(&*env::PITCHFORK_STATE_FILE) {
        Ok(state) => ids.extend(state.daemons.keys().cloned()),
        Err(e) => warn!("Failed to read state for log daemon discovery: {e}"),
    }

    match PitchforkToml::all_merged() {
        Ok(config) => ids.extend(config.daemons.keys().cloned()),
        Err(e) => warn!("Failed to read config for log daemon discovery: {e}"),
    }

    let logged_ids: std::collections::HashSet<String> =
        LOG_STORE.list_daemon_ids()?.into_iter().collect();
    Ok(ids
        .into_iter()
        .filter(|id| logged_ids.contains(&id.qualified()))
        .collect())
}

#[allow(clippy::too_many_arguments)]
pub async fn tail_logs(
    names: &[DaemonId],
    single_daemon: bool,
    start_from_end: bool,
    message_filters: Vec<MessageFilter>,
    field_filters: Vec<FieldFilter>,
    jq_filter: Option<&crate::log_jq::JqFilter>,
    show_timestamp: bool,
    raw: bool,
) -> Result<()> {
    // Poll SQLite log store for new entries since last known row id.
    let strip_ansi = raw || !console::colors_enabled();

    let mut states: std::collections::HashMap<String, i64> = names
        .iter()
        .map(|id| {
            let since = if start_from_end {
                // Anchor to the last entry overall, not the last filtered entry,
                // so --tail combined with a filter does not rescan from row 1
                // on every poll when no message matches yet.
                LOG_STORE.last_id(id).unwrap_or(None).unwrap_or(0)
            } else {
                0
            };
            (id.qualified(), since)
        })
        .collect();

    let interval = tokio::time::interval(Duration::from_millis(200));
    tokio::pin!(interval);

    loop {
        interval.tick().await;

        let mut out = vec![];
        for id in names {
            let after_id = states.get(&id.qualified()).copied();
            match LOG_STORE.query(&LogQuery {
                daemon_ids: vec![id.qualified()],
                from: None,
                to: None,
                limit: None,
                order_desc: false,
                after_id,
                before_id: None,
                message_filters: message_filters.clone(),
                field_filters: field_filters.clone(),
                include_structured: jq_filter.is_some() || !raw,
            }) {
                Ok(raw_entries) => {
                    // Save the max id from SQL results (before jq filtering)
                    // so the cursor always advances past evaluated rows, even
                    // when jq rejects all of them.
                    let last_raw_id = raw_entries.last().map(|e| e.id);

                    // Apply jq filter if present.
                    let entries = match jq_filter {
                        Some(jq) => jq.filter(raw_entries),
                        None => raw_entries,
                    };
                    out.extend(entries);
                    // Advance the cursor past rows already evaluated.
                    //
                    // Use the max id from the raw SQL result (not the jq-filtered
                    // subset) so the cursor advances past all rows the database
                    // returned, preventing re-evaluation on every poll.
                    //
                    // When the filter is active and no entries matched, fall
                    // back to last_id to advance past non-matching rows so they
                    // are not re-evaluated on every poll. The narrow race here
                    // (a matching row written between query and last_id) is
                    // accepted as the cost of bounded re-scanning.
                    let has_sql_filter = !message_filters.is_empty() || !field_filters.is_empty();
                    let new_cursor = if let Some(id) = last_raw_id {
                        Some(id)
                    } else if has_sql_filter || jq_filter.is_some() {
                        LOG_STORE.last_id(id).ok().flatten()
                    } else {
                        None
                    };
                    if let Some(last_id) = new_cursor {
                        states.insert(id.qualified(), last_id);
                    }
                }
                Err(e) => {
                    error!("Failed to tail logs for {}: {e}", id.qualified());
                }
            }
        }

        if !out.is_empty() {
            // Single-daemon: entries are already in chronological order from
            // the SQL query (order_desc: false). Skip the sort.
            let out: Vec<LogEntry> = if single_daemon {
                out
            } else {
                out.into_iter()
                    .sorted_by(|a, b| a.timestamp.cmp(&b.timestamp).then(a.id.cmp(&b.id)))
                    .collect()
            };
            let stdout = io::stdout();
            let mut buf = io::BufWriter::new(stdout.lock());
            let ts_format = &settings().logs.timestamp_format;
            let mut date_buf = String::with_capacity(ts_format.len() + 6);
            for entry in &out {
                date_buf.clear();
                write!(date_buf, "{}", entry.timestamp.format(ts_format))
                    .map_err(io::Error::other)
                    .into_diagnostic()?;
                write_formatted_log(
                    &mut buf,
                    entry,
                    &date_buf,
                    single_daemon,
                    strip_ansi,
                    show_timestamp,
                    raw,
                )
                .into_diagnostic()?;
            }
        }
    }
}

fn parse_datetime(s: &str) -> Result<DateTime<Local>> {
    let naive_dt = NaiveDateTime::parse_from_str(s, "%Y-%m-%d %H:%M:%S").into_diagnostic()?;
    Local
        .from_local_datetime(&naive_dt)
        .single()
        .ok_or_else(|| miette::miette!("Invalid or ambiguous datetime: '{}'. ", s))
}

/// Parse time input string into DateTime.
///
/// `is_since` indicates whether this is for --since (true) or --until (false).
/// The "yesterday fallback" only applies to --since: if the time is in the future,
/// assume the user meant yesterday. For --until, future times are kept as-is.
fn parse_time_input(s: &str, is_since: bool) -> Result<DateTime<Local>> {
    let s = s.trim();

    // Try full datetime first (YYYY-MM-DD HH:MM:SS)
    if let Ok(dt) = parse_datetime(s) {
        return Ok(dt);
    }

    // Try datetime without seconds (YYYY-MM-DD HH:MM)
    if let Ok(naive_dt) = NaiveDateTime::parse_from_str(s, "%Y-%m-%d %H:%M") {
        return Local
            .from_local_datetime(&naive_dt)
            .single()
            .ok_or_else(|| miette::miette!("Invalid or ambiguous datetime: '{}'", s));
    }

    // Try time-only format (HH:MM:SS or HH:MM)
    // Note: This branch won't be reached for inputs like "10:30" that could match
    // parse_datetime, because parse_datetime expects a full date prefix and will fail.
    if let Ok(time) = parse_time_only(s) {
        let now = Local::now();
        let today = now.date_naive();
        let mut naive_dt = NaiveDateTime::new(today, time);
        let mut dt = Local
            .from_local_datetime(&naive_dt)
            .single()
            .ok_or_else(|| miette::miette!("Invalid or ambiguous datetime: '{}'", s))?;

        // If the interpreted time for today is in the future, assume the user meant yesterday
        // BUT only for --since. For --until, a future time today is valid.
        if is_since
            && dt > now
            && let Some(yesterday) = today.pred_opt()
        {
            naive_dt = NaiveDateTime::new(yesterday, time);
            dt = Local
                .from_local_datetime(&naive_dt)
                .single()
                .ok_or_else(|| miette::miette!("Invalid or ambiguous datetime: '{}'", s))?;
        }
        return Ok(dt);
    }

    if let Ok(duration) = humantime::parse_duration(s) {
        let now = Local::now();
        let target = now - chrono::Duration::from_std(duration).into_diagnostic()?;
        return Ok(target);
    }

    Err(miette::miette!(
        "Invalid time format: '{}'. Expected formats:\n\
         - Full datetime: \"YYYY-MM-DD HH:MM:SS\" or \"YYYY-MM-DD HH:MM\"\n\
         - Time only: \"HH:MM:SS\" or \"HH:MM\" (uses today's date)\n\
         - Relative time: \"5min\", \"2h\", \"1d\" (e.g., last 5 minutes)",
        s
    ))
}

fn parse_time_only(s: &str) -> Result<NaiveTime> {
    if let Ok(time) = NaiveTime::parse_from_str(s, "%H:%M:%S") {
        return Ok(time);
    }

    if let Ok(time) = NaiveTime::parse_from_str(s, "%H:%M") {
        return Ok(time);
    }

    Err(miette::miette!("Invalid time format: '{}'", s))
}

/// Prints error log lines in a styled block matching the startup logs format.
///
/// Format:
/// ```text
///  ERROR LOGS
///  12:00:00 error message
/// ```
///
/// Timestamps use dimmed red. The tag uses white text on red background.
pub fn print_error_logs_block(log_lines: &[(String, String, String)]) {
    if log_lines.is_empty() {
        return;
    }

    let is_tty = std::io::stderr().is_terminal();
    let format_msg = |msg: &str| -> String {
        let stripped = strip_pty_controls(msg);
        if is_tty {
            stripped
        } else {
            console::strip_ansi_codes(&stripped).to_string()
        }
    };

    let tag = estyle(" ERROR LOGS ").white().on_red();
    eprintln!("\n{tag}");

    // Determine if we need to show daemon IDs (same logic as startup logs)
    let unique_ids: BTreeSet<&str> = log_lines.iter().map(|(_, id, _)| id.as_str()).collect();
    let show_id = unique_ids.len() > 1;

    if show_id {
        for (date, id, msg) in log_lines {
            let time = date.split(' ').nth(1).unwrap_or(date);
            let colored = colored_id_label(id, is_tty && console::colors_enabled_stderr());
            eprintln!(
                "{} {} {}",
                estyle(time).red().dim(),
                colored,
                format_msg(msg)
            );
        }
    } else {
        for (date, _, msg) in log_lines {
            let time = date.split(' ').nth(1).unwrap_or(date);
            eprintln!("{} {}", estyle(time).red().dim(), format_msg(msg));
        }
    }
}

/// Describes the type of ready check being performed for display purposes.
pub enum ReadyCheckType {
    Output(String),
    Http(String),
    Port(u16),
    Cmd(String),
    Delay(u64),
    /// A `oneshot` daemon: ready when the process exits 0.
    Completion,
    Default,
}

impl std::fmt::Display for ReadyCheckType {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            ReadyCheckType::Output(pattern) => write!(f, "output matching '{pattern}'"),
            ReadyCheckType::Http(url) => write!(f, "HTTP {url}"),
            ReadyCheckType::Port(port) => write!(f, "TCP port {port}"),
            ReadyCheckType::Cmd(cmd) => write!(f, "command '{cmd}'"),
            ReadyCheckType::Delay(secs) => write!(f, "delay ({secs}s)"),
            ReadyCheckType::Completion => write!(f, "completion"),
            ReadyCheckType::Default => write!(f, "default readiness check"),
        }
    }
}

/// Creates a progress job showing a spinner while waiting for a ready check.
///
/// Returns a `Arc<ProgressJob>` that the caller should update:
/// - Set body to success message and status to `Done` when the daemon is ready
/// - Set body to failure message and status to `Failed` when the daemon fails
pub fn create_ready_check_job(
    daemon_id: &DaemonId,
    check_type: &ReadyCheckType,
) -> std::sync::Arc<clx::progress::ProgressJob> {
    use clx::progress::{ProgressJobBuilder, ProgressJobDoneBehavior, ProgressStatus};

    let is_tty = std::io::stderr().is_terminal();
    let colors_enabled = is_tty && console::colors_enabled_stderr();
    let id_label = colored_id_label(&daemon_id.qualified(), colors_enabled);
    let show_ts = crate::settings::settings().general.startup_log_timestamps;

    // When timestamps are off, {{spinner()}} renders as an animated spinner
    // (1 char wide) matching the "•" prefix used by println.  When on,
    // we show a dim timestamp instead.
    let prefix = if show_ts {
        // The timestamp updates each refresh via the now() tera function.
        // We use a fixed-width format (HH:MM:SS = 8 chars) for alignment.
        edim(chrono::Local::now().format("%H:%M:%S").to_string()).to_string()
    } else {
        "{{spinner()}}".to_string()
    };

    ProgressJobBuilder::new()
        .body(format!(
            "{} {} waiting for {{{{ check_type }}}}...",
            prefix, id_label
        ))
        .prop("check_type", &check_type.to_string())
        .status(ProgressStatus::Running)
        .on_done(ProgressJobDoneBehavior::Keep)
        .start()
}

/// Collects startup log lines for a single daemon (does not print).
///
/// Returns a list of `(time, daemon_id_qualified, message)` tuples for log
/// entries written after `from`.
pub fn collect_startup_logs(
    daemon_id: &DaemonId,
    from: DateTime<Local>,
) -> Result<Vec<(String, String, String)>> {
    let entries = LOG_STORE.query(&LogQuery {
        daemon_ids: vec![daemon_id.qualified()],
        from: Some(from),
        to: None,
        limit: None,
        order_desc: false,
        after_id: None,
        before_id: None,
        message_filters: Vec::new(),
        field_filters: Vec::new(),
        include_structured: false,
    })?;
    let log_lines = entries
        .into_iter()
        .map(|e| {
            let ts = e.timestamp.format("%Y-%m-%d %H:%M:%S").to_string();
            (ts, e.daemon_id, e.message)
        })
        .collect();

    Ok(log_lines)
}

/// Stream startup logs for a daemon to a progress job in real-time.
///
/// Spawns a background tokio task that polls the daemon's log store
/// and calls `job.println()` for each new line. Returns a watch sender
/// that stops the streaming when sent `true`.
pub fn stream_startup_logs(
    daemon_id: &DaemonId,
    job: std::sync::Arc<clx::progress::ProgressJob>,
) -> (
    tokio::sync::watch::Sender<bool>,
    tokio::task::JoinHandle<()>,
) {
    let (tx, mut rx) = tokio::sync::watch::channel(false);
    let id = daemon_id.clone();

    let show_ts = crate::settings::settings().general.startup_log_timestamps;

    // Anchor to the daemon's current max log id *synchronously* before
    // spawning the streaming task. This must happen before ipc.run() starts
    // the daemon, otherwise early output could be written to the log store
    // before the anchor is established and get skipped as "already seen".
    let anchor_id: i64 = LOG_STORE
        .query(&LogQuery {
            daemon_ids: vec![id.qualified()],
            limit: Some(1),
            order_desc: true,
            ..Default::default()
        })
        .ok()
        .and_then(|entries| entries.last().map(|e| e.id))
        .unwrap_or(0);

    let handle = tokio::spawn(async move {
        let is_tty = std::io::stderr().is_terminal();
        let colors_enabled = is_tty && console::colors_enabled_stderr();
        let id_label = colored_id_label(&id.qualified(), colors_enabled);
        let prefix = if show_ts {
            String::new()
        } else {
            edim("•").to_string()
        };

        let mut last_id = anchor_id;

        // Initial fetch: catch any logs already written since the anchor.
        if let Ok(entries) = LOG_STORE.tail(&id, Some(last_id)) {
            for entry in &entries {
                let time = entry.timestamp.format("%H:%M:%S").to_string();
                let msg = strip_pty_controls(&entry.message);
                let msg = if is_tty {
                    msg
                } else {
                    console::strip_ansi_codes(&msg).to_string()
                };
                let line_prefix = if show_ts {
                    edim(time).to_string()
                } else {
                    prefix.clone()
                };
                job.println(&format!("{} {} {}", line_prefix, id_label, msg));
            }
            if let Some(last) = entries.last() {
                last_id = last.id;
            }
        }

        loop {
            tokio::select! {
                _ = tokio::time::sleep(Duration::from_millis(200)) => {
                    if let Ok(entries) = LOG_STORE.tail(&id, Some(last_id)) {
                        for entry in &entries {
                            let time = entry.timestamp.format("%H:%M:%S").to_string();
                            let msg = strip_pty_controls(&entry.message);
                            let msg = if is_tty {
                                msg
                            } else {
                                console::strip_ansi_codes(&msg).to_string()
                            };
                            let line_prefix = if show_ts {
                                edim(time).to_string()
                            } else {
                                prefix.clone()
                            };
                            job.println(&format!("{} {} {}", line_prefix, id_label, msg));
                        }
                        if let Some(last) = entries.last() {
                            last_id = last.id;
                        }
                    }
                }
                _ = rx.changed() => {
                    break;
                }
            }
        }

        // Final drain
        if let Ok(entries) = LOG_STORE.tail(&id, Some(last_id)) {
            for entry in &entries {
                let time = entry.timestamp.format("%H:%M:%S").to_string();
                let msg = strip_pty_controls(&entry.message);
                let msg = if is_tty {
                    msg
                } else {
                    console::strip_ansi_codes(&msg).to_string()
                };
                let line_prefix = if show_ts {
                    edim(time).to_string()
                } else {
                    prefix.clone()
                };
                job.println(&format!("{} {} {}", line_prefix, id_label, msg));
            }
        }
    });

    (tx, handle)
}

/// Strips PTY control sequences from a string while preserving SGR (color/style) codes.
///
/// Removes CSI sequences that control cursor movement, screen clearing, erasing, etc.,
/// but keeps `\x1b[...m` (SGR) sequences so colors are retained.
fn strip_pty_controls(s: &str) -> String {
    struct Stripper {
        result: String,
    }

    impl vte::Perform for Stripper {
        fn print(&mut self, c: char) {
            self.result.push(c);
        }

        fn execute(&mut self, byte: u8) {
            // Keep \n and \t; drop other control characters (BEL, BS, CR, etc.)
            if byte == b'\n' || byte == b'\t' {
                self.result.push(byte as char);
            }
        }

        fn csi_dispatch(
            &mut self,
            params: &vte::Params,
            _intermediates: &[u8],
            _ignore: bool,
            action: char,
        ) {
            // Keep SGR sequences (final byte 'm')
            if action == 'm' {
                self.result.push_str("\x1b[");
                let mut first = true;
                for sub in params.iter() {
                    if !first {
                        self.result.push(';');
                    }
                    first = false;
                    for (i, &p) in sub.iter().enumerate() {
                        if i > 0 {
                            self.result.push(':');
                        }
                        self.result.push_str(&p.to_string());
                    }
                }
                self.result.push('m');
            }
            // All other CSI sequences (cursor move, clear, erase, etc.) are dropped
        }

        fn osc_dispatch(&mut self, _params: &[&[u8]], _bell_terminated: bool) {
            // Drop OSC sequences (e.g. window title)
        }

        fn esc_dispatch(&mut self, _intermediates: &[u8], _ignore: bool, _byte: u8) {
            // Drop ESC sequences (e.g. ESC c = reset terminal)
        }

        fn hook(
            &mut self,
            _params: &vte::Params,
            _intermediates: &[u8],
            _ignore: bool,
            _action: char,
        ) {
            // Drop DCS hooks
        }

        fn put(&mut self, _byte: u8) {
            // Drop DCS data
        }

        fn unhook(&mut self) {
            // Drop DCS unhook
        }
    }

    let mut parser = vte::Parser::new();
    let mut stripper = Stripper {
        result: String::with_capacity(s.len()),
    };
    parser.advance(&mut stripper, s.as_bytes());
    stripper.result
}

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

    /// Newest first, as the query returns them: row `i` is the `i`-th newest.
    fn fetch_from(rows: &[u32], reads: &mut Vec<usize>) -> impl FnMut(usize) -> Result<Vec<u32>> {
        move |window| {
            reads.push(window);
            Ok(rows.iter().copied().take(window).collect())
        }
    }

    #[test]
    fn last_matches_reads_one_window_when_the_newest_rows_match() {
        let rows: Vec<u32> = (0..100_000).collect();
        let mut reads = Vec::new();
        let found = last_matches(2, fetch_from(&rows, &mut reads), |r| {
            r.into_iter().filter(|i| i % 10 == 0).collect()
        })
        .unwrap();
        assert_eq!(found, vec![0, 10]);
        assert_eq!(reads, vec![MIN_MATCH_WINDOW]);
    }

    #[test]
    fn last_matches_widens_the_window_until_it_holds_n_matches() {
        let rows: Vec<u32> = (0..100_000).collect();
        let mut reads = Vec::new();
        let found = last_matches(2, fetch_from(&rows, &mut reads), |r| {
            r.into_iter().filter(|i| *i >= 3000).collect()
        })
        .unwrap();
        assert_eq!(found, vec![3000, 3001]);
        assert_eq!(reads, vec![1000, 2000, 4000]);
    }

    #[test]
    fn last_matches_stops_once_every_row_is_read() {
        let rows: Vec<u32> = (0..2500).collect();
        let mut reads = Vec::new();
        let found = last_matches(5, fetch_from(&rows, &mut reads), |r| {
            r.into_iter().filter(|i| *i == 2400).collect()
        })
        .unwrap();
        assert_eq!(found, vec![2400]);
        assert_eq!(reads, vec![1000, 2000, 4000]);
    }
}