ottyl/src/engine.rs

use crate::alerts::{self, AlertRuntimeState};
use crate::data_value::{DataValue, MetricPrimitiveKind, ParamRole, RegexToMetricPrimitiveMatcher};
use crate::dto::{
    BucketValueDto, LogEntryDto, LogsParams, MetricDeclarationDto, ParamDeclarationDto,
    PromqlSeriesDto, StatsDto, TimelineBucketDto, TimelineParams,
};
use crate::filter_datetime;
use crate::filter_parser::{self, CompareOp, FilterExpr, FilterValue};
use crate::index_key::IndexKey;
use crate::metric_primitive::{MetricPrimitive, Number};
use crate::promql::{self, AggOp, LabelMatcher, PromqlQuery};
use crate::settings::SettingsHandle;
use crate::timestamp::Timestamp;
use regex::Regex;
use serde::Deserialize;
use std::cmp::Ordering;
use std::collections::{BTreeMap, HashMap, HashSet};
use tokio::sync::{mpsc, oneshot};

pub struct IngestedLine {
    pub sequence: u64,
    pub timestamp: Timestamp,
    pub text: String,
}

/// One sample scraped from a Prometheus/OpenMetrics endpoint, pushed straight into a metric's bucket
/// without going through log-line regex matching — see `crate::ingest::run_metrics_poll`.
pub struct MetricSample {
    pub metric_name: String,
    pub labels: BTreeMap<String, String>,
    pub value: f64,
    pub timestamp: Timestamp,
}

/// Also the request body for `POST /api/params` — `Deserialize` is derived directly here rather than adding a separate DTO, since the shape already matches.
#[derive(Debug, Clone, Deserialize)]
pub struct DeclareParamRequest {
    pub name: String,
    pub message_match_regex: String,
    pub capture_group: String,
    /// See `RegexToMetricPrimitiveMatcher::json_field_path`; `#[serde(default)]` so existing
    /// regex-only callers are unaffected.
    #[serde(default)]
    pub json_field_path: Option<String>,
    pub kind: MetricPrimitiveKind,
}

/// Params are derived purely from inline named capture groups on this metric's own regex — referencing an already-declared `ParamDeclaration` by name isn't supported yet. Also the request body for `POST /api/metrics` (hence `Deserialize`).
#[derive(Debug, Clone, Deserialize)]
pub struct DeclareMetricRequest {
    pub name: String,
    pub message_match_regex: String,
    pub params: Vec<RegexToMetricPrimitiveMatcher>,
    /// Fixed labels layered on top of what `params` captured, so e.g. a case-insensitive match doesn't fragment into "ERROR"/"Error" series; `#[serde(default)]` so existing declarations are unaffected.
    #[serde(default)]
    pub matched_labels: Option<BTreeMap<String, String>>,
    /// When a line doesn't match, registers one occurrence labeled with these fixed values instead of being invisible to this metric — plain regex alternation can't express "didn't match" without lookaround; `#[serde(default)]` so existing declarations keep matching-only behavior.
    #[serde(default)]
    pub unmatched_labels: Option<BTreeMap<String, String>>,
}

#[allow(dead_code)]
pub enum EngineQuery {
    Logs(LogsParams),
    Timeline(TimelineParams),
    Stats,
    ParamDeclarations,
    MetricDeclarations,
    /// `labels_filter` compares against each label's `Display` form (not a typed value), since the filter arrives as plain strings off the query string.
    BucketValues {
        metric_name: String,
        labels_filter: Option<BTreeMap<String, String>>,
        start_ms: Option<i64>,
        end_ms: Option<i64>,
        since_sequence: Option<u64>,
        limit: Option<usize>,
    },
    /// `query` is parsed inside the engine (not the API handler) so it stays in the single-owner loop with no copying; a malformed query is only discovered here, hence `EngineResponse::Promql` carries a `Result`.
    Promql { query: String, start_ms: Option<i64>, end_ms: Option<i64> },
}

#[allow(dead_code)]
pub enum EngineMessage {
    Ingest(IngestedLine),
    IngestMetricSample(MetricSample),
    DeclareParam {
        request: DeclareParamRequest,
        respond_to: oneshot::Sender<Result<(), String>>,
    },
    RemoveParamDeclaration {
        name: String,
        respond_to: oneshot::Sender<bool>,
    },
    DeclareMetric {
        request: DeclareMetricRequest,
        respond_to: oneshot::Sender<Result<(), String>>,
    },
    /// Updates an existing metric declaration named `name` in place — same validation as `DeclareMetric`, but fails if no declaration named `name` exists (rather than silently creating one), and re-backfills bucket data from scratch so edited params/regex don't leave stale buckets from the old declaration mixed in.
    UpdateMetric {
        name: String,
        request: DeclareMetricRequest,
        respond_to: oneshot::Sender<Result<(), String>>,
    },
    RemoveMetricDeclaration {
        name: String,
        respond_to: oneshot::Sender<bool>,
    },
    /// `None` means unlimited; sent on startup and whenever the setting changes so the cap applies without a restart.
    SetMaxLogs(Option<usize>),
    Query {
        query: EngineQuery,
        respond_to: oneshot::Sender<EngineResponse>,
    },
}

#[allow(dead_code)]
pub enum EngineResponse {
    Logs(Vec<LogEntryDto>),
    Timeline(Vec<TimelineBucketDto>),
    Stats(StatsDto),
    ParamDeclarations(Vec<ParamDeclarationDto>),
    MetricDeclarations(Vec<MetricDeclarationDto>),
    BucketValues(Vec<BucketValueDto>),
    /// `Err` holds a human-readable message for a malformed query, which the API layer maps to a 400 response.
    Promql(Result<Vec<PromqlSeriesDto>, String>),
    /// Kept as a variant (rather than making `handle_query` return a `Result`) so every query path still funnels through the same `oneshot::Sender<EngineResponse>`.
    Error(String),
}

pub type EngineHandle = mpsc::Sender<EngineMessage>;

/// Sole owner of the primary index; runs single-threaded so the map never needs a lock, and builds DTOs itself so callers never clone the index.
pub async fn run(mut rx: mpsc::Receiver<EngineMessage>, settings: SettingsHandle) {
    let mut index: BTreeMap<IndexKey, DataValue> = BTreeMap::new();
    let mut max_logs: Option<usize> = None;
    let mut log_count: usize = 0;
    // Its own counter, independent of log `sequence` — scraped samples aren't tied to any ingested line.
    let mut metric_sample_sequence: u64 = 0;
    // Per-rule rolling windows/seen-value sets, keyed by rule id; lives only as long as this loop (no restart persistence).
    let mut alert_states: HashMap<String, AlertRuntimeState> = HashMap::new();
    // Ticket 011: running total of `approx_log_bytes` (see that function) across
    // every currently-ingested log, kept in lockstep with `log_count` — incremented
    // here on ingest, decremented in `evict_excess_logs` when a log is dropped.
    let mut approx_bytes_total: u64 = 0;

    while let Some(message) = rx.recv().await {
        match message {
            EngineMessage::Ingest(line) => {
                let IngestedLine { sequence, timestamp, text } = line;
                approx_bytes_total += approx_log_bytes(&text);
                index.insert(
                    IndexKey::IngestedLog { sequence, timestamp },
                    DataValue::LogLiteral { text: text.clone() },
                );
                log_count += 1;
                apply_all_metrics(&mut index, sequence, timestamp, &text);
                evict_excess_logs(&mut index, &mut log_count, max_logs, &mut approx_bytes_total);

                // Evaluate alert rules against just this line, then hand any fired rule off to delivery via tokio::spawn so it stays off the hot path.
                let (rules, ntfy_topic, ntfy_server) = {
                    let settings = settings.lock().unwrap();
                    (settings.alert_rules.clone(), settings.ntfy_topic.clone(), settings.ntfy_server.clone())
                };
                let fired = alerts::evaluate_alerts(&rules, &mut alert_states, sequence, timestamp, &text);
                for (rule, message) in fired {
                    tokio::spawn(alerts::deliver_for_rule(rule.name, message, ntfy_topic.clone(), ntfy_server.clone()));
                }
            }
            EngineMessage::IngestMetricSample(sample) => {
                let labels = sample
                    .labels
                    .into_iter()
                    .map(|(name, value)| (name, MetricPrimitive::String(value)))
                    .collect();
                push_bucket_value(
                    &mut index,
                    &sample.metric_name,
                    labels,
                    metric_sample_sequence,
                    sample.timestamp,
                    MetricPrimitive::Number(Number::Float(sample.value)),
                );
                metric_sample_sequence += 1;
            }
            EngineMessage::SetMaxLogs(new_max) => {
                max_logs = new_max;
                evict_excess_logs(&mut index, &mut log_count, max_logs, &mut approx_bytes_total);
            }
            EngineMessage::DeclareParam { request, respond_to } => {
                let result = declare_param(&mut index, request);
                let _ = respond_to.send(result);
            }
            EngineMessage::RemoveParamDeclaration { name, respond_to } => {
                let existed = index.remove(&IndexKey::ParamDeclaration { name }).is_some();
                let _ = respond_to.send(existed);
            }
            EngineMessage::DeclareMetric { request, respond_to } => {
                let result = declare_metric(&mut index, request);
                let _ = respond_to.send(result);
            }
            EngineMessage::UpdateMetric { name, request, respond_to } => {
                let result = update_metric(&mut index, &name, request);
                let _ = respond_to.send(result);
            }
            EngineMessage::RemoveMetricDeclaration { name, respond_to } => {
                let existed = index.remove(&IndexKey::MetricDeclaration { name }).is_some();
                let _ = respond_to.send(existed);
            }
            EngineMessage::Query { query, respond_to } => {
                let response = handle_query(&index, query, approx_bytes_total);
                let _ = respond_to.send(response);
            }
        }
    }
}

/// Validates the regex and capture group before storing, so a bad declaration is rejected up front rather than discovered later.
fn declare_param(
    index: &mut BTreeMap<IndexKey, DataValue>,
    request: DeclareParamRequest,
) -> Result<(), String> {
    let re = Regex::new(&request.message_match_regex)
        .map_err(|err| format!("invalid regex: {err}"))?;
    // A Boolean-kind param never uses a capture group — its value is just whether the regex matched.
    let uses_json_field = request.json_field_path.as_deref().is_some_and(|path| !path.is_empty());
    if request.kind != MetricPrimitiveKind::Boolean
        && !uses_json_field
        && !re.capture_names().flatten().any(|name| name == request.capture_group)
    {
        return Err(format!(
            "regex has no capture group named '{}'",
            request.capture_group
        ));
    }

    index.insert(
        IndexKey::ParamDeclaration { name: request.name },
        DataValue::ParamDeclaration {
            message_match_regex: request.message_match_regex,
            capture_group: request.capture_group,
            json_field_path: request.json_field_path,
            kind: request.kind,
        },
    );
    Ok(())
}

/// Validates every param matcher's capture-group reference (not just the first bad one) so a rejection reports every problem at once, and the at-most-one-Value-param rule. Shared by `declare_metric` and `update_metric` so both reject a bad declaration the same way.
fn validate_metric_request(request: &DeclareMetricRequest) -> Result<Regex, String> {
    let re = Regex::new(&request.message_match_regex)
        .map_err(|err| format!("invalid regex: {err}"))?;

    let errors: Vec<String> = request
        .params
        .iter()
        .filter(|matcher| {
            let uses_json_field =
                matcher.json_field_path.as_deref().is_some_and(|path| !path.is_empty());
            !uses_json_field && !re.capture_names().flatten().any(|name| name == matcher.capture_group)
        })
        .map(|matcher| {
            format!(
                "param '{}' references capture group '{}', which the regex does not have",
                matcher.param_name, matcher.capture_group
            )
        })
        .collect();
    if !errors.is_empty() {
        return Err(errors.join("; "));
    }

    let value_params = request.params.iter().filter(|matcher| matcher.role == ParamRole::Value).count();
    if value_params > 1 {
        return Err(format!(
            "a metric may declare at most one Value-role param, found {value_params}"
        ));
    }

    Ok(re)
}

fn declare_metric(
    index: &mut BTreeMap<IndexKey, DataValue>,
    request: DeclareMetricRequest,
) -> Result<(), String> {
    let re = validate_metric_request(&request)?;

    backfill_metric(
        index,
        &request.name,
        &re,
        &request.params,
        request.matched_labels.as_ref(),
        request.unmatched_labels.as_ref(),
    );
    index.insert(
        IndexKey::MetricDeclaration { name: request.name },
        DataValue::MetricDeclaration {
            message_match_regex: request.message_match_regex,
            params: request.params,
            matched_labels: request.matched_labels,
            unmatched_labels: request.unmatched_labels,
        },
    );
    Ok(())
}

/// Updates the metric declaration named `existing_name` in place, reusing `declare_metric`'s validation so an edit is rejected the same way a bad new declaration would be. Fails (rather than creating a fresh declaration) if no metric named `existing_name` exists yet. Existing bucket/last-event data for this metric is cleared first and rebuilt from the full log history under the new regex/params, so edited matchers don't leave stale buckets from the previous declaration mixed in with fresh ones. Supports renaming (when `request.name` differs from `existing_name`) as long as the new name isn't already taken by another declaration.
fn update_metric(
    index: &mut BTreeMap<IndexKey, DataValue>,
    existing_name: &str,
    request: DeclareMetricRequest,
) -> Result<(), String> {
    if !index.contains_key(&IndexKey::MetricDeclaration { name: existing_name.to_string() }) {
        return Err(format!("no metric declaration named '{existing_name}'"));
    }
    if request.name != existing_name
        && index.contains_key(&IndexKey::MetricDeclaration { name: request.name.clone() })
    {
        return Err(format!("a metric named '{}' already exists", request.name));
    }

    let re = validate_metric_request(&request)?;

    clear_metric_data(index, existing_name);
    if request.name != existing_name {
        index.remove(&IndexKey::MetricDeclaration { name: existing_name.to_string() });
    }

    backfill_metric(
        index,
        &request.name,
        &re,
        &request.params,
        request.matched_labels.as_ref(),
        request.unmatched_labels.as_ref(),
    );
    index.insert(
        IndexKey::MetricDeclaration { name: request.name.clone() },
        DataValue::MetricDeclaration {
            message_match_regex: request.message_match_regex,
            params: request.params,
            matched_labels: request.matched_labels,
            unmatched_labels: request.unmatched_labels,
        },
    );
    Ok(())
}

/// Removes every `Bucket` and `LastMetricEvent` entry belonging to `metric_name`, ahead of a full re-backfill in `update_metric` — otherwise the old declaration's buckets would sit alongside freshly backfilled ones under the same metric name.
fn clear_metric_data(index: &mut BTreeMap<IndexKey, DataValue>, metric_name: &str) {
    index.retain(|key, _| match key {
        IndexKey::Bucket { metric_name: name, .. } => name != metric_name,
        IndexKey::LastMetricEvent { metric_name: name, .. } => name != metric_name,
        _ => true,
    });
}

/// Returns `None` on a parse failure, which the caller treats as "this match doesn't apply".
fn parse_metric_primitive(raw: &str, kind: MetricPrimitiveKind) -> Option<MetricPrimitive> {
    match kind {
        MetricPrimitiveKind::String => Some(MetricPrimitive::String(raw.to_string())),
        MetricPrimitiveKind::Number => raw
            .parse::<i64>()
            .map(Number::Int)
            .or_else(|_| raw.parse::<f64>().map(Number::Float))
            .ok()
            .map(MetricPrimitive::Number),
        MetricPrimitiveKind::Timestamp => {
            crate::filter_datetime::parse_datetime(raw, Timestamp::now())
                .ok()
                .map(MetricPrimitive::Timestamp)
        }
        // A Value-role param never reaches this branch (see apply_metric_to_log); reached only for a Label-role param or standalone ParamDeclaration, parsed as a literal millisecond count.
        MetricPrimitiveKind::Duration => raw.parse::<i64>().ok().map(MetricPrimitive::Duration),
        // A standalone Boolean param never reaches this branch (see derive_log_params, which derives its
        // value from whether the regex matched rather than parsing captured text); reached only if a metric
        // matcher references a Boolean-kind param, where it's parsed as a literal "true"/"false" capture.
        MetricPrimitiveKind::Boolean => raw.parse::<bool>().ok().map(MetricPrimitive::Boolean),
    }
}

/// At most one param may be `Value`-role (its primitive becomes the event's value, otherwise a fixed `1` marker is used); any param failing to match/parse skips the whole match rather than storing a partial one. A `Value`-role `Duration` param is elapsed time since the previous same-labeled event, so the first occurrence pushes no bucket (nothing to diff against) rather than a bogus zero. When `regex` doesn't match and `unmatched_labels` is `Some`, an occurrence is still registered with those fixed labels.
fn apply_metric_to_log(
    index: &mut BTreeMap<IndexKey, DataValue>,
    metric_name: &str,
    regex: &Regex,
    params: &[RegexToMetricPrimitiveMatcher],
    matched_labels: Option<&BTreeMap<String, String>>,
    unmatched_labels: Option<&BTreeMap<String, String>>,
    sequence: u64,
    timestamp: Timestamp,
    text: &str,
) {
    let Some(captures) = regex.captures(text) else {
        if let Some(unmatched_labels) = unmatched_labels {
            let labels = unmatched_labels
                .iter()
                .map(|(name, value)| (name.clone(), MetricPrimitive::String(value.clone())))
                .collect();
            push_bucket_value(index, metric_name, labels, sequence, timestamp, MetricPrimitive::Number(Number::Int(1)));
        }
        return;
    };

    let mut labels = BTreeMap::new();
    let mut value = None;
    let mut is_duration = false;
    for matcher in params {
        let raw: String = match matcher.json_field_path.as_deref().filter(|path| !path.is_empty()) {
            Some(field_path) => match crate::data_value::extract_json_field(text, field_path) {
                Some(value) => value,
                None => return,
            },
            None => match captures.name(&matcher.capture_group) {
                Some(m) => m.as_str().to_string(),
                None => return,
            },
        };
        if matcher.role == ParamRole::Value && matcher.kind == MetricPrimitiveKind::Duration {
            is_duration = true;
            continue;
        }
        let Some(parsed) = parse_metric_primitive(&raw, matcher.kind) else {
            return;
        };
        match matcher.role {
            ParamRole::Label => {
                labels.insert(matcher.param_name.clone(), parsed);
            }
            ParamRole::Value => {
                value = Some(parsed);
            }
        }
    }
    if let Some(matched_labels) = matched_labels {
        for (name, value) in matched_labels {
            labels.insert(name.clone(), MetricPrimitive::String(value.clone()));
        }
    }

    let value = if is_duration {
        // Tracked separately from Bucket: a Duration's first occurrence pushes no Bucket entry, so the last-event timestamp can't live in Bucket.values.
        let last_event_key =
            IndexKey::LastMetricEvent { metric_name: metric_name.to_string(), labels: labels.clone() };
        let previous_timestamp = match index.get(&last_event_key) {
            Some(DataValue::LastEventTimestamp { timestamp: previous }) => Some(*previous),
            _ => None,
        };
        index.insert(last_event_key, DataValue::LastEventTimestamp { timestamp });
        match previous_timestamp {
            Some(previous) => {
                MetricPrimitive::Duration(timestamp.as_millis() - previous.as_millis())
            }
            None => return,
        }
    } else {
        value.unwrap_or(MetricPrimitive::Number(Number::Int(1)))
    };

    push_bucket_value(index, metric_name, labels, sequence, timestamp, value);
}

fn push_bucket_value(
    index: &mut BTreeMap<IndexKey, DataValue>,
    metric_name: &str,
    labels: BTreeMap<String, MetricPrimitive>,
    sequence: u64,
    timestamp: Timestamp,
    value: MetricPrimitive,
) {
    let key = IndexKey::Bucket { metric_name: metric_name.to_string(), labels };
    let entry = index
        .entry(key)
        .or_insert_with(|| DataValue::BucketValue { values: Vec::new() });
    if let DataValue::BucketValue { values } = entry {
        values.push((sequence, timestamp, value));
    }
}

/// Backfills buckets for a newly-declared metric against every already-ingested log, not just future ones.
fn backfill_metric(
    index: &mut BTreeMap<IndexKey, DataValue>,
    metric_name: &str,
    regex: &Regex,
    params: &[RegexToMetricPrimitiveMatcher],
    matched_labels: Option<&BTreeMap<String, String>>,
    unmatched_labels: Option<&BTreeMap<String, String>>,
) {
    let lines: Vec<(u64, Timestamp, String)> = logs_range(index)
        .map(|(sequence, timestamp, text)| (sequence, timestamp, text.to_string()))
        .collect();
    for (sequence, timestamp, text) in lines {
        apply_metric_to_log(
            index,
            metric_name,
            regex,
            params,
            matched_labels,
            unmatched_labels,
            sequence,
            timestamp,
            &text,
        );
    }
}

fn apply_all_metrics(
    index: &mut BTreeMap<IndexKey, DataValue>,
    sequence: u64,
    timestamp: Timestamp,
    text: &str,
) {
    type SeededMetric = (
        String,
        String,
        Vec<RegexToMetricPrimitiveMatcher>,
        Option<BTreeMap<String, String>>,
        Option<BTreeMap<String, String>>,
    );
    let metrics: Vec<SeededMetric> = metric_declarations_range(index)
        .map(|(name, message_match_regex, params, matched_labels, unmatched_labels)| {
            (
                name.to_string(),
                message_match_regex.to_string(),
                params.to_vec(),
                matched_labels.cloned(),
                unmatched_labels.cloned(),
            )
        })
        .collect();
    for (name, message_match_regex, params, matched_labels, unmatched_labels) in metrics {
        if let Ok(re) = Regex::new(&message_match_regex) {
            apply_metric_to_log(
                index,
                &name,
                &re,
                &params,
                matched_labels.as_ref(),
                unmatched_labels.as_ref(),
                sequence,
                timestamp,
                text,
            );
        }
    }
}

/// The metrics API layers its own textual label-filter and range filtering on top rather than using this function's `labels_filter` param (used directly by tests).
fn bucket_values<'a>(
    index: &'a BTreeMap<IndexKey, DataValue>,
    metric_name: &'a str,
    labels_filter: Option<&'a BTreeMap<String, MetricPrimitive>>,
) -> impl Iterator<Item = (&'a BTreeMap<String, MetricPrimitive>, &'a (u64, Timestamp, MetricPrimitive))> + 'a
{
    index
        .range(IndexKey::MetricBucketSplit..IndexKey::End)
        .filter_map(move |(key, value)| match (key, value) {
            (IndexKey::Bucket { metric_name: name, labels }, DataValue::BucketValue { values })
                if name == metric_name && labels_filter.is_none_or(|filter| filter == labels) =>
            {
                Some(values.iter().map(move |entry| (labels, entry)))
            }
            _ => None,
        })
        .flatten()
}

/// A [`FilterExpr`] compiled against a snapshot of the index, so evaluating per log line touches the index only once: a `metric[param]` comparison is pre-resolved into a `HashSet` of matching sequences ([`CompiledFilter::MetricMatch`]) for O(1) per-line lookup instead of a regex re-run; a comparison with no `[param]` is rejected since v1 buckets have no single "the metric's value", only a per-label-combo occurrence marker.
enum CompiledFilter {
    Regex(Regex),
    Sequence { op: CompareOp, target: f64 },
    Timestamp { op: CompareOp, target_ms: i64 },
    MetricMatch(HashSet<u64>),
    /// `json[field] <op> value` — unlike `MetricMatch`, this can't be pre-resolved against the
    /// index (the JSON lives in the log line itself, not a declared metric/param), so it's
    /// evaluated per line in `eval_compiled` by parsing `text` as JSON.
    Json { field: String, op: CompareOp, value: FilterValue },
    And(Box<CompiledFilter>, Box<CompiledFilter>),
    Or(Box<CompiledFilter>, Box<CompiledFilter>),
}

/// Resolves every regex/metric-comparison/datetime-literal up front so per-line evaluation ([`eval_compiled`]) never touches the index; `now` anchors relative literals so every comparison in one query resolves against the same instant.
fn compile_filter_query(
    index: &BTreeMap<IndexKey, DataValue>,
    filter: &str,
    now: Timestamp,
) -> Result<CompiledFilter, String> {
    let expr = filter_parser::parse(filter).map_err(|err| err.to_string())?;
    compile_filter(index, &expr, now)
}

fn compile_filter(
    index: &BTreeMap<IndexKey, DataValue>,
    expr: &FilterExpr,
    now: Timestamp,
) -> Result<CompiledFilter, String> {
    match expr {
        FilterExpr::Regex(pattern) => Regex::new(pattern)
            .map(CompiledFilter::Regex)
            .map_err(|err| format!("invalid regex in filter: {err}")),
        FilterExpr::Comparison { metric, param, op, value } => match metric.as_str() {
            "sequence" => compile_sequence_comparison(*op, value),
            "timestamp" => compile_timestamp_comparison(*op, value, now),
            other => compile_metric_comparison(index, other, param.as_deref(), *op, value, now),
        },
        FilterExpr::JsonComparison { field, op, value } => Ok(CompiledFilter::Json {
            field: field.clone(),
            op: *op,
            value: value.clone(),
        }),
        FilterExpr::And(lhs, rhs) => Ok(CompiledFilter::And(
            Box::new(compile_filter(index, lhs, now)?),
            Box::new(compile_filter(index, rhs, now)?),
        )),
        FilterExpr::Or(lhs, rhs) => Ok(CompiledFilter::Or(
            Box::new(compile_filter(index, lhs, now)?),
            Box::new(compile_filter(index, rhs, now)?),
        )),
    }
}

/// A `Raw` value is accepted defensively (parsed as a number) even though the grammar shouldn't produce one for `sequence`.
fn compile_sequence_comparison(op: CompareOp, value: &FilterValue) -> Result<CompiledFilter, String> {
    let target = match value {
        FilterValue::Number(n) => *n,
        FilterValue::Raw(raw) => raw.parse::<f64>().map_err(|_| {
            format!("'sequence' comparisons require a numeric value, got {raw:?}")
        })?,
    };
    Ok(CompiledFilter::Sequence { op, target })
}

/// A `Number` is treated as raw milliseconds; a `Raw` token is resolved via `filter_datetime::parse_datetime` against `now`.
fn compile_timestamp_comparison(
    op: CompareOp,
    value: &FilterValue,
    now: Timestamp,
) -> Result<CompiledFilter, String> {
    let target_ms = match value {
        FilterValue::Number(n) => *n as i64,
        FilterValue::Raw(raw) => filter_datetime::parse_datetime(raw, now)
            .map_err(|err| err.to_string())?
            .as_millis(),
    };
    Ok(CompiledFilter::Timestamp { op, target_ms })
}

/// `metric[param] <op> value` for any metric name other than `sequence`/`timestamp` — see [`CompiledFilter`]'s doc comment.
fn compile_metric_comparison(
    index: &BTreeMap<IndexKey, DataValue>,
    metric: &str,
    param: Option<&str>,
    op: CompareOp,
    value: &FilterValue,
    now: Timestamp,
) -> Result<CompiledFilter, String> {
    let param = param.ok_or_else(|| {
        format!(
            "metric comparison on '{metric}' requires a [param] (e.g. '{metric}[some_param]{}...'); \
             v1 buckets only carry per-label occurrence data, not a single metric value",
            match op {
                CompareOp::Lt => "<",
                CompareOp::Gt => ">",
                CompareOp::Le => "<=",
                CompareOp::Ge => ">=",
                CompareOp::Eq => "==",
                CompareOp::Ne => "!=",
            }
        )
    })?;

    let mut matching = HashSet::new();
    for (labels, (sequence, _, _)) in bucket_values(index, metric, None) {
        let Some(primitive) = labels.get(param) else {
            continue;
        };
        if compare_metric_primitive(primitive, op, value, now)? {
            matching.insert(*sequence);
        }
    }
    Ok(CompiledFilter::MetricMatch(matching))
}

/// Mismatched kinds that can't be meaningfully compared (e.g. a string label against a bare number) simply never match, rather than erroring.
fn compare_metric_primitive(
    primitive: &MetricPrimitive,
    op: CompareOp,
    value: &FilterValue,
    now: Timestamp,
) -> Result<bool, String> {
    let ordering = match (primitive, value) {
        (MetricPrimitive::Number(n), FilterValue::Number(v)) => n.as_f64().partial_cmp(v),
        (MetricPrimitive::Number(n), FilterValue::Raw(raw)) => {
            let v: f64 = raw
                .parse()
                .map_err(|_| format!("expected a numeric value, got {raw:?}"))?;
            n.as_f64().partial_cmp(&v)
        }
        (MetricPrimitive::Timestamp(ts), FilterValue::Number(v)) => {
            ts.as_millis().partial_cmp(&(*v as i64))
        }
        (MetricPrimitive::Timestamp(ts), FilterValue::Raw(raw)) => {
            let target = filter_datetime::parse_datetime(raw, now).map_err(|err| err.to_string())?;
            ts.as_millis().partial_cmp(&target.as_millis())
        }
        (MetricPrimitive::String(s), FilterValue::Raw(raw)) => s.as_str().partial_cmp(raw.as_str()),
        (MetricPrimitive::String(s), FilterValue::Number(n)) => {
            s.as_str().partial_cmp(n.to_string().as_str())
        }
        (MetricPrimitive::Duration(ms), FilterValue::Number(v)) => (*ms as f64).partial_cmp(v),
        (MetricPrimitive::Duration(ms), FilterValue::Raw(raw)) => {
            let v: f64 = raw
                .parse()
                .map_err(|_| format!("expected a numeric value, got {raw:?}"))?;
            (*ms as f64).partial_cmp(&v)
        }
        (MetricPrimitive::Boolean(b), FilterValue::Raw(raw)) => {
            let v: bool = raw
                .parse()
                .map_err(|_| format!("expected true or false, got {raw:?}"))?;
            b.partial_cmp(&v)
        }
        (MetricPrimitive::Boolean(_), FilterValue::Number(_)) => None,
    };
    Ok(ordering.is_some_and(|ordering| apply_compare_op(ordering, op)))
}

/// `json[field] <op> value`: the log line is parsed as JSON fresh on every call (no per-line
/// cache), since this doesn't go through the index-backed pre-resolution `MetricMatch` uses.
/// Any line that isn't a JSON object, or lacks `field`, simply doesn't match — no error.
pub(crate) fn eval_json_comparison(text: &str, field: &str, op: CompareOp, value: &FilterValue) -> bool {
    let Ok(parsed) = serde_json::from_str::<serde_json::Value>(text) else {
        return false;
    };
    let Some(field_value) = parsed.as_object().and_then(|obj| obj.get(field)) else {
        return false;
    };
    compare_json_value(field_value, op, value)
}

/// Mirrors `compare_metric_primitive`'s "mismatched kinds never match" behavior. Only `Eq`/`Ne`
/// are meaningful for booleans, and `null`/array/object fields aren't scalar-comparable at all.
fn compare_json_value(field_value: &serde_json::Value, op: CompareOp, value: &FilterValue) -> bool {
    match field_value {
        serde_json::Value::Number(n) => {
            let target = match value {
                FilterValue::Number(v) => Some(*v),
                FilterValue::Raw(raw) => raw.parse::<f64>().ok(),
            };
            match (n.as_f64(), target) {
                (Some(f), Some(v)) => f.partial_cmp(&v).is_some_and(|ordering| apply_compare_op(ordering, op)),
                _ => false,
            }
        }
        serde_json::Value::String(s) => {
            let target = match value {
                FilterValue::Raw(raw) => raw.clone(),
                FilterValue::Number(n) => n.to_string(),
            };
            s.as_str()
                .partial_cmp(target.as_str())
                .is_some_and(|ordering| apply_compare_op(ordering, op))
        }
        serde_json::Value::Bool(b) => {
            let FilterValue::Raw(raw) = value else {
                return false;
            };
            match raw.parse::<bool>() {
                Ok(v) => match op {
                    CompareOp::Eq => *b == v,
                    CompareOp::Ne => *b != v,
                    _ => false,
                },
                Err(_) => false,
            }
        }
        serde_json::Value::Null | serde_json::Value::Array(_) | serde_json::Value::Object(_) => false,
    }
}

/// `pub(crate)` so `crate::alerts` can reuse it for `Threshold`/`Ratio` comparisons instead of duplicating the match.
pub(crate) fn apply_compare_op(ordering: Ordering, op: CompareOp) -> bool {
    match op {
        CompareOp::Lt => ordering == Ordering::Less,
        CompareOp::Gt => ordering == Ordering::Greater,
        CompareOp::Le => ordering != Ordering::Greater,
        CompareOp::Ge => ordering != Ordering::Less,
        CompareOp::Eq => ordering == Ordering::Equal,
        CompareOp::Ne => ordering != Ordering::Equal,
    }
}

/// Cheap: everything was already resolved by [`compile_filter_query`], so this is just pattern matching and `HashSet` lookups — no index access.
fn eval_compiled(compiled: &CompiledFilter, sequence: u64, timestamp: Timestamp, text: &str) -> bool {
    match compiled {
        CompiledFilter::Regex(re) => re.is_match(text),
        CompiledFilter::Sequence { op, target } => {
            apply_compare_op((sequence as f64).total_cmp(target), *op)
        }
        CompiledFilter::Timestamp { op, target_ms } => {
            apply_compare_op(timestamp.as_millis().cmp(target_ms), *op)
        }
        CompiledFilter::MetricMatch(sequences) => sequences.contains(&sequence),
        CompiledFilter::Json { field, op, value } => eval_json_comparison(text, field, *op, value),
        CompiledFilter::And(lhs, rhs) => {
            eval_compiled(lhs, sequence, timestamp, text) && eval_compiled(rhs, sequence, timestamp, text)
        }
        CompiledFilter::Or(lhs, rhs) => {
            eval_compiled(lhs, sequence, timestamp, text) || eval_compiled(rhs, sequence, timestamp, text)
        }
    }
}

fn metric_declarations_range(
    index: &BTreeMap<IndexKey, DataValue>,
) -> impl Iterator<
    Item = (
        &str,
        &str,
        &[RegexToMetricPrimitiveMatcher],
        Option<&BTreeMap<String, String>>,
        Option<&BTreeMap<String, String>>,
    ),
> {
    index
        .range(IndexKey::ParamMetricSplit..IndexKey::MetricBucketSplit)
        .filter_map(|(key, value)| match (key, value) {
            (
                IndexKey::MetricDeclaration { name },
                DataValue::MetricDeclaration { message_match_regex, params, matched_labels, unmatched_labels },
            ) => Some((
                name.as_str(),
                message_match_regex.as_str(),
                params.as_slice(),
                matched_labels.as_ref(),
                unmatched_labels.as_ref(),
            )),
            _ => None,
        })
}

fn param_declarations_range(
    index: &BTreeMap<IndexKey, DataValue>,
) -> impl Iterator<Item = (&str, &str, &str, Option<&str>, MetricPrimitiveKind)> {
    index
        .range(IndexKey::LogParamSplit..IndexKey::ParamMetricSplit)
        .filter_map(|(key, value)| match (key, value) {
            (
                IndexKey::ParamDeclaration { name },
                DataValue::ParamDeclaration { message_match_regex, capture_group, json_field_path, kind },
            ) => Some((
                name.as_str(),
                message_match_regex.as_str(),
                capture_group.as_str(),
                json_field_path.as_deref(),
                *kind,
            )),
            _ => None,
        })
}

/// A param whose stored regex fails to compile (shouldn't happen — `declare_param` validates it) is skipped rather than panicking.
fn compiled_param_declarations(
    index: &BTreeMap<IndexKey, DataValue>,
) -> Vec<(String, Regex, String, Option<String>, MetricPrimitiveKind)> {
    param_declarations_range(index)
        .filter_map(|(name, message_match_regex, capture_group, json_field_path, kind)| {
            Regex::new(message_match_regex).ok().map(|re| {
                (
                    name.to_string(),
                    re,
                    capture_group.to_string(),
                    json_field_path.map(|path| path.to_string()),
                    kind,
                )
            })
        })
        .collect()
}

/// A param is omitted (not included as null/empty) when its regex/capture group (or JSON field
/// path) doesn't match, or the extracted text fails to parse — see `LogEntryDto::params`. The one
/// exception is a `Boolean`-kind param, which never uses a capture group and is instead present as
/// `true`/`false` on every line based purely on whether `message_match_regex` matches
/// (case-insensitively or per the regex's own flags).
fn derive_log_params(
    text: &str,
    declarations: &[(String, Regex, String, Option<String>, MetricPrimitiveKind)],
) -> BTreeMap<String, MetricPrimitive> {
    let mut params = BTreeMap::new();
    for (name, regex, capture_group, json_field_path, kind) in declarations {
        if *kind == MetricPrimitiveKind::Boolean {
            params.insert(name.clone(), MetricPrimitive::Boolean(regex.is_match(text)));
            continue;
        }
        let Some(captures) = regex.captures(text) else {
            continue;
        };
        let raw: String = match json_field_path.as_deref().filter(|path| !path.is_empty()) {
            Some(field_path) => match crate::data_value::extract_json_field(text, field_path) {
                Some(value) => value,
                None => continue,
            },
            None => match captures.name(capture_group) {
                Some(m) => m.as_str().to_string(),
                None => continue,
            },
        };
        let Some(value) = parse_metric_primitive(&raw, *kind) else {
            continue;
        };
        params.insert(name.clone(), value);
    }
    params
}

/// Ticket 011's approximate per-log byte size: see `StatsDto::approx_bytes_total`'s
/// doc comment for why the raw UTF-8 text length is the chosen proxy rather than a
/// full struct-size estimate.
fn approx_log_bytes(text: &str) -> u64 {
    text.len() as u64
}

/// Evicts the oldest (lowest-`sequence`) ingested logs until `log_count` is
/// back at or under `max_logs`. `sequence` numbering is not renumbered on
/// eviction, and any `Bucket` values already derived from an evicted log are
/// left in place — buckets are aggregate data, not raw logs, so they're not
/// retroactively pruned. `approx_bytes_total` (ticket 011) is decremented by
/// each evicted log's approximate size, mirroring the increment on ingest.
fn evict_excess_logs(
    index: &mut BTreeMap<IndexKey, DataValue>,
    log_count: &mut usize,
    max_logs: Option<usize>,
    approx_bytes_total: &mut u64,
) {
    let Some(max_logs) = max_logs else { return };
    while *log_count > max_logs {
        let Some(oldest_key) = index
            .range(IndexKey::Start..IndexKey::LogParamSplit)
            .next()
            .map(|(key, _)| key.clone())
        else {
            break;
        };
        if let Some(DataValue::LogLiteral { text }) = index.get(&oldest_key) {
            *approx_bytes_total = approx_bytes_total.saturating_sub(approx_log_bytes(text));
        }
        index.remove(&oldest_key);
        *log_count -= 1;
    }
}

fn logs_range(index: &BTreeMap<IndexKey, DataValue>) -> impl Iterator<Item = (u64, Timestamp, &str)> {
    index.range(IndexKey::Start..IndexKey::LogParamSplit).filter_map(|(key, value)| {
        match (key, value) {
            (IndexKey::IngestedLog { sequence, timestamp }, DataValue::LogLiteral { text }) => {
                Some((*sequence, *timestamp, text.as_str()))
            }
            _ => None,
        }
    })
}

fn handle_query(
    index: &BTreeMap<IndexKey, DataValue>,
    query: EngineQuery,
    approx_bytes_total: u64,
) -> EngineResponse {
    match query {
        EngineQuery::Logs(params) => {
            let compiled_filter = match params.filter.as_deref() {
                Some(filter) => match compile_filter_query(index, filter, Timestamp::now()) {
                    Ok(compiled) => Some(compiled),
                    Err(err) => return EngineResponse::Error(err),
                },
                None => None,
            };
            let limit = params.limit.unwrap_or(500).min(10_000);
            // O(lines_returned * declared_params) per query — fine at v1 scale; if volume/params grow, move to incremental derivation at ingest time like apply_all_metrics does for buckets.
            let compiled_params = compiled_param_declarations(index);
            let entries: Vec<LogEntryDto> = logs_range(index)
                .filter(|(sequence, timestamp, _)| {
                    params.since_sequence.is_none_or(|since| *sequence > since)
                        && params.start_ms.is_none_or(|start| timestamp.as_millis() >= start)
                        && params.end_ms.is_none_or(|end| timestamp.as_millis() <= end)
                })
                .filter(|(sequence, timestamp, text)| {
                    compiled_filter
                        .as_ref()
                        .is_none_or(|filter| eval_compiled(filter, *sequence, *timestamp, text))
                })
                .take(limit)
                .map(|(sequence, timestamp, text)| LogEntryDto {
                    sequence,
                    timestamp_ms: timestamp.as_millis(),
                    text: text.to_string(),
                    params: derive_log_params(text, &compiled_params),
                })
                .collect();
            EngineResponse::Logs(entries)
        }
        EngineQuery::Timeline(params) => {
            let compiled_filter = match params.filter.as_deref() {
                Some(filter) => match compile_filter_query(index, filter, Timestamp::now()) {
                    Ok(compiled) => Some(compiled),
                    Err(err) => return EngineResponse::Error(err),
                },
                None => None,
            };
            let bucket_ms = params.bucket_ms.unwrap_or(1_000).max(1);
            let mut buckets: BTreeMap<i64, u64> = BTreeMap::new();
            for (sequence, timestamp, text) in logs_range(index).filter(|(_, timestamp, _)| {
                params.start_ms.is_none_or(|start| timestamp.as_millis() >= start)
                    && params.end_ms.is_none_or(|end| timestamp.as_millis() <= end)
            }) {
                if compiled_filter
                    .as_ref()
                    .is_some_and(|filter| !eval_compiled(filter, sequence, timestamp, text))
                {
                    continue;
                }
                let bucket_start = (timestamp.as_millis() / bucket_ms) * bucket_ms;
                *buckets.entry(bucket_start).or_insert(0) += 1;
            }
            let out = buckets
                .into_iter()
                .map(|(start_ms, count)| TimelineBucketDto { start_ms, count })
                .collect();
            EngineResponse::Timeline(out)
        }
        EngineQuery::Stats => {
            let mut log_count = 0u64;
            let mut first_sequence = None;
            let mut last_sequence = None;
            let mut first_timestamp_ms = None;
            let mut last_timestamp_ms = None;
            for (sequence, timestamp, _) in logs_range(index) {
                log_count += 1;
                first_sequence.get_or_insert(sequence);
                last_sequence = Some(sequence);
                first_timestamp_ms.get_or_insert(timestamp.as_millis());
                last_timestamp_ms = Some(timestamp.as_millis());
            }
            EngineResponse::Stats(StatsDto {
                log_count,
                first_sequence,
                last_sequence,
                first_timestamp_ms,
                last_timestamp_ms,
                approx_bytes_total,
            })
        }
        EngineQuery::ParamDeclarations => {
            let declarations = param_declarations_range(index)
                .map(|(name, message_match_regex, capture_group, json_field_path, kind)| ParamDeclarationDto {
                    name: name.to_string(),
                    message_match_regex: message_match_regex.to_string(),
                    capture_group: capture_group.to_string(),
                    json_field_path: json_field_path.map(|path| path.to_string()),
                    kind,
                })
                .collect();
            EngineResponse::ParamDeclarations(declarations)
        }
        EngineQuery::MetricDeclarations => {
            let declarations = metric_declarations_range(index)
                .map(|(name, message_match_regex, params, matched_labels, unmatched_labels)| MetricDeclarationDto {
                    name: name.to_string(),
                    message_match_regex: message_match_regex.to_string(),
                    params: params.to_vec(),
                    matched_labels: matched_labels.cloned(),
                    unmatched_labels: unmatched_labels.cloned(),
                })
                .collect();
            EngineResponse::MetricDeclarations(declarations)
        }
        EngineQuery::BucketValues {
            metric_name,
            labels_filter,
            start_ms,
            end_ms,
            since_sequence,
            limit,
        } => {
            let limit = limit.unwrap_or(500).min(10_000);
            let values: Vec<BucketValueDto> = bucket_values(index, &metric_name, None)
                .filter(|(labels, _)| {
                    labels_filter.as_ref().is_none_or(|filter| {
                        filter.iter().all(|(name, expected)| {
                            labels
                                .get(name)
                                .is_some_and(|value| &value.to_string() == expected)
                        })
                    })
                })
                .filter(|(_, (sequence, timestamp, _))| {
                    since_sequence.is_none_or(|since| *sequence > since)
                        && start_ms.is_none_or(|start| timestamp.as_millis() >= start)
                        && end_ms.is_none_or(|end| timestamp.as_millis() <= end)
                })
                .take(limit)
                .map(|(labels, (sequence, timestamp, value))| BucketValueDto {
                    labels: labels.clone(),
                    sequence: *sequence,
                    timestamp_ms: timestamp.as_millis(),
                    value: value.clone(),
                })
                .collect();
            EngineResponse::BucketValues(values)
        }
        EngineQuery::Promql { query, start_ms, end_ms } => {
            let response = promql::parse_query(&query)
                .map(|parsed| evaluate_promql(index, &parsed, start_ms, end_ms));
            EngineResponse::Promql(response)
        }
    }
}

/// Coerces a bucket's `MetricPrimitive` into a plottable `f64`: Number/Timestamp/Duration use their natural numeric value; `String` has no numeric reading, so it falls back to `1.0` (matching the presence/occurrence marker used for every bucket) rather than being skipped, keeping every declared metric queryable.
fn coerce_to_f64(value: &MetricPrimitive) -> f64 {
    match value {
        MetricPrimitive::Number(n) => match n {
            Number::Int(v) => *v as f64,
            Number::Float(v) => *v,
        },
        MetricPrimitive::Timestamp(ts) => ts.as_millis() as f64,
        MetricPrimitive::Duration(ms) => *ms as f64,
        MetricPrimitive::String(_) => 1.0,
        MetricPrimitive::Boolean(b) => {
            if *b {
                1.0
            } else {
                0.0
            }
        }
    }
}

/// Compared against each label's `Display` form; extra labels on the bucket are ignored — the matcher list only needs to be a subset.
fn matches_label_matchers(
    labels: &BTreeMap<String, MetricPrimitive>,
    matchers: &[LabelMatcher],
) -> bool {
    matchers.iter().all(|matcher| {
        labels.get(&matcher.name).is_some_and(|value| value.to_string() == matcher.value)
    })
}

/// Evaluates a parsed PromQL query against `index`, producing chart-ready series: baseline is one series per label combination; `sum()`/`avg()` merge series by timestamp into one label-less series; `rate()` is simplified to a single point over `(last_ts - range, last_ts]` (an occurrence-count-per-second reading, not PromQL's true sliding-window-per-step) since bucket values are occurrence markers, not cumulative counters.
fn evaluate_promql(
    index: &BTreeMap<IndexKey, DataValue>,
    parsed: &PromqlQuery,
    start_ms: Option<i64>,
    end_ms: Option<i64>,
) -> Vec<PromqlSeriesDto> {
    let mut series_points: BTreeMap<BTreeMap<String, String>, Vec<(i64, f64)>> = BTreeMap::new();
    for (labels, (_, timestamp, value)) in bucket_values(index, &parsed.metric_name, None)
        .filter(|(labels, _)| matches_label_matchers(labels, &parsed.matchers))
        .filter(|(_, (_, timestamp, _))| {
            start_ms.is_none_or(|start| timestamp.as_millis() >= start)
                && end_ms.is_none_or(|end| timestamp.as_millis() <= end)
        })
    {
        let display_labels: BTreeMap<String, String> =
            labels.iter().map(|(name, value)| (name.clone(), value.to_string())).collect();
        series_points
            .entry(display_labels)
            .or_default()
            .push((timestamp.as_millis(), coerce_to_f64(value)));
    }
    for points in series_points.values_mut() {
        points.sort_by_key(|(timestamp_ms, _)| *timestamp_ms);
    }

    match parsed.agg {
        None => series_points
            .into_iter()
            .map(|(labels, points)| PromqlSeriesDto { labels, points })
            .collect(),
        Some(AggOp::Sum) | Some(AggOp::Avg) => {
            let mut grouped: BTreeMap<i64, Vec<f64>> = BTreeMap::new();
            for points in series_points.values() {
                for (timestamp_ms, value) in points {
                    grouped.entry(*timestamp_ms).or_default().push(*value);
                }
            }
            let is_avg = parsed.agg == Some(AggOp::Avg);
            let points = grouped
                .into_iter()
                .map(|(timestamp_ms, values)| {
                    let sum: f64 = values.iter().sum();
                    let value = if is_avg { sum / values.len() as f64 } else { sum };
                    (timestamp_ms, value)
                })
                .collect();
            vec![PromqlSeriesDto { labels: BTreeMap::new(), points }]
        }
        Some(AggOp::Rate) => {
            // Validated by `promql::parse_query`: `rate()` never parses without one.
            let range_ms = parsed.range.expect("rate() always carries a range").as_millis() as i64;
            let range_seconds = range_ms as f64 / 1000.0;
            series_points
                .into_iter()
                .filter_map(|(labels, points)| {
                    let last_ts = points.last()?.0;
                    let window_start = last_ts - range_ms;
                    let count_in_window =
                        points.iter().filter(|(ts, _)| *ts > window_start && *ts <= last_ts).count();
                    let rate = count_in_window as f64 / range_seconds;
                    Some(PromqlSeriesDto { labels, points: vec![(last_ts, rate)] })
                })
                .collect()
        }
        Some(AggOp::Gauge) => {
            // Collapse each label combination's events down to their most-recent value in range — "value at time T", not a distinct stored gauge kind.
            series_points
                .into_iter()
                .filter_map(|(labels, points)| {
                    let last = *points.last()?;
                    Some(PromqlSeriesDto { labels, points: vec![last] })
                })
                .collect()
        }
        Some(AggOp::Histogram) => {
            // Validated by `promql::parse_query`: `histogram()` never parses
            // without one.
            let bucket_width =
                parsed.bucket_width.expect("histogram() always carries a bucket width");
            let last_ts = series_points
                .values()
                .flat_map(|points| points.iter().map(|(ts, _)| *ts))
                .max()
                .unwrap_or(0);
            // Buckets every value (across all label combinations, collapsing the label dimension like sum()/avg()) into fixed-width buckets, labeled `le` (Prometheus's less-than-or-equal convention).
            let mut counts: BTreeMap<i64, u64> = BTreeMap::new();
            for points in series_points.values() {
                for (_, value) in points {
                    let bucket_index = (value / bucket_width).floor() as i64;
                    *counts.entry(bucket_index).or_insert(0) += 1;
                }
            }
            counts
                .into_iter()
                .map(|(bucket_index, count)| {
                    let upper_bound = (bucket_index + 1) as f64 * bucket_width;
                    let labels = BTreeMap::from([("le".to_string(), upper_bound.to_string())]);
                    PromqlSeriesDto { labels, points: vec![(last_ts, count as f64)] }
                })
                .collect()
        }
    }
}

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

    fn request(name: &str, pattern: &str, capture_group: &str) -> DeclareParamRequest {
        DeclareParamRequest {
            name: name.to_string(),
            message_match_regex: pattern.to_string(),
            capture_group: capture_group.to_string(),
            json_field_path: None,
            kind: MetricPrimitiveKind::String,
        }
    }

    #[test]
    fn declares_valid_param() {
        let mut index = BTreeMap::new();
        let result = declare_param(&mut index, request("level", r"level=(?P<level>\w+)", "level"));
        assert!(result.is_ok());

        let declarations: Vec<_> = param_declarations_range(&index).collect();
        assert_eq!(declarations.len(), 1);
        assert_eq!(declarations[0].0, "level");
        assert_eq!(declarations[0].1, r"level=(?P<level>\w+)");
        assert_eq!(declarations[0].2, "level");
    }

    #[test]
    fn rejects_invalid_regex() {
        let mut index = BTreeMap::new();
        let result = declare_param(&mut index, request("level", "level=(", "level"));
        assert!(result.is_err());
        assert!(index.is_empty());
    }

    #[test]
    fn rejects_unknown_capture_group() {
        let mut index = BTreeMap::new();
        let result = declare_param(
            &mut index,
            request("level", r"level=(?P<level>\w+)", "missing"),
        );
        assert!(result.is_err());
        assert!(index.is_empty());
    }

    #[test]
    fn declares_valid_param_from_json_field_path() {
        let mut index = BTreeMap::new();
        let result = declare_param(
            &mut index,
            DeclareParamRequest {
                name: "level".to_string(),
                // No named capture group needed for JSON-field extraction, just something that
                // matches the lines this param applies to.
                message_match_regex: r".".to_string(),
                capture_group: String::new(),
                json_field_path: Some("foo.bar".to_string()),
                kind: MetricPrimitiveKind::String,
            },
        );
        assert!(result.is_ok());

        let declarations: Vec<_> = param_declarations_range(&index).collect();
        assert_eq!(declarations.len(), 1);
        assert_eq!(declarations[0].3, Some("foo.bar"));
    }

    #[test]
    fn derives_log_param_from_json_field_path() {
        let mut index = BTreeMap::new();
        declare_param(
            &mut index,
            DeclareParamRequest {
                name: "bar".to_string(),
                message_match_regex: r".".to_string(),
                capture_group: String::new(),
                json_field_path: Some("foo.bar".to_string()),
                kind: MetricPrimitiveKind::Number,
            },
        )
        .unwrap();

        let declarations = compiled_param_declarations(&index);
        let params = derive_log_params(r#"{"foo":{"bar":42}}"#, &declarations);
        assert_eq!(params.get("bar"), Some(&MetricPrimitive::Number(Number::Int(42))));

        // A line that doesn't parse as JSON, or is missing the field, yields no param.
        assert!(derive_log_params("not json", &declarations).is_empty());
        assert!(derive_log_params(r#"{"foo":{}}"#, &declarations).is_empty());
    }

    #[test]
    fn evict_excess_logs_drops_oldest_sequences_first() {
        let mut index = BTreeMap::new();
        let mut log_count = 0;
        let mut approx_bytes_total = 0u64;
        for sequence in 0..5u64 {
            let text = format!("line {sequence}");
            approx_bytes_total += approx_log_bytes(&text);
            index.insert(
                IndexKey::IngestedLog { sequence, timestamp: Timestamp::from_millis(sequence as i64) },
                DataValue::LogLiteral { text },
            );
            log_count += 1;
        }

        evict_excess_logs(&mut index, &mut log_count, Some(3), &mut approx_bytes_total);

        assert_eq!(log_count, 3);
        let remaining: Vec<_> = logs_range(&index).map(|(seq, _, _)| seq).collect();
        assert_eq!(remaining, vec![2, 3, 4]);
        let expected_bytes: u64 =
            remaining.iter().map(|seq| approx_log_bytes(&format!("line {seq}"))).sum();
        assert_eq!(approx_bytes_total, expected_bytes);
    }

    #[test]
    fn evict_excess_logs_is_noop_when_unlimited_or_under_cap() {
        let mut index = BTreeMap::new();
        let mut log_count = 0;
        let mut approx_bytes_total = 0u64;
        for sequence in 0..3u64 {
            approx_bytes_total += approx_log_bytes("x");
            index.insert(
                IndexKey::IngestedLog { sequence, timestamp: Timestamp::from_millis(sequence as i64) },
                DataValue::LogLiteral { text: "x".to_string() },
            );
            log_count += 1;
        }

        evict_excess_logs(&mut index, &mut log_count, None, &mut approx_bytes_total);
        assert_eq!(log_count, 3);
        assert_eq!(approx_bytes_total, 3);

        evict_excess_logs(&mut index, &mut log_count, Some(10), &mut approx_bytes_total);
        assert_eq!(log_count, 3);
        assert_eq!(approx_bytes_total, 3);
    }

    #[test]
    fn declarations_stay_within_their_range() {
        let mut index = BTreeMap::new();
        index.insert(
            IndexKey::IngestedLog { sequence: 0, timestamp: Timestamp::from_millis(0) },
            DataValue::LogLiteral { text: "hello".to_string() },
        );
        declare_param(&mut index, request("level", r"level=(?P<level>\w+)", "level")).unwrap();
        index.insert(
            IndexKey::MetricDeclaration { name: "requests".to_string() },
            DataValue::MetricDeclaration {
                message_match_regex: "x".to_string(),
                params: vec![],
                matched_labels: None,
                unmatched_labels: None,
            },
        );

        let declarations: Vec<_> = param_declarations_range(&index).collect();
        assert_eq!(declarations.len(), 1);
        assert_eq!(declarations[0].0, "level");

        for key in index.range(IndexKey::LogParamSplit..IndexKey::ParamMetricSplit).map(|(k, _)| k) {
            assert!(matches!(key, IndexKey::ParamDeclaration { .. }));
        }
    }

    #[test]
    fn removes_declaration() {
        let mut index = BTreeMap::new();
        declare_param(&mut index, request("level", r"level=(?P<level>\w+)", "level")).unwrap();
        assert!(index.remove(&IndexKey::ParamDeclaration { name: "level".to_string() }).is_some());
        assert!(param_declarations_range(&index).next().is_none());
    }

    fn metric_request(
        name: &str,
        pattern: &str,
        params: Vec<(&str, &str, MetricPrimitiveKind)>,
    ) -> DeclareMetricRequest {
        metric_request_with_roles(
            name,
            pattern,
            params
                .into_iter()
                .map(|(param_name, capture_group, kind)| (param_name, capture_group, kind, ParamRole::Label))
                .collect(),
        )
    }

    fn metric_request_with_roles(
        name: &str,
        pattern: &str,
        params: Vec<(&str, &str, MetricPrimitiveKind, ParamRole)>,
    ) -> DeclareMetricRequest {
        DeclareMetricRequest {
            name: name.to_string(),
            message_match_regex: pattern.to_string(),
            matched_labels: None,
            unmatched_labels: None,
            params: params
                .into_iter()
                .map(|(param_name, capture_group, kind, role)| RegexToMetricPrimitiveMatcher {
                    param_name: param_name.to_string(),
                    capture_group: capture_group.to_string(),
                    json_field_path: None,
                    kind,
                    role,
                })
                .collect(),
        }
    }

    #[test]
    fn declares_valid_metric() {
        let mut index = BTreeMap::new();
        let result = declare_metric(
            &mut index,
            metric_request(
                "requests",
                r"request path=(?P<path>\S+) duration=(?P<duration>\d+)",
                vec![
                    ("path", "path", MetricPrimitiveKind::String),
                    ("duration", "duration", MetricPrimitiveKind::Number),
                ],
            ),
        );
        assert!(result.is_ok());

        let declarations: Vec<_> = metric_declarations_range(&index).collect();
        assert_eq!(declarations.len(), 1);
        assert_eq!(declarations[0].0, "requests");
        assert_eq!(
            declarations[0].1,
            r"request path=(?P<path>\S+) duration=(?P<duration>\d+)"
        );
        assert_eq!(declarations[0].2.len(), 2);
    }

    #[test]
    fn rejects_invalid_metric_regex() {
        let mut index = BTreeMap::new();
        let result = declare_metric(&mut index, metric_request("requests", "path=(", vec![]));
        assert!(result.is_err());
        assert!(index.is_empty());
    }

    #[test]
    fn rejects_metric_matcher_with_unknown_capture_group() {
        let mut index = BTreeMap::new();
        let result = declare_metric(
            &mut index,
            metric_request(
                "requests",
                r"request path=(?P<path>\S+)",
                vec![
                    ("path", "path", MetricPrimitiveKind::String),
                    ("duration", "duration", MetricPrimitiveKind::Number),
                ],
            ),
        );
        assert!(result.is_err());
        // Both matchers are validated: the error mentions the one that's actually bad.
        assert!(result.unwrap_err().contains("duration"));
        assert!(index.is_empty());
    }

    #[test]
    fn metric_declarations_stay_within_their_range() {
        let mut index = BTreeMap::new();
        index.insert(
            IndexKey::IngestedLog { sequence: 0, timestamp: Timestamp::from_millis(0) },
            DataValue::LogLiteral { text: "hello".to_string() },
        );
        declare_param(&mut index, request("level", r"level=(?P<level>\w+)", "level")).unwrap();
        declare_metric(
            &mut index,
            metric_request(
                "requests",
                r"duration=(?P<duration>\d+)",
                vec![("duration", "duration", MetricPrimitiveKind::Number)],
            ),
        )
        .unwrap();

        let declarations: Vec<_> = metric_declarations_range(&index).collect();
        assert_eq!(declarations.len(), 1);
        assert_eq!(declarations[0].0, "requests");

        for key in index.range(IndexKey::ParamMetricSplit..IndexKey::MetricBucketSplit).map(|(k, _)| k) {
            assert!(matches!(key, IndexKey::MetricDeclaration { .. }));
        }
    }

    #[test]
    fn removes_metric_declaration() {
        let mut index = BTreeMap::new();
        declare_metric(
            &mut index,
            metric_request(
                "requests",
                r"duration=(?P<duration>\d+)",
                vec![("duration", "duration", MetricPrimitiveKind::Number)],
            ),
        )
        .unwrap();
        assert!(index.remove(&IndexKey::MetricDeclaration { name: "requests".to_string() }).is_some());
        assert!(metric_declarations_range(&index).next().is_none());
    }

    #[test]
    fn update_metric_rejects_when_no_declaration_exists_yet() {
        let mut index = BTreeMap::new();
        let result = update_metric(
            &mut index,
            "requests",
            metric_request("requests", r"request path=(?P<path>\S+)", vec![]),
        );
        assert!(result.is_err());
        assert!(result.unwrap_err().contains("requests"));
    }

    #[test]
    fn update_metric_reuses_declare_metric_validation() {
        let mut index = BTreeMap::new();
        declare_metric(&mut index, metric_request("requests", r"request path=(?P<path>\S+)", vec![])).unwrap();

        let result = update_metric(
            &mut index,
            "requests",
            metric_request(
                "requests",
                r"request path=(?P<path>\S+)",
                vec![("duration", "duration", MetricPrimitiveKind::Number)],
            ),
        );
        assert!(result.is_err());
        assert!(result.unwrap_err().contains("duration"));
    }

    #[test]
    fn update_metric_replaces_declaration_in_place_without_duplicating() {
        let mut index = BTreeMap::new();
        declare_metric(
            &mut index,
            metric_request(
                "requests",
                r"request path=(?P<path>\S+)",
                vec![("path", "path", MetricPrimitiveKind::String)],
            ),
        )
        .unwrap();

        let result = update_metric(
            &mut index,
            "requests",
            metric_request(
                "requests",
                r"request path=(?P<path>\S+) duration=(?P<duration>\d+)",
                vec![
                    ("path", "path", MetricPrimitiveKind::String),
                    ("duration", "duration", MetricPrimitiveKind::Number),
                ],
            ),
        );
        assert!(result.is_ok());

        // Still one declaration named "requests", now with the updated regex/params.
        let declarations: Vec<_> = metric_declarations_range(&index).collect();
        assert_eq!(declarations.len(), 1);
        assert_eq!(declarations[0].0, "requests");
        assert_eq!(declarations[0].2.len(), 2);
    }

    #[test]
    fn update_metric_rebackfills_bucket_data_from_the_new_declaration() {
        let mut index = BTreeMap::new();
        ingest_log(&mut index, 0, "request path=/a");
        ingest_log(&mut index, 1, "request path=/b");
        declare_metric(
            &mut index,
            metric_request(
                "requests",
                r"request path=(?P<path>\S+)",
                vec![("path", "path", MetricPrimitiveKind::String)],
            ),
        )
        .unwrap();
        assert_eq!(bucket_values(&index, "requests", None).count(), 2);

        // Editing to a narrower regex should leave exactly the buckets matching the *new* declaration — not the old ones layered underneath.
        update_metric(
            &mut index,
            "requests",
            metric_request(
                "requests",
                r"request path=(?P<path>/a)",
                vec![("path", "path", MetricPrimitiveKind::String)],
            ),
        )
        .unwrap();
        assert_eq!(bucket_values(&index, "requests", None).count(), 1);
    }

    #[test]
    fn update_metric_can_rename_and_rejects_a_rename_onto_an_existing_name() {
        let mut index = BTreeMap::new();
        declare_metric(&mut index, metric_request("requests", r"request path=(?P<path>\S+)", vec![])).unwrap();
        declare_metric(&mut index, metric_request("errors", r"error code=(?P<code>\d+)", vec![])).unwrap();

        // Renaming onto an already-taken name is rejected.
        let result = update_metric(
            &mut index,
            "requests",
            metric_request("errors", r"request path=(?P<path>\S+)", vec![]),
        );
        assert!(result.is_err());

        // Renaming onto a fresh name succeeds and drops the old declaration.
        update_metric(
            &mut index,
            "requests",
            metric_request("http_requests", r"request path=(?P<path>\S+)", vec![]),
        )
        .unwrap();
        let names: Vec<_> = metric_declarations_range(&index).map(|(name, ..)| name.to_string()).collect();
        assert!(names.contains(&"http_requests".to_string()));
        assert!(!names.contains(&"requests".to_string()));
    }

    fn ingest_log(index: &mut BTreeMap<IndexKey, DataValue>, sequence: u64, text: &str) {
        let timestamp = Timestamp::from_millis(sequence as i64);
        index.insert(
            IndexKey::IngestedLog { sequence, timestamp },
            DataValue::LogLiteral { text: text.to_string() },
        );
        apply_all_metrics(index, sequence, timestamp, text);
    }

    #[test]
    fn declaring_metric_backfills_matching_logs() {
        let mut index = BTreeMap::new();
        ingest_log(&mut index, 0, "request path=/a duration=1");
        ingest_log(&mut index, 1, "request path=/b duration=2");
        ingest_log(&mut index, 2, "unrelated line");

        declare_metric(
            &mut index,
            metric_request(
                "requests",
                r"request path=(?P<path>\S+) duration=(?P<duration>\d+)",
                vec![("path", "path", MetricPrimitiveKind::String)],
            ),
        )
        .unwrap();

        let values: Vec<_> = bucket_values(&index, "requests", None).collect();
        assert_eq!(values.len(), 2);
    }

    #[test]
    fn metric_param_extracts_from_json_field_path() {
        let mut index = BTreeMap::new();
        ingest_log(&mut index, 0, r#"{"request":{"path":"/a"}}"#);
        // Doesn't parse as JSON at all — should be skipped, not panic.
        ingest_log(&mut index, 1, "not json");
        // Parses as JSON but is missing the field — also skipped.
        ingest_log(&mut index, 2, r#"{"request":{}}"#);

        declare_metric(
            &mut index,
            DeclareMetricRequest {
                name: "requests".to_string(),
                message_match_regex: r".".to_string(),
                matched_labels: None,
                unmatched_labels: None,
                params: vec![RegexToMetricPrimitiveMatcher {
                    param_name: "path".to_string(),
                    capture_group: String::new(),
                    json_field_path: Some("request.path".to_string()),
                    kind: MetricPrimitiveKind::String,
                    role: ParamRole::Label,
                }],
            },
        )
        .unwrap();

        let values: Vec<_> = bucket_values(&index, "requests", None).collect();
        assert_eq!(values.len(), 1);
        assert_eq!(
            values[0].0.get("path"),
            Some(&MetricPrimitive::String("/a".to_string()))
        );
    }

    #[test]
    fn ingesting_after_declaration_updates_buckets_live() {
        let mut index = BTreeMap::new();
        declare_metric(
            &mut index,
            metric_request(
                "requests",
                r"request path=(?P<path>\S+)",
                vec![("path", "path", MetricPrimitiveKind::String)],
            ),
        )
        .unwrap();

        ingest_log(&mut index, 0, "request path=/a");

        let values: Vec<_> = bucket_values(&index, "requests", None).collect();
        assert_eq!(values.len(), 1);
        assert_eq!(values[0].1.0, 0);
    }

    #[test]
    fn non_matching_log_leaves_bucket_range_untouched() {
        let mut index = BTreeMap::new();
        declare_metric(
            &mut index,
            metric_request(
                "requests",
                r"request path=(?P<path>\S+)",
                vec![("path", "path", MetricPrimitiveKind::String)],
            ),
        )
        .unwrap();

        ingest_log(&mut index, 0, "totally unrelated line");

        assert!(
            index
                .range(IndexKey::MetricBucketSplit..IndexKey::End)
                .find(|(key, _)| matches!(key, IndexKey::Bucket { .. }))
                .is_none()
        );
    }

    #[test]
    fn bucket_values_preserve_insertion_order() {
        let mut index = BTreeMap::new();
        declare_metric(
            &mut index,
            metric_request(
                "requests",
                r"request path=(?P<path>\S+)",
                vec![("path", "path", MetricPrimitiveKind::String)],
            ),
        )
        .unwrap();

        ingest_log(&mut index, 5, "request path=/a");
        ingest_log(&mut index, 2, "request path=/a");
        ingest_log(&mut index, 9, "request path=/a");

        let sequences: Vec<u64> = bucket_values(&index, "requests", None)
            .map(|(_, (sequence, _, _))| *sequence)
            .collect();
        assert_eq!(sequences, vec![5, 2, 9]);
    }

    #[test]
    fn differing_label_combinations_produce_distinct_buckets() {
        let mut index = BTreeMap::new();
        declare_metric(
            &mut index,
            metric_request(
                "requests",
                r"request path=(?P<path>\S+)",
                vec![("path", "path", MetricPrimitiveKind::String)],
            ),
        )
        .unwrap();

        ingest_log(&mut index, 0, "request path=/a");
        ingest_log(&mut index, 1, "request path=/b");

        let bucket_keys: Vec<_> = index
            .range(IndexKey::MetricBucketSplit..IndexKey::End)
            .filter(|(key, _)| matches!(key, IndexKey::Bucket { .. }))
            .collect();
        assert_eq!(bucket_keys.len(), 2);
        for (_, value) in &bucket_keys {
            match value {
                DataValue::BucketValue { values } => assert_eq!(values.len(), 1),
                _ => panic!("expected BucketValue"),
            }
        }
    }

    #[test]
    fn single_log_line_can_match_multiple_metrics() {
        let mut index = BTreeMap::new();
        declare_metric(
            &mut index,
            metric_request(
                "requests",
                r"request path=(?P<path>\S+) duration=(?P<duration>\d+)",
                vec![("path", "path", MetricPrimitiveKind::String)],
            ),
        )
        .unwrap();
        declare_metric(
            &mut index,
            metric_request(
                "durations",
                r"request path=(?P<path>\S+) duration=(?P<duration>\d+)",
                vec![("duration", "duration", MetricPrimitiveKind::Number)],
            ),
        )
        .unwrap();

        ingest_log(&mut index, 0, "request path=/a duration=5");

        assert_eq!(bucket_values(&index, "requests", None).count(), 1);
        assert_eq!(bucket_values(&index, "durations", None).count(), 1);
    }

    fn declare_requests_metric(index: &mut BTreeMap<IndexKey, DataValue>) {
        declare_metric(
            index,
            metric_request(
                "requests",
                r"request path=(?P<path>\S+)",
                vec![("path", "path", MetricPrimitiveKind::String)],
            ),
        )
        .unwrap();
    }

    #[test]
    fn promql_query_selects_matching_label_combination() {
        let mut index = BTreeMap::new();
        declare_requests_metric(&mut index);
        ingest_log(&mut index, 0, "request path=/a");
        ingest_log(&mut index, 1, "request path=/b");
        ingest_log(&mut index, 2, "request path=/a");

        let parsed = promql::parse_query(r#"requests{path="/a"}"#).unwrap();
        let series = evaluate_promql(&index, &parsed, None, None);

        assert_eq!(series.len(), 1);
        assert_eq!(series[0].labels.get("path").map(String::as_str), Some("/a"));
        // Occurrence markers coerce to 1.0 (see coerce_to_f64); timestamps come from ingest_log's sequence-as-millis helper.
        assert_eq!(series[0].points, vec![(0, 1.0), (2, 1.0)]);
    }

    #[test]
    fn promql_query_with_no_matching_labels_returns_no_series() {
        let mut index = BTreeMap::new();
        declare_requests_metric(&mut index);
        ingest_log(&mut index, 0, "request path=/a");

        let parsed = promql::parse_query(r#"requests{path="/nope"}"#).unwrap();
        let series = evaluate_promql(&index, &parsed, None, None);
        assert!(series.is_empty());
    }

    #[test]
    fn promql_sum_aggregates_across_label_combinations() {
        let mut index = BTreeMap::new();
        declare_requests_metric(&mut index);
        ingest_log(&mut index, 0, "request path=/a");
        ingest_log(&mut index, 0, "request path=/b");

        let parsed = promql::parse_query("sum(requests)").unwrap();
        let series = evaluate_promql(&index, &parsed, None, None);

        assert_eq!(series.len(), 1);
        assert!(series[0].labels.is_empty());
        assert_eq!(series[0].points, vec![(0, 2.0)]);
    }

    #[test]
    fn promql_range_bounds_filter_points_via_start_end_ms() {
        let mut index = BTreeMap::new();
        declare_requests_metric(&mut index);
        ingest_log(&mut index, 0, "request path=/a");
        ingest_log(&mut index, 5, "request path=/a");
        ingest_log(&mut index, 10, "request path=/a");

        let parsed = promql::parse_query("requests").unwrap();
        let series = evaluate_promql(&index, &parsed, Some(1), Some(9));

        assert_eq!(series.len(), 1);
        assert_eq!(series[0].points, vec![(5, 1.0)]);
    }

    #[test]
    fn declares_metric_with_value_role_param() {
        let mut index = BTreeMap::new();
        let result = declare_metric(
            &mut index,
            metric_request_with_roles(
                "durations",
                r"request path=(?P<path>\S+) duration=(?P<duration>\d+)",
                vec![
                    ("path", "path", MetricPrimitiveKind::String, ParamRole::Label),
                    ("duration", "duration", MetricPrimitiveKind::Number, ParamRole::Value),
                ],
            ),
        );
        assert!(result.is_ok());
    }

    #[test]
    fn rejects_metric_with_multiple_value_role_params() {
        let mut index = BTreeMap::new();
        let result = declare_metric(
            &mut index,
            metric_request_with_roles(
                "bad",
                r"a=(?P<a>\d+) b=(?P<b>\d+)",
                vec![
                    ("a", "a", MetricPrimitiveKind::Number, ParamRole::Value),
                    ("b", "b", MetricPrimitiveKind::Number, ParamRole::Value),
                ],
            ),
        );
        assert!(result.is_err());
        assert!(result.unwrap_err().contains("at most one Value-role param"));
    }

    #[test]
    fn value_role_param_becomes_the_bucket_value_not_a_label() {
        let mut index = BTreeMap::new();
        declare_metric(
            &mut index,
            metric_request_with_roles(
                "durations",
                r"request path=(?P<path>\S+) duration=(?P<duration>\d+)",
                vec![
                    ("path", "path", MetricPrimitiveKind::String, ParamRole::Label),
                    ("duration", "duration", MetricPrimitiveKind::Number, ParamRole::Value),
                ],
            ),
        )
        .unwrap();

        ingest_log(&mut index, 0, "request path=/a duration=5");

        let values: Vec<_> = bucket_values(&index, "durations", None).collect();
        assert_eq!(values.len(), 1);
        let (labels, (_, _, value)) = &values[0];
        assert!(!labels.contains_key("duration"));
        assert_eq!(labels.get("path").map(|v| v.to_string()), Some("/a".to_string()));
        assert_eq!(*value, MetricPrimitive::Number(Number::Int(5)));
    }

    fn declare_durations_metric(index: &mut BTreeMap<IndexKey, DataValue>) {
        declare_metric(
            index,
            metric_request_with_roles(
                "durations",
                r"request path=(?P<path>\S+) duration=(?P<duration>\d+)",
                vec![
                    ("path", "path", MetricPrimitiveKind::String, ParamRole::Label),
                    ("duration", "duration", MetricPrimitiveKind::Number, ParamRole::Value),
                ],
            ),
        )
        .unwrap();
    }

    #[test]
    fn gauge_returns_most_recent_value_per_label_combination() {
        let mut index = BTreeMap::new();
        declare_durations_metric(&mut index);
        ingest_log(&mut index, 0, "request path=/a duration=5");
        ingest_log(&mut index, 1, "request path=/a duration=9");
        ingest_log(&mut index, 2, "request path=/b duration=42");

        let parsed = promql::parse_query("gauge(durations)").unwrap();
        let mut series = evaluate_promql(&index, &parsed, None, None);
        series.sort_by(|a, b| a.labels.get("path").cmp(&b.labels.get("path")));

        assert_eq!(series.len(), 2);
        assert_eq!(series[0].labels.get("path").map(String::as_str), Some("/a"));
        assert_eq!(series[0].points, vec![(1, 9.0)]);
        assert_eq!(series[1].labels.get("path").map(String::as_str), Some("/b"));
        assert_eq!(series[1].points, vec![(2, 42.0)]);
    }

    #[test]
    fn gauge_respects_time_range_bounds() {
        let mut index = BTreeMap::new();
        declare_durations_metric(&mut index);
        ingest_log(&mut index, 0, "request path=/a duration=5");
        ingest_log(&mut index, 5, "request path=/a duration=9");
        ingest_log(&mut index, 10, "request path=/a duration=42");

        let parsed = promql::parse_query("gauge(durations)").unwrap();
        let series = evaluate_promql(&index, &parsed, None, Some(9));

        assert_eq!(series.len(), 1);
        assert_eq!(series[0].points, vec![(5, 9.0)]);
    }

    #[test]
    fn histogram_buckets_captured_values_by_fixed_width() {
        let mut index = BTreeMap::new();
        declare_durations_metric(&mut index);
        for (seq, duration) in [(0, 2), (1, 4), (2, 12), (3, 15), (4, 21)] {
            ingest_log(&mut index, seq, &format!("request path=/a duration={duration}"));
        }

        let parsed = promql::parse_query("histogram(durations[10])").unwrap();
        let mut series = evaluate_promql(&index, &parsed, None, None);
        series.sort_by(|a, b| a.labels.get("le").cmp(&b.labels.get("le")));

        // durations 2,4 -> le=10 (count 2); 12,15 -> le=20 (count 2); 21 -> le=30 (count 1).
        assert_eq!(series.len(), 3);
        assert_eq!(series[0].labels.get("le").map(String::as_str), Some("10"));
        assert_eq!(series[0].points[0].1, 2.0);
        assert_eq!(series[1].labels.get("le").map(String::as_str), Some("20"));
        assert_eq!(series[1].points[0].1, 2.0);
        assert_eq!(series[2].labels.get("le").map(String::as_str), Some("30"));
        assert_eq!(series[2].points[0].1, 1.0);
    }

    fn declare_login_duration_metric(index: &mut BTreeMap<IndexKey, DataValue>) {
        declare_metric(
            index,
            metric_request_with_roles(
                "login_gap",
                r"user=(?P<user>\S+) logged in",
                vec![
                    ("user", "user", MetricPrimitiveKind::String, ParamRole::Label),
                    ("gap", "user", MetricPrimitiveKind::Duration, ParamRole::Value),
                ],
            ),
        )
        .unwrap();
    }

    #[test]
    fn duration_value_param_computes_elapsed_time_since_previous_same_labels_event() {
        let mut index = BTreeMap::new();
        declare_login_duration_metric(&mut index);

        ingest_log(&mut index, 0, "user=alice logged in");
        ingest_log(&mut index, 1, "user=bob logged in");
        ingest_log(&mut index, 2, "user=alice logged in");

        let values: Vec<_> = bucket_values(&index, "login_gap", None).collect();
        // First occurrences (alice@0, bob@1) produce no duration value — only alice's second occurrence (seq 2) has something to diff against.
        assert_eq!(values.len(), 1);
        let (labels, (sequence, timestamp, value)) = &values[0];
        assert_eq!(labels.get("user").map(|v| v.to_string()), Some("alice".to_string()));
        assert_eq!(*sequence, 2);
        // `ingest_log`'s helper timestamps by `sequence as millis`.
        assert_eq!(timestamp.as_millis(), 2);
        assert_eq!(*value, MetricPrimitive::Duration(2));
    }

    #[test]
    fn duration_diffs_against_the_immediately_preceding_event_not_the_first() {
        let mut index = BTreeMap::new();
        declare_login_duration_metric(&mut index);

        ingest_log(&mut index, 0, "user=alice logged in");
        ingest_log(&mut index, 3, "user=alice logged in");
        ingest_log(&mut index, 9, "user=alice logged in");

        let values: Vec<_> = bucket_values(&index, "login_gap", None).collect();
        assert_eq!(values.len(), 2);
        assert_eq!(values[0].1.2, MetricPrimitive::Duration(3));
        assert_eq!(values[1].1.2, MetricPrimitive::Duration(6));
    }

    #[test]
    fn duration_values_are_queryable_via_promql_like_any_other_metric() {
        let mut index = BTreeMap::new();
        declare_login_duration_metric(&mut index);
        ingest_log(&mut index, 0, "user=alice logged in");
        ingest_log(&mut index, 10, "user=alice logged in");

        let parsed = promql::parse_query("login_gap").unwrap();
        let series = evaluate_promql(&index, &parsed, None, None);

        assert_eq!(series.len(), 1);
        assert_eq!(series[0].points, vec![(10, 10.0)]);
    }

    // Fixed labels (not a captured param) so a case-insensitive match collapses into one clean "error" series regardless of source-text case.
    fn declare_error_category_metric(index: &mut BTreeMap<IndexKey, DataValue>) {
        declare_metric(
            index,
            DeclareMetricRequest {
                name: "log_category".to_string(),
                message_match_regex: r"(?i)error".to_string(),
                params: vec![],
                matched_labels: Some(BTreeMap::from([("category".to_string(), "error".to_string())])),
                unmatched_labels: Some(BTreeMap::from([("category".to_string(), "other".to_string())])),
            },
        )
        .unwrap();
    }

    #[test]
    fn unmatched_labels_registers_occurrence_for_non_matching_lines() {
        let mut index = BTreeMap::new();
        declare_error_category_metric(&mut index);

        ingest_log(&mut index, 0, "everything is fine");
        ingest_log(&mut index, 1, "an ERROR occurred");
        ingest_log(&mut index, 2, "still fine");

        let error_labels = BTreeMap::from([("category".to_string(), MetricPrimitive::String("error".to_string()))]);
        let error_values: Vec<_> = bucket_values(&index, "log_category", Some(&error_labels)).collect();
        assert_eq!(error_values.len(), 1);
        assert_eq!(error_values[0].1.0, 1);

        let other_labels = BTreeMap::from([("category".to_string(), MetricPrimitive::String("other".to_string()))]);
        let other_values: Vec<_> = bucket_values(&index, "log_category", Some(&other_labels)).collect();
        assert_eq!(other_values.len(), 2);
    }

    #[test]
    fn unmatched_labels_produces_two_series_from_one_bare_promql_query() {
        let mut index = BTreeMap::new();
        declare_error_category_metric(&mut index);
        ingest_log(&mut index, 0, "an ERROR occurred");
        ingest_log(&mut index, 1, "all good");
        ingest_log(&mut index, 2, "all good again");

        let parsed = promql::parse_query("log_category").unwrap();
        let mut series = evaluate_promql(&index, &parsed, None, None);
        series.sort_by(|a, b| a.labels.get("category").cmp(&b.labels.get("category")));

        assert_eq!(series.len(), 2);
        assert_eq!(series[0].labels.get("category").map(String::as_str), Some("error"));
        assert_eq!(series[0].points.len(), 1);
        assert_eq!(series[1].labels.get("category").map(String::as_str), Some("other"));
        assert_eq!(series[1].points.len(), 2);
    }

    #[test]
    fn no_unmatched_labels_means_non_matching_lines_are_untouched() {
        let mut index = BTreeMap::new();
        declare_requests_metric(&mut index);
        ingest_log(&mut index, 0, "not a request line at all");
        assert_eq!(bucket_values(&index, "requests", None).count(), 0);
    }

    #[test]
    fn handle_query_promql_rejects_malformed_query() {
        let index = BTreeMap::new();
        let response = handle_query(
            &index,
            EngineQuery::Promql {
                query: "not a valid promql query(".to_string(),
                start_ms: None,
                end_ms: None,
            },
            0,
        );
        match response {
            EngineResponse::Promql(Err(message)) => assert!(!message.is_empty()),
            _ => panic!("expected EngineResponse::Promql(Err(_))"),
        }
    }

    #[test]
    fn handle_query_promql_returns_series_for_valid_query() {
        let mut index = BTreeMap::new();
        declare_requests_metric(&mut index);
        ingest_log(&mut index, 0, "request path=/a");

        let response = handle_query(
            &index,
            EngineQuery::Promql {
                query: r#"requests{path="/a"}"#.to_string(),
                start_ms: None,
                end_ms: None,
            },
            0,
        );
        match response {
            EngineResponse::Promql(Ok(series)) => {
                assert_eq!(series.len(), 1);
                assert_eq!(series[0].points, vec![(0, 1.0)]);
            }
            _ => panic!("expected EngineResponse::Promql(Ok(_))"),
        }
    }

    // -- FilterBar query integration --

    fn logs_query(filter: Option<&str>) -> EngineQuery {
        EngineQuery::Logs(LogsParams {
            since_sequence: None,
            start_ms: None,
            end_ms: None,
            limit: None,
            filter: filter.map(str::to_string),
        })
    }

    fn timeline_query(filter: Option<&str>, bucket_ms: i64) -> EngineQuery {
        EngineQuery::Timeline(TimelineParams {
            start_ms: None,
            end_ms: None,
            bucket_ms: Some(bucket_ms),
            filter: filter.map(str::to_string),
        })
    }

    #[test]
    fn logs_query_with_regex_filter_returns_only_matching_lines() {
        let mut index = BTreeMap::new();
        ingest_log(&mut index, 0, "everything is fine");
        ingest_log(&mut index, 1, "an error occurred");
        ingest_log(&mut index, 2, "still fine");
        ingest_log(&mut index, 3, "another error here");

        match handle_query(&index, logs_query(Some(r#"regex("error")"#)), 0) {
            EngineResponse::Logs(entries) => {
                let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
                assert_eq!(sequences, vec![1, 3]);
            }
            _ => panic!("expected Logs response"),
        }
    }

    #[test]
    fn logs_query_with_sequence_comparison_filters_by_sequence() {
        let mut index = BTreeMap::new();
        for i in 0..5u64 {
            ingest_log(&mut index, i, &format!("line {i}"));
        }

        match handle_query(&index, logs_query(Some("sequence<2")), 0) {
            EngineResponse::Logs(entries) => {
                let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
                assert_eq!(sequences, vec![0, 1]);
            }
            _ => panic!("expected Logs response"),
        }
    }

    #[test]
    fn logs_query_with_timestamp_comparison_filters_by_timestamp() {
        let mut index = BTreeMap::new();
        for i in 0..5u64 {
            let timestamp = Timestamp::from_millis(i as i64 * 1000);
            index.insert(
                IndexKey::IngestedLog { sequence: i, timestamp },
                DataValue::LogLiteral { text: format!("line {i}") },
            );
        }

        match handle_query(&index, logs_query(Some("timestamp>2000")), 0) {
            EngineResponse::Logs(entries) => {
                let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
                assert_eq!(sequences, vec![3, 4]);
            }
            _ => panic!("expected Logs response"),
        }
    }

    #[test]
    fn logs_query_with_json_field_comparison_filters_by_json_field() {
        let mut index = BTreeMap::new();
        ingest_log(&mut index, 0, r#"{"level":"error","code":500}"#);
        ingest_log(&mut index, 1, r#"{"level":"info","code":200}"#);
        ingest_log(&mut index, 2, "not json at all");

        match handle_query(&index, logs_query(Some(r#"json[level]==error"#)), 0) {
            EngineResponse::Logs(entries) => {
                let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
                assert_eq!(sequences, vec![0]);
            }
            _ => panic!("expected Logs response"),
        }

        match handle_query(&index, logs_query(Some("json[code]>=300")), 0) {
            EngineResponse::Logs(entries) => {
                let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
                assert_eq!(sequences, vec![0]);
            }
            _ => panic!("expected Logs response"),
        }
    }

    #[test]
    fn logs_query_combines_regex_and_metric_comparison_with_and() {
        let mut index = BTreeMap::new();
        declare_metric(
            &mut index,
            metric_request(
                "requests",
                r"level=(?P<level>\w+) duration=(?P<duration>\d+)",
                vec![("level", "level", MetricPrimitiveKind::String)],
            ),
        )
        .unwrap();

        ingest_log(&mut index, 0, "level=error duration=1 boom");
        ingest_log(&mut index, 1, "level=info duration=2 boom");
        ingest_log(&mut index, 2, "level=error duration=3 all good");

        // Only sequence 0 has both; note comparison values are bare tokens (level==error), not quoted string literals like regex()'s argument.
        match handle_query(
            &index,
            logs_query(Some(r#"regex("boom") and requests[level]==error"#)),
            0,
        ) {
            EngineResponse::Logs(entries) => {
                let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
                assert_eq!(sequences, vec![0]);
            }
            _ => panic!("expected Logs response"),
        }

        // With `or`, sequences 0 (boom) and 2 (level=error) both match.
        match handle_query(
            &index,
            logs_query(Some(r#"regex("boom") or requests[level]==error"#)),
            0,
        ) {
            EngineResponse::Logs(entries) => {
                let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
                assert_eq!(sequences, vec![0, 1, 2]);
            }
            _ => panic!("expected Logs response"),
        }
    }

    #[test]
    fn logs_query_includes_declared_params_on_matching_lines_only() {
        let mut index = BTreeMap::new();
        declare_param(&mut index, request("level", r"level=(?P<level>\w+)", "level")).unwrap();
        ingest_log(&mut index, 0, "level=info hello");
        ingest_log(&mut index, 1, "no match here");

        let EngineResponse::Logs(entries) = handle_query(&index, logs_query(None), 0) else {
            panic!("expected Logs response");
        };

        assert_eq!(entries.len(), 2);
        assert_eq!(
            entries[0].params.get("level"),
            Some(&MetricPrimitive::String("info".to_string()))
        );
        assert!(entries[1].params.get("level").is_none());
    }

    #[test]
    fn boolean_param_declares_without_capture_group_and_is_always_present() {
        let mut index = BTreeMap::new();
        let mut req = request("IsError", r"(?i)error", "");
        req.kind = MetricPrimitiveKind::Boolean;
        assert!(declare_param(&mut index, req).is_ok());

        ingest_log(&mut index, 0, "an ERROR occurred");
        ingest_log(&mut index, 1, "everything is fine");

        let EngineResponse::Logs(entries) = handle_query(&index, logs_query(None), 0) else {
            panic!("expected Logs response");
        };

        assert_eq!(entries.len(), 2);
        assert_eq!(entries[0].params.get("IsError"), Some(&MetricPrimitive::Boolean(true)));
        assert_eq!(entries[1].params.get("IsError"), Some(&MetricPrimitive::Boolean(false)));
    }

    #[test]
    fn timeline_query_applies_filter_before_bucketing() {
        let mut index = BTreeMap::new();
        ingest_log(&mut index, 0, "everything is fine");
        ingest_log(&mut index, 1, "an error occurred");
        ingest_log(&mut index, 2, "another error here");

        match handle_query(&index, timeline_query(Some(r#"regex("error")"#), 1_000_000), 0) {
            EngineResponse::Timeline(buckets) => {
                let total: u64 = buckets.iter().map(|b| b.count).sum();
                assert_eq!(total, 2);
            }
            _ => panic!("expected Timeline response"),
        }
    }

    #[test]
    fn malformed_filter_string_surfaces_as_engine_error() {
        let mut index = BTreeMap::new();
        ingest_log(&mut index, 0, "hello");

        match handle_query(&index, logs_query(Some("@@@ not a valid filter @@@")), 0) {
            EngineResponse::Error(message) => assert!(!message.is_empty()),
            _ => panic!("expected an Error response for a malformed filter string"),
        }

        match handle_query(&index, timeline_query(Some("@@@ not a valid filter @@@"), 1000), 0) {
            EngineResponse::Error(message) => assert!(!message.is_empty()),
            _ => panic!("expected an Error response for a malformed filter string"),
        }
    }

    #[test]
    fn metric_comparison_without_param_is_a_compile_error() {
        let mut index = BTreeMap::new();
        declare_metric(
            &mut index,
            metric_request(
                "requests",
                r"level=(?P<level>\w+)",
                vec![("level", "level", MetricPrimitiveKind::String)],
            ),
        )
        .unwrap();
        ingest_log(&mut index, 0, "level=error");

        match handle_query(&index, logs_query(Some("requests==5")), 0) {
            EngineResponse::Error(message) => assert!(message.contains("param")),
            _ => panic!("expected an Error response for a metric comparison without [param]"),
        }
    }
}