hamelin_lib 0.21.3

//! Tests for MATCH command
//!
//! MATCH performs ordered pattern matching over event sequences.
//! These tests validate schema propagation and type checking for MATCH commands.

use crate::err::TranslationErrors;
use crate::tree::ast::query::Query;
use crate::tree::ast::ParseWithErrors;
use crate::tree::tests::shared::{catalog_with_tables, type_check_with_catalog};
use crate::type_check;
use crate::types::array::Array;
use crate::types::struct_type::Struct;
use crate::types::{Type, STRING, TIMESTAMP};
use rstest::rstest;

// Helper to create the simba:sysmon_events schema based on the actual catalog
fn sysmon_events_schema() -> Vec<(&'static str, Type)> {
    vec![
        ("timestamp", TIMESTAMP),
        (
            "event",
            Struct::default()
                .with_str("id", STRING)
                .with_str("code", STRING)
                .with_str("kind", STRING)
                .with_str("provider", STRING)
                .with_str("action", STRING)
                .with_str("created", TIMESTAMP)
                .with_str("ingested", TIMESTAMP)
                .with_str("original", STRING)
                .into(),
        ),
        (
            "host",
            Struct::default()
                .with_str(
                    "os",
                    Struct::default()
                        .with_str("kernel", STRING)
                        .with_str("build", STRING)
                        .with_str("type", STRING)
                        .with_str("platform", STRING)
                        .with_str("version", STRING)
                        .with_str("family", STRING)
                        .with_str("name", STRING)
                        .into(),
                )
                .with_str("id", STRING)
                .with_str("ip", Array::new(STRING).into())
                .with_str("mac", Array::new(STRING).into())
                .with_str("name", STRING)
                .with_str("hostname", STRING)
                .with_str("architecture", STRING)
                .into(),
        ),
        ("message", STRING),
    ]
}

#[rstest]
#[case::basic_pattern(
    r#"
MATCH x=simba:sysmon_events y=simba:sysmon_events+
BY host.name
"#
)]
#[case::with_cte(
    r#"
DEF events = FROM simba:sysmon_events
| WHERE event.code IN ["1", "3"];

MATCH x=events y=events+
BY host.name
"#
)]
#[case::with_within_clause_single_pattern(
    r#"
DEF events = FROM simba:sysmon_events;

MATCH events WITHIN 5m
"#
)]
#[case::with_within_clause_multiple_patterns(
    r#"
DEF events = FROM simba:sysmon_events;

MATCH x=events y=events+ WITHIN 5m
"#
)]
#[case::with_aggregation(
    r#"
MATCH x=simba:sysmon_events y=simba:sysmon_events+
BY host.name
| AGG count = count() BY host.name
"#
)]
#[case::aggregating_match_results(
    r#"
DEF events = FROM simba:sysmon_events
| SET host = host.name;

MATCH x=events y=events+
BY host
| AGG
    match_count = count()
  BY host
"#
)]
#[case::agg_count(
    r#"
DEF events = FROM simba:sysmon_events;

MATCH x=events y=events+
AGG event_count = count()
BY host.name
WITHIN 5m
"#
)]
#[case::agg_first_last(
    r#"
DEF events = FROM simba:sysmon_events;

MATCH x=events y=events+
AGG first_ts = first(timestamp), last_ts = last(timestamp)
BY host.name
WITHIN 5m
"#
)]
#[case::agg_multiple_aggregations(
    r#"
DEF events = FROM simba:sysmon_events;

MATCH x=events y=events+
AGG event_count = count(), first_event = first(event.code), last_event = last(event.code)
BY host.name
WITHIN 5m
"#
)]
fn test_match_commands(#[case] query_str: &str) -> Result<(), TranslationErrors> {
    let query = Query::parse_result(query_str)?;
    let catalog = catalog_with_tables(vec![("simba:sysmon_events", sysmon_events_schema())]);
    let _typed = type_check_with_catalog(query, catalog).into_result()?;
    Ok(())
}

// ======================== Pattern Quantifier Validation ========================

/// {0} quantifier is rejected — matching exactly zero times is meaningless
#[rstest]
#[case::exactly_zero_alone(
    r#"
DEF events = FROM simba:sysmon_events;
MATCH b=events{0} WITHIN 5m
"#
)]
#[case::exactly_zero_with_required(
    r#"
DEF events = FROM simba:sysmon_events;
MATCH a=events b=events{0} WITHIN 5m
"#
)]
fn test_exactly_zero_rejected(#[case] query_str: &str) {
    let query = Query::parse_result(query_str).expect("should parse");
    let catalog = catalog_with_tables(vec![("simba:sysmon_events", sysmon_events_schema())]);
    let err = type_check_with_catalog(query, catalog)
        .into_result()
        .expect_err("expected error for {0} quantifier");
    let msg = err.to_string();
    assert!(
        msg.contains("matching exactly zero times is meaningless"),
        "unexpected error message: {msg}"
    );
}

/// All-optional patterns (*, ?) are rejected — they don't require matching anything
#[rstest]
#[case::single_star(
    r#"
DEF events = FROM simba:sysmon_events;
MATCH b=events* WITHIN 5m
"#
)]
#[case::single_question(
    r#"
DEF events = FROM simba:sysmon_events;
MATCH b=events? WITHIN 5m
"#
)]
#[case::two_stars(
    r#"
DEF events = FROM simba:sysmon_events;
MATCH a=events* b=events* WITHIN 5m
"#
)]
#[case::mixed_optional(
    r#"
DEF events = FROM simba:sysmon_events;
MATCH a=events? b=events* WITHIN 5m
"#
)]
#[case::three_optionals(
    r#"
DEF events = FROM simba:sysmon_events;
MATCH a=events? b=events? c=events? WITHIN 5m
"#
)]
fn test_all_optional_patterns_rejected(#[case] query_str: &str) {
    let query = Query::parse_result(query_str).expect("should parse");
    let catalog = catalog_with_tables(vec![("simba:sysmon_events", sysmon_events_schema())]);
    let err = type_check_with_catalog(query, catalog)
        .into_result()
        .expect_err("expected error for all-optional pattern");
    let msg = err.to_string();
    assert!(
        msg.contains("at least one required pattern element"),
        "unexpected error message: {msg}"
    );
}

#[rstest]
#[case::required_then_optional(
    r#"
DEF events = FROM simba:sysmon_events;
MATCH a=events b=events? WITHIN 5m
"#
)]
#[case::at_least_one_then_star(
    r#"
DEF events = FROM simba:sysmon_events;
MATCH a=events+ b=events* WITHIN 5m
"#
)]
fn test_patterns_with_required_element_accepted(#[case] query_str: &str) {
    let query = Query::parse_result(query_str).expect("should parse");
    let catalog = catalog_with_tables(vec![("simba:sysmon_events", sysmon_events_schema())]);
    let result = type_check_with_catalog(query, catalog).into_result();
    assert!(result.is_ok(), "expected success: {result:?}");
}

// ======================== MATCH Duplicate Field Error Tests ========================

#[rstest]
#[case::agg_duplicate_fields(
    r#"
DEF events = FROM simba:sysmon_events;

MATCH x=events y=events+
AGG a = count(), a = sum(x.message)
BY host.name
WITHIN 5m
"#
)]
#[case::agg_groupby_conflict(
    r#"
DEF events = FROM simba:sysmon_events;

MATCH x=events y=events+
AGG host.name = count()
BY host.name
WITHIN 5m
"#
)]
fn test_match_duplicate_field_errors(#[case] query_str: &str) {
    let query = Query::parse_result(query_str).expect("should parse");
    let catalog = catalog_with_tables(vec![("simba:sysmon_events", sysmon_events_schema())]);
    let err = type_check_with_catalog(query, catalog)
        .into_result()
        .expect_err("expected error for duplicate fields in MATCH");
    let msg = err.to_string();
    assert!(
        msg.contains("conflicting definition"),
        "unexpected error message: {msg}"
    );
}

/// Test that MATCH implicitly sorts by timestamp when no explicit SORT is specified
#[test]
fn test_match_includes_timestamp_in_output() -> Result<(), TranslationErrors> {
    // Query uses DEF clauses (no catalog timestamp) and references timestamp after MATCH
    // MATCH should implicitly add timestamp to output schema even without explicit SORT
    let query_str = r#"
DEF A = ROWS[{timestamp: '2024-01-01T00:00:00Z' AS timestamp, x: 1}];
DEF B = ROWS[{timestamp: '2024-01-01T00:00:00Z' AS timestamp, y: 2}];

MATCH A B
| AGG count() BY timestamp@h
"#;

    let query = Query::parse_result(query_str)?;
    let _typed = type_check(query).into_result()?;
    Ok(())
}