netring 0.21.0

//! Per-event context passed to handlers.
//!
//! `Ctx` lives on the dispatch stack — never heap-allocated.
//! Handlers receive a single `&mut Ctx<'_>` (alongside the typed
//! event payload) and pull what they need via methods on `Ctx`:
//!
//! ```ignore
//! Monitor::builder()
//!     .state::<MyState>()
//!     .counter::<IpAddr>(Duration::from_secs(10), Duration::from_secs(1))
//!     .on_ctx::<Http>(|req, ctx| {
//!         let counters = ctx.counter_mut::<IpAddr>();
//!         counters.bump(req.client_ip(), ctx.ts);
//!         ctx.state_mut::<MyState>().requests += 1;
//!         Ok(())
//!     });
//! ```
//!
//! ## Why method-style instead of axum-style extractors
//!
//! The first cut of this module shipped an `FromCtx` trait with
//! multi-extractor blanket impls (1..=8 arities). It doesn't
//! compile: every `<P as FromCtx>::from_ctx(&mut ctx)` call holds
//! `&mut Ctx` for as long as its return value lives, so the
//! second extractor can't re-borrow. axum gets away with this
//! because async-await sequences the borrows; sync Rust can't.
//!
//! Method accessors on `Ctx` give the same ergonomics without
//! the borrow-checker headache: each call is its own bounded
//! borrow, and the compiler tracks disjoint field accesses
//! (`state_map`, `counters`, `sink`) correctly.

use flowscope::Timestamp;

mod from_ctx;
mod split;

pub use from_ctx::{CounterRegistry, FlowStateRegistry, StateMap};

use crate::correlate::TimeBucketedCounter;
use crate::protocol::FlowKey;

/// Tag for which capture source this event came from.
/// `SourceIdx(0)` for single-interface monitors; multi-interface
/// (Phase E) increments per registered iface.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct SourceIdx(pub u8);

/// Runtime context passed to every handler invocation.
///
/// The `flow` / `ts` / `source` fields are `pub` for direct read.
/// State / counter / sink access goes through the typed accessor
/// methods so the storage maps can stay `pub(crate)`.
pub struct Ctx<'a> {
    /// The flow key for the current event, if any.
    ///
    /// Held by value (`FlowKey` is `Copy`) so dispatch sites can
    /// stamp a per-message key without lifetime gymnastics —
    /// borrowing through `Option<&'a FlowKey>` would require a
    /// place to anchor the borrow that outlives the `Ctx`.
    ///
    /// **Prefer `payload.key` over `ctx.flow` when the event
    /// payload carries one.** For lifecycle events
    /// (`FlowStarted/Ended/Established<P>`) and message events
    /// (`HttpMessage`, `DnsMessage`, …), `payload.key` is the
    /// typed value with no `Option` indirection. For non-flow
    /// events (`Tick`) `ctx.flow` is `None`. The same key is
    /// available through both paths for flow events; the field
    /// is preserved as a stable accessor for cross-cutting
    /// handlers (e.g. a tick handler that logs the most recent
    /// flow, where the typed payload doesn't carry one).
    pub flow: Option<FlowKey>,

    /// Timestamp of the current event.
    pub ts: Timestamp,

    /// Source-interface index.
    pub source: SourceIdx,

    /// 0.21 D.4: optional human-readable name set on the parent
    /// monitor via [`crate::monitor::MonitorBuilder::name`].
    /// Handlers running under multiple monitors in the same
    /// process can branch on this to disambiguate (e.g. log
    /// `target = monitor_name` or stamp `with("monitor", name)`
    /// on emitted anomalies). Borrowed from the
    /// [`Monitor`](crate::monitor::Monitor)'s
    /// owned `Box<str>` storage at dispatch time; `None` when
    /// `.name(...)` was not called.
    pub monitor_name: Option<&'a str>,

    /// Per-monitor user state, keyed by `TypeId`.
    pub(crate) state_map: &'a mut StateMap,

    /// The anomaly sink (Phase C fills in the trait body).
    pub(crate) sink: &'a mut dyn crate::anomaly::sink::AnomalySink,

    /// Per-monitor counter storage.
    pub(crate) counters: &'a mut CounterRegistry,

    /// 0.21 I.7: per-flow user state. Each `T: Default + Send +
    /// 'static` registered via
    /// [`crate::monitor::MonitorBuilder::flow_state`] gets a
    /// per-flow slot lazily created on
    /// [`Self::flow_state_mut::<T>`] access. Backed by
    /// [`flowscope::correlate::FlowStateMap`].
    pub(crate) flow_states: &'a mut FlowStateRegistry,
}

impl<'a> Ctx<'a> {
    /// Bench-only constructor — `benches/zero_alloc.rs` needs to
    /// manually assemble a `Ctx` to drive the dispatcher without
    /// a full `Monitor`. Not part of the public API outside the
    /// bench feature.
    #[cfg(feature = "bench-zero-alloc")]
    pub fn new_for_bench(
        ts: Timestamp,
        state_map: &'a mut StateMap,
        sink: &'a mut dyn crate::anomaly::sink::AnomalySink,
        counters: &'a mut CounterRegistry,
        flow_states: &'a mut FlowStateRegistry,
    ) -> Self {
        Self {
            flow: None,
            ts,
            source: SourceIdx(0),
            monitor_name: None,
            state_map,
            sink,
            counters,
            flow_states,
        }
    }

    /// Constructor exposed for integration tests that need to
    /// drive the dispatcher with a custom `Ctx`. Not part of the
    /// documented public API (`#[doc(hidden)]`). `monitor_name`
    /// defaults to `None`; production callers in the run loop
    /// set it via the [`Ctx`] struct literal directly.
    #[doc(hidden)]
    pub fn new(
        flow: Option<FlowKey>,
        ts: Timestamp,
        source: SourceIdx,
        state_map: &'a mut StateMap,
        sink: &'a mut dyn crate::anomaly::sink::AnomalySink,
        counters: &'a mut CounterRegistry,
        flow_states: &'a mut FlowStateRegistry,
    ) -> Self {
        Self {
            flow,
            ts,
            source,
            monitor_name: None,
            state_map,
            sink,
            counters,
            flow_states,
        }
    }

    /// Borrow per-monitor state `T` mutably.
    ///
    /// `T: Default` so the slot is lazy-created on first access.
    /// Pre-register via `MonitorBuilder::state::<T>()` to surface
    /// typos at build time.
    #[inline]
    pub fn state_mut<T: Default + Send + 'static>(&mut self) -> &mut T {
        self.state_map.get_or_init_mut::<T>()
    }

    /// Borrow the `K`-keyed sliding-window counter mutably.
    ///
    /// # Panics
    ///
    /// Panics if no `MonitorBuilder::counter::<K>(...)` call
    /// registered this key — a programmer error caught early in
    /// development.
    #[inline]
    pub fn counter_mut<K>(&mut self) -> &mut TimeBucketedCounter<K>
    where
        K: std::hash::Hash + Eq + Clone + Send + 'static,
    {
        self.counters.get_mut::<K>()
    }

    /// Borrow the anomaly sink mutably.
    #[inline]
    pub fn sink_mut(&mut self) -> &mut dyn crate::anomaly::sink::AnomalySink {
        self.sink
    }

    /// 0.21 I.7: borrow per-flow user state `T` mutably.
    ///
    /// Returns `None` when:
    /// - `T` was not registered via
    ///   [`crate::monitor::MonitorBuilder::flow_state::<T>`].
    /// - This `Ctx` doesn't carry a flow key
    ///   ([`Tick`](crate::protocol::event_typed::Tick) and
    ///   `TrackerAnomaly` are the only such events).
    ///
    /// Lazy-creates `T::default()` on first access for a given
    /// flow key; subsequent accesses return the same `&mut T`.
    /// The slot's last-seen timestamp refreshes on every access.
    /// Eviction happens via `FlowStateMap::feed(FlowEvent::Ended)`
    /// — when `FlowEnded<P>` lands, the slot for that key is
    /// freed.
    ///
    /// Typical use: aggregate per-flow stats inside a
    /// `FlowStarted<P>` handler, read+emit in `FlowEnded<P>`:
    ///
    /// ```ignore
    /// .on_ctx::<FlowStarted<Tcp>>(|_e, ctx| {
    ///     let s = ctx.flow_state_mut::<MyState>().unwrap();
    ///     s.bytes = 0;
    ///     Ok(())
    /// })
    /// .on_ctx::<FlowEnded<Tcp>>(|e, ctx| {
    ///     let s = ctx.flow_state_mut::<MyState>().unwrap();
    ///     ctx.emit("FlowDone", Severity::Info)
    ///         .with_metric("bytes", s.bytes as f64)
    ///         .emit();
    ///     Ok(())
    /// })
    /// ```
    #[inline]
    pub fn flow_state_mut<T>(&mut self) -> Option<&mut T>
    where
        T: Default + Send + 'static,
    {
        let key = self.flow?;
        let ts = self.ts;
        let map = self.flow_states.get_mut::<T>()?;
        Some(map.get_or_default(&key, ts))
    }

    /// Shortcut: begin an anomaly emission keyed by `kind` + `severity`
    /// using `self.ts` as the timestamp. Equivalent to
    /// `self.sink_mut().begin(kind, severity, self.ts)` but reads
    /// cleaner:
    ///
    /// ```ignore
    /// ctx.emit("FlowStartedTcp", Severity::Info)
    ///     .with_key(&evt.key)
    ///     .with_metric("count", n as f64)
    ///     .emit();
    /// ```
    ///
    /// The returned [`AnomalyWriter`](crate::anomaly::sink::AnomalyWriter)
    /// borrows the sink for its lifetime; no temporaries needed.
    #[inline]
    pub fn emit(
        &mut self,
        kind: &'static str,
        severity: crate::anomaly::Severity,
    ) -> crate::anomaly::sink::AnomalyWriter<'_> {
        let ts = self.ts;
        self.sink.begin(kind, severity, ts)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::anomaly::sink::NoopSink;

    #[derive(Default)]
    struct DemoState {
        n: u64,
    }

    fn make_ctx<'a>(
        state: &'a mut StateMap,
        sink: &'a mut NoopSink,
        counters: &'a mut CounterRegistry,
        flow_states: &'a mut FlowStateRegistry,
    ) -> Ctx<'a> {
        Ctx {
            flow: None,
            ts: Timestamp::new(0, 0),
            source: SourceIdx(0),
            monitor_name: None,
            state_map: state,
            sink,
            counters,
            flow_states,
        }
    }

    #[test]
    fn ctx_constructs_from_borrowed_fields() {
        let mut state = StateMap::default();
        let mut counters = CounterRegistry::default();
        let mut sink = NoopSink;
        let mut flow_states = FlowStateRegistry::default();
        let _ctx = make_ctx(&mut state, &mut sink, &mut counters, &mut flow_states);
    }

    #[test]
    fn state_mut_lazy_creates_then_returns_same() {
        let mut state = StateMap::default();
        let mut counters = CounterRegistry::default();
        let mut sink = NoopSink;
        let mut flow_states = FlowStateRegistry::default();
        let mut ctx = make_ctx(&mut state, &mut sink, &mut counters, &mut flow_states);
        ctx.state_mut::<DemoState>().n = 7;
        assert_eq!(ctx.state_mut::<DemoState>().n, 7);
    }

    #[test]
    fn sink_mut_returns_dyn_anomalysink() {
        let mut state = StateMap::default();
        let mut counters = CounterRegistry::default();
        let mut sink = NoopSink;
        let mut flow_states = FlowStateRegistry::default();
        let mut ctx = make_ctx(&mut state, &mut sink, &mut counters, &mut flow_states);
        let _: &mut dyn crate::anomaly::sink::AnomalySink = ctx.sink_mut();
    }

    #[test]
    fn counter_mut_returns_registered_counter() {
        use std::time::Duration;
        let mut state = StateMap::default();
        let mut counters = CounterRegistry::default();
        counters.register::<u16>(TimeBucketedCounter::<u16>::new_unbounded(
            Duration::from_secs(60),
            Duration::from_secs(1),
        ));
        let mut sink = NoopSink;
        let mut flow_states = FlowStateRegistry::default();
        let mut ctx = make_ctx(&mut state, &mut sink, &mut counters, &mut flow_states);
        ctx.counter_mut::<u16>().bump(42u16, Timestamp::new(0, 0));
    }

    #[test]
    fn source_idx_roundtrip() {
        assert_eq!(SourceIdx(3), SourceIdx(3));
        assert_ne!(SourceIdx(1), SourceIdx(2));
    }
}