flowscope 0.10.1

//! Plan 116 PR 1 — unified `Driver<E, M>` + `Event<K, M>` shape.
//!
//! Verifies the new types ship alongside the legacy
//! `FlowSessionDriver` / `FlowMultiSessionDriver` without
//! interfering. Covers builder shape, single-parser session
//! dispatch, port-routed dispatch, broadcast dispatch, and the
//! merged-event-stream contract.

#![cfg(all(feature = "test-helpers", feature = "extractors", feature = "session"))]

use flowscope::driver_unified::{Driver, Event};
use flowscope::extract::FiveTuple;
use flowscope::extract::parse::test_frames::{ipv4_tcp, ipv4_udp};
use flowscope::{DatagramParser, FlowSide, PacketView, SessionParser, Timestamp};

/// Toy session parser: counts bytes per side, emits one
/// "byte-count" message per feed.
#[derive(Default, Clone)]
struct CountParser {
    name: &'static str,
}

impl CountParser {
    fn named(name: &'static str) -> Self {
        Self { name }
    }
}

impl SessionParser for CountParser {
    type Message = (FlowSide, usize);

    fn feed_initiator(&mut self, b: &[u8], _ts: Timestamp) -> Vec<Self::Message> {
        if b.is_empty() {
            Vec::new()
        } else {
            vec![(FlowSide::Initiator, b.len())]
        }
    }

    fn feed_responder(&mut self, b: &[u8], _ts: Timestamp) -> Vec<Self::Message> {
        if b.is_empty() {
            Vec::new()
        } else {
            vec![(FlowSide::Responder, b.len())]
        }
    }

    fn parser_kind(&self) -> &'static str {
        if self.name.is_empty() {
            "count"
        } else {
            self.name
        }
    }
}

#[derive(Debug, Clone)]
enum Msg {
    Counted {
        name: &'static str,
        side: FlowSide,
        len: usize,
    },
}

impl Msg {
    /// Accessor used by assertions to keep `side` + `len` live.
    #[track_caller]
    fn unwrap_counted(&self) -> (&'static str, FlowSide, usize) {
        match self {
            Msg::Counted { name, side, len } => (*name, *side, *len),
        }
    }
}

#[test]
fn builder_shape_compiles_and_no_op_is_empty() {
    let mut driver = Driver::<_, Msg>::builder(FiveTuple::bidirectional()).build();
    // No registered parsers; an empty input run produces no events.
    let out = driver.finish();
    assert!(out.is_empty());
}

#[test]
fn single_session_parser_emits_flow_lifecycle_and_messages() {
    let mut driver = Driver::<_, Msg>::builder(FiveTuple::bidirectional())
        .session_broadcast(CountParser::named("count"), |(side, len)| Msg::Counted {
            name: "count",
            side,
            len,
        })
        .build();

    // SYN — first packet of a new flow.
    let syn = ipv4_tcp(
        [1; 6],
        [2; 6],
        [10, 0, 0, 1],
        [10, 0, 0, 2],
        1234,
        80,
        1000,
        0,
        0x02,
        b"",
    );
    let events = driver.track(PacketView::new(&syn, Timestamp::new(0, 0)));

    let started = events
        .iter()
        .filter(|e| matches!(e, Event::FlowStarted { .. }))
        .count();
    assert_eq!(started, 1, "expected one FlowStarted: {events:?}");
}

#[test]
fn port_routed_parser_fires_only_on_matching_flows() {
    let mut driver = Driver::<_, Msg>::builder(FiveTuple::bidirectional())
        .session_on_ports(CountParser::named("http"), [80], |(side, len)| {
            Msg::Counted {
                name: "http",
                side,
                len,
            }
        })
        .session_on_ports(CountParser::named("ssh"), [22], |(side, len)| {
            Msg::Counted {
                name: "ssh",
                side,
                len,
            }
        })
        .build();

    // Port 80 frame with payload — only the HTTP slot should
    // see it (the SSH slot's port filter rejects the frame
    // before even hitting its inner driver).
    let frame = ipv4_tcp(
        [1; 6],
        [2; 6],
        [10, 0, 0, 1],
        [10, 0, 0, 2],
        33000,
        80,
        1000,
        0,
        0x18,
        b"GET /\r\n",
    );
    let events = driver.track(PacketView::new(&frame, Timestamp::new(0, 0)));

    let http_msgs: Vec<_> = events
        .iter()
        .filter_map(|e| match e {
            Event::Message {
                message: Msg::Counted { name: "http", .. },
                ..
            } => Some(()),
            _ => None,
        })
        .collect();
    let ssh_msgs: Vec<_> = events
        .iter()
        .filter_map(|e| match e {
            Event::Message {
                message: Msg::Counted { name: "ssh", .. },
                ..
            } => Some(()),
            _ => None,
        })
        .collect();
    assert!(
        !http_msgs.is_empty(),
        "HTTP slot didn't fire on port-80 flow"
    );
    assert!(
        ssh_msgs.is_empty(),
        "SSH slot wrongly fired on port-80 flow"
    );
    // Verify the HTTP-routed message carries the expected metadata
    // (uses the Msg accessor so side + len fields aren't dead).
    let first_http = events
        .iter()
        .find_map(|e| match e {
            Event::Message {
                message: m @ Msg::Counted { name: "http", .. },
                ..
            } => Some(m.unwrap_counted()),
            _ => None,
        })
        .expect("at least one HTTP message");
    assert_eq!(first_http.0, "http");
    assert!(matches!(
        first_http.1,
        FlowSide::Initiator | FlowSide::Responder
    ));
    assert!(first_http.2 > 0);
}

#[test]
fn broadcast_parser_fires_regardless_of_port() {
    let mut driver = Driver::<_, Msg>::builder(FiveTuple::bidirectional())
        .session_broadcast(CountParser::named("any"), |(side, len)| Msg::Counted {
            name: "any",
            side,
            len,
        })
        .build();

    let frame = ipv4_tcp(
        [1; 6],
        [2; 6],
        [10, 0, 0, 1],
        [10, 0, 0, 2],
        33000,
        9999,
        1000,
        0,
        0x18,
        b"hello",
    );
    let events = driver.track(PacketView::new(&frame, Timestamp::new(0, 0)));
    let any: usize = events
        .iter()
        .filter(|e| {
            matches!(
                e,
                Event::Message {
                    message: Msg::Counted { name: "any", .. },
                    ..
                }
            )
        })
        .count();
    assert!(any > 0, "broadcast slot didn't fire on arbitrary-port flow");
}

#[test]
fn event_accessors_smoke_test() {
    let ev: Event<u32, Msg> = Event::FlowStarted {
        key: 7,
        ts: Timestamp::new(0, 0),
        l4: None,
    };
    assert_eq!(ev.key().copied(), Some(7));
    assert!(ev.is_flow_event());
    assert!(!ev.is_parser_event());
    assert!(ev.parser_kind().is_none());

    let ev: Event<u32, Msg> = Event::Message {
        key: 7,
        side: FlowSide::Initiator,
        message: Msg::Counted {
            name: "x",
            side: FlowSide::Initiator,
            len: 1,
        },
        ts: Timestamp::new(0, 0),
        parser_kind: "x",
    };
    assert!(ev.is_parser_event());
    assert_eq!(ev.parser_kind(), Some("x"));
}

/// Toy datagram parser that emits one event per non-empty payload.
#[derive(Default, Clone)]
struct UdpEcho;

impl DatagramParser for UdpEcho {
    type Message = usize;
    fn parse(&mut self, payload: &[u8], _side: FlowSide, _ts: Timestamp) -> Vec<Self::Message> {
        if payload.is_empty() {
            Vec::new()
        } else {
            vec![payload.len()]
        }
    }
    fn parser_kind(&self) -> &'static str {
        "udp-echo"
    }
}

#[test]
fn datagram_on_ports_fires_on_matching_udp_flows() {
    let mut driver = Driver::<_, usize>::builder(FiveTuple::bidirectional())
        .datagram_on_ports(UdpEcho, [53], |n| n)
        .build();
    let frame = ipv4_udp([10, 0, 0, 1], [10, 0, 0, 2], 33000, 53, b"hello");
    let events = driver.track(PacketView::new(&frame, Timestamp::new(0, 0)));
    let msgs: Vec<usize> = events
        .iter()
        .filter_map(|e| match e {
            Event::Message { message, .. } => Some(*message),
            _ => None,
        })
        .collect();
    assert_eq!(msgs, vec![5]);
}

#[test]
fn datagram_broadcast_fires_regardless_of_port() {
    let mut driver = Driver::<_, usize>::builder(FiveTuple::bidirectional())
        .datagram_broadcast(UdpEcho, |n| n)
        .build();
    let frame = ipv4_udp([10, 0, 0, 1], [10, 0, 0, 2], 33000, 7777, b"abcd");
    let events = driver.track(PacketView::new(&frame, Timestamp::new(0, 0)));
    let msgs: Vec<usize> = events
        .iter()
        .filter_map(|e| match e {
            Event::Message { message, .. } => Some(*message),
            _ => None,
        })
        .collect();
    assert_eq!(msgs, vec![4]);
}

#[test]
fn datagram_on_ports_skips_non_matching() {
    let mut driver = Driver::<_, usize>::builder(FiveTuple::bidirectional())
        .datagram_on_ports(UdpEcho, [53], |n| n)
        .build();
    let frame = ipv4_udp([10, 0, 0, 1], [10, 0, 0, 2], 33000, 7777, b"abcd");
    let events = driver.track(PacketView::new(&frame, Timestamp::new(0, 0)));
    let msgs: Vec<usize> = events
        .iter()
        .filter_map(|e| match e {
            Event::Message { message, .. } => Some(*message),
            _ => None,
        })
        .collect();
    assert!(
        msgs.is_empty(),
        "wrongly fired on non-matching port: {msgs:?}"
    );
}

#[test]
fn session_and_datagram_slots_coexist() {
    let mut driver = Driver::<_, MixedMsg>::builder(FiveTuple::bidirectional())
        .session_broadcast(CountParser::named("tcp"), |(s, n)| MixedMsg::Tcp {
            side: s,
            len: n,
        })
        .datagram_broadcast(UdpEcho, MixedMsg::Udp)
        .build();
    let tcp = ipv4_tcp(
        [1; 6],
        [2; 6],
        [10, 0, 0, 1],
        [10, 0, 0, 2],
        33000,
        80,
        0,
        0,
        0x18,
        b"x",
    );
    let udp = ipv4_udp([10, 0, 0, 3], [10, 0, 0, 4], 5353, 53, b"yz");
    let mut events = driver.track(PacketView::new(&tcp, Timestamp::new(0, 0)));
    events.extend(driver.track(PacketView::new(&udp, Timestamp::new(1, 0))));
    let tcp_msgs = events
        .iter()
        .filter(|e| {
            matches!(
                e,
                Event::Message {
                    message: MixedMsg::Tcp { .. },
                    ..
                }
            )
        })
        .count();
    let udp_msgs = events
        .iter()
        .filter(|e| {
            matches!(
                e,
                Event::Message {
                    message: MixedMsg::Udp(_),
                    ..
                }
            )
        })
        .count();
    assert!(tcp_msgs > 0, "no TCP messages emitted");
    assert!(udp_msgs > 0, "no UDP messages emitted");
    // Read the inner data via fingerprint() to keep the variant
    // fields live (otherwise side/len are dead-code).
    let fingerprints: Vec<_> = events
        .iter()
        .filter_map(|e| match e {
            Event::Message { message, .. } => Some(message.fingerprint()),
            _ => None,
        })
        .collect();
    assert!(!fingerprints.is_empty());
}

#[derive(Debug, Clone)]
enum MixedMsg {
    Tcp { side: FlowSide, len: usize },
    Udp(usize),
}

impl MixedMsg {
    /// Side-arm accessor consumed by `session_and_datagram_slots_coexist`
    /// so the inner fields stay live.
    fn fingerprint(&self) -> (usize, FlowSide) {
        match self {
            MixedMsg::Tcp { side, len } => (*len, *side),
            MixedMsg::Udp(n) => (*n, FlowSide::Initiator),
        }
    }
}

// ── heuristic routing (plan 116 PR 2b) ─────────────────────────────

/// Toy "always matches" signature.
fn always_match(_b: &[u8]) -> flowscope::detect::signatures::SignatureMatch {
    flowscope::detect::signatures::SignatureMatch::Match
}

/// Toy "never matches" signature.
fn never_match(_b: &[u8]) -> flowscope::detect::signatures::SignatureMatch {
    flowscope::detect::signatures::SignatureMatch::NoMatch
}

#[test]
fn session_heuristic_dispatches_on_match() {
    let mut driver = Driver::<_, Msg>::builder(FiveTuple::bidirectional())
        .session_heuristic(CountParser::named("h"), always_match, |(side, len)| {
            Msg::Counted {
                name: "h",
                side,
                len,
            }
        })
        .build();
    let frame = ipv4_tcp(
        [1; 6],
        [2; 6],
        [10, 0, 0, 1],
        [10, 0, 0, 2],
        33000,
        9999,
        0,
        0,
        0x18,
        b"hello",
    );
    let events = driver.track(PacketView::new(&frame, Timestamp::new(0, 0)));
    let messages = events
        .iter()
        .filter(|e| {
            matches!(
                e,
                Event::Message {
                    message: Msg::Counted { name: "h", .. },
                    ..
                }
            )
        })
        .count();
    assert!(
        messages > 0,
        "heuristic slot didn't fire on Match signature"
    );
}

#[test]
fn session_heuristic_skips_on_never_match() {
    let mut driver = Driver::<_, Msg>::builder(FiveTuple::bidirectional())
        .session_heuristic_with_budget(CountParser::named("h"), never_match, 2, |(side, len)| {
            Msg::Counted {
                name: "h",
                side,
                len,
            }
        })
        .build();
    let frame = ipv4_tcp(
        [1; 6],
        [2; 6],
        [10, 0, 0, 1],
        [10, 0, 0, 2],
        33000,
        9999,
        0,
        0,
        0x18,
        b"X",
    );
    // First packet — probing; second packet — still probing; third packet — gave up.
    let frame2 = ipv4_tcp(
        [1; 6],
        [2; 6],
        [10, 0, 0, 1],
        [10, 0, 0, 2],
        33000,
        9999,
        0,
        0,
        0x18,
        b"Y",
    );
    let frame3 = ipv4_tcp(
        [1; 6],
        [2; 6],
        [10, 0, 0, 1],
        [10, 0, 0, 2],
        33000,
        9999,
        0,
        0,
        0x18,
        b"Z",
    );
    let mut events = driver.track(PacketView::new(&frame, Timestamp::new(0, 0)));
    events.extend(driver.track(PacketView::new(&frame2, Timestamp::new(1, 0))));
    events.extend(driver.track(PacketView::new(&frame3, Timestamp::new(2, 0))));
    let messages = events
        .iter()
        .filter(|e| {
            matches!(
                e,
                Event::Message {
                    message: Msg::Counted { name: "h", .. },
                    ..
                }
            )
        })
        .count();
    assert_eq!(
        messages, 0,
        "heuristic slot wrongly fired on NoMatch signature"
    );
}

#[test]
fn datagram_heuristic_dispatches_on_match() {
    let mut driver = Driver::<_, usize>::builder(FiveTuple::bidirectional())
        .datagram_heuristic(UdpEcho, always_match, |n| n)
        .build();
    let frame = ipv4_udp([10, 0, 0, 1], [10, 0, 0, 2], 33000, 9999, b"yo");
    let events = driver.track(PacketView::new(&frame, Timestamp::new(0, 0)));
    let msgs: Vec<usize> = events
        .iter()
        .filter_map(|e| match e {
            Event::Message { message, .. } => Some(*message),
            _ => None,
        })
        .collect();
    assert_eq!(msgs, vec![2]);
}

#[test]
fn datagram_heuristic_skips_on_never_match() {
    let mut driver = Driver::<_, usize>::builder(FiveTuple::bidirectional())
        .datagram_heuristic_with_budget(UdpEcho, never_match, 2, |n| n)
        .build();
    let frame = ipv4_udp([10, 0, 0, 1], [10, 0, 0, 2], 33000, 9999, b"yo");
    let events = driver.track(PacketView::new(&frame, Timestamp::new(0, 0)));
    let msgs: Vec<usize> = events
        .iter()
        .filter_map(|e| match e {
            Event::Message { message, .. } => Some(*message),
            _ => None,
        })
        .collect();
    assert!(
        msgs.is_empty(),
        "datagram heuristic wrongly fired: {msgs:?}"
    );
}

#[test]
fn idle_timeout_fn_applies_to_central_tracker() {
    use std::time::Duration;
    // Custom idle timeout of 1 s; sweep at t=10 should evict any
    // flow whose last packet was older than 1 s.
    let mut driver = Driver::<_, Msg>::builder(FiveTuple::bidirectional())
        .idle_timeout_fn(|_, _| Some(Duration::from_secs(1)))
        .build();
    let frame = ipv4_tcp(
        [1; 6],
        [2; 6],
        [10, 0, 0, 1],
        [10, 0, 0, 2],
        33000,
        80,
        0,
        0,
        0x02,
        b"",
    );
    let _ = driver.track(PacketView::new(&frame, Timestamp::new(0, 0)));
    let evs = driver.sweep(Timestamp::new(5, 0));
    let ended = evs
        .iter()
        .filter(|e| matches!(e, Event::FlowEnded { .. }))
        .count();
    assert_eq!(ended, 1, "expected idle-timeout to fire FlowEnded by t=5s");
}

#[test]
fn emit_anomalies_off_by_default_yields_no_anomalies() {
    let mut driver = Driver::<_, Msg>::builder(FiveTuple::bidirectional()).build();
    let frame = ipv4_tcp(
        [1; 6],
        [2; 6],
        [10, 0, 0, 1],
        [10, 0, 0, 2],
        33000,
        80,
        0,
        0,
        0x02,
        b"",
    );
    let evs = driver.track(PacketView::new(&frame, Timestamp::new(0, 0)));
    let anomalies = evs
        .iter()
        .filter(|e| matches!(e, Event::FlowAnomaly { .. } | Event::TrackerAnomaly { .. }))
        .count();
    assert_eq!(anomalies, 0);
}

#[test]
fn one_flow_started_per_flow_regardless_of_parser_count() {
    // Plan 116 spec: "one FlowStarted per flow, not N per parser".
    // The central tracker is the single source of flow lifecycle;
    // slots filter their own SessionEvent::Started variants out.
    let mut driver = Driver::<_, Msg>::builder(FiveTuple::bidirectional())
        .session_broadcast(CountParser::named("a"), |(s, n)| Msg::Counted {
            name: "a",
            side: s,
            len: n,
        })
        .session_broadcast(CountParser::named("b"), |(s, n)| Msg::Counted {
            name: "b",
            side: s,
            len: n,
        })
        .session_broadcast(CountParser::named("c"), |(s, n)| Msg::Counted {
            name: "c",
            side: s,
            len: n,
        })
        .build();

    let frame = ipv4_tcp(
        [1; 6],
        [2; 6],
        [10, 0, 0, 1],
        [10, 0, 0, 2],
        33000,
        80,
        1000,
        0,
        0x02,
        b"",
    );
    let events = driver.track(PacketView::new(&frame, Timestamp::new(0, 0)));
    let started = events
        .iter()
        .filter(|e| matches!(e, Event::FlowStarted { .. }))
        .count();
    assert_eq!(
        started, 1,
        "with 3 broadcast parsers, expected 1 FlowStarted; got {started}",
    );
}

#[test]
fn parser_closed_fires_per_parser_on_flow_end() {
    // Plan 116 spec: ParserClosed fires per (parser, flow). With
    // 2 broadcast parsers registered, ending a flow should yield
    // 2 ParserClosed events (one per parser) and 1 FlowEnded
    // (single source).
    let mut driver = Driver::<_, Msg>::builder(FiveTuple::bidirectional())
        .session_broadcast(CountParser::named("a"), |(s, n)| Msg::Counted {
            name: "a",
            side: s,
            len: n,
        })
        .session_broadcast(CountParser::named("b"), |(s, n)| Msg::Counted {
            name: "b",
            side: s,
            len: n,
        })
        .build();

    // SYN to start the flow.
    let syn = ipv4_tcp(
        [1; 6],
        [2; 6],
        [10, 0, 0, 1],
        [10, 0, 0, 2],
        33000,
        80,
        1000,
        0,
        0x02,
        b"",
    );
    let _ = driver.track(PacketView::new(&syn, Timestamp::new(0, 0)));

    // finish() flushes idle flows.
    let finish_events = driver.finish();
    let parser_closed: Vec<&'static str> = finish_events
        .iter()
        .filter_map(|e| match e {
            Event::ParserClosed { parser_kind, .. } => Some(*parser_kind),
            _ => None,
        })
        .collect();
    let flow_ended = finish_events
        .iter()
        .filter(|e| matches!(e, Event::FlowEnded { .. }))
        .count();

    // 2 ParserClosed (one per registered parser).
    assert_eq!(
        parser_closed.len(),
        2,
        "expected 2 ParserClosed events; got {parser_closed:?}",
    );
    assert!(parser_closed.contains(&"a"));
    assert!(parser_closed.contains(&"b"));
    // 1 FlowEnded (central tracker only).
    assert_eq!(flow_ended, 1, "expected 1 FlowEnded; got {flow_ended}");
}

#[test]
fn emit_packet_details_populates_tcp_and_frame() {
    let mut driver = Driver::<_, Msg>::builder(FiveTuple::bidirectional())
        .emit_packet_details(true)
        .build();

    let frame = ipv4_tcp(
        [1; 6],
        [2; 6],
        [10, 0, 0, 1],
        [10, 0, 0, 2],
        33000,
        80,
        1000,
        0,
        0x18,
        b"data",
    );
    let events = driver.track(PacketView::new(&frame, Timestamp::new(0, 0)));
    let packet_events: Vec<_> = events
        .iter()
        .filter_map(|e| match e {
            Event::FlowPacket { tcp, frame, .. } => Some((tcp.is_some(), frame.is_some())),
            _ => None,
        })
        .collect();
    assert!(
        packet_events.iter().any(|(t, f)| *t && *f),
        "emit_packet_details=true should populate tcp + frame: {packet_events:?}",
    );
}

#[test]
fn emit_packet_details_off_leaves_tcp_and_frame_none() {
    let mut driver = Driver::<_, Msg>::builder(FiveTuple::bidirectional()).build();
    let frame = ipv4_tcp(
        [1; 6],
        [2; 6],
        [10, 0, 0, 1],
        [10, 0, 0, 2],
        33000,
        80,
        1000,
        0,
        0x18,
        b"data",
    );
    let events = driver.track(PacketView::new(&frame, Timestamp::new(0, 0)));
    let packet_events: Vec<_> = events
        .iter()
        .filter_map(|e| match e {
            Event::FlowPacket { tcp, frame, .. } => Some((tcp.is_some(), frame.is_some())),
            _ => None,
        })
        .collect();
    assert!(
        packet_events.iter().all(|(t, f)| !*t && !*f),
        "default emit_packet_details=false should leave tcp/frame None: {packet_events:?}",
    );
}

#[test]
fn udp_flow_does_not_emit_established() {
    let mut driver = Driver::<_, Msg>::builder(FiveTuple::bidirectional())
        .session_broadcast(CountParser::default(), |(side, len)| Msg::Counted {
            name: "count",
            side,
            len,
        })
        .build();
    let frame = ipv4_udp([10, 0, 0, 1], [10, 0, 0, 2], 5353, 53, b"x");
    let events = driver.track(PacketView::new(&frame, Timestamp::new(0, 0)));
    let established = events
        .iter()
        .filter(|e| matches!(e, Event::FlowEstablished { .. }))
        .count();
    assert_eq!(established, 0);
}