//! `EngineSnapshot` is the actor's live state: ingested log lines, declared params/metrics,
//! computed metric buckets, and the sort-helper indexes built on demand — all stored in one
//! flat `BTreeMap<IndexKey, DataValue>` and read/written through the left-right index in
//! `super::handle`. The submodules below hold everything that isn't part of `EngineSnapshot`'s
//! own impl (types, free functions, tests) split out by concern; every `impl EngineSnapshot`
//! block itself lives here, in one file, rather than being scattered across those submodules.

pub(crate) mod filter;
pub(crate) mod metrics;
pub(crate) mod params;
pub(crate) mod query;
pub(crate) mod sort;

pub(crate) use filter::eval_json_comparison;

use crate::api::dto::PromqlSeriesDto;
use crate::consts::{
    MAX_AUTO_DECLARED_PARAMS, MAX_NEW_AUTO_PARAMS_PER_LINE, OTTYL_RESERVED_PARAM_NAMES,
    SORT_HELPER_INDEX_PURGE_AFTER,
};
use crate::flavour::LogLiteralHelper;
use crate::engine::data_value::{DataValue, MetricPrimitiveKind, ParamDeclaration};
use crate::engine::filter_parser::{self, CompareOp, FilterExpr, FilterValue};
use crate::engine::index_key::IndexKey;
use crate::engine::message::{EngineQuery, EngineResponse};
use crate::engine::metric_primitive::{MetricPrimitive, Number};
use crate::engine::promql::{AggOp, PromqlQuery};
use crate::engine::timestamp::Timestamp;
use crate::engine::{DeclareMetricRequest, DeclareParamRequest};
use filter::CompiledFilter;
use metrics::parse_metric_primitive;
use params::{build_auto_param_pattern, extract_param_raw, infer_kind, sanitize_group_name, AutoParamCandidate, ResolvedParam};
use query::{
    coerce_to_f64, handle_query_inner, matches_label_matchers, merge_series_locf,
    merge_series_windowed, ottyl_reserved_numeric_value_fn, project_labels, windowed_rate_series,
};
use regex::Regex;
use sort::{derive_single_param, CompiledParamDeclaration, SortHelperAccess, SortHelperIndexEntry, SortHelperKey};
use std::collections::{BTreeMap, HashMap, HashSet};

#[derive(Debug, Clone, Default)]
pub(crate) struct EngineSnapshot {
    pub(crate) index: BTreeMap<IndexKey, DataValue>,
    pub(crate) sort_helper_indexes: HashMap<String, SortHelperIndexEntry>,
    /// Compiled regexes for param/metric declarations, keyed by declaration name. Compiling a
    /// regex is far more expensive than matching with one, and `apply_all_metrics`/
    /// `update_sort_helper_indexes_on_ingest` run per ingested line, so every insertion or
    /// removal of a declaration (see `params.rs`/`metrics.rs`) must keep these in sync rather
    /// than recompiling from the stored pattern string on every line.
    pub(crate) param_regex_cache: HashMap<String, Regex>,
    pub(crate) metric_regex_cache: HashMap<String, Regex>,
    pub(crate) log_count: usize,
    pub(crate) approx_bytes_total: u64,
    pub(crate) ingest_lag_high_watermark_ms: i64,
}

// --- core: snapshot construction and log range access ---
impl EngineSnapshot {
    pub(crate) fn metric_declarations_range(
        &self,
    ) -> impl Iterator<
        Item = (
            &str,
            &str,
            &[String],
            Option<&str>,
            Option<&BTreeMap<String, String>>,
            Option<&BTreeMap<String, String>>,
        ),
    > {
        self.index
            .range(IndexKey::ParamMetricSplit..IndexKey::MetricBucketSplit)
            .filter_map(|(key, value)| match (key, value) {
                (
                    IndexKey::MetricDeclaration { name },
                    DataValue::MetricDeclaration {
                        message_match_regex,
                        label_params,
                        value_param,
                        matched_labels,
                        unmatched_labels,
                    },
                ) => Some((
                    name.as_str(),
                    message_match_regex.as_str(),
                    label_params.as_slice(),
                    value_param.as_deref(),
                    matched_labels.as_ref(),
                    unmatched_labels.as_ref(),
                )),
                _ => None,
            })
    }

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

    pub(crate) fn from_index(index: BTreeMap<IndexKey, DataValue>) -> Self {
        let mut snapshot = EngineSnapshot { index, ..Default::default() };
        snapshot.rebuild_regex_caches();
        snapshot
    }

    /// Recompiles `param_regex_cache`/`metric_regex_cache` from the declarations currently in
    /// `index`. Needed whenever a snapshot's declarations are populated by inserting straight
    /// into `index` rather than through the normal declare/insert methods that keep the caches
    /// in sync incrementally (e.g. `from_index` building a snapshot from a bare index).
    pub(crate) fn rebuild_regex_caches(&mut self) {
        self.param_regex_cache = self
            .param_declarations_range()
            .filter_map(|(name, message_match_regex, ..)| {
                regex::Regex::new(message_match_regex).ok().map(|re| (name.to_string(), re))
            })
            .collect();
        self.metric_regex_cache = self
            .metric_declarations_range()
            .filter_map(|(name, message_match_regex, ..)| {
                regex::Regex::new(message_match_regex).ok().map(|re| (name.to_string(), re))
            })
            .collect();
    }

    pub(crate) fn evict_excess_logs(&mut self, max_logs: Option<usize>) {
        let Some(max_logs) = max_logs else { return };
        while self.log_count > max_logs {
            let Some(oldest_key) = self
                .index
                .range(IndexKey::Start..IndexKey::LogParamSplit)
                .next()
                .map(|(key, _)| key.clone())
            else {
                break;
            };
            if let Some(DataValue::LogLiteral { text }) = self.index.get(&oldest_key) {
                self.approx_bytes_total = self.approx_bytes_total.saturating_sub(text.approx_log_bytes());
            }
            self.index.remove(&oldest_key);
            self.log_count -= 1;
        }
    }

    pub(crate) fn logs_range(&self) -> impl Iterator<Item = (u64, Timestamp, &str)> {
        self.logs_in_sequence_range(..)
    }

    /// Seeks directly to `sequence_range` within the `IngestedLog` key range instead of
    /// scanning from `IndexKey::Start` and filtering after the fact. `IngestedLog` sorts by
    /// `(sequence, timestamp)`, and `sequence` is assigned monotonically at ingest
    /// (`handle.rs`), so a bound on `sequence` alone is an exact seek — `Timestamp::MIN` is
    /// used as the synthetic sub-key since it sorts below any real timestamp for that sequence.
    pub(crate) fn logs_in_sequence_range(
        &self,
        sequence_range: impl std::ops::RangeBounds<u64>,
    ) -> impl Iterator<Item = (u64, Timestamp, &str)> {
        let lower = match sequence_range.start_bound() {
            std::ops::Bound::Included(&sequence) => {
                IndexKey::IngestedLog { sequence, timestamp: Timestamp::MIN }
            }
            std::ops::Bound::Excluded(&sequence) => {
                IndexKey::IngestedLog { sequence: sequence.saturating_add(1), timestamp: Timestamp::MIN }
            }
            std::ops::Bound::Unbounded => IndexKey::Start,
        };
        let upper = match sequence_range.end_bound() {
            std::ops::Bound::Included(&sequence) => {
                IndexKey::IngestedLog { sequence: sequence.saturating_add(1), timestamp: Timestamp::MIN }
            }
            std::ops::Bound::Excluded(&sequence) => {
                IndexKey::IngestedLog { sequence, timestamp: Timestamp::MIN }
            }
            std::ops::Bound::Unbounded => IndexKey::LogParamSplit,
        };
        self.index.range(lower..upper).filter_map(|(key, value)| match (key, value) {
            (IndexKey::IngestedLog { sequence, timestamp }, DataValue::LogLiteral { text }) => {
                Some((*sequence, *timestamp, text.as_str()))
            }
            _ => None,
        })
    }

    pub(crate) fn logs_since(
        &self,
        before_sequence: u64,
        since_ms: i64,
        now_ms: i64,
    ) -> impl Iterator<Item = (u64, Timestamp, &str)> {
        self.logs_in_sequence_range(..before_sequence).filter(move |(_, timestamp, _)| {
            timestamp.as_millis() > since_ms && timestamp.as_millis() <= now_ms
        })
    }
}

// --- filter: compiling search-bar filter expressions ---
impl EngineSnapshot {
    pub(in crate::engine::snapshot) fn compile_filter_query(&self, filter: &str, now: Timestamp) -> Result<CompiledFilter, String> {
        let expr = filter_parser::parse(filter).map_err(|err| err.to_string())?;
        self.compile_filter(&expr, now)
    }

    fn compile_filter(&self, 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() {
                "ottyl_seq" => CompiledFilter::from_seq_comparison(*op, value),
                "ottyl_ts" => CompiledFilter::from_ts_comparison(*op, value, now),
                "ottyl_fmt" => Ok(CompiledFilter::from_fmt_comparison(*op, value)),
                other => self.compile_metric_comparison(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(self.compile_filter(lhs, now)?),
                Box::new(self.compile_filter(rhs, now)?),
            )),
            FilterExpr::Or(lhs, rhs) => Ok(CompiledFilter::Or(
                Box::new(self.compile_filter(lhs, now)?),
                Box::new(self.compile_filter(rhs, now)?),
            )),
        }
    }

    fn compile_metric_comparison(
        &self,
        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 => "!=",
                }
            )
        })?;

        if let Some(DataValue::MetricDeclaration { label_params, .. }) =
            self.index.get(&IndexKey::MetricDeclaration { name: metric.to_string() })
        {
            if !label_params.iter().any(|p| p == param) {
                return Err(format!(
                    "param '{param}' is not a label param of metric '{metric}'; declared label params: [{}]",
                    label_params.join(", ")
                ));
            }
        }

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

// --- metrics: declaring, applying, and querying metric buckets ---
impl EngineSnapshot {
    pub(crate) fn validate_metric_request(&self, request: &DeclareMetricRequest) -> Result<Regex, String> {
        let re = Regex::new(&request.message_match_regex)
            .map_err(|err| format!("invalid regex: {err}"))?;

        let declared =
            |name: &str| self.index.contains_key(&IndexKey::ParamDeclaration { name: name.to_string() });

        let mut errors: Vec<String> = request
            .label_params
            .iter()
            .filter(|name| !declared(name))
            .map(|name| format!("label param '{name}' references an undeclared param"))
            .collect();
        if let Some(value_param) = &request.value_param {
            if !declared(value_param) {
                errors.push(format!("value param '{value_param}' references an undeclared param"));
            }
        }
        if !errors.is_empty() {
            return Err(errors.join("; "));
        }

        Ok(re)
    }

    pub(crate) fn declare_metric(&mut self, request: DeclareMetricRequest) -> Result<(), String> {
        let re = self.validate_metric_request(&request)?;
        self.metric_regex_cache.insert(request.name.clone(), re.clone());
        self.resolve_backfill_and_store_metric(request, &re);
        Ok(())
    }

    /// Shared by `declare_metric` and `update_metric`: resolves param declarations, backfills
    /// the metric's buckets from existing log history, then writes the `MetricDeclaration` entry.
    /// Callers differ only in what happens before this (rename bookkeeping for updates).
    fn resolve_backfill_and_store_metric(&mut self, request: DeclareMetricRequest, re: &Regex) {
        let resolved_params = self.resolved_param_declarations();
        self.backfill_metric(
            &request.name,
            re,
            &request.label_params,
            request.value_param.as_deref(),
            &resolved_params,
            request.matched_labels.as_ref(),
            request.unmatched_labels.as_ref(),
        );
        self.index.insert(
            IndexKey::MetricDeclaration { name: request.name.clone() },
            DataValue::MetricDeclaration {
                message_match_regex: request.message_match_regex,
                label_params: request.label_params,
                value_param: request.value_param,
                matched_labels: request.matched_labels,
                unmatched_labels: request.unmatched_labels,
            },
        );
    }

    pub(crate) fn validate_update_metric(
        &self,
        existing_name: &str,
        request: &DeclareMetricRequest,
    ) -> Result<Regex, String> {
        if !self.index.contains_key(&IndexKey::MetricDeclaration { name: existing_name.to_string() }) {
            return Err(format!("no metric declaration named '{existing_name}'"));
        }
        if request.name != existing_name
            && self.index.contains_key(&IndexKey::MetricDeclaration { name: request.name.clone() })
        {
            return Err(format!("a metric named '{}' already exists", request.name));
        }
        self.validate_metric_request(request)
    }

    pub(crate) fn update_metric(&mut self, existing_name: &str, request: DeclareMetricRequest) -> Result<(), String> {
        let re = self.validate_update_metric(existing_name, &request)?;

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

        self.resolve_backfill_and_store_metric(request, &re);
        Ok(())
    }

    fn clear_metric_data(&mut self, metric_name: &str) {
        self.index.retain(|key, _| match key {
            IndexKey::Bucket { metric_name: name, .. } => name != metric_name,
            IndexKey::LastMetricEvent { metric_name: name, .. } => name != metric_name,
            _ => true,
        });
    }

    pub(crate) fn remove_metric_declaration(&mut self, name: &str) -> bool {
        let existed = self.index.remove(&IndexKey::MetricDeclaration { name: name.to_string() }).is_some();
        self.metric_regex_cache.remove(name);
        self.clear_metric_data(name);
        existed
    }

    #[allow(clippy::too_many_arguments)]
    fn apply_metric_to_log(
        &mut self,
        metric_name: &str,
        regex: &Regex,
        label_params: &[String],
        value_param: Option<&str>,
        resolved_params: &HashMap<String, ResolvedParam>,
        matched_labels: Option<&BTreeMap<String, String>>,
        unmatched_labels: Option<&BTreeMap<String, String>>,
        sequence: u64,
        timestamp: Timestamp,
        text: &str,
    ) {
        if regex.captures(text).is_none() {
            if let Some(unmatched_labels) = unmatched_labels {
                let labels = unmatched_labels
                    .iter()
                    .map(|(name, value)| (name.clone(), MetricPrimitive::String(value.clone())))
                    .collect();
                self.push_bucket_value(metric_name, labels, sequence, timestamp, MetricPrimitive::Number(Number::Int(1)));
            }
            return;
        }

        let mut labels = BTreeMap::new();
        for param_name in label_params {
            let Some(resolved) = resolved_params.get(param_name) else { return };
            let Some(parsed) = self.resolve_param_value(metric_name, resolved, &labels, text, timestamp) else {
                return;
            };
            labels.insert(param_name.clone(), parsed);
        }

        if let Some(matched_labels) = matched_labels {
            for (name, value) in matched_labels {
                labels.insert(name.clone(), MetricPrimitive::String(value.clone()));
            }
        }

        let value = match value_param {
            Some(value_param) => {
                let Some(resolved) = resolved_params.get(value_param) else { return };
                match self.resolve_param_value(metric_name, resolved, &labels, text, timestamp) {
                    Some(parsed) => parsed,
                    None => return,
                }
            }
            None => MetricPrimitive::Number(Number::Int(1)),
        };

        self.push_bucket_value(metric_name, labels, sequence, timestamp, value);
    }

    pub(crate) fn push_bucket_value(
        &mut self,
        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 = self
            .index
            .entry(key)
            .or_insert_with(|| DataValue::BucketValue { values: Vec::new() });
        if let DataValue::BucketValue { values } = entry {
            values.push((sequence, timestamp, value));
        }
    }

    #[allow(clippy::too_many_arguments)]
    fn backfill_metric(
        &mut self,
        metric_name: &str,
        regex: &Regex,
        label_params: &[String],
        value_param: Option<&str>,
        resolved_params: &HashMap<String, ResolvedParam>,
        matched_labels: Option<&BTreeMap<String, String>>,
        unmatched_labels: Option<&BTreeMap<String, String>>,
    ) {
        let lines: Vec<(u64, Timestamp, String)> = self
            .logs_range()
            .map(|(sequence, timestamp, text)| (sequence, timestamp, text.to_string()))
            .collect();
        for (sequence, timestamp, text) in lines {
            self.apply_metric_to_log(
                metric_name,
                regex,
                label_params,
                value_param,
                resolved_params,
                matched_labels,
                unmatched_labels,
                sequence,
                timestamp,
                &text,
            );
        }
    }

    pub(crate) fn apply_all_metrics(&mut self, sequence: u64, timestamp: Timestamp, text: &str) {
        type SeededMetric = (
            String,
            Vec<String>,
            Option<String>,
            Option<BTreeMap<String, String>>,
            Option<BTreeMap<String, String>>,
        );
        let metrics: Vec<SeededMetric> = self
            .metric_declarations_range()
            .map(|(name, _, label_params, value_param, matched_labels, unmatched_labels)| {
                (
                    name.to_string(),
                    label_params.to_vec(),
                    value_param.map(|s| s.to_string()),
                    matched_labels.cloned(),
                    unmatched_labels.cloned(),
                )
            })
            .collect();
        let resolved_params = self.resolved_param_declarations();
        for (name, label_params, value_param, matched_labels, unmatched_labels) in metrics {
            if let Some(re) = self.metric_regex_cache.get(&name).cloned() {
                self.apply_metric_to_log(
                    &name,
                    &re,
                    &label_params,
                    value_param.as_deref(),
                    &resolved_params,
                    matched_labels.as_ref(),
                    unmatched_labels.as_ref(),
                    sequence,
                    timestamp,
                    text,
                );
            }
        }
    }

    pub(crate) fn bucket_values<'a>(
        &'a self,
        metric_name: &'a str,
        labels_filter: Option<&'a BTreeMap<String, MetricPrimitive>>,
    ) -> impl Iterator<Item = (&'a BTreeMap<String, MetricPrimitive>, &'a (u64, Timestamp, MetricPrimitive))> + 'a
    {
        // IndexKey::Bucket sorts by (metric_name, labels), so bound the range to just this
        // metric's buckets instead of ranging MetricBucketSplit..End over every declared
        // metric. '\u{10FFFF}' is the maximum Unicode scalar value, so `metric_name` followed
        // by it sorts after every real key for this metric (any labels) and before any key
        // for a lexicographically greater metric name.
        let lower = IndexKey::Bucket { metric_name: metric_name.to_string(), labels: BTreeMap::new() };
        let upper = IndexKey::Bucket {
            metric_name: format!("{metric_name}\u{10FFFF}"),
            labels: BTreeMap::new(),
        };
        self.index
            .range(lower..upper)
            .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()
    }

    pub(crate) fn declared_value_kind(&self, metric_name: &str) -> Option<MetricPrimitiveKind> {
        match self.index.get(&IndexKey::MetricDeclaration { name: metric_name.to_string() }) {
            Some(DataValue::MetricDeclaration { value_param, .. }) => Some(
                value_param
                    .as_ref()
                    .and_then(|name| {
                        self.index.get(&IndexKey::ParamDeclaration { name: name.clone() })
                    })
                    .map(|value| match value {
                        DataValue::ParamDeclaration(param) => param.kind,
                        _ => MetricPrimitiveKind::Number,
                    })
                    .unwrap_or(MetricPrimitiveKind::Number),
            ),
            _ => None,
        }
    }
}

// --- params: declaring and resolving param values ---
impl EngineSnapshot {
    pub(crate) fn validate_param_declaration(&self, request: &DeclareParamRequest) -> Result<(), String> {
        if OTTYL_RESERVED_PARAM_NAMES.contains(&request.name.as_str()) {
            return Err(format!(
                "'{}' is a reserved name, present on every log line without declaring anything; \
                 it can't be redeclared as a param",
                request.name
            ));
        }
        if let Some(DataValue::ParamDeclaration(existing)) =
            self.index.get(&IndexKey::ParamDeclaration { name: request.name.clone() })
        {
            if existing.auto_derived {
                return Err(format!(
                    "param '{}' was auto-detected from the log stream and can't be edited",
                    request.name
                ));
            }
        }
        let re = Regex::new(&request.message_match_regex)
            .map_err(|err| format!("invalid regex: {err}"))?;
        if request.elapsed_since_last && request.kind != MetricPrimitiveKind::Duration {
            return Err("elapsed_since_last params must declare kind 'duration'".to_string());
        }
        let uses_json_field = request.json_field_path.as_deref().is_some_and(|path| !path.is_empty());
        if request.kind != MetricPrimitiveKind::Boolean
            && !request.elapsed_since_last
            && !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
            ));
        }
        Ok(())
    }

    pub(crate) fn insert_param_declaration(&mut self, request: DeclareParamRequest) {
        if let Ok(re) = Regex::new(&request.message_match_regex) {
            self.param_regex_cache.insert(request.name.clone(), re);
        }
        self.index.insert(
            IndexKey::ParamDeclaration { name: request.name.clone() },
            DataValue::ParamDeclaration(ParamDeclaration { auto_derived: false, ..request }),
        );
    }

    fn auto_derived_param_count(&self) -> usize {
        self.param_declarations_range().filter(|declaration| declaration.6).count()
    }

    pub(crate) fn auto_declare_params(&mut self, text: &str) {
        let mut total = self.auto_derived_param_count();
        if total >= MAX_AUTO_DECLARED_PARAMS {
            return;
        }
        let candidates: Vec<AutoParamCandidate> = if crate::engine::logfmt::detect_format(text) == "json" {
            crate::engine::data_value::flatten_json_object(text)
                .into_iter()
                .map(|(path, value)| AutoParamCandidate {
                    name: path.clone(),
                    value,
                    message_match_regex: String::new(),
                    capture_group: String::new(),
                    json_field_path: Some(path),
                })
                .collect()
        } else {
            crate::engine::logfmt::extract_pairs(text)
                .into_iter()
                .map(|(key, value, quoted)| {
                    let group = sanitize_group_name(key);
                    AutoParamCandidate {
                        name: key.to_string(),
                        value: value.to_string(),
                        message_match_regex: build_auto_param_pattern(key, &group, quoted),
                        capture_group: group,
                        json_field_path: None,
                    }
                })
                .collect()
        };

        let mut new_this_line = 0;
        for candidate in candidates {
            if new_this_line >= MAX_NEW_AUTO_PARAMS_PER_LINE || total >= MAX_AUTO_DECLARED_PARAMS {
                break;
            }
            if self.try_insert_auto_param(candidate, &mut total) {
                new_this_line += 1;
            }
        }
    }

    /// Inserts `candidate` as an auto-derived `ParamDeclaration` unless a param with that name
    /// is already declared, bumping `*total` and returning whether it inserted.
    fn try_insert_auto_param(&mut self, candidate: AutoParamCandidate, total: &mut usize) -> bool {
        let index_key = IndexKey::ParamDeclaration { name: candidate.name.clone() };
        if self.index.contains_key(&index_key) {
            return false;
        }
        if let Ok(re) = Regex::new(&candidate.message_match_regex) {
            self.param_regex_cache.insert(candidate.name.clone(), re);
        }
        self.index.insert(
            index_key,
            DataValue::ParamDeclaration(ParamDeclaration {
                name: candidate.name,
                message_match_regex: candidate.message_match_regex,
                capture_group: candidate.capture_group,
                json_field_path: candidate.json_field_path,
                kind: infer_kind(&candidate.value),
                elapsed_since_last: false,
                auto_derived: true,
                log_stream_gated: false,
            }),
        );
        *total += 1;
        true
    }

    pub(crate) fn derive_log_stream_params_from_row(
        &mut self,
        header: &crate::engine::logfmt::LogStreamHeader,
        text: &str,
    ) {
        if !crate::engine::logfmt::is_log_stream_line(text) {
            return;
        }
        let row = header.split_row(text);

        for (i, column) in header.columns.iter().enumerate() {
            let Some(value) = row.get(i) else { continue };
            let index_key = IndexKey::ParamDeclaration { name: column.name.clone() };
            if self.index.contains_key(&index_key) {
                continue;
            }
            let message_match_regex = header.column_pattern(i, &column.name);
            if let Ok(re) = Regex::new(&message_match_regex) {
                self.param_regex_cache.insert(column.name.clone(), re);
            }
            self.index.insert(
                index_key,
                DataValue::ParamDeclaration(ParamDeclaration {
                    name: column.name.clone(),
                    message_match_regex,
                    capture_group: column.name.clone(),
                    json_field_path: None,
                    kind: infer_kind(value.trim()),
                    elapsed_since_last: false,
                    auto_derived: true,
                    log_stream_gated: true,
                }),
            );
        }

        if let Some(message) = row.get(header.columns.len()).filter(|m| !m.trim().is_empty()) {
            let index_key = IndexKey::ParamDeclaration { name: "message".to_string() };
            if !self.index.contains_key(&index_key) {
                let message_match_regex = header.message_pattern("message");
                if let Ok(re) = Regex::new(&message_match_regex) {
                    self.param_regex_cache.insert("message".to_string(), re);
                }
                self.index.insert(
                    index_key,
                    DataValue::ParamDeclaration(ParamDeclaration {
                        name: "message".to_string(),
                        message_match_regex,
                        capture_group: "message".to_string(),
                        json_field_path: None,
                        kind: infer_kind(message.trim()),
                        elapsed_since_last: false,
                        auto_derived: true,
                        log_stream_gated: true,
                    }),
                );
            }
        }
    }

    pub(crate) fn clear_all_param_declarations(&mut self) -> usize {
        let names: Vec<String> =
            self.param_declarations_range().map(|(name, ..)| name.to_string()).collect();
        let removed = names.len();
        for name in names {
            self.index.remove(&IndexKey::ParamDeclaration { name: name.clone() });
            self.param_regex_cache.remove(&name);
        }
        removed
    }

    pub(crate) fn reextract_params_from_retained_logs(&mut self) -> (usize, usize) {
        let texts: Vec<String> = self.logs_range().map(|(_, _, text)| text.to_string()).collect();
        let logs_considered = texts.len();
        for text in &texts {
            self.auto_declare_params(text);
        }
        let reextracted = self.param_declarations_range().count();
        (reextracted, logs_considered)
    }

    pub(in crate::engine::snapshot) fn resolved_param_declarations(&self) -> HashMap<String, ResolvedParam> {
        self.param_declarations_range()
            .filter_map(|(name, _, capture_group, json_field_path, kind, elapsed_since_last, _, log_stream_gated)| {
                self.param_regex_cache.get(name).map(|regex| {
                    (
                        name.to_string(),
                        ResolvedParam {
                            regex: regex.clone(),
                            capture_group: capture_group.to_string(),
                            json_field_path: json_field_path.map(|path| path.to_string()),
                            kind,
                            elapsed_since_last,
                            log_stream_gated,
                        },
                    )
                })
            })
            .collect()
    }

    pub(in crate::engine::snapshot) fn resolve_param_value(
        &mut self,
        metric_name: &str,
        resolved: &ResolvedParam,
        labels_so_far: &BTreeMap<String, MetricPrimitive>,
        text: &str,
        timestamp: Timestamp,
    ) -> Option<MetricPrimitive> {
        if resolved.elapsed_since_last {
            let last_event_key = IndexKey::LastMetricEvent {
                metric_name: metric_name.to_string(),
                labels: labels_so_far.clone(),
            };
            let previous_timestamp = match self.index.get(&last_event_key) {
                Some(DataValue::LastEventTimestamp { timestamp: previous }) => Some(*previous),
                _ => None,
            };
            self.index.insert(last_event_key, DataValue::LastEventTimestamp { timestamp });
            return previous_timestamp
                .map(|previous| MetricPrimitive::Duration(timestamp.as_millis() - previous.as_millis()));
        }

        let raw = extract_param_raw(
            text,
            &resolved.regex,
            &resolved.capture_group,
            resolved.json_field_path.as_deref(),
            resolved.log_stream_gated,
        )?;
        parse_metric_primitive(&raw, resolved.kind)
    }
}

// --- query: EngineQuery/PromQL evaluation ---
impl EngineSnapshot {
    pub(crate) fn handle_query_ro(&self, query: EngineQuery) -> EngineResponse {
        handle_query_inner(self, query, SortHelperAccess::Immutable(&self.sort_helper_indexes))
    }

    pub(crate) fn has_sort_helper_index(&self, sort_by: &str) -> bool {
        self.sort_helper_indexes.contains_key(sort_by)
    }

    pub(crate) fn evaluate_promql(
        &self,
        parsed: &PromqlQuery,
        start_ms: Option<i64>,
        end_ms: Option<i64>,
    ) -> Result<Vec<PromqlSeriesDto>, String> {
        if parsed.metric_name == "ottyl_fmt" {
            return Err(
                "'ottyl_fmt' is a categorical value (json/logfmt/none), not a number, so it can't be \
                 used in a dashboard PromQL query; filter on it in the search bar instead, e.g. \
                 ottyl_fmt==json"
                    .to_string(),
            );
        }

        if ottyl_reserved_numeric_value_fn(&parsed.metric_name).is_none()
            && !self.index.contains_key(&IndexKey::MetricDeclaration { name: parsed.metric_name.clone() })
            && self.bucket_values(&parsed.metric_name, None).next().is_none()
        {
            return Err(format!(
                "unknown metric \"{}\": no declaration or recorded data for it (was it deleted?)",
                parsed.metric_name
            ));
        }
        if matches!(parsed.agg, Some(AggOp::Sum) | Some(AggOp::Avg))
            && !parsed.inner_rate
            && self.declared_value_kind(&parsed.metric_name) == Some(MetricPrimitiveKind::String)
        {
            return Err(format!(
                "cannot sum/avg a string-typed metric \"{}\"; use count() or a numeric-typed metric",
                parsed.metric_name
            ));
        }

        let mut series_points: BTreeMap<BTreeMap<String, String>, Vec<(i64, f64)>> = BTreeMap::new();
        if let Some(reserved_value) = ottyl_reserved_numeric_value_fn(&parsed.metric_name) {
            let no_labels: BTreeMap<String, MetricPrimitive> = BTreeMap::new();
            if matches_label_matchers(&no_labels, &parsed.matchers) {
                for (sequence, timestamp, _text) in self.logs_range() {
                    let timestamp_ms = timestamp.as_millis();
                    if start_ms.is_none_or(|start| timestamp_ms >= start)
                        && end_ms.is_none_or(|end| timestamp_ms <= end)
                    {
                        series_points
                            .entry(BTreeMap::new())
                            .or_default()
                            .push((timestamp_ms, reserved_value(sequence, timestamp_ms)));
                    }
                }
            }
        } else {
            for (labels, (_, timestamp, value)) in self.bucket_values(&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);
        }

        Ok(match parsed.agg {
            None => series_points
                .into_iter()
                .map(|(labels, points)| PromqlSeriesDto { labels, points })
                .collect(),
            Some(AggOp::Count) => {
                let mut groups: BTreeMap<BTreeMap<String, String>, Vec<Vec<(i64, f64)>>> = BTreeMap::new();
                for (labels, points) in series_points {
                    let count_points: Vec<(i64, f64)> =
                        points.iter().map(|(timestamp_ms, _)| (*timestamp_ms, 1.0)).collect();
                    let group_key = project_labels(&labels, &parsed.by_labels);
                    groups.entry(group_key).or_default().push(count_points);
                }
                groups
                    .into_iter()
                    .map(|(group_labels, sub_series)| {
                        let points = merge_series_windowed(&sub_series, false);
                        PromqlSeriesDto { labels: group_labels, points }
                    })
                    .collect()
            }
            Some(AggOp::Sum) | Some(AggOp::Avg) => {
                let is_avg = parsed.agg == Some(AggOp::Avg);
                let rate_range_ms = parsed
                    .inner_rate
                    .then(|| parsed.range.expect("inner rate() always carries a range").as_millis() as i64);

                let mut groups: BTreeMap<BTreeMap<String, String>, Vec<Vec<(i64, f64)>>> = BTreeMap::new();
                for (labels, points) in series_points {
                    let points = match rate_range_ms {
                        Some(range_ms) => windowed_rate_series(&points, range_ms, start_ms, end_ms),
                        None => points,
                    };
                    let group_key = project_labels(&labels, &parsed.by_labels);
                    groups.entry(group_key).or_default().push(points);
                }

                groups
                    .into_iter()
                    .map(|(group_labels, sub_series)| {
                        let points = if rate_range_ms.is_some() {
                            merge_series_locf(&sub_series, is_avg)
                        } else {
                            merge_series_windowed(&sub_series, is_avg)
                        };
                        PromqlSeriesDto { labels: group_labels, points }
                    })
                    .collect()
            }
            Some(AggOp::Rate) => {
                let range_ms = parsed.range.expect("rate() always carries a range").as_millis() as i64;
                series_points
                    .into_iter()
                    .map(|(labels, points)| PromqlSeriesDto {
                        labels,
                        points: windowed_rate_series(&points, range_ms, start_ms, end_ms),
                    })
                    .collect()
            }
            Some(AggOp::Gauge) => {
                series_points
                    .into_iter()
                    .filter_map(|(labels, points)| {
                        let last = *points.last()?;
                        Some(PromqlSeriesDto { labels, points: vec![last] })
                    })
                    .collect()
            }
            Some(AggOp::Histogram) => {
                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);
                let mut groups: BTreeMap<BTreeMap<String, String>, Vec<&(i64, f64)>> = BTreeMap::new();
                for (labels, points) in &series_points {
                    let group_key = project_labels(labels, &parsed.by_labels);
                    groups.entry(group_key).or_default().extend(points.iter());
                }

                groups
                    .into_iter()
                    .flat_map(|(group_labels, points)| {
                        let mut counts: BTreeMap<i64, u64> = BTreeMap::new();
                        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(move |(bucket_index, count)| {
                            let upper_bound = (bucket_index + 1) as f64 * bucket_width;
                            let mut labels = group_labels.clone();
                            labels.insert("le".to_string(), upper_bound.to_string());
                            PromqlSeriesDto { labels, points: vec![(last_ts, count as f64)] }
                        })
                    })
                    .collect()
            }
        })
    }
}

// --- sort: sort-helper index maintenance ---
impl EngineSnapshot {
    pub(super) fn build_sort_helper_index(&self, sort_by: &str) -> BTreeMap<SortHelperKey, Timestamp> {
        let compiled_params = self.compiled_param_declarations();
        self.logs_range()
            .map(|(sequence, timestamp, text)| {
                let key = match derive_single_param(text, &compiled_params, sort_by) {
                    Some(value) => SortHelperKey::WithParam(value, sequence),
                    None => SortHelperKey::WithoutParam(sequence),
                };
                (key, timestamp)
            })
            .collect()
    }

    pub(crate) fn update_sort_helper_indexes_on_ingest(&mut self, sequence: u64, timestamp: Timestamp, text: &str) {
        if self.sort_helper_indexes.is_empty() {
            return;
        }
        let compiled_params = self.compiled_param_declarations();
        for (sort_by, entry) in self.sort_helper_indexes.iter_mut() {
            let key = match derive_single_param(text, &compiled_params, sort_by) {
                Some(value) => SortHelperKey::WithParam(value, sequence),
                None => SortHelperKey::WithoutParam(sequence),
            };
            entry.index.insert(key, timestamp);
        }
    }

    pub(crate) fn purge_stale_sort_helper_indexes(&mut self) {
        self.sort_helper_indexes
            .retain(|_, entry| entry.last_accessed.elapsed() < SORT_HELPER_INDEX_PURGE_AFTER);
    }

    pub(crate) fn ensure_sort_helper_index_built(&mut self, sort_by: &str) {
        if !self.sort_helper_indexes.contains_key(sort_by) {
            let built = self.build_sort_helper_index(sort_by);
            self.sort_helper_indexes.insert(
                sort_by.to_string(),
                SortHelperIndexEntry { index: built, last_accessed: std::time::Instant::now() },
            );
        }
        let entry = self.sort_helper_indexes.get_mut(sort_by).expect("just inserted above if missing");
        entry.last_accessed = std::time::Instant::now();
    }

    pub(crate) fn compiled_param_declarations(&self) -> Vec<CompiledParamDeclaration> {
        self.param_declarations_range()
            .filter(|(_, _, _, _, _, elapsed_since_last, _, _)| !elapsed_since_last)
            .filter_map(|(name, _, capture_group, json_field_path, kind, _, _, log_stream_gated)| {
                self.param_regex_cache.get(name).map(|re| {
                    (
                        name.to_string(),
                        re.clone(),
                        capture_group.to_string(),
                        json_field_path.map(|path| path.to_string()),
                        kind,
                        log_stream_gated,
                    )
                })
            })
            .collect()
    }
}