pipeflow 0.0.4

//! Filter step for conditional filtering
//!
//! Allows messages to pass through based on configurable condition expressions.
//! Supports single condition or multiple conditions with AND/OR logic.

use serde::{Deserialize, Serialize};

use crate::common::message::Message;
use crate::error::{Error, Result};
use crate::transform::json_path;
use crate::transform::step::Step;

/// Comparison operators for filter conditions
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum FilterOperator {
    /// Equal (==)
    #[default]
    Eq,
    /// Not equal (!=)
    Ne,
    /// Greater than (>)
    Gt,
    /// Greater than or equal (>=)
    Ge,
    /// Less than (<)
    Lt,
    /// Less than or equal (<=)
    Le,
    /// String contains substring
    Contains,
    /// String matches regex pattern
    Matches,
    /// Absolute value greater than (|a| > |b|)
    AbsGt,
    /// Absolute value greater than or equal (|a| >= |b|)
    AbsGe,
    /// Absolute value less than (|a| < |b|)
    AbsLt,
    /// Absolute value less than or equal (|a| <= |b|)
    AbsLe,
}

/// Condition mode for combining multiple conditions
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum FilterMode {
    /// All conditions must match (AND logic)
    #[default]
    And,
    /// At least one condition must match (OR logic)
    Or,
}

/// A single filter condition
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FilterCondition {
    /// JSONPath-like field accessor (e.g., `$.status`, `$.data.value`)
    pub field: String,
    /// Comparison operator
    #[serde(default)]
    pub operator: FilterOperator,
    /// Expected value for comparison
    pub value: serde_json::Value,
}

/// Filter step configuration
///
/// Supports either a single condition or multiple conditions with AND/OR mode.
///
/// # Deserialization
///
/// Uses `#[serde(untagged)]` to allow both formats without a discriminator field.
/// serde tries variants in declaration order, so `Multi` is attempted first.
///
/// **Single condition** (simpler syntax):
/// ```yaml
/// field: "$.status"
/// operator: eq
/// value: "success"
/// ```
///
/// **Multiple conditions** (with mode):
/// ```yaml
/// mode: and  # or "or"
/// conditions:
///   - field: "$.status"
///     value: "active"
///   - field: "$.score"
///     operator: ge
///     value: 50
/// ```
///
/// # Edge Cases
///
/// Since this uses untagged deserialization, avoid using `mode` or `conditions`
/// as field names in your single-condition payload, as this may cause ambiguity.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(untagged)]
pub enum FilterStepConfig {
    /// Multiple conditions with mode (tried first during deserialization)
    Multi {
        /// Condition combining mode (and/or)
        #[serde(default)]
        mode: FilterMode,
        /// List of conditions to evaluate
        conditions: Vec<FilterCondition>,
    },
    /// Single condition (for simpler configs)
    Single(FilterCondition),
}

/// Compiled condition with pre-compiled path and optional regex
struct CompiledCondition {
    /// Pre-compiled JSONPath for efficient field access
    path: json_path::CompiledPath,
    /// Comparison operator
    operator: FilterOperator,
    /// Expected value to compare against
    expected: serde_json::Value,
    /// Pre-compiled regex (only for Matches operator)
    regex: Option<regex::Regex>,
}

impl CompiledCondition {
    fn new(condition: FilterCondition) -> Result<Self> {
        let path = json_path::CompiledPath::compile(&condition.field)?;

        let regex = if condition.operator == FilterOperator::Matches {
            let pattern = condition
                .value
                .as_str()
                .ok_or_else(|| Error::config("Matches operator requires a string pattern"))?;
            Some(
                regex::Regex::new(pattern)
                    .map_err(|e| Error::config(format!("Invalid regex pattern: {}", e)))?,
            )
        } else {
            None
        };

        Ok(Self {
            path,
            operator: condition.operator,
            expected: condition.value,
            regex,
        })
    }

    /// Evaluate this condition against a payload (no string parsing at runtime)
    fn evaluate(&self, payload: &serde_json::Value) -> bool {
        let Some(field_value) = self.path.extract(payload) else {
            return false; // Field not found = condition fails
        };

        use FilterOperator::*;
        use std::cmp::Ordering::{Equal, Greater, Less};

        match self.operator {
            Eq => field_value == &self.expected,
            Ne => field_value != &self.expected,
            Gt => compare_values(field_value, &self.expected) == Some(Greater),
            Ge => matches!(
                compare_values(field_value, &self.expected),
                Some(Greater | Equal)
            ),
            Lt => compare_values(field_value, &self.expected) == Some(Less),
            Le => matches!(
                compare_values(field_value, &self.expected),
                Some(Less | Equal)
            ),
            Contains => matches!(
                (field_value.as_str(), self.expected.as_str()),
                (Some(haystack), Some(needle)) if haystack.contains(needle)
            ),
            Matches => matches!(
                (field_value.as_str(), &self.regex),
                (Some(text), Some(re)) if re.is_match(text)
            ),
            AbsGt => compare_abs_values(field_value, &self.expected) == Some(Greater),
            AbsGe => matches!(
                compare_abs_values(field_value, &self.expected),
                Some(Greater | Equal)
            ),
            AbsLt => compare_abs_values(field_value, &self.expected) == Some(Less),
            AbsLe => matches!(
                compare_abs_values(field_value, &self.expected),
                Some(Less | Equal)
            ),
        }
    }
}

/// Compare absolute values of two numbers
fn compare_abs_values(a: &serde_json::Value, b: &serde_json::Value) -> Option<std::cmp::Ordering> {
    match (a, b) {
        (serde_json::Value::Number(a), serde_json::Value::Number(b)) => {
            let a_f64 = a.as_f64()?.abs();
            let b_f64 = b.as_f64()?.abs();
            a_f64.partial_cmp(&b_f64)
        }
        _ => None,
    }
}

/// Compare two JSON values, returns ordering for numeric comparisons
fn compare_values(a: &serde_json::Value, b: &serde_json::Value) -> Option<std::cmp::Ordering> {
    match (a, b) {
        (serde_json::Value::Number(a), serde_json::Value::Number(b)) => {
            let a_f64 = a.as_f64()?;
            let b_f64 = b.as_f64()?;
            a_f64.partial_cmp(&b_f64)
        }
        (serde_json::Value::String(a), serde_json::Value::String(b)) => Some(a.cmp(b)),
        _ => None,
    }
}

/// Filter step that conditionally passes or filters out messages
pub struct FilterStep {
    mode: FilterMode,
    conditions: Vec<CompiledCondition>,
}

impl FilterStep {
    /// Create a new filter step from configuration
    pub fn new(config: FilterStepConfig) -> Result<Self> {
        let (mode, raw_conditions) = match config {
            FilterStepConfig::Multi { mode, conditions } => (mode, conditions),
            FilterStepConfig::Single(cond) => (FilterMode::And, vec![cond]),
        };

        if raw_conditions.is_empty() {
            return Err(Error::config("Filter step requires at least one condition"));
        }

        let mut conditions = Vec::with_capacity(raw_conditions.len());
        for cond in raw_conditions {
            conditions.push(CompiledCondition::new(cond)?);
        }

        Ok(Self { mode, conditions })
    }

    /// Evaluate all conditions against a message payload
    fn evaluate(&self, payload: &serde_json::Value) -> bool {
        match self.mode {
            FilterMode::And => self.conditions.iter().all(|c| c.evaluate(payload)),
            FilterMode::Or => self.conditions.iter().any(|c| c.evaluate(payload)),
        }
    }
}

impl Step for FilterStep {
    fn step_type(&self) -> &'static str {
        "filter"
    }

    fn process(&self, msg: Message) -> Result<Option<Message>> {
        let passes = self.evaluate(&msg.payload);

        tracing::debug!(
            conditions_count = self.conditions.len(),
            mode = ?self.mode,
            result = if passes { "passed" } else { "rejected" },
            "Filter step evaluation"
        );

        Ok(if passes { Some(msg) } else { None })
    }
}

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

    // ========== Config Deserialize Tests ==========

    #[test]
    fn test_single_condition_config() {
        let yaml = r#"
field: $.status
operator: eq
value: "success"
"#;
        let config: FilterStepConfig = serde_yaml::from_str(yaml).unwrap();
        match config {
            FilterStepConfig::Single(c) => {
                assert_eq!(c.field, "$.status");
                assert_eq!(c.operator, FilterOperator::Eq);
            }
            _ => panic!("Expected Single variant"),
        }
    }

    #[test]
    fn test_multi_condition_config() {
        let yaml = r#"
mode: and
conditions:
  - field: $.status
    operator: eq
    value: "active"
  - field: $.score
    operator: ge
    value: 50
"#;
        let config: FilterStepConfig = serde_yaml::from_str(yaml).unwrap();
        match config {
            FilterStepConfig::Multi { mode, conditions } => {
                assert_eq!(mode, FilterMode::And);
                assert_eq!(conditions.len(), 2);
            }
            _ => panic!("Expected Multi variant"),
        }
    }

    #[test]
    fn test_multi_condition_or_mode() {
        let yaml = r#"
mode: or
conditions:
  - field: $.type
    value: "premium"
  - field: $.score
    operator: gt
    value: 100
"#;
        let config: FilterStepConfig = serde_yaml::from_str(yaml).unwrap();
        match config {
            FilterStepConfig::Multi { mode, .. } => {
                assert_eq!(mode, FilterMode::Or);
            }
            _ => panic!("Expected Multi variant"),
        }
    }

    // ========== FilterStep Tests ==========

    fn make_msg(payload: serde_json::Value) -> Message {
        Message::new("test", payload)
    }

    #[test]
    fn test_filter_single_eq_pass() {
        let config = FilterStepConfig::Single(FilterCondition {
            field: "$.status".into(),
            operator: FilterOperator::Eq,
            value: json!("success"),
        });
        let step = FilterStep::new(config).unwrap();
        let msg = make_msg(json!({"status": "success"}));

        let result = step.process(msg).unwrap();
        assert!(result.is_some());
    }

    #[test]
    fn test_filter_single_eq_reject() {
        let config = FilterStepConfig::Single(FilterCondition {
            field: "$.status".into(),
            operator: FilterOperator::Eq,
            value: json!("success"),
        });
        let step = FilterStep::new(config).unwrap();
        let msg = make_msg(json!({"status": "failure"}));

        let result = step.process(msg).unwrap();
        assert!(result.is_none());
    }

    #[test]
    fn test_filter_multi_and_all_pass() {
        let config = FilterStepConfig::Multi {
            mode: FilterMode::And,
            conditions: vec![
                FilterCondition {
                    field: "$.status".into(),
                    operator: FilterOperator::Eq,
                    value: json!("active"),
                },
                FilterCondition {
                    field: "$.score".into(),
                    operator: FilterOperator::Ge,
                    value: json!(50),
                },
            ],
        };
        let step = FilterStep::new(config).unwrap();

        // Both conditions pass
        let msg = make_msg(json!({"status": "active", "score": 75}));
        assert!(step.process(msg).unwrap().is_some());
    }

    #[test]
    fn test_filter_multi_and_one_fails() {
        let config = FilterStepConfig::Multi {
            mode: FilterMode::And,
            conditions: vec![
                FilterCondition {
                    field: "$.status".into(),
                    operator: FilterOperator::Eq,
                    value: json!("active"),
                },
                FilterCondition {
                    field: "$.score".into(),
                    operator: FilterOperator::Ge,
                    value: json!(50),
                },
            ],
        };
        let step = FilterStep::new(config).unwrap();

        // Second condition fails (score < 50)
        let msg = make_msg(json!({"status": "active", "score": 30}));
        assert!(step.process(msg).unwrap().is_none());
    }

    #[test]
    fn test_filter_multi_or_one_passes() {
        let config = FilterStepConfig::Multi {
            mode: FilterMode::Or,
            conditions: vec![
                FilterCondition {
                    field: "$.type".into(),
                    operator: FilterOperator::Eq,
                    value: json!("premium"),
                },
                FilterCondition {
                    field: "$.score".into(),
                    operator: FilterOperator::Gt,
                    value: json!(100),
                },
            ],
        };
        let step = FilterStep::new(config).unwrap();

        // First condition passes, second fails
        let msg = make_msg(json!({"type": "premium", "score": 50}));
        assert!(step.process(msg).unwrap().is_some());

        // First fails, second passes
        let msg = make_msg(json!({"type": "basic", "score": 150}));
        assert!(step.process(msg).unwrap().is_some());
    }

    #[test]
    fn test_filter_multi_or_all_fail() {
        let config = FilterStepConfig::Multi {
            mode: FilterMode::Or,
            conditions: vec![
                FilterCondition {
                    field: "$.type".into(),
                    operator: FilterOperator::Eq,
                    value: json!("premium"),
                },
                FilterCondition {
                    field: "$.score".into(),
                    operator: FilterOperator::Gt,
                    value: json!(100),
                },
            ],
        };
        let step = FilterStep::new(config).unwrap();

        // Both conditions fail
        let msg = make_msg(json!({"type": "basic", "score": 50}));
        assert!(step.process(msg).unwrap().is_none());
    }

    #[test]
    fn test_filter_nested_field() {
        let config = FilterStepConfig::Single(FilterCondition {
            field: "$.data.user.role".into(),
            operator: FilterOperator::Eq,
            value: json!("admin"),
        });
        let step = FilterStep::new(config).unwrap();

        let msg = make_msg(json!({"data": {"user": {"role": "admin"}}}));
        assert!(step.process(msg).unwrap().is_some());
    }

    #[test]
    fn test_filter_array_index() {
        let config = FilterStepConfig::Single(FilterCondition {
            field: "$.items[0].id".into(),
            operator: FilterOperator::Eq,
            value: json!(1),
        });
        let step = FilterStep::new(config).unwrap();

        let msg = make_msg(json!({"items": [{"id": 1}, {"id": 2}]}));
        assert!(step.process(msg).unwrap().is_some());
    }

    #[test]
    fn test_filter_contains_operator() {
        let config = FilterStepConfig::Single(FilterCondition {
            field: "$.message".into(),
            operator: FilterOperator::Contains,
            value: json!("error"),
        });
        let step = FilterStep::new(config).unwrap();

        let msg = make_msg(json!({"message": "An error occurred"}));
        assert!(step.process(msg).unwrap().is_some());

        let msg = make_msg(json!({"message": "All good"}));
        assert!(step.process(msg).unwrap().is_none());
    }

    #[test]
    fn test_filter_matches_operator() {
        let config = FilterStepConfig::Single(FilterCondition {
            field: "$.email".into(),
            operator: FilterOperator::Matches,
            value: json!(r"^[a-z]+@example\.com$"),
        });
        let step = FilterStep::new(config).unwrap();

        let msg = make_msg(json!({"email": "alice@example.com"}));
        assert!(step.process(msg).unwrap().is_some());

        let msg = make_msg(json!({"email": "alice@other.com"}));
        assert!(step.process(msg).unwrap().is_none());
    }

    #[test]
    fn test_filter_abs_gt_operator() {
        let config = FilterStepConfig::Single(FilterCondition {
            field: "$.change".into(),
            operator: FilterOperator::AbsGt,
            value: json!(2.0),
        });
        let step = FilterStep::new(config).unwrap();

        // 3.0 > 2.0 -> pass
        let msg = make_msg(json!({"change": 3.0}));
        assert!(step.process(msg).unwrap().is_some());

        // -5.0 (abs 5.0) > 2.0 -> pass
        let msg = make_msg(json!({"change": -5.0}));
        assert!(step.process(msg).unwrap().is_some());

        // 1.0 < 2.0 -> fail
        let msg = make_msg(json!({"change": 1.0}));
        assert!(step.process(msg).unwrap().is_none());

        // -1.0 (abs 1.0) < 2.0 -> fail
        let msg = make_msg(json!({"change": -1.0}));
        assert!(step.process(msg).unwrap().is_none());
    }

    #[test]
    fn test_filter_invalid_regex_error() {
        let config = FilterStepConfig::Single(FilterCondition {
            field: "$.x".into(),
            operator: FilterOperator::Matches,
            value: json!("[invalid regex"),
        });
        assert!(FilterStep::new(config).is_err());
    }

    #[test]
    fn test_filter_empty_conditions_error() {
        let config = FilterStepConfig::Multi {
            mode: FilterMode::And,
            conditions: vec![],
        };
        assert!(FilterStep::new(config).is_err());
    }

    #[test]
    fn test_filter_missing_field_fails() {
        let config = FilterStepConfig::Single(FilterCondition {
            field: "$.nonexistent".into(),
            operator: FilterOperator::Eq,
            value: json!("value"),
        });
        let step = FilterStep::new(config).unwrap();

        let msg = make_msg(json!({"other": "field"}));
        assert!(step.process(msg).unwrap().is_none());
    }
}