// Project:   scalo
// File:      src/transport/filter/classify.rs
// Purpose:   CEL expression classification into performance tiers
// Language:  Rust
//
// License:   Apache-2.0
// Copyright: (c) 2026 HYPERI PTY LIMITED

//! Classify CEL expressions into performance tiers via text pattern matching.
//!
//! Tier 1 patterns are detected by regex and executed as SIMD field operations
//! (no CEL engine). Expressions that don't match Tier 1 are classified as
//! Tier 2 (standard CEL) or Tier 3 (complex CEL with restricted functions).

use std::sync::LazyLock;

use regex::Regex;

use super::config::FilterTier;

/// Recognised Tier 1 operation extracted from the expression text.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Tier1Op {
    FieldExists { field: String },
    FieldNotExists { field: String },
    FieldEquals { field: String, value: String },
    FieldNotEquals { field: String, value: String },
    FieldStartsWith { field: String, prefix: String },
    FieldEndsWith { field: String, suffix: String },
    FieldContains { field: String, substring: String },
}

/// Result of classifying an expression.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ClassifyResult {
    /// Expression matches a Tier 1 SIMD pattern.
    Tier1(Tier1Op),
    /// Expression is valid CEL without restricted functions (Tier 2).
    Tier2 { fields: Vec<String> },
    /// Expression uses restricted functions (Tier 3).
    Tier3 { fields: Vec<String> },
}

impl ClassifyResult {
    #[must_use]
    pub fn tier(&self) -> FilterTier {
        match self {
            Self::Tier1(_) => FilterTier::Tier1,
            Self::Tier2 { .. } => FilterTier::Tier2,
            Self::Tier3 { .. } => FilterTier::Tier3,
        }
    }
}

// ---------------------------------------------------------------------------
// Tier 1 regex patterns (compiled once via LazyLock)
// ---------------------------------------------------------------------------

// Field name: word chars + dots for nested paths
static RE_HAS: LazyLock<Regex> =
    LazyLock::new(|| Regex::new(r"^\s*has\(\s*([\w.]+)\s*\)\s*$").unwrap());

static RE_NOT_HAS: LazyLock<Regex> =
    LazyLock::new(|| Regex::new(r"^\s*!\s*has\(\s*([\w.]+)\s*\)\s*$").unwrap());

static RE_EQ_STR: LazyLock<Regex> =
    LazyLock::new(|| Regex::new(r#"^\s*([\w.]+)\s*==\s*"([^"]*)"\s*$"#).unwrap());

static RE_NEQ_STR: LazyLock<Regex> =
    LazyLock::new(|| Regex::new(r#"^\s*([\w.]+)\s*!=\s*"([^"]*)"\s*$"#).unwrap());

static RE_STARTS_WITH: LazyLock<Regex> = LazyLock::new(|| {
    Regex::new(r#"^\s*([\w.]+)\s*\.\s*startsWith\(\s*"([^"]*)"\s*\)\s*$"#).unwrap()
});

static RE_ENDS_WITH: LazyLock<Regex> =
    LazyLock::new(|| Regex::new(r#"^\s*([\w.]+)\s*\.\s*endsWith\(\s*"([^"]*)"\s*\)\s*$"#).unwrap());

static RE_CONTAINS: LazyLock<Regex> =
    LazyLock::new(|| Regex::new(r#"^\s*([\w.]+)\s*\.\s*contains\(\s*"([^"]*)"\s*\)\s*$"#).unwrap());

// Restricted function names (Tier 3)
const RESTRICTED_FUNCTIONS: &[&str] = &[
    "matches",
    "map",
    "filter",
    "exists",
    "all",
    "exists_one",
    "timestamp",
    "duration",
];

// CEL keywords and built-in function names (NOT field references)
const CEL_KEYWORDS: &[&str] = &[
    "true",
    "false",
    "null",
    "in",
    "has",
    "size",
    "int",
    "uint",
    "double",
    "string",
    "bool",
    "type",
    "contains",
    "startsWith",
    "endsWith",
    "matches",
    "map",
    "filter",
    "exists",
    "all",
    "exists_one",
    "timestamp",
    "duration",
];

/// Classify a CEL expression into a performance tier.
///
/// Returns `Err` if the expression is syntactically invalid (can't even be
/// parsed as a potential CEL expression -- empty, unbalanced quotes, etc.).
///
/// # Examples
///
/// ```rust,ignore
/// let result = classify("has(_table)");
/// assert!(matches!(result, Ok(ClassifyResult::Tier1(..))));
///
/// let result = classify("severity > 3 && source != \"internal\"");
/// assert!(matches!(result, Ok(ClassifyResult::Tier2 { .. })));
/// ```
pub fn classify(expr: &str) -> Result<ClassifyResult, String> {
    let trimmed = expr.trim();
    if trimmed.is_empty() {
        return Err("empty expression".into());
    }

    // Try Tier 1 patterns first (ordered by expected frequency)
    if let Some(op) = try_tier1(trimmed) {
        return Ok(ClassifyResult::Tier1(op));
    }

    // Not Tier 1 -- check for restricted functions (Tier 3) vs standard (Tier 2)
    let has_restricted = check_restricted_functions(trimmed);
    let fields = extract_field_references(trimmed);

    if has_restricted {
        Ok(ClassifyResult::Tier3 { fields })
    } else {
        Ok(ClassifyResult::Tier2 { fields })
    }
}

/// Try to match a Tier 1 pattern. Returns `None` if no pattern matches.
fn try_tier1(expr: &str) -> Option<Tier1Op> {
    // has(field)
    if let Some(caps) = RE_HAS.captures(expr) {
        return Some(Tier1Op::FieldExists {
            field: caps[1].to_string(),
        });
    }

    // !has(field)
    if let Some(caps) = RE_NOT_HAS.captures(expr) {
        return Some(Tier1Op::FieldNotExists {
            field: caps[1].to_string(),
        });
    }

    // field == "value"
    if let Some(caps) = RE_EQ_STR.captures(expr) {
        return Some(Tier1Op::FieldEquals {
            field: caps[1].to_string(),
            value: caps[2].to_string(),
        });
    }

    // field != "value"
    if let Some(caps) = RE_NEQ_STR.captures(expr) {
        return Some(Tier1Op::FieldNotEquals {
            field: caps[1].to_string(),
            value: caps[2].to_string(),
        });
    }

    // field.startsWith("prefix")
    if let Some(caps) = RE_STARTS_WITH.captures(expr) {
        return Some(Tier1Op::FieldStartsWith {
            field: caps[1].to_string(),
            prefix: caps[2].to_string(),
        });
    }

    // field.endsWith("suffix")
    if let Some(caps) = RE_ENDS_WITH.captures(expr) {
        return Some(Tier1Op::FieldEndsWith {
            field: caps[1].to_string(),
            suffix: caps[2].to_string(),
        });
    }

    // field.contains("substring")
    if let Some(caps) = RE_CONTAINS.captures(expr) {
        return Some(Tier1Op::FieldContains {
            field: caps[1].to_string(),
            substring: caps[2].to_string(),
        });
    }

    None
}

/// Check if the expression uses any restricted functions (Tier 3).
///
/// Walks all occurrences of each `func(` pattern; returns true on
/// the first match that lands outside a string literal. The previous
/// "find first occurrence + quote-count" shape missed real calls
/// when an earlier same-name occurrence sat inside a string.
fn check_restricted_functions(expr: &str) -> bool {
    for func in RESTRICTED_FUNCTIONS {
        let pattern = format!("{func}(");
        for (pos, _) in expr.match_indices(&pattern) {
            if !position_is_in_string(expr, pos) {
                return true;
            }
        }
    }
    false
}

/// `true` if byte `pos` falls inside a `"..."` string literal,
/// counting `\"` as escaped.
fn position_is_in_string(expr: &str, pos: usize) -> bool {
    let mut in_string = false;
    let mut prev_was_escape = false;
    for (i, ch) in expr.char_indices() {
        if i >= pos {
            return in_string;
        }
        if ch == '"' && !prev_was_escape {
            in_string = !in_string;
        }
        prev_was_escape = ch == '\\' && !prev_was_escape;
    }
    in_string
}

/// Extract field references from an expression (for Tier 2/3 CEL context building).
///
/// Scans for identifier patterns that aren't CEL keywords or function names.
/// For method calls like `field.matches("...")`, extracts only the receiver field.
/// Returns unique field names (may include dotted paths for nested access).
fn extract_field_references(expr: &str) -> Vec<String> {
    // Match dotted identifier (potentially nested) -- we'll trim trailing method call
    // segments after matching.
    static RE_IDENT: LazyLock<Regex> = LazyLock::new(|| Regex::new(r"[a-zA-Z_][\w.]*").unwrap());

    let mut fields: Vec<String> = Vec::new();

    // Build a mask of which byte positions are inside string literals
    let mut in_string_mask = vec![false; expr.len()];
    let mut in_string = false;
    let mut prev_was_escape = false;
    for (i, ch) in expr.char_indices() {
        if in_string {
            in_string_mask[i] = true;
        }
        if ch == '"' && !prev_was_escape {
            in_string = !in_string;
        }
        prev_was_escape = ch == '\\' && !prev_was_escape;
    }

    for m in RE_IDENT.find_iter(expr) {
        if in_string_mask.get(m.start()).copied().unwrap_or(false) {
            continue;
        }

        let mut ident = m.as_str().to_string();

        // If this identifier is immediately followed by '(' (a function call),
        // strip the last dotted segment (the method name) -- the receiver is
        // the actual field reference.
        let after = &expr[m.end()..];
        if after.trim_start().starts_with('(') {
            if let Some(dot_pos) = ident.rfind('.') {
                // Method call on a field: keep the receiver
                ident.truncate(dot_pos);
            } else {
                // Bare function call (e.g., has(), size()) -- not a field
                continue;
            }
        }

        if ident.is_empty() {
            continue;
        }

        // Skip CEL keywords (check the leading segment)
        let base = ident.split('.').next().unwrap_or(&ident);
        if CEL_KEYWORDS.contains(&base) {
            continue;
        }

        if !fields.contains(&ident) {
            fields.push(ident);
        }
    }

    fields
}

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

    #[test]
    fn classify_has_field() {
        let result = classify("has(_table)").unwrap();
        assert_eq!(result.tier(), FilterTier::Tier1);
        assert!(matches!(
            result,
            ClassifyResult::Tier1(Tier1Op::FieldExists { ref field }) if field == "_table"
        ));
    }

    #[test]
    fn classify_not_has_field() {
        let result = classify("!has(_internal)").unwrap();
        assert_eq!(result.tier(), FilterTier::Tier1);
        assert!(matches!(
            result,
            ClassifyResult::Tier1(Tier1Op::FieldNotExists { ref field }) if field == "_internal"
        ));
    }

    #[test]
    fn classify_field_equals_string() {
        let result = classify(r#"status == "poison""#).unwrap();
        assert_eq!(result.tier(), FilterTier::Tier1);
        assert!(matches!(
            result,
            ClassifyResult::Tier1(Tier1Op::FieldEquals { ref field, ref value })
                if field == "status" && value == "poison"
        ));
    }

    #[test]
    fn classify_field_not_equals() {
        let result = classify(r#"source != "internal""#).unwrap();
        assert_eq!(result.tier(), FilterTier::Tier1);
        assert!(matches!(
            result,
            ClassifyResult::Tier1(Tier1Op::FieldNotEquals { ref field, ref value })
                if field == "source" && value == "internal"
        ));
    }

    #[test]
    fn classify_starts_with() {
        let result = classify(r#"host.startsWith("prod-")"#).unwrap();
        assert_eq!(result.tier(), FilterTier::Tier1);
        assert!(matches!(
            result,
            ClassifyResult::Tier1(Tier1Op::FieldStartsWith { ref field, ref prefix })
                if field == "host" && prefix == "prod-"
        ));
    }

    #[test]
    fn classify_ends_with() {
        let result = classify(r#"name.endsWith(".log")"#).unwrap();
        assert_eq!(result.tier(), FilterTier::Tier1);
        assert!(matches!(
            result,
            ClassifyResult::Tier1(Tier1Op::FieldEndsWith { ref field, ref suffix })
                if field == "name" && suffix == ".log"
        ));
    }

    #[test]
    fn classify_contains() {
        let result = classify(r#"path.contains("/api/")"#).unwrap();
        assert_eq!(result.tier(), FilterTier::Tier1);
        assert!(matches!(
            result,
            ClassifyResult::Tier1(Tier1Op::FieldContains { ref field, ref substring })
                if field == "path" && substring == "/api/"
        ));
    }

    #[test]
    fn classify_dotted_path() {
        let result = classify(r#"metadata.source == "aws""#).unwrap();
        assert_eq!(result.tier(), FilterTier::Tier1);
        assert!(matches!(
            result,
            ClassifyResult::Tier1(Tier1Op::FieldEquals { ref field, ref value })
                if field == "metadata.source" && value == "aws"
        ));
    }

    #[test]
    fn classify_compound_expression_is_tier2() {
        let result = classify(r#"severity > 3 && source != "internal""#).unwrap();
        assert_eq!(result.tier(), FilterTier::Tier2);
    }

    #[test]
    fn classify_regex_is_tier3() {
        let result = classify(r#"field.matches("^prod-.*")"#).unwrap();
        assert_eq!(result.tier(), FilterTier::Tier3);
    }

    #[test]
    fn classify_iteration_is_tier3() {
        let result = classify(r#"tags.exists(t, t == "pii")"#).unwrap();
        assert_eq!(result.tier(), FilterTier::Tier3);
    }

    #[test]
    fn classify_empty_expression_errors() {
        assert!(classify("").is_err());
        assert!(classify("   ").is_err());
    }

    #[test]
    fn classify_whitespace_tolerance() {
        let result = classify("  has( _table )  ").unwrap();
        assert_eq!(result.tier(), FilterTier::Tier1);
    }

    #[test]
    fn classify_tier2_extracts_fields() {
        let result = classify(r#"severity > 3 && source != "internal""#).unwrap();
        if let ClassifyResult::Tier2 { fields } = result {
            assert!(fields.contains(&"severity".to_string()));
            assert!(fields.contains(&"source".to_string()));
        } else {
            panic!("Expected Tier2");
        }
    }

    #[test]
    fn restricted_function_in_string_not_detected() {
        // "matches" inside a string literal should NOT trigger Tier 3
        let result = classify(r#"field == "matches""#).unwrap();
        assert_eq!(result.tier(), FilterTier::Tier1);
    }

    /// Regression: real `.matches()` after a same-name
    /// occurrence inside a string must still classify as Tier 3.
    /// Pre-fix `expr.find` returned the in-string position and
    /// the function never looked further.
    #[test]
    fn restricted_call_after_string_literal_decoy_classifies_tier3() {
        let result = classify(r#"message == "matches(" || source.matches("^prod")"#).unwrap();
        assert_eq!(result.tier(), FilterTier::Tier3);
    }
}
