//! 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::{IntoTyped, ParseWithErrors, TypeCheckExecutor};
use crate::tree::tests::shared::catalog_with_tables;
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#"
WITH 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#"
WITH events = FROM simba.sysmon_events

MATCH events WITHIN 5m
"#
)]
#[case::with_within_clause_multiple_patterns(
    r#"
WITH 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#"
WITH events = FROM simba.sysmon_events
| LET host = host.name

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

MATCH x=events y=events+
AGG event_count = count()
BY host.name
WITHIN 5m
"#
)]
#[case::agg_first_last(
    r#"
WITH 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#"
WITH 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 = query.typed_with().with_provider(catalog).typed_result()?;
    Ok(())
}

/// 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 WITH 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#"
WITH A = ROWS[{timestamp: '2024-01-01T00:00:00Z' AS timestamp, x: 1}]
WITH 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 = query.typed_result()?;
    Ok(())
}
