dakera-engine 0.10.2

//! Metadata filter evaluation
//!
//! Evaluates filter expressions against JSON metadata values.

use common::{FilterCondition, FilterExpression, FilterValue};
use regex::Regex;
use serde_json::Value;

/// Evaluates a filter expression against a metadata JSON value
pub fn evaluate_filter(filter: &FilterExpression, metadata: Option<&Value>) -> bool {
    match filter {
        FilterExpression::And { conditions } => {
            conditions.iter().all(|c| evaluate_filter(c, metadata))
        }
        FilterExpression::Or { conditions } => {
            conditions.iter().any(|c| evaluate_filter(c, metadata))
        }
        FilterExpression::Field { field } => {
            // Evaluate each field condition (typically just one)
            for (field_name, condition) in field {
                if !evaluate_field_condition(field_name, condition, metadata) {
                    return false;
                }
            }
            true
        }
    }
}

/// Evaluates a single field condition against metadata
fn evaluate_field_condition(
    field_name: &str,
    condition: &FilterCondition,
    metadata: Option<&Value>,
) -> bool {
    // Handle nested field access with dot notation (e.g., "user.name")
    let field_value = get_nested_field(metadata, field_name);

    match condition {
        FilterCondition::Exists(should_exist) => field_value.is_some() == *should_exist,
        FilterCondition::Eq(filter_val) => {
            field_value.is_some_and(|v| compare_values_eq(v, filter_val))
        }
        FilterCondition::Ne(filter_val) => {
            field_value.is_none_or(|v| !compare_values_eq(v, filter_val))
        }
        FilterCondition::Gt(filter_val) => field_value.is_some_and(|v| {
            compare_values_ord(v, filter_val) == Some(std::cmp::Ordering::Greater)
        }),
        FilterCondition::Gte(filter_val) => field_value.is_some_and(|v| {
            matches!(
                compare_values_ord(v, filter_val),
                Some(std::cmp::Ordering::Greater | std::cmp::Ordering::Equal)
            )
        }),
        FilterCondition::Lt(filter_val) => field_value
            .is_some_and(|v| compare_values_ord(v, filter_val) == Some(std::cmp::Ordering::Less)),
        FilterCondition::Lte(filter_val) => field_value.is_some_and(|v| {
            matches!(
                compare_values_ord(v, filter_val),
                Some(std::cmp::Ordering::Less | std::cmp::Ordering::Equal)
            )
        }),
        FilterCondition::In(values) => {
            field_value.is_some_and(|v| values.iter().any(|fv| compare_values_eq(v, fv)))
        }
        FilterCondition::NotIn(values) => {
            field_value.is_none_or(|v| !values.iter().any(|fv| compare_values_eq(v, fv)))
        }
        // Enhanced string operators (Turbopuffer-inspired)
        FilterCondition::Contains(substring) => field_value.is_some_and(|v| {
            if let Value::String(s) = v {
                s.contains(substring.as_str())
            } else {
                false
            }
        }),
        FilterCondition::IContains(substring) => field_value.is_some_and(|v| {
            if let Value::String(s) = v {
                s.to_lowercase().contains(&substring.to_lowercase())
            } else {
                false
            }
        }),
        FilterCondition::StartsWith(prefix) => field_value.is_some_and(|v| {
            if let Value::String(s) = v {
                s.starts_with(prefix.as_str())
            } else {
                false
            }
        }),
        FilterCondition::EndsWith(suffix) => field_value.is_some_and(|v| {
            if let Value::String(s) = v {
                s.ends_with(suffix.as_str())
            } else {
                false
            }
        }),
        FilterCondition::Glob(pattern) => field_value.is_some_and(|v| {
            if let Value::String(s) = v {
                glob_match(pattern, s)
            } else {
                false
            }
        }),
        FilterCondition::Regex(pattern) => field_value.is_some_and(|v| {
            if let Value::String(s) = v {
                Regex::new(pattern)
                    .map(|re| re.is_match(s))
                    .unwrap_or(false)
            } else {
                false
            }
        }),
    }
}

/// Simple glob pattern matching supporting * (any chars) and ? (single char)
fn glob_match(pattern: &str, text: &str) -> bool {
    let pattern_chars: Vec<char> = pattern.chars().collect();
    let text_chars: Vec<char> = text.chars().collect();

    glob_match_recursive(&pattern_chars, &text_chars, 0, 0)
}

fn glob_match_recursive(pattern: &[char], text: &[char], pi: usize, ti: usize) -> bool {
    // Both exhausted - match
    if pi >= pattern.len() && ti >= text.len() {
        return true;
    }

    // Pattern exhausted but text remains - no match (unless pattern ends with *)
    if pi >= pattern.len() {
        return false;
    }

    let pc = pattern[pi];

    match pc {
        '*' => {
            // Try matching zero or more characters
            // First try matching zero characters (move pattern forward)
            if glob_match_recursive(pattern, text, pi + 1, ti) {
                return true;
            }
            // Then try matching one character and continue with *
            if ti < text.len() && glob_match_recursive(pattern, text, pi, ti + 1) {
                return true;
            }
            false
        }
        '?' => {
            // Match exactly one character
            if ti < text.len() {
                glob_match_recursive(pattern, text, pi + 1, ti + 1)
            } else {
                false
            }
        }
        _ => {
            // Regular character - must match exactly
            if ti < text.len() && pc == text[ti] {
                glob_match_recursive(pattern, text, pi + 1, ti + 1)
            } else {
                false
            }
        }
    }
}

/// Get a potentially nested field from metadata using dot notation
fn get_nested_field<'a>(metadata: Option<&'a Value>, field_path: &str) -> Option<&'a Value> {
    let metadata = metadata?;

    let parts: Vec<&str> = field_path.split('.').collect();
    let mut current = metadata;

    for part in parts {
        match current {
            Value::Object(map) => {
                current = map.get(part)?;
            }
            _ => return None,
        }
    }

    Some(current)
}

/// Compare a JSON value with a filter value for equality
fn compare_values_eq(json_val: &Value, filter_val: &FilterValue) -> bool {
    match (json_val, filter_val) {
        (Value::String(s), FilterValue::String(fs)) => s == fs,
        (Value::Number(n), FilterValue::Number(fn_)) => {
            n.as_f64().is_some_and(|nf| (nf - fn_).abs() < f64::EPSILON)
        }
        (Value::Number(n), FilterValue::Integer(fi)) => n.as_i64() == Some(*fi),
        (Value::Bool(b), FilterValue::Boolean(fb)) => b == fb,
        // Check if json value is in string array
        (Value::String(s), FilterValue::StringArray(arr)) => arr.contains(s),
        // Check if json value is in number array
        (Value::Number(n), FilterValue::NumberArray(arr)) => n
            .as_f64()
            .is_some_and(|nf| arr.iter().any(|&af| (nf - af).abs() < f64::EPSILON)),
        _ => false,
    }
}

/// Compare a JSON value with a filter value for ordering
fn compare_values_ord(json_val: &Value, filter_val: &FilterValue) -> Option<std::cmp::Ordering> {
    match (json_val, filter_val) {
        (Value::String(s), FilterValue::String(fs)) => Some(s.cmp(fs)),
        (Value::Number(n), FilterValue::Number(fn_)) => {
            n.as_f64().and_then(|nf| nf.partial_cmp(fn_))
        }
        (Value::Number(n), FilterValue::Integer(fi)) => {
            n.as_f64().and_then(|nf| nf.partial_cmp(&(*fi as f64)))
        }
        _ => None,
    }
}

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

    fn make_field_filter(field: &str, condition: FilterCondition) -> FilterExpression {
        let mut map = HashMap::new();
        map.insert(field.to_string(), condition);
        FilterExpression::Field { field: map }
    }

    #[test]
    fn test_eq_string() {
        let metadata = json!({"category": "electronics"});
        let filter = make_field_filter(
            "category",
            FilterCondition::Eq(FilterValue::String("electronics".to_string())),
        );
        assert!(evaluate_filter(&filter, Some(&metadata)));

        let filter = make_field_filter(
            "category",
            FilterCondition::Eq(FilterValue::String("books".to_string())),
        );
        assert!(!evaluate_filter(&filter, Some(&metadata)));
    }

    #[test]
    fn test_eq_number() {
        let metadata = json!({"price": 99.99});
        let filter = make_field_filter("price", FilterCondition::Eq(FilterValue::Number(99.99)));
        assert!(evaluate_filter(&filter, Some(&metadata)));

        let filter = make_field_filter("price", FilterCondition::Eq(FilterValue::Number(100.0)));
        assert!(!evaluate_filter(&filter, Some(&metadata)));
    }

    #[test]
    fn test_ne() {
        let metadata = json!({"status": "active"});
        let filter = make_field_filter(
            "status",
            FilterCondition::Ne(FilterValue::String("inactive".to_string())),
        );
        assert!(evaluate_filter(&filter, Some(&metadata)));

        let filter = make_field_filter(
            "status",
            FilterCondition::Ne(FilterValue::String("active".to_string())),
        );
        assert!(!evaluate_filter(&filter, Some(&metadata)));
    }

    #[test]
    fn test_gt_lt() {
        let metadata = json!({"price": 50.0});

        // Greater than
        let filter = make_field_filter("price", FilterCondition::Gt(FilterValue::Number(40.0)));
        assert!(evaluate_filter(&filter, Some(&metadata)));

        let filter = make_field_filter("price", FilterCondition::Gt(FilterValue::Number(50.0)));
        assert!(!evaluate_filter(&filter, Some(&metadata)));

        // Less than
        let filter = make_field_filter("price", FilterCondition::Lt(FilterValue::Number(60.0)));
        assert!(evaluate_filter(&filter, Some(&metadata)));

        let filter = make_field_filter("price", FilterCondition::Lt(FilterValue::Number(50.0)));
        assert!(!evaluate_filter(&filter, Some(&metadata)));
    }

    #[test]
    fn test_gte_lte() {
        let metadata = json!({"count": 10});

        // Greater than or equal
        let filter = make_field_filter("count", FilterCondition::Gte(FilterValue::Integer(10)));
        assert!(evaluate_filter(&filter, Some(&metadata)));

        let filter = make_field_filter("count", FilterCondition::Gte(FilterValue::Integer(11)));
        assert!(!evaluate_filter(&filter, Some(&metadata)));

        // Less than or equal
        let filter = make_field_filter("count", FilterCondition::Lte(FilterValue::Integer(10)));
        assert!(evaluate_filter(&filter, Some(&metadata)));

        let filter = make_field_filter("count", FilterCondition::Lte(FilterValue::Integer(9)));
        assert!(!evaluate_filter(&filter, Some(&metadata)));
    }

    #[test]
    fn test_in() {
        let metadata = json!({"category": "electronics"});
        let filter = make_field_filter(
            "category",
            FilterCondition::In(vec![
                FilterValue::String("electronics".to_string()),
                FilterValue::String("computers".to_string()),
            ]),
        );
        assert!(evaluate_filter(&filter, Some(&metadata)));

        let filter = make_field_filter(
            "category",
            FilterCondition::In(vec![
                FilterValue::String("books".to_string()),
                FilterValue::String("clothing".to_string()),
            ]),
        );
        assert!(!evaluate_filter(&filter, Some(&metadata)));
    }

    #[test]
    fn test_not_in() {
        let metadata = json!({"status": "active"});
        let filter = make_field_filter(
            "status",
            FilterCondition::NotIn(vec![
                FilterValue::String("deleted".to_string()),
                FilterValue::String("archived".to_string()),
            ]),
        );
        assert!(evaluate_filter(&filter, Some(&metadata)));

        let filter = make_field_filter(
            "status",
            FilterCondition::NotIn(vec![
                FilterValue::String("active".to_string()),
                FilterValue::String("pending".to_string()),
            ]),
        );
        assert!(!evaluate_filter(&filter, Some(&metadata)));
    }

    #[test]
    fn test_exists() {
        let metadata = json!({"name": "test", "value": null});

        let filter = make_field_filter("name", FilterCondition::Exists(true));
        assert!(evaluate_filter(&filter, Some(&metadata)));

        let filter = make_field_filter("missing", FilterCondition::Exists(true));
        assert!(!evaluate_filter(&filter, Some(&metadata)));

        let filter = make_field_filter("missing", FilterCondition::Exists(false));
        assert!(evaluate_filter(&filter, Some(&metadata)));
    }

    #[test]
    fn test_nested_field() {
        let metadata = json!({
            "user": {
                "name": "Alice",
                "profile": {
                    "age": 30
                }
            }
        });

        let filter = make_field_filter(
            "user.name",
            FilterCondition::Eq(FilterValue::String("Alice".to_string())),
        );
        assert!(evaluate_filter(&filter, Some(&metadata)));

        let filter = make_field_filter(
            "user.profile.age",
            FilterCondition::Gte(FilterValue::Integer(18)),
        );
        assert!(evaluate_filter(&filter, Some(&metadata)));
    }

    #[test]
    fn test_and_combinator() {
        let metadata = json!({"category": "electronics", "price": 50.0});

        let filter = FilterExpression::And {
            conditions: vec![
                make_field_filter(
                    "category",
                    FilterCondition::Eq(FilterValue::String("electronics".to_string())),
                ),
                make_field_filter("price", FilterCondition::Lt(FilterValue::Number(100.0))),
            ],
        };
        assert!(evaluate_filter(&filter, Some(&metadata)));

        // One condition fails
        let filter = FilterExpression::And {
            conditions: vec![
                make_field_filter(
                    "category",
                    FilterCondition::Eq(FilterValue::String("electronics".to_string())),
                ),
                make_field_filter("price", FilterCondition::Gt(FilterValue::Number(100.0))),
            ],
        };
        assert!(!evaluate_filter(&filter, Some(&metadata)));
    }

    #[test]
    fn test_or_combinator() {
        let metadata = json!({"category": "electronics", "price": 150.0});

        // One condition passes
        let filter = FilterExpression::Or {
            conditions: vec![
                make_field_filter(
                    "category",
                    FilterCondition::Eq(FilterValue::String("books".to_string())),
                ),
                make_field_filter("price", FilterCondition::Gt(FilterValue::Number(100.0))),
            ],
        };
        assert!(evaluate_filter(&filter, Some(&metadata)));

        // Both conditions fail
        let filter = FilterExpression::Or {
            conditions: vec![
                make_field_filter(
                    "category",
                    FilterCondition::Eq(FilterValue::String("books".to_string())),
                ),
                make_field_filter("price", FilterCondition::Lt(FilterValue::Number(100.0))),
            ],
        };
        assert!(!evaluate_filter(&filter, Some(&metadata)));
    }

    #[test]
    fn test_no_metadata() {
        let filter = make_field_filter(
            "category",
            FilterCondition::Eq(FilterValue::String("test".to_string())),
        );
        assert!(!evaluate_filter(&filter, None));

        // $exists false should pass when no metadata
        let filter = make_field_filter("anything", FilterCondition::Exists(false));
        assert!(evaluate_filter(&filter, None));
    }

    #[test]
    fn test_boolean() {
        let metadata = json!({"active": true, "verified": false});

        let filter = make_field_filter("active", FilterCondition::Eq(FilterValue::Boolean(true)));
        assert!(evaluate_filter(&filter, Some(&metadata)));

        let filter =
            make_field_filter("verified", FilterCondition::Eq(FilterValue::Boolean(false)));
        assert!(evaluate_filter(&filter, Some(&metadata)));
    }

    #[test]
    fn test_contains() {
        let metadata = json!({"description": "Hello World Example"});

        // Case-sensitive contains
        let filter = make_field_filter(
            "description",
            FilterCondition::Contains("World".to_string()),
        );
        assert!(evaluate_filter(&filter, Some(&metadata)));

        let filter = make_field_filter(
            "description",
            FilterCondition::Contains("world".to_string()),
        );
        assert!(!evaluate_filter(&filter, Some(&metadata))); // Case mismatch

        let filter = make_field_filter(
            "description",
            FilterCondition::Contains("NotFound".to_string()),
        );
        assert!(!evaluate_filter(&filter, Some(&metadata)));
    }

    #[test]
    fn test_icontains() {
        let metadata = json!({"description": "Hello World Example"});

        // Case-insensitive contains
        let filter = make_field_filter(
            "description",
            FilterCondition::IContains("world".to_string()),
        );
        assert!(evaluate_filter(&filter, Some(&metadata)));

        let filter = make_field_filter(
            "description",
            FilterCondition::IContains("WORLD".to_string()),
        );
        assert!(evaluate_filter(&filter, Some(&metadata)));

        let filter = make_field_filter(
            "description",
            FilterCondition::IContains("notfound".to_string()),
        );
        assert!(!evaluate_filter(&filter, Some(&metadata)));
    }

    #[test]
    fn test_starts_with() {
        let metadata = json!({"filename": "document.pdf"});

        let filter = make_field_filter(
            "filename",
            FilterCondition::StartsWith("document".to_string()),
        );
        assert!(evaluate_filter(&filter, Some(&metadata)));

        let filter = make_field_filter("filename", FilterCondition::StartsWith("doc".to_string()));
        assert!(evaluate_filter(&filter, Some(&metadata)));

        let filter = make_field_filter("filename", FilterCondition::StartsWith("pdf".to_string()));
        assert!(!evaluate_filter(&filter, Some(&metadata)));
    }

    #[test]
    fn test_ends_with() {
        let metadata = json!({"filename": "document.pdf"});

        let filter = make_field_filter("filename", FilterCondition::EndsWith(".pdf".to_string()));
        assert!(evaluate_filter(&filter, Some(&metadata)));

        let filter = make_field_filter("filename", FilterCondition::EndsWith("pdf".to_string()));
        assert!(evaluate_filter(&filter, Some(&metadata)));

        let filter = make_field_filter("filename", FilterCondition::EndsWith(".txt".to_string()));
        assert!(!evaluate_filter(&filter, Some(&metadata)));
    }

    #[test]
    fn test_glob() {
        let metadata = json!({"path": "src/main/java/App.java"});

        // Test * wildcard (any characters)
        let filter = make_field_filter("path", FilterCondition::Glob("*.java".to_string()));
        assert!(evaluate_filter(&filter, Some(&metadata)));

        let filter = make_field_filter("path", FilterCondition::Glob("src/*".to_string()));
        assert!(evaluate_filter(&filter, Some(&metadata)));

        let filter = make_field_filter("path", FilterCondition::Glob("*App*".to_string()));
        assert!(evaluate_filter(&filter, Some(&metadata)));

        // Test ? wildcard (single character)
        let metadata2 = json!({"code": "A1B2"});
        let filter = make_field_filter("code", FilterCondition::Glob("A?B?".to_string()));
        assert!(evaluate_filter(&filter, Some(&metadata2)));

        let filter = make_field_filter("code", FilterCondition::Glob("A?B".to_string()));
        assert!(!evaluate_filter(&filter, Some(&metadata2))); // Too short

        // No match
        let filter = make_field_filter("path", FilterCondition::Glob("*.rs".to_string()));
        assert!(!evaluate_filter(&filter, Some(&metadata)));
    }

    #[test]
    fn test_regex() {
        let metadata = json!({"email": "user@example.com"});

        // Valid regex match
        let filter = make_field_filter(
            "email",
            FilterCondition::Regex(r"^[\w.]+@[\w.]+\.\w+$".to_string()),
        );
        assert!(evaluate_filter(&filter, Some(&metadata)));

        let filter = make_field_filter(
            "email",
            FilterCondition::Regex(r"@example\.com$".to_string()),
        );
        assert!(evaluate_filter(&filter, Some(&metadata)));

        // No match
        let filter = make_field_filter("email", FilterCondition::Regex(r"^admin@".to_string()));
        assert!(!evaluate_filter(&filter, Some(&metadata)));

        // Invalid regex should return false, not panic
        let filter = make_field_filter("email", FilterCondition::Regex(r"[invalid".to_string()));
        assert!(!evaluate_filter(&filter, Some(&metadata)));
    }

    #[test]
    fn test_string_operators_on_non_string() {
        let metadata = json!({"count": 42});

        // String operators on non-string fields should return false
        let filter = make_field_filter("count", FilterCondition::Contains("4".to_string()));
        assert!(!evaluate_filter(&filter, Some(&metadata)));

        let filter = make_field_filter("count", FilterCondition::StartsWith("4".to_string()));
        assert!(!evaluate_filter(&filter, Some(&metadata)));

        let filter = make_field_filter("count", FilterCondition::Glob("*".to_string()));
        assert!(!evaluate_filter(&filter, Some(&metadata)));
    }
}