rsigma-convert 0.15.0

//! PostgreSQL/TimescaleDB backend for Sigma rule conversion.
//!
//! Converts Sigma detection rules into PostgreSQL SQL queries, leveraging
//! PostgreSQL-native features: `ILIKE` for case-insensitive matching,
//! `~*`/`~` for regex, `inet`/`cidr` for network address matching,
//! `tsvector`/`tsquery` for full-text keyword search, and JSONB for
//! semi-structured event data.

mod correlation;
#[cfg(test)]
mod tests;

use std::collections::HashMap;
use std::sync::LazyLock;

use regex::Regex;
use rsigma_eval::pipeline::state::PipelineState;
use rsigma_parser::*;

use crate::backend::*;
use crate::condition::convert_condition_expr;
use crate::convert::{default_convert_detection, default_convert_detection_item};
use crate::error::{ConvertError, Result};
use crate::state::{ConversionState, ConvertResult};

fn validate_sql_identifier(s: &str) -> Result<()> {
    static RE: LazyLock<Regex> =
        LazyLock::new(|| Regex::new(r"^[A-Za-z_][A-Za-z0-9_$]*$").unwrap());
    if RE.is_match(s) {
        Ok(())
    } else {
        Err(ConvertError::InvalidIdentifier(s.to_string()))
    }
}

/// A JSONB field-path navigation op: an object key or a positional index
/// (possibly negative, counting from the end of the array).
enum FieldOp {
    Key(String),
    Index(i64),
}

/// Return a fresh, query-unique sequence number for naming array-element
/// aliases (`__sigma_e0`, `__sigma_e1`, ...). Stored in the conversion state so
/// nested object-scope blocks never collide.
fn next_array_alias_seq(state: &mut ConversionState) -> u64 {
    let cur = state
        .processing_state
        .get("_array_alias_seq")
        .and_then(|v| v.as_u64())
        .unwrap_or(0);
    state.processing_state.insert(
        "_array_alias_seq".to_string(),
        serde_json::Value::from(cur + 1),
    );
    cur
}

// =============================================================================
// PostgreSQL TextQueryConfig
// =============================================================================

pub static POSTGRES_CONFIG: TextQueryConfig = TextQueryConfig {
    precedence: (TokenType::NOT, TokenType::AND, TokenType::OR),
    group_expression: "({expr})",
    token_separator: " ",

    and_token: "AND",
    or_token: "OR",
    not_token: "NOT",
    eq_token: " = ",

    not_eq_token: Some(" <> "),
    eq_expression: None,
    not_eq_expression: None,
    convert_not_as_not_eq: false,

    wildcard_multi: "%",
    wildcard_single: "_",

    str_quote: "'",
    str_quote_pattern: None,
    str_quote_pattern_negation: false,
    escape_char: "'",
    add_escaped: &[],
    filter_chars: &[],

    field_quote: Some("\""),
    field_quote_pattern: Some(r"^[a-z_][a-z0-9_]*$"),
    field_quote_pattern_negation: true,
    field_escape: None,
    field_escape_pattern: None,

    startswith_expression: Some("{field} ILIKE {value}"),
    not_startswith_expression: Some("{field} NOT ILIKE {value}"),
    startswith_expression_allow_special: false,
    endswith_expression: Some("{field} ILIKE {value}"),
    not_endswith_expression: Some("{field} NOT ILIKE {value}"),
    endswith_expression_allow_special: false,
    contains_expression: Some("{field} ILIKE {value}"),
    not_contains_expression: Some("{field} NOT ILIKE {value}"),
    contains_expression_allow_special: false,
    wildcard_match_expression: Some("{field} ILIKE {value}"),

    case_sensitive_match_expression: Some("{field} LIKE {value}"),
    case_sensitive_startswith_expression: Some("{field} LIKE {value}"),
    case_sensitive_endswith_expression: Some("{field} LIKE {value}"),
    case_sensitive_contains_expression: Some("{field} LIKE {value}"),

    re_expression: Some("{field} ~* {regex}"),
    not_re_expression: Some("{field} !~* {regex}"),
    re_escape_char: None,
    re_escape: &[],
    re_escape_escape_char: None,

    cidr_expression: Some("({field})::inet <<= {value}::cidr"),
    not_cidr_expression: Some("NOT (({field})::inet <<= {value}::cidr)"),

    field_null_expression: "{field} IS NULL",
    field_exists_expression: Some("{field} IS NOT NULL"),
    field_not_exists_expression: Some("{field} IS NULL"),

    compare_op_expression: Some("{field} {op} {value}"),
    compare_ops: &[("gt", ">"), ("gte", ">="), ("lt", "<"), ("lte", "<=")],

    convert_or_as_in: true,
    convert_and_as_in: false,
    in_expressions_allow_wildcards: false,
    field_in_list_expression: Some("{field} {op} ({list})"),
    or_in_operator: Some("IN"),
    and_in_operator: None,
    list_separator: ", ",

    unbound_value_str_expression: None,
    unbound_value_num_expression: None,
    unbound_value_re_expression: None,

    field_eq_field_expression: Some("{field1} = {field2}"),
    field_eq_field_escaping_quoting: true,

    deferred_start: None,
    deferred_separator: None,
    deferred_only_query: "",

    bool_true: "true",
    bool_false: "false",
    query_expression: "SELECT * FROM {table} WHERE {query}",
    state_defaults: &[("table", "security_events")],
};

// =============================================================================
// PostgresBackend
// =============================================================================

/// PostgreSQL/TimescaleDB backend for Sigma rule conversion.
pub struct PostgresBackend {
    pub config: &'static TextQueryConfig,
    /// Default table name (overridden by pipeline state `table` key).
    pub table: String,
    /// Timestamp column name for time-windowed queries.
    pub timestamp_field: String,
    /// If set, fields are accessed via JSONB extraction (`metadata->>'fieldName'`).
    pub json_field: Option<String>,
    /// Use case-sensitive regex (`~`) instead of case-insensitive (`~*`).
    pub case_sensitive_re: bool,
    /// PostgreSQL schema name (e.g. `public`).
    pub schema: Option<String>,
    /// PostgreSQL database name (connection-level metadata, not used in queries).
    pub database: Option<String>,
    /// Enable TimescaleDB-specific features.
    pub timescaledb: bool,
    /// Correlation method selected via the `correlation_method` option
    /// (`sliding`/`tumbling`/`session`), mirroring pySigma's
    /// `correlation_methods`. `None` falls back to each rule's own `window`.
    pub correlation_method: Option<String>,
    /// Default session gap in seconds, from the `gap` option (e.g. `gap=5m`).
    /// Used when a session window is requested (via the rule or
    /// `correlation_method=session`) and the rule does not declare its own
    /// `gap`. A rule's own `gap` always wins.
    pub session_gap_secs: Option<u64>,
}

impl PostgresBackend {
    pub fn new() -> Self {
        Self {
            config: &POSTGRES_CONFIG,
            table: "security_events".to_string(),
            timestamp_field: "time".to_string(),
            json_field: None,
            case_sensitive_re: false,
            schema: None,
            database: None,
            timescaledb: false,
            correlation_method: None,
            session_gap_secs: None,
        }
    }

    /// Create a backend from CLI-style key=value option pairs.
    ///
    /// Recognized keys: `table`, `schema`, `database`, `timestamp_field`,
    /// `json_field`, `case_sensitive_re` (true/false), `correlation_method`
    /// (sliding/tumbling/session), `gap` (default session gap, e.g. `5m`).
    /// Unknown keys are silently ignored so forward-compatible options can be
    /// added without breaking existing invocations.
    pub fn from_options(options: &HashMap<String, String>) -> Self {
        let mut backend = Self::new();
        if let Some(v) = options.get("table") {
            backend.table = v.clone();
        }
        if let Some(v) = options.get("schema") {
            backend.schema = Some(v.clone());
        }
        if let Some(v) = options.get("database") {
            backend.database = Some(v.clone());
        }
        if let Some(v) = options.get("timestamp_field") {
            backend.timestamp_field = v.clone();
        }
        if let Some(v) = options.get("json_field") {
            backend.json_field = Some(v.clone());
        }
        if let Some(v) = options.get("case_sensitive_re") {
            backend.case_sensitive_re = v == "true";
        }
        if let Some(v) = options.get("correlation_method") {
            backend.correlation_method = Some(v.clone());
        }
        if let Some(v) = options.get("gap") {
            // Invalid durations are ignored here (consistent with the lenient
            // option handling above); the CLI validates the format up front.
            backend.session_gap_secs = Timespan::parse(v).ok().map(|t| t.seconds);
        }
        backend
    }

    /// Resolve the effective window mode for a correlation conversion.
    ///
    /// The `correlation_method` backend option, when set, is the converting
    /// user's explicit choice and overrides the rule's own `window` hint;
    /// otherwise the rule's `window` (default `sliding`) is used. An option that
    /// is not one of [`Self::correlation_methods`] is rejected.
    fn resolve_window_mode(&self, rule: &CorrelationRule) -> Result<WindowMode> {
        match self.correlation_method.as_deref() {
            None => Ok(rule.window),
            Some(method) => {
                if !self.correlation_methods().iter().any(|(n, _)| *n == method) {
                    let available = self
                        .correlation_methods()
                        .iter()
                        .map(|(n, _)| *n)
                        .collect::<Vec<_>>()
                        .join(", ");
                    return Err(ConvertError::UnsupportedCorrelation(format!(
                        "unknown correlation_method '{method}'; available: {available}"
                    )));
                }
                method.parse::<WindowMode>().map_err(|_| {
                    ConvertError::UnsupportedCorrelation(format!(
                        "correlation_method '{method}' has no window mapping"
                    ))
                })
            }
        }
    }

    /// Resolve the fully qualified table name `[schema.]table` using this
    /// precedence for each component:
    ///
    /// **table**: `custom_attributes["postgres.table"]` > `state["table"]` > `self.table`
    /// **schema**: `custom_attributes["postgres.schema"]` > `state["schema"]` > `self.schema`
    fn resolve_table(
        &self,
        custom_attrs: &HashMap<String, yaml_serde::Value>,
        state: &HashMap<String, serde_json::Value>,
    ) -> Result<String> {
        let table = custom_attrs
            .get("postgres.table")
            .and_then(|v| v.as_str())
            .or(state.get("table").and_then(|v| v.as_str()))
            .unwrap_or(&self.table);
        validate_sql_identifier(table)?;

        let schema = custom_attrs
            .get("postgres.schema")
            .and_then(|v| v.as_str())
            .or(state.get("schema").and_then(|v| v.as_str()))
            .or(self.schema.as_deref())
            .filter(|s| !s.is_empty());

        if let Some(s) = schema {
            validate_sql_identifier(s)?;
            Ok(format!("{s}.{table}"))
        } else {
            Ok(table.to_string())
        }
    }

    /// Qualify a raw table name with schema. Precedence:
    /// `per_rule_schema` > `state["schema"]` > `self.schema` > none.
    fn qualify_table_name(
        &self,
        table: &str,
        state: &HashMap<String, serde_json::Value>,
        per_rule_schema: Option<&str>,
    ) -> Result<String> {
        validate_sql_identifier(table)?;

        let schema = per_rule_schema
            .or(state.get("schema").and_then(|v| v.as_str()))
            .or(self.schema.as_deref())
            .filter(|s| !s.is_empty());

        if let Some(s) = schema {
            validate_sql_identifier(s)?;
            Ok(format!("{s}.{table}"))
        } else {
            Ok(table.to_string())
        }
    }

    fn field_expr(&self, field: &str) -> Result<String> {
        match &self.json_field {
            // JSONB navigation, with text extraction (`->>`) on the final hop.
            // Supports dotted paths and positional `[N]` indices.
            Some(json_col) => Self::jsonb_path(json_col, field, true),
            None => Ok(text_escape_and_quote_field(self.config, field)),
        }
    }

    /// Escape a string value for use in a SQL single-quoted literal.
    /// PostgreSQL uses `''` to escape single quotes inside string literals.
    fn escape_sql_str(&self, s: &str) -> String {
        s.replace('\'', "''")
    }

    /// Build a SigmaString value into a SQL string literal with proper escaping
    /// and wildcard translation for LIKE/ILIKE.
    fn build_like_value(&self, value: &SigmaString) -> String {
        let mut result = String::with_capacity(value.original.len() + 2);
        result.push('\'');
        for part in &value.parts {
            match part {
                StringPart::Plain(s) => {
                    for ch in s.chars() {
                        match ch {
                            '\'' => result.push_str("''"),
                            '%' => result.push_str("\\%"),
                            '_' => result.push_str("\\_"),
                            '\\' => result.push_str("\\\\"),
                            _ => result.push(ch),
                        }
                    }
                }
                StringPart::Special(SpecialChar::WildcardMulti) => result.push('%'),
                StringPart::Special(SpecialChar::WildcardSingle) => result.push('_'),
            }
        }
        result.push('\'');
        result
    }

    /// Add `%` wildcards to a LIKE value based on modifier semantics.
    /// The value is already a quoted `'...'` string from `build_like_value`.
    fn wrap_like_wildcards(
        &self,
        quoted: &str,
        is_contains: bool,
        is_startswith: bool,
        is_endswith: bool,
    ) -> String {
        if !is_contains && !is_startswith && !is_endswith {
            return quoted.to_string();
        }
        let inner = &quoted[1..quoted.len() - 1];
        let prefix = if is_contains || is_endswith { "%" } else { "" };
        let suffix = if is_contains || is_startswith {
            "%"
        } else {
            ""
        };
        format!("'{prefix}{inner}{suffix}'")
    }

    /// Build a plain SQL string literal from a SigmaString (no wildcards).
    fn build_plain_value(&self, value: &SigmaString) -> String {
        let plain = value.as_plain().unwrap_or_else(|| value.original.clone());
        format!("'{}'", self.escape_sql_str(&plain))
    }

    // --- Array object-scope matching (JSONB) ---

    /// Build a JSONB navigation expression `base->'a'->0->'b'...`, supporting
    /// dotted object keys and positional `[N]` indices. When `as_text` the
    /// final hop uses `->>` (text); otherwise every hop uses `->` (jsonb),
    /// which is what `jsonb_array_elements` needs.
    fn jsonb_path(base: &str, field: &str, as_text: bool) -> Result<String> {
        let ops = Self::parse_field_ops(field)?;
        let last = ops.len() - 1;
        let mut expr = base.to_string();
        for (i, op) in ops.iter().enumerate() {
            let text = as_text && i == last;
            match op {
                FieldOp::Key(k) => {
                    let escaped = k.replace('\'', "''");
                    if text {
                        expr.push_str(&format!("->>'{escaped}'"));
                    } else {
                        expr.push_str(&format!("->'{escaped}'"));
                    }
                }
                FieldOp::Index(n) => {
                    if text {
                        expr.push_str(&format!("->>{n}"));
                    } else {
                        expr.push_str(&format!("->{n}"));
                    }
                }
            }
        }
        Ok(expr)
    }

    /// Parse a JSONB field path into object-key and positional-index ops,
    /// validating each key as a SQL identifier.
    fn parse_field_ops(field: &str) -> Result<Vec<FieldOp>> {
        let mut ops = Vec::new();
        for part in field.split('.') {
            let bpos = part.find('[');
            let name = match bpos {
                Some(p) => &part[..p],
                None => part,
            };
            if !name.is_empty() {
                validate_sql_identifier(name)?;
                ops.push(FieldOp::Key(name.to_string()));
            }
            if let Some(p) = bpos {
                let mut rem = &part[p..];
                while !rem.is_empty() {
                    let rest = rem
                        .strip_prefix('[')
                        .ok_or_else(|| ConvertError::InvalidIdentifier(part.to_string()))?;
                    let close = rest
                        .find(']')
                        .ok_or_else(|| ConvertError::InvalidIdentifier(part.to_string()))?;
                    let n: i64 = rest[..close]
                        .parse()
                        .map_err(|_| ConvertError::InvalidIdentifier(part.to_string()))?;
                    ops.push(FieldOp::Index(n));
                    rem = &rest[close + 1..];
                }
            }
        }
        if ops.is_empty() {
            return Err(ConvertError::InvalidIdentifier(field.to_string()));
        }
        Ok(ops)
    }

    /// Clone this backend with a different JSONB base column. Used to convert an
    /// array object-scope body relative to the per-element alias.
    fn with_json_field(&self, json_field: Option<String>) -> Self {
        PostgresBackend {
            config: self.config,
            table: self.table.clone(),
            timestamp_field: self.timestamp_field.clone(),
            json_field,
            case_sensitive_re: self.case_sensitive_re,
            schema: self.schema.clone(),
            database: self.database.clone(),
            timescaledb: self.timescaledb,
            correlation_method: self.correlation_method.clone(),
            session_gap_secs: self.session_gap_secs,
        }
    }

    /// If `body` matches the array member value itself (every item is
    /// field-less), return those items so the element can be bound as a scalar
    /// text column via `jsonb_array_elements_text`.
    fn scalar_self_body(body: &Detection) -> Option<&[DetectionItem]> {
        if let Detection::AllOf(items) = body
            && !items.is_empty()
            && items.iter().all(|it| it.field.name.is_none())
        {
            return Some(items);
        }
        None
    }

    /// Lower an array object-scope match given the JSONB expression for the
    /// array. `EXISTS` for `any`; non-empty `NOT EXISTS(... NOT ...)` for `all`.
    fn array_exists_from_expr(
        &self,
        array_expr: &str,
        quantifier: ArrayQuantifier,
        body: &Detection,
        state: &mut ConversionState,
    ) -> Result<String> {
        let seq = next_array_alias_seq(state);
        let alias = format!("__sigma_e{seq}");

        let (srf, body_sql) = if let Some(items) = Self::scalar_self_body(body) {
            // Scalar members: bind each element as a text column and reuse the
            // normal item conversion by renaming the field-less items to the
            // alias (a plain, unquoted identifier in non-JSONB mode).
            let renamed = Detection::AllOf(
                items
                    .iter()
                    .map(|it| DetectionItem {
                        field: FieldSpec::new(Some(alias.clone()), it.field.modifiers.clone()),
                        values: it.values.clone(),
                    })
                    .collect(),
            );
            let elem = self.with_json_field(None);
            (
                "jsonb_array_elements_text",
                default_convert_detection(&elem, &renamed, state)?,
            )
        } else {
            // Object members: bind each element as a JSONB column and convert
            // the body relative to it (nested blocks recurse via the trait).
            let elem = self.with_json_field(Some(alias.clone()));
            (
                "jsonb_array_elements",
                default_convert_detection(&elem, body, state)?,
            )
        };

        Ok(match quantifier {
            ArrayQuantifier::Any => format!(
                "(jsonb_typeof({array_expr}) = 'array' AND EXISTS \
                 (SELECT 1 FROM {srf}({array_expr}) AS {alias} WHERE {body_sql}))"
            ),
            ArrayQuantifier::All => format!(
                "(jsonb_typeof({array_expr}) = 'array' AND jsonb_array_length({array_expr}) > 0 \
                 AND NOT EXISTS \
                 (SELECT 1 FROM {srf}({array_expr}) AS {alias} WHERE NOT ({body_sql})))"
            ),
            // `all_or_empty` is `all` minus the non-empty guard, and it also
            // matches a missing/null array. Same CASE shape as `none` so the
            // empty/missing cases match without running jsonb_array_elements on
            // a scalar.
            ArrayQuantifier::AllOrEmpty => format!(
                "(CASE WHEN jsonb_typeof({array_expr}) = 'array' \
                 THEN NOT EXISTS (SELECT 1 FROM {srf}({array_expr}) AS {alias} WHERE NOT ({body_sql})) \
                 ELSE {array_expr} IS NULL OR jsonb_typeof({array_expr}) = 'null' END)"
            ),
            // `none` matches an empty or missing array, so it cannot use the
            // `typeof = 'array' AND ...` guard (that would reject empty/missing).
            // A CASE guarantees `jsonb_array_elements` is only evaluated on an
            // actual array, avoiding a runtime error on a scalar value.
            ArrayQuantifier::None => format!(
                "(CASE WHEN jsonb_typeof({array_expr}) = 'array' \
                 THEN NOT EXISTS (SELECT 1 FROM {srf}({array_expr}) AS {alias} WHERE {body_sql}) \
                 ELSE {array_expr} IS NULL OR jsonb_typeof({array_expr}) = 'null' END)"
            ),
        })
    }
}

impl Default for PostgresBackend {
    fn default() -> Self {
        Self::new()
    }
}

impl Backend for PostgresBackend {
    fn name(&self) -> &str {
        "postgres"
    }

    fn formats(&self) -> &[(&str, &str)] {
        &[
            ("default", "Plain PostgreSQL SQL"),
            ("view", "CREATE OR REPLACE VIEW for each rule"),
            (
                "timescaledb",
                "TimescaleDB-optimized queries with time_bucket()",
            ),
            (
                "continuous_aggregate",
                "CREATE MATERIALIZED VIEW ... WITH (timescaledb.continuous)",
            ),
            (
                "sliding_window",
                "Correlation queries using window functions for per-row sliding detection",
            ),
        ]
    }

    fn requires_pipeline(&self) -> bool {
        false
    }

    // --- Detection rule conversion ---

    fn convert_rule(
        &self,
        rule: &SigmaRule,
        output_format: &str,
        pipeline_state: &PipelineState,
    ) -> Result<Vec<String>> {
        let mut queries = Vec::new();
        for (idx, cond_expr) in rule.detection.conditions.iter().enumerate() {
            let mut state = ConversionState::new(pipeline_state.state.clone());
            state
                .processing_state
                .insert("_output_format".to_string(), output_format.into());
            let query = self.convert_condition(cond_expr, &rule.detection.named, &mut state)?;
            let finished = self.finish_query(rule, query, &state)?;
            let finalized = self.finalize_query(rule, finished, idx, &state, output_format)?;
            queries.push(finalized);
        }
        Ok(queries)
    }

    // --- Condition tree dispatch ---

    fn convert_condition(
        &self,
        expr: &ConditionExpr,
        detections: &HashMap<String, Detection>,
        state: &mut ConversionState,
    ) -> Result<String> {
        convert_condition_expr(self, expr, detections, state)
    }

    fn convert_condition_and(&self, exprs: &[String]) -> Result<String> {
        Ok(text_convert_condition_and(self.config, exprs))
    }

    fn convert_condition_or(&self, exprs: &[String]) -> Result<String> {
        Ok(text_convert_condition_or(self.config, exprs))
    }

    fn convert_condition_not(&self, expr: &str) -> Result<String> {
        Ok(text_convert_condition_not(self.config, expr))
    }

    // --- Detection ---

    fn convert_detection(&self, det: &Detection, state: &mut ConversionState) -> Result<String> {
        default_convert_detection(self, det, state)
    }

    fn convert_detection_item(
        &self,
        item: &DetectionItem,
        state: &mut ConversionState,
    ) -> Result<String> {
        default_convert_detection_item(self, item, state)
    }

    fn convert_array_match(
        &self,
        field: &str,
        quantifier: ArrayQuantifier,
        body: &Detection,
        state: &mut ConversionState,
    ) -> Result<String> {
        // Array matching requires JSONB-backed event storage; in flat-column
        // mode there is no array to unnest.
        let json_col = self
            .json_field
            .as_deref()
            .ok_or(ConvertError::UnsupportedArrayMatching)?;
        let array_expr = Self::jsonb_path(json_col, field, false)?;
        self.array_exists_from_expr(&array_expr, quantifier, body, state)
    }

    fn supports_field_index(&self) -> bool {
        // Positional `field[N]` lowers to JSONB `->n` / `->>n`, but only when
        // events are stored in a JSONB column; flat columns have no array.
        self.json_field.is_some()
    }

    // --- Field/value escaping ---

    fn escape_and_quote_field(&self, field: &str) -> String {
        self.field_expr(field)
            .unwrap_or_else(|_| text_escape_and_quote_field(self.config, field))
    }

    fn convert_value_str(&self, value: &SigmaString, _state: &ConversionState) -> String {
        self.build_like_value(value)
    }

    fn convert_value_re(&self, regex: &str, _state: &ConversionState) -> String {
        format!("'{}'", self.escape_sql_str(regex))
    }

    // --- Value-type-specific methods ---

    fn convert_field_eq_str(
        &self,
        field: &str,
        value: &SigmaString,
        modifiers: &[Modifier],
        _state: &mut ConversionState,
    ) -> Result<ConvertResult> {
        let f = self.field_expr(field)?;
        let is_cased = modifiers.contains(&Modifier::Cased);
        let is_contains = modifiers.contains(&Modifier::Contains);
        let is_startswith = modifiers.contains(&Modifier::StartsWith);
        let is_endswith = modifiers.contains(&Modifier::EndsWith);
        let has_wildcards = value.contains_wildcards();

        let like_op = if is_cased { "LIKE" } else { "ILIKE" };

        if is_contains || is_startswith || is_endswith || has_wildcards {
            let inner = self.build_like_value(value);
            let val = self.wrap_like_wildcards(&inner, is_contains, is_startswith, is_endswith);
            return Ok(ConvertResult::Query(format!("{f} {like_op} {val}")));
        }

        let val = self.build_plain_value(value);
        Ok(ConvertResult::Query(format!("{f} = {val}")))
    }

    fn convert_field_eq_str_case_sensitive(
        &self,
        field: &str,
        value: &SigmaString,
        modifiers: &[Modifier],
        state: &mut ConversionState,
    ) -> Result<ConvertResult> {
        let mut mods = modifiers.to_vec();
        if !mods.contains(&Modifier::Cased) {
            mods.push(Modifier::Cased);
        }
        self.convert_field_eq_str(field, value, &mods, state)
    }

    fn convert_field_eq_num(
        &self,
        field: &str,
        value: f64,
        _state: &mut ConversionState,
    ) -> Result<String> {
        let f = self.field_expr(field)?;
        if value.fract() == 0.0 && (i64::MIN as f64..=i64::MAX as f64).contains(&value) {
            Ok(format!("{f} = {}", value as i64))
        } else {
            Ok(format!("{f} = {value}"))
        }
    }

    fn convert_field_eq_bool(
        &self,
        field: &str,
        value: bool,
        _state: &mut ConversionState,
    ) -> Result<String> {
        let f = self.field_expr(field)?;
        let v = if value {
            self.config.bool_true
        } else {
            self.config.bool_false
        };
        Ok(format!("{f} = {v}"))
    }

    fn convert_field_eq_null(&self, field: &str, _state: &mut ConversionState) -> Result<String> {
        let f = self.field_expr(field)?;
        Ok(format!("{f} IS NULL"))
    }

    fn convert_field_eq_re(
        &self,
        field: &str,
        pattern: &str,
        flags: &[Modifier],
        _state: &mut ConversionState,
    ) -> Result<ConvertResult> {
        let f = self.field_expr(field)?;
        let escaped_pattern = self.escape_sql_str(pattern);
        let is_cased = flags.contains(&Modifier::Cased) || self.case_sensitive_re;
        let op = if is_cased { "~" } else { "~*" };
        Ok(ConvertResult::Query(format!(
            "{f} {op} '{escaped_pattern}'"
        )))
    }

    fn convert_field_eq_cidr(
        &self,
        field: &str,
        cidr: &str,
        _state: &mut ConversionState,
    ) -> Result<ConvertResult> {
        let f = self.field_expr(field)?;
        Ok(ConvertResult::Query(format!(
            "({f})::inet <<= '{cidr}'::cidr"
        )))
    }

    fn convert_field_compare(
        &self,
        field: &str,
        op: &Modifier,
        value: f64,
        _state: &mut ConversionState,
    ) -> Result<String> {
        let f = self.field_expr(field)?;
        let op_token = match op {
            Modifier::Lt => "<",
            Modifier::Lte => "<=",
            Modifier::Gt => ">",
            Modifier::Gte => ">=",
            _ => {
                return Err(ConvertError::UnsupportedModifier(format!(
                    "compare op {op:?}"
                )));
            }
        };
        let val_str =
            if value.fract() == 0.0 && (i64::MIN as f64..=i64::MAX as f64).contains(&value) {
                (value as i64).to_string()
            } else {
                value.to_string()
            };
        Ok(format!("{f} {op_token} {val_str}"))
    }

    fn convert_field_exists(
        &self,
        field: &str,
        exists: bool,
        _state: &mut ConversionState,
    ) -> Result<String> {
        let f = self.field_expr(field)?;
        if exists {
            Ok(format!("{f} IS NOT NULL"))
        } else {
            Ok(format!("{f} IS NULL"))
        }
    }

    fn convert_field_eq_query_expr(
        &self,
        field: &str,
        expr: &str,
        _id: &str,
        _state: &mut ConversionState,
    ) -> Result<String> {
        let f = self.field_expr(field)?;
        let resolved = expr.replace("{field}", &f);
        Ok(resolved)
    }

    fn convert_field_ref(
        &self,
        field1: &str,
        field2: &str,
        _state: &mut ConversionState,
    ) -> Result<ConvertResult> {
        let f1 = self.field_expr(field1)?;
        let f2 = self.field_expr(field2)?;
        Ok(ConvertResult::Query(format!("{f1} = {f2}")))
    }

    fn convert_keyword(&self, value: &SigmaValue, _state: &mut ConversionState) -> Result<String> {
        let search_target = match &self.json_field {
            Some(json_col) => format!("{json_col}::text"),
            None => "ROW(*)::text".to_string(),
        };
        match value {
            SigmaValue::String(s) => {
                let plain = s.as_plain().unwrap_or_else(|| s.original.clone());
                if plain.is_empty() {
                    return Err(ConvertError::UnsupportedKeyword);
                }
                let escaped = self.escape_sql_str(&plain);
                Ok(format!(
                    "to_tsvector('simple', {search_target}) @@ plainto_tsquery('simple', '{escaped}')"
                ))
            }
            SigmaValue::Integer(n) => Ok(format!(
                "to_tsvector('simple', {search_target}) @@ plainto_tsquery('simple', '{n}')"
            )),
            SigmaValue::Float(f) => Ok(format!(
                "to_tsvector('simple', {search_target}) @@ plainto_tsquery('simple', '{f}')"
            )),
            _ => Err(ConvertError::UnsupportedKeyword),
        }
    }

    fn convert_condition_as_in_expression(
        &self,
        field: &str,
        values: &[&SigmaValue],
        is_or: bool,
        _state: &mut ConversionState,
    ) -> Result<String> {
        if !is_or {
            return Err(ConvertError::UnsupportedModifier(
                "AND IN-list not supported for PostgreSQL".into(),
            ));
        }
        let f = self.field_expr(field)?;
        let items: Vec<String> = values
            .iter()
            .map(|v| match v {
                SigmaValue::String(s) => self.build_plain_value(s),
                SigmaValue::Integer(n) => n.to_string(),
                SigmaValue::Float(f) => f.to_string(),
                _ => String::new(),
            })
            .collect();
        let list = items.join(", ");
        Ok(format!("{f} IN ({list})"))
    }

    // --- Query finalization ---

    fn finish_query(
        &self,
        rule: &SigmaRule,
        query: String,
        state: &ConversionState,
    ) -> Result<String> {
        let qualified = self.resolve_table(&rule.custom_attributes, &state.processing_state)?;

        let is_timescaledb = state
            .get_state_str("_output_format")
            .is_some_and(|f| f == "timescaledb" || f == "continuous_aggregate");

        let base_cols = if rule.fields.is_empty() {
            "*".to_string()
        } else {
            rule.fields
                .iter()
                .map(|f| self.format_select_field(f))
                .collect::<Result<Vec<_>>>()?
                .join(", ")
        };

        let select_cols = if is_timescaledb {
            format!(
                "time_bucket('1 hour', {}) AS bucket, {}",
                self.timestamp_field, base_cols
            )
        } else {
            base_cols
        };

        let custom_tmpl = state.get_state_str("query_expression_template");

        let effective_tmpl = match custom_tmpl {
            Some(t) => t.to_string(),
            None => format!("SELECT {select_cols} FROM {{table}} WHERE {{query}}"),
        };

        let mut result = effective_tmpl.replace("{query}", &query);
        result = result.replace("{table}", &qualified);
        result = result.replace("{rule.title}", &rule.title);
        if let Some(id) = &rule.id {
            result = result.replace("{rule.id}", id);
        }

        Ok(result)
    }

    fn finalize_query(
        &self,
        rule: &SigmaRule,
        query: String,
        _index: usize,
        _state: &ConversionState,
        output_format: &str,
    ) -> Result<String> {
        let view_name = || {
            let raw = match &rule.id {
                Some(id) => id.replace('-', "_"),
                None => rule.title.to_lowercase().replace([' ', '-'], "_"),
            };
            let sanitized: String = raw
                .chars()
                .filter(|c| c.is_ascii_alphanumeric() || *c == '_')
                .collect();
            if sanitized.is_empty() {
                "sigma_rule".to_string()
            } else {
                format!("sigma_{sanitized}")
            }
        };

        match output_format {
            "default" | "timescaledb" => Ok(query),
            "view" => Ok(format!("CREATE OR REPLACE VIEW {} AS {query}", view_name())),
            "continuous_aggregate" => Ok(format!(
                "CREATE MATERIALIZED VIEW {} \
                 WITH (timescaledb.continuous) AS {query} \
                 WITH NO DATA",
                view_name()
            )),
            other => Err(ConvertError::RuleConversion(format!(
                "unknown output format: {other}"
            ))),
        }
    }

    fn finalize_output(&self, queries: Vec<String>, output_format: &str) -> Result<String> {
        let sep = match output_format {
            "view" | "continuous_aggregate" => ";\n\n",
            _ => "\n",
        };
        Ok(queries.join(sep))
    }

    // --- Correlation ---

    fn supports_correlation(&self) -> bool {
        true
    }

    fn correlation_methods(&self) -> &[(&str, &str)] {
        &[
            (
                "sliding",
                "Trailing per-event window (default; preserves existing SQL)",
            ),
            (
                "tumbling",
                "Fixed boundary-aligned buckets (time_bucket/date_bin)",
            ),
            (
                "session",
                "Gaps-and-islands sessionization (requires a gap)",
            ),
        ]
    }

    fn default_correlation_method(&self) -> &str {
        "sliding"
    }

    fn convert_correlation_rule_with_warnings(
        &self,
        rule: &CorrelationRule,
        output_format: &str,
        pipeline_state: &PipelineState,
        warnings: &mut Vec<String>,
    ) -> Result<Vec<String>> {
        let table = self.resolve_table(&rule.custom_attributes, &pipeline_state.state)?;
        let ts = &self.timestamp_field;
        let use_time_bucket =
            output_format == "timescaledb" || output_format == "continuous_aggregate";

        let mut group_by_cols: Vec<String> = rule
            .group_by
            .iter()
            .map(|g| self.field_expr(g))
            .collect::<Result<_>>()?;
        if use_time_bucket {
            group_by_cols.insert(0, format!("time_bucket('1 hour', {ts})"));
        }
        let group_by_clause = if group_by_cols.is_empty() {
            String::new()
        } else {
            format!(" GROUP BY {}", group_by_cols.join(", "))
        };
        let group_by_select = if group_by_cols.is_empty() {
            String::new()
        } else {
            format!("{}, ", group_by_cols.join(", "))
        };

        let window_secs = rule.timespan.seconds;
        let having_clause = self.build_having_clause(&rule.condition)?;

        let field_from_condition = match &rule.condition {
            CorrelationCondition::Threshold { field, .. } => {
                field.as_ref().and_then(|f| f.first().cloned())
            }
            _ => None,
        };
        let value_field = field_from_condition.as_deref().or_else(|| {
            rule.aliases
                .first()
                .and_then(|a| a.mapping.values().next().map(|s| s.as_str()))
        });

        // Build per-rule table mapping from _rule_tables injected by convert_collection
        let rule_tables: HashMap<String, String> = pipeline_state
            .state
            .get("_rule_tables")
            .and_then(|v| serde_json::from_value(v.clone()).ok())
            .unwrap_or_default();

        // Per-rule converted queries for CTE-based pre-filtering
        let rule_queries: HashMap<String, String> = pipeline_state
            .state
            .get("_rule_queries")
            .and_then(|v| serde_json::from_value(v.clone()).ok())
            .unwrap_or_default();

        let (cte_prefix, source_table, time_filter) =
            self.build_correlation_source(&rule.rules, &rule_queries, &table, ts, window_secs);

        // The windowing strategy (sliding/tumbling/session). The
        // `correlation_method` option is the converting user's explicit choice
        // (pySigma-style) and overrides the rule's own `window` hint; otherwise
        // the rule's `window` is used. `sliding` (the default, also what an
        // absent `window` resolves to) preserves the existing per-output_format
        // behavior, so existing rules are unaffected. `tumbling` and `session`
        // are opt-in.
        let window = self.resolve_window_mode(rule)?;
        let temporal = matches!(
            rule.correlation_type,
            CorrelationType::Temporal | CorrelationType::TemporalOrdered
        );
        if matches!(window, WindowMode::Tumbling) {
            let query = if temporal {
                self.build_temporal_tumbling(
                    rule,
                    &table,
                    ts,
                    window_secs,
                    use_time_bucket,
                    &having_clause,
                    &rule_tables,
                    pipeline_state,
                )?
            } else {
                self.build_tumbling_correlation(
                    rule,
                    &cte_prefix,
                    &source_table,
                    ts,
                    window_secs,
                    value_field,
                    use_time_bucket,
                )?
            };
            return Ok(vec![query]);
        }
        if matches!(window, WindowMode::Session) {
            // The rule's own `gap` wins; the `gap` backend option provides the
            // default so `correlation_method=session` works for rules that do
            // not declare one.
            let gap_secs = rule
                .gap
                .as_ref()
                .map(|g| g.seconds)
                .or(self.session_gap_secs)
                .ok_or_else(|| {
                    ConvertError::UnsupportedCorrelation(
                        "session window requires a 'gap' (set it on the rule or pass \
                         -O gap=5m as a conversion default)"
                            .into(),
                    )
                })?;
            let query = if temporal {
                self.build_temporal_session(
                    rule,
                    &table,
                    ts,
                    window_secs,
                    gap_secs,
                    &having_clause,
                    &rule_tables,
                    pipeline_state,
                    warnings,
                )?
            } else {
                self.build_session_correlation(
                    rule,
                    &cte_prefix,
                    &source_table,
                    ts,
                    window_secs,
                    gap_secs,
                    value_field,
                    warnings,
                )?
            };
            return Ok(vec![query]);
        }

        let query = match rule.correlation_type {
            CorrelationType::EventCount if output_format == "sliding_window" => self
                .build_sliding_window_query(
                    &cte_prefix,
                    &source_table,
                    &time_filter,
                    &rule.group_by,
                    ts,
                    window_secs,
                    &rule.condition,
                )?,
            CorrelationType::EventCount => {
                format!(
                    "{cte_prefix}SELECT {group_by_select}COUNT(*) AS event_count \
                     FROM {source_table}\
                     {time_filter}\
                     {group_by_clause} \
                     HAVING {having_clause}",
                    having_clause = having_clause.replace("{agg}", "COUNT(*)")
                )
            }
            CorrelationType::ValueCount => {
                let field = match value_field {
                    Some(f) => self.field_expr(f)?,
                    None => "'unknown_field'".to_string(),
                };
                let agg = format!("COUNT(DISTINCT {field})");
                format!(
                    "{cte_prefix}SELECT {group_by_select}{agg} AS value_count \
                     FROM {source_table}\
                     {time_filter}\
                     {group_by_clause} \
                     HAVING {having_clause}",
                    having_clause = having_clause.replace("{agg}", &agg)
                )
            }
            CorrelationType::Temporal | CorrelationType::TemporalOrdered => self
                .build_temporal_query(
                    rule,
                    &table,
                    ts,
                    window_secs,
                    &group_by_select,
                    &group_by_clause,
                    &having_clause,
                    &rule_tables,
                    pipeline_state,
                )?,
            CorrelationType::ValueSum => {
                let field = match value_field {
                    Some(f) => self.field_expr(f)?,
                    None => "'unknown_field'".to_string(),
                };
                let agg = format!("SUM({field})");
                format!(
                    "{cte_prefix}SELECT {group_by_select}{agg} AS value_sum \
                     FROM {source_table}\
                     {time_filter}\
                     {group_by_clause} \
                     HAVING {having_clause}",
                    having_clause = having_clause.replace("{agg}", &agg)
                )
            }
            CorrelationType::ValueAvg => {
                let field = match value_field {
                    Some(f) => self.field_expr(f)?,
                    None => "'unknown_field'".to_string(),
                };
                let agg = format!("AVG({field})");
                format!(
                    "{cte_prefix}SELECT {group_by_select}{agg} AS value_avg \
                     FROM {source_table}\
                     {time_filter}\
                     {group_by_clause} \
                     HAVING {having_clause}",
                    having_clause = having_clause.replace("{agg}", &agg)
                )
            }
            CorrelationType::ValuePercentile | CorrelationType::ValueMedian => {
                let field = match value_field {
                    Some(f) => self.field_expr(f)?,
                    None => "'unknown_field'".to_string(),
                };
                let percentile = if rule.correlation_type == CorrelationType::ValueMedian {
                    0.5
                } else {
                    match &rule.condition {
                        CorrelationCondition::Threshold { percentile, .. } => {
                            percentile.map(|p| p as f64 / 100.0).unwrap_or(0.95)
                        }
                        _ => 0.95,
                    }
                };
                let agg = format!("PERCENTILE_CONT({percentile}) WITHIN GROUP (ORDER BY {field})");
                format!(
                    "{cte_prefix}SELECT {group_by_select}\
                     {agg} AS pct_value \
                     FROM {source_table}\
                     {time_filter}\
                     {group_by_clause} \
                     HAVING {having_clause}",
                    having_clause = having_clause.replace("{agg}", &agg)
                )
            }
        };

        Ok(vec![query])
    }
}