Skip to main content

freeswitch_log_parser/
session.rs

1use std::collections::{HashMap, HashSet};
2use std::str::FromStr;
3
4use freeswitch_types::{BridgeDialString, CallDirection, DialString};
5
6use crate::line::parse_line;
7use crate::message::{classify_message, MessageKind};
8use crate::stream::{Block, LogEntry, LogStream, ParseStats, UnclassifiedLine};
9
10type SessionHook = Box<dyn Fn(&LogEntry, &mut SessionState) + Send>;
11
12/// Mutable per-UUID state accumulator, updated as entries are processed.
13///
14/// Fields are `None` until the corresponding data is first seen in the stream.
15/// Variables accumulate from CHANNEL_DATA dumps, `set()`/`export()` executions,
16/// `SET`/`EXPORT` log lines, and inline `variable_*` lines.
17#[derive(Debug, Clone, Default)]
18pub struct SessionState {
19    /// `None` until a `Channel-Name` field is encountered.
20    pub channel_name: Option<String>,
21    /// `None` until a state change or `Channel-State` field is encountered.
22    pub channel_state: Option<String>,
23    /// First dialplan context seen; set once and never overwritten.
24    pub initial_context: Option<String>,
25    /// Current dialplan context; updated on each transfer/continue.
26    pub dialplan_context: Option<String>,
27    /// Source extension in the dialplan routing; `None` until a dialplan line is processed.
28    pub dialplan_from: Option<String>,
29    /// Target extension in the dialplan routing; `None` until a dialplan line is processed.
30    pub dialplan_to: Option<String>,
31    /// Call direction from `Call-Direction` CHANNEL_DATA field; `None` until seen.
32    pub call_direction: Option<CallDirection>,
33    /// Caller ID number from `Caller-Caller-ID-Number` CHANNEL_DATA field; `None` until seen.
34    pub caller_id_number: Option<String>,
35    /// Caller ID name from `Caller-Caller-ID-Name` CHANNEL_DATA field; `None` until seen.
36    pub caller_id_name: Option<String>,
37    /// Destination number from `Caller-Destination-Number` CHANNEL_DATA field; `None` until seen.
38    pub destination_number: Option<String>,
39    /// Hangup cause extracted from ChannelLifecycle Hangup detail; `None` until hangup seen.
40    pub hangup_cause: Option<String>,
41    /// Timestamp when "has been answered" lifecycle event was seen; `None` until answered.
42    pub answered_at: Option<String>,
43    /// Other leg's UUID; `None` until bridged. Set from `Originate Resulted in Success` on A-leg,
44    /// and from `New Channel` on B-leg (back-pointing to A-leg via originate context).
45    pub other_leg_uuid: Option<String>,
46    /// Pending bridge target channel from `EXECUTE bridge()`, consumed when B-leg `New Channel` matches.
47    pub(crate) pending_bridge_target: Option<String>,
48    /// All variables learned so far, with the `variable_` prefix stripped from names.
49    pub variables: HashMap<String, String>,
50}
51
52/// Changes to indexed fields reported by `update_from_entry` for index maintenance.
53#[derive(Default)]
54struct IndexedFieldChanges {
55    channel_name: Option<(Option<String>, Option<String>)>,
56    pending_bridge_target: Option<(Option<String>, Option<String>)>,
57    other_leg_uuid: Option<(Option<String>, Option<String>)>,
58}
59
60/// Immutable point-in-time copy of a session's state, attached to each [`EnrichedEntry`].
61///
62/// Does not include `variables` to keep snapshots lightweight — access the full
63/// variable map via [`SessionTracker::sessions()`].
64#[derive(Debug, Clone)]
65pub struct SessionSnapshot {
66    pub channel_name: Option<String>,
67    pub channel_state: Option<String>,
68    pub initial_context: Option<String>,
69    pub dialplan_context: Option<String>,
70    pub dialplan_from: Option<String>,
71    pub dialplan_to: Option<String>,
72    pub call_direction: Option<CallDirection>,
73    pub caller_id_number: Option<String>,
74    pub caller_id_name: Option<String>,
75    pub destination_number: Option<String>,
76    pub hangup_cause: Option<String>,
77    pub answered_at: Option<String>,
78    pub other_leg_uuid: Option<String>,
79}
80
81impl SessionState {
82    fn snapshot(&self) -> SessionSnapshot {
83        SessionSnapshot {
84            channel_name: self.channel_name.clone(),
85            channel_state: self.channel_state.clone(),
86            initial_context: self.initial_context.clone(),
87            dialplan_context: self.dialplan_context.clone(),
88            dialplan_from: self.dialplan_from.clone(),
89            dialplan_to: self.dialplan_to.clone(),
90            call_direction: self.call_direction,
91            caller_id_number: self.caller_id_number.clone(),
92            caller_id_name: self.caller_id_name.clone(),
93            destination_number: self.destination_number.clone(),
94            hangup_cause: self.hangup_cause.clone(),
95            answered_at: self.answered_at.clone(),
96            other_leg_uuid: self.other_leg_uuid.clone(),
97        }
98    }
99
100    fn update_from_entry(&mut self, entry: &LogEntry) -> IndexedFieldChanges {
101        let old_channel_name = self.channel_name.clone();
102        let old_pending_bridge_target = self.pending_bridge_target.clone();
103        let old_other_leg_uuid = self.other_leg_uuid.clone();
104
105        if let Some(Block::ChannelData { fields, variables }) = &entry.block {
106            for (name, value) in fields {
107                match name.as_str() {
108                    "Channel-Name" => self.channel_name = Some(value.clone()),
109                    "Channel-State" => self.channel_state = Some(value.clone()),
110                    "Call-Direction" => {
111                        self.call_direction = CallDirection::from_str(value).ok();
112                    }
113                    "Caller-Caller-ID-Number" => {
114                        self.caller_id_number = Some(value.clone());
115                    }
116                    "Caller-Caller-ID-Name" => {
117                        self.caller_id_name = Some(value.clone());
118                    }
119                    "Caller-Destination-Number" => {
120                        self.destination_number = Some(value.clone());
121                    }
122                    "Other-Leg-Unique-ID" => {
123                        self.other_leg_uuid = Some(value.clone());
124                    }
125                    _ => {}
126                }
127            }
128            for (name, value) in variables {
129                let var_name = name.strip_prefix("variable_").unwrap_or(name);
130                self.variables.insert(var_name.to_string(), value.clone());
131            }
132        }
133
134        match &entry.message_kind {
135            MessageKind::Dialplan { detail, .. } => {
136                if let Some(dp) = parse_dialplan_context(detail) {
137                    self.initial_context.get_or_insert(dp.context.clone());
138                    self.dialplan_context = Some(dp.context);
139                    self.dialplan_from = Some(dp.from);
140                    self.dialplan_to = Some(dp.to);
141                }
142            }
143            MessageKind::Execute {
144                application,
145                arguments,
146                ..
147            } => match application.as_str() {
148                "set" | "export" => {
149                    if let Some((name, value)) = arguments.split_once('=') {
150                        self.variables.insert(name.to_string(), value.to_string());
151                    }
152                }
153                "bridge" => {
154                    if let Some(info) = parse_bridge_args(arguments) {
155                        if let Some(uuid) = &info.origination_uuid {
156                            self.other_leg_uuid = Some(uuid.clone());
157                        }
158                        self.pending_bridge_target = Some(info.target_channel);
159                    }
160                }
161                _ => {}
162            },
163            MessageKind::Variable { name, value } => {
164                let var_name = name.strip_prefix("variable_").unwrap_or(name);
165                self.variables.insert(var_name.to_string(), value.clone());
166            }
167            MessageKind::ChannelField { name, value } => match name.as_str() {
168                "Channel-Name" => self.channel_name = Some(value.clone()),
169                "Channel-State" => self.channel_state = Some(value.clone()),
170                _ => {}
171            },
172            MessageKind::StateChange { detail } => {
173                if let Some(new_state) = parse_state_change(detail) {
174                    self.channel_state = Some(new_state);
175                }
176            }
177            MessageKind::ChannelLifecycle { detail } => {
178                if let Some(name) = parse_new_channel(detail) {
179                    if self.channel_name.is_none() {
180                        self.channel_name = Some(name);
181                    }
182                }
183                if let Some(cause) = parse_hangup(detail) {
184                    self.hangup_cause = Some(cause);
185                }
186                if is_answered(detail) && self.answered_at.is_none() {
187                    self.answered_at = Some(entry.timestamp.clone());
188                }
189            }
190            _ => {}
191        }
192
193        if entry.message.contains("Processing ") && entry.message.contains(" in context ") {
194            if let Some(dp) = parse_processing_line(&entry.message) {
195                self.initial_context.get_or_insert(dp.context.clone());
196                self.dialplan_context = Some(dp.context);
197                self.dialplan_from = Some(dp.from);
198                self.dialplan_to = Some(dp.to);
199            }
200        }
201
202        for attached in &entry.attached {
203            let parsed = parse_line(attached);
204            self.update_from_message(parsed.message);
205        }
206
207        let mut changes = IndexedFieldChanges::default();
208        if self.channel_name != old_channel_name {
209            changes.channel_name = Some((old_channel_name, self.channel_name.clone()));
210        }
211        if self.pending_bridge_target != old_pending_bridge_target {
212            changes.pending_bridge_target = Some((
213                old_pending_bridge_target,
214                self.pending_bridge_target.clone(),
215            ));
216        }
217        if self.other_leg_uuid != old_other_leg_uuid {
218            changes.other_leg_uuid = Some((old_other_leg_uuid, self.other_leg_uuid.clone()));
219        }
220        changes
221    }
222
223    fn update_from_message(&mut self, msg: &str) {
224        let kind = classify_message(msg);
225        match &kind {
226            MessageKind::Dialplan { detail, .. } => {
227                if let Some(dp) = parse_dialplan_context(detail) {
228                    self.initial_context.get_or_insert(dp.context.clone());
229                    self.dialplan_context = Some(dp.context);
230                    self.dialplan_from = Some(dp.from);
231                    self.dialplan_to = Some(dp.to);
232                }
233            }
234            MessageKind::Variable { name, value } => {
235                let var_name = name.strip_prefix("variable_").unwrap_or(name);
236                self.variables.insert(var_name.to_string(), value.clone());
237            }
238            MessageKind::ChannelField { name, value } => match name.as_str() {
239                "Channel-Name" => self.channel_name = Some(value.clone()),
240                "Channel-State" => self.channel_state = Some(value.clone()),
241                _ => {}
242            },
243            MessageKind::StateChange { detail } => {
244                if let Some(new_state) = parse_state_change(detail) {
245                    self.channel_state = Some(new_state);
246                }
247            }
248            _ => {}
249        }
250    }
251}
252
253struct DialplanContext {
254    from: String,
255    to: String,
256    context: String,
257}
258
259fn parse_dialplan_context(detail: &str) -> Option<DialplanContext> {
260    if !detail.starts_with("parsing [") {
261        return None;
262    }
263    let rest = &detail["parsing [".len()..];
264    let bracket_end = rest.find(']')?;
265    let inner = &rest[..bracket_end];
266
267    let arrow = inner.find("->")?;
268    let from_part = &inner[..arrow];
269    let to_part = &inner[arrow + 2..];
270
271    let context = if rest.len() > bracket_end + 1 {
272        let after = rest[bracket_end + 1..].trim();
273        if let Some(stripped) = after.strip_prefix("continue=") {
274            let _ = stripped;
275        }
276        from_part.to_string()
277    } else {
278        from_part.to_string()
279    };
280
281    Some(DialplanContext {
282        from: from_part.to_string(),
283        to: to_part.to_string(),
284        context,
285    })
286}
287
288fn parse_processing_line(msg: &str) -> Option<DialplanContext> {
289    let proc_idx = msg.find("Processing ")?;
290    let rest = &msg[proc_idx + "Processing ".len()..];
291
292    let arrow = rest.find("->")?;
293    let from = &rest[..arrow];
294
295    let after_arrow = &rest[arrow + 2..];
296    let space = after_arrow.find(' ')?;
297    let to = &after_arrow[..space];
298
299    let ctx_idx = after_arrow.find("in context ")?;
300    let ctx_rest = &after_arrow[ctx_idx + "in context ".len()..];
301    let context = ctx_rest.split_whitespace().next()?;
302
303    Some(DialplanContext {
304        from: from.to_string(),
305        to: to.to_string(),
306        context: context.to_string(),
307    })
308}
309
310fn parse_new_channel(detail: &str) -> Option<String> {
311    let rest = detail.strip_prefix("New Channel ")?;
312    let bracket = rest.rfind(" [")?;
313    Some(rest[..bracket].to_string())
314}
315
316fn parse_state_change(detail: &str) -> Option<String> {
317    let arrow = detail.find(" -> ")?;
318    Some(detail[arrow + 4..].trim().to_string())
319}
320
321fn parse_hangup(detail: &str) -> Option<String> {
322    if !detail.contains("Hangup ") {
323        return None;
324    }
325    let start = detail.rfind('[')?;
326    let end = detail[start..].find(']')?;
327    Some(detail[start + 1..start + end].to_string())
328}
329
330fn is_answered(detail: &str) -> bool {
331    detail.contains("has been answered")
332}
333
334/// Extract `origination_uuid` and the bridge target channel from bridge() arguments.
335/// Uses `BridgeDialString` from freeswitch-types for correct parsing of `[]`, `{}`,
336/// `|` failover, and `,` simultaneous ring syntax.
337fn parse_bridge_args(arguments: &str) -> Option<BridgeInfo> {
338    let dial = BridgeDialString::from_str(arguments).ok()?;
339    let first_ep = dial.groups().first()?.first()?;
340    let origination_uuid = first_ep
341        .variables()
342        .and_then(|v| v.get("origination_uuid"))
343        .map(|s| s.to_string());
344    let mut bare = first_ep.clone();
345    bare.set_variables(None);
346    let target_channel = bare.to_string();
347    Some(BridgeInfo {
348        origination_uuid,
349        target_channel,
350    })
351}
352
353struct BridgeInfo {
354    origination_uuid: Option<String>,
355    target_channel: String,
356}
357
358/// Parse "Originate Resulted in Success: [channel] Peer UUID: uuid"
359fn parse_originate_success(msg: &str) -> Option<String> {
360    let marker = "Peer UUID: ";
361    let idx = msg.find(marker)?;
362    let uuid = msg[idx + marker.len()..].trim();
363    if uuid.is_empty() {
364        None
365    } else {
366        Some(uuid.to_string())
367    }
368}
369
370/// Parse the bracketed channel name from "Originate Resulted in Success: [<chan>] …".
371/// Used as a fallback when the `Peer UUID:` suffix is absent (FS 1.10.5-dev and
372/// similar builds). Returns the channel name borrowed from `msg`.
373fn parse_originate_channel(msg: &str) -> Option<&str> {
374    let start = msg.find(" [")? + 2;
375    let end = msg[start..].find(']')?;
376    let chan = &msg[start..start + end];
377    if chan.is_empty() {
378        None
379    } else {
380        Some(chan)
381    }
382}
383
384/// Terminal channel-/callstate values — sessions left in one of these are
385/// stragglers from prior calls and must not be considered candidates when
386/// disambiguating channel-name collisions in the originate-success fallback.
387///
388/// Covers both `Channel-State` (`CS_*`) and `Callstate` (`HANGUP`). `DOWN` is
389/// excluded because it doubles as the initial Callstate before any change is
390/// observed.
391fn is_terminal_channel_state(state: Option<&str>) -> bool {
392    matches!(
393        state,
394        Some("CS_HANGUP" | "CS_REPORTING" | "CS_DESTROY" | "CS_NONE" | "HANGUP")
395    )
396}
397
398/// A [`LogEntry`] paired with the session's state snapshot at that point in time.
399#[derive(Debug)]
400pub struct EnrichedEntry {
401    pub entry: LogEntry,
402    /// `None` for system lines (entries with an empty UUID).
403    pub session: Option<SessionSnapshot>,
404}
405
406/// Layer 3 per-session state machine — tracks per-UUID state (dialplan context,
407/// channel state, variables) across entries and yields [`EnrichedEntry`] values.
408///
409/// Wraps a [`LogStream`] and maintains a `HashMap<String, SessionState>` keyed by UUID.
410/// Sessions are never automatically cleaned up; call [`remove_session()`](SessionTracker::remove_session)
411/// when a call ends.
412pub struct SessionTracker<I> {
413    inner: LogStream<I>,
414    sessions: HashMap<String, SessionState>,
415    by_channel_name: HashMap<String, HashSet<String>>,
416    by_pending_target: HashMap<String, String>,
417    by_other_leg: HashMap<String, String>,
418    pre_hook: Option<SessionHook>,
419    post_hook: Option<SessionHook>,
420}
421
422impl<I: Iterator<Item = String>> SessionTracker<I> {
423    /// Wrap a [`LogStream`] to add per-session state tracking.
424    pub fn new(inner: LogStream<I>) -> Self {
425        SessionTracker {
426            inner,
427            sessions: HashMap::new(),
428            by_channel_name: HashMap::new(),
429            by_pending_target: HashMap::new(),
430            by_other_leg: HashMap::new(),
431            pre_hook: None,
432            post_hook: None,
433        }
434    }
435
436    /// Register a hook that runs BEFORE built-in field extraction.
437    ///
438    /// Use this to override how specific fields are extracted. Fields set
439    /// by the pre-hook may be preserved by built-in extraction if it uses
440    /// `is_none()` guards.
441    pub fn with_pre_hook<F>(mut self, hook: F) -> Self
442    where
443        F: Fn(&LogEntry, &mut SessionState) + Send + 'static,
444    {
445        self.pre_hook = Some(Box::new(hook));
446        self
447    }
448
449    /// Register a hook that runs AFTER all built-in processing.
450    ///
451    /// Use this for custom field extraction and relationship detection.
452    /// The hook can read fields populated by built-in extraction and
453    /// fill gaps with application-specific patterns (e.g., `uuid_bridge`
454    /// API results, custom SIP headers).
455    ///
456    /// # Example
457    ///
458    /// ```
459    /// use freeswitch_log_parser::{LogStream, SessionTracker, MessageKind};
460    ///
461    /// let stream = LogStream::new(std::iter::empty::<String>());
462    /// let tracker = SessionTracker::new(stream)
463    ///     .with_post_hook(|entry, state| {
464    ///         if let MessageKind::Execute { application, arguments, .. } = &entry.message_kind {
465    ///             if application == "set" && arguments.starts_with("api_result=+OK ") {
466    ///                 // extract UUID and set state.other_leg_uuid
467    ///             }
468    ///         }
469    ///     });
470    /// ```
471    pub fn with_post_hook<F>(mut self, hook: F) -> Self
472    where
473        F: Fn(&LogEntry, &mut SessionState) + Send + 'static,
474    {
475        self.post_hook = Some(Box::new(hook));
476        self
477    }
478
479    /// All currently tracked sessions, keyed by UUID.
480    pub fn sessions(&self) -> &HashMap<String, SessionState> {
481        &self.sessions
482    }
483
484    /// Remove and return a session's accumulated state. Call this when a call ends
485    /// (e.g. `CS_DESTROY` or hangup) to free memory.
486    pub fn remove_session(&mut self, uuid: &str) -> Option<SessionState> {
487        let state = self.sessions.remove(uuid)?;
488        if let Some(chan) = &state.channel_name {
489            if let Some(set) = self.by_channel_name.get_mut(chan) {
490                set.remove(uuid);
491                if set.is_empty() {
492                    self.by_channel_name.remove(chan);
493                }
494            }
495        }
496        if let Some(target) = &state.pending_bridge_target {
497            self.by_pending_target.remove(target);
498        }
499        if let Some(other) = &state.other_leg_uuid {
500            self.by_other_leg.remove(other);
501        }
502        Some(state)
503    }
504
505    /// Delegates to [`LogStream::stats()`].
506    pub fn stats(&self) -> &ParseStats {
507        self.inner.stats()
508    }
509
510    /// Delegates to [`LogStream::drain_unclassified()`].
511    pub fn drain_unclassified(&mut self) -> Vec<UnclassifiedLine> {
512        self.inner.drain_unclassified()
513    }
514
515    fn apply_index_changes(&mut self, uuid: &str, changes: &IndexedFieldChanges) {
516        if let Some((old, new)) = &changes.channel_name {
517            if let Some(old_name) = old {
518                if let Some(set) = self.by_channel_name.get_mut(old_name) {
519                    set.remove(uuid);
520                    if set.is_empty() {
521                        self.by_channel_name.remove(old_name);
522                    }
523                }
524            }
525            if let Some(new_name) = new {
526                self.by_channel_name
527                    .entry(new_name.clone())
528                    .or_default()
529                    .insert(uuid.to_string());
530            }
531        }
532        if let Some((old, new)) = &changes.pending_bridge_target {
533            if let Some(old_target) = old {
534                self.by_pending_target.remove(old_target);
535            }
536            if let Some(new_target) = new {
537                self.by_pending_target
538                    .insert(new_target.clone(), uuid.to_string());
539            }
540        }
541        if let Some((old, new)) = &changes.other_leg_uuid {
542            if let Some(old_leg) = old {
543                self.by_other_leg.remove(old_leg);
544            }
545            if let Some(new_leg) = new {
546                self.by_other_leg.insert(new_leg.clone(), uuid.to_string());
547            }
548        }
549    }
550
551    /// Cross-session leg linking. Called after `update_from_entry` so per-session
552    /// state (bridge target, channel name) is already populated.
553    fn link_legs(&mut self, uuid: &str, entry: &LogEntry) {
554        // 1. "Originate Resulted in Success ... Peer UUID: BLEG" — authoritative
555        if entry.message.contains("Originate Resulted in Success") {
556            let a_uuid = uuid.to_string();
557            if let Some(peer_uuid) = parse_originate_success(&entry.message) {
558                let a_old_pending = self
559                    .sessions
560                    .get(&a_uuid)
561                    .and_then(|s| s.pending_bridge_target.clone());
562
563                if let Some(a_state) = self.sessions.get_mut(&a_uuid) {
564                    a_state.other_leg_uuid = Some(peer_uuid.clone());
565                    a_state.pending_bridge_target = None;
566                }
567                self.by_other_leg.insert(peer_uuid.clone(), a_uuid.clone());
568                if let Some(old_target) = a_old_pending {
569                    self.by_pending_target.remove(&old_target);
570                }
571
572                let b_state = self.sessions.entry(peer_uuid.clone()).or_default();
573                b_state.other_leg_uuid = Some(a_uuid.clone());
574                self.by_other_leg.insert(a_uuid, peer_uuid);
575            } else if let Some(chan) = parse_originate_channel(&entry.message) {
576                // Fallback for FS builds without `Peer UUID:` suffix (e.g. 1.10.5-dev):
577                // link via unique non-terminated b-leg session whose channel_name
578                // matches. Candidates in terminal states are stragglers; if zero or
579                // multiple live candidates remain, skip (correctness over coverage).
580                let candidates: Vec<String> = self
581                    .by_channel_name
582                    .get(chan)
583                    .map(|set| {
584                        set.iter()
585                            .filter(|u| *u != &a_uuid)
586                            .filter(|u| {
587                                self.sessions
588                                    .get(*u)
589                                    .map(|s| !is_terminal_channel_state(s.channel_state.as_deref()))
590                                    .unwrap_or(false)
591                            })
592                            .cloned()
593                            .collect()
594                    })
595                    .unwrap_or_default();
596
597                if candidates.len() == 1 {
598                    let b_uuid = candidates.into_iter().next().unwrap();
599                    let a_old_pending = self
600                        .sessions
601                        .get(&a_uuid)
602                        .and_then(|s| s.pending_bridge_target.clone());
603
604                    if let Some(a_state) = self.sessions.get_mut(&a_uuid) {
605                        a_state.other_leg_uuid = Some(b_uuid.clone());
606                        a_state.pending_bridge_target = None;
607                    }
608                    if let Some(b_state) = self.sessions.get_mut(&b_uuid) {
609                        b_state.other_leg_uuid = Some(a_uuid.clone());
610                    }
611
612                    self.by_other_leg.insert(b_uuid.clone(), a_uuid.clone());
613                    self.by_other_leg.insert(a_uuid, b_uuid);
614                    if let Some(old_target) = a_old_pending {
615                        self.by_pending_target.remove(&old_target);
616                    }
617                }
618            }
619            return;
620        }
621
622        // 2. New Channel on this UUID — check if any other session has a pending bridge
623        //    with origination_uuid matching this UUID, or target matching this channel name.
624        if let MessageKind::ChannelLifecycle { detail } = &entry.message_kind {
625            if let Some(channel_name) = parse_new_channel(detail) {
626                let b_uuid = uuid.to_string();
627
628                // O(1) index lookups instead of full scan
629                let a_uuid_found = self
630                    .by_other_leg
631                    .get(&b_uuid)
632                    .cloned()
633                    .or_else(|| self.by_pending_target.get(&channel_name).cloned())
634                    .filter(|a| a != &b_uuid);
635
636                if let Some(a_uuid) = a_uuid_found {
637                    let a_old_pending = self
638                        .sessions
639                        .get(&a_uuid)
640                        .and_then(|s| s.pending_bridge_target.clone());
641
642                    if let Some(a_state) = self.sessions.get_mut(&a_uuid) {
643                        a_state.other_leg_uuid = Some(b_uuid.clone());
644                        a_state.pending_bridge_target = None;
645                    }
646                    if let Some(b_state) = self.sessions.get_mut(&b_uuid) {
647                        b_state.other_leg_uuid = Some(a_uuid.clone());
648                    }
649
650                    self.by_other_leg.insert(b_uuid.clone(), a_uuid.clone());
651                    self.by_other_leg.insert(a_uuid, b_uuid);
652                    if let Some(old_target) = a_old_pending {
653                        self.by_pending_target.remove(&old_target);
654                    }
655                }
656            }
657        }
658    }
659}
660
661impl<I: Iterator<Item = String>> Iterator for SessionTracker<I> {
662    type Item = EnrichedEntry;
663
664    fn next(&mut self) -> Option<EnrichedEntry> {
665        let entry = self.inner.next()?;
666
667        if entry.uuid.is_empty() {
668            return Some(EnrichedEntry {
669                entry,
670                session: None,
671            });
672        }
673
674        let uuid = entry.uuid.clone();
675        self.sessions.entry(uuid.clone()).or_default();
676
677        if let Some(hook) = &self.pre_hook {
678            let state = self.sessions.get_mut(&uuid).unwrap();
679            hook(&entry, state);
680        }
681
682        let changes = self
683            .sessions
684            .get_mut(&uuid)
685            .unwrap()
686            .update_from_entry(&entry);
687        self.apply_index_changes(&uuid, &changes);
688
689        self.link_legs(&uuid, &entry);
690
691        if let Some(hook) = &self.post_hook {
692            let state = self.sessions.get_mut(&uuid).unwrap();
693            hook(&entry, state);
694        }
695
696        let snapshot = self.sessions.get(&uuid).unwrap().snapshot();
697
698        Some(EnrichedEntry {
699            entry,
700            session: Some(snapshot),
701        })
702    }
703}
704
705#[cfg(test)]
706mod tests {
707    use super::*;
708
709    const UUID1: &str = "a1b2c3d4-e5f6-7890-abcd-ef1234567890";
710    const UUID2: &str = "b2c3d4e5-f6a7-8901-bcde-f12345678901";
711    const UUID3: &str = "c3d4e5f6-a7b8-9012-cdef-234567890123";
712    const TS1: &str = "2025-01-15 10:30:45.123456";
713    const TS2: &str = "2025-01-15 10:30:46.234567";
714
715    fn full_line(uuid: &str, ts: &str, msg: &str) -> String {
716        format!("{uuid} {ts} 95.97% [DEBUG] sofia.c:100 {msg}")
717    }
718
719    fn collect_enriched(lines: Vec<String>) -> Vec<EnrichedEntry> {
720        let stream = LogStream::new(lines.into_iter());
721        SessionTracker::new(stream).collect()
722    }
723
724    #[test]
725    fn system_line_no_session() {
726        let lines = vec![format!(
727            "{TS1} 95.97% [INFO] mod_event_socket.c:1772 Event Socket command"
728        )];
729        let entries = collect_enriched(lines);
730        assert_eq!(entries.len(), 1);
731        assert!(entries[0].session.is_none());
732    }
733
734    #[test]
735    fn dialplan_context_propagation() {
736        let lines = vec![
737            full_line(UUID1, TS1, "CHANNEL_DATA:"),
738            format!("{UUID1} Channel-Name: [sofia/internal/+15550001234@192.0.2.1]"),
739            format!("{UUID1} EXECUTE [depth=0] sofia/internal/+15550001234@192.0.2.1 answer"),
740            format!("{UUID1} Dialplan: sofia/internal/+15550001234@192.0.2.1 parsing [public->global] continue=true"),
741            full_line(UUID1, TS2, "Some later event"),
742        ];
743        let entries = collect_enriched(lines);
744        let last = entries.last().unwrap();
745        let session = last.session.as_ref().unwrap();
746        assert_eq!(session.dialplan_context.as_deref(), Some("public"));
747        assert_eq!(session.dialplan_from.as_deref(), Some("public"));
748        assert_eq!(session.dialplan_to.as_deref(), Some("global"));
749    }
750
751    #[test]
752    fn processing_line_extracts_context() {
753        let lines = vec![full_line(
754            UUID1,
755            TS1,
756            "Processing 5551234567->5559876543 in context public",
757        )];
758        let entries = collect_enriched(lines);
759        let session = entries[0].session.as_ref().unwrap();
760        assert_eq!(session.dialplan_context.as_deref(), Some("public"));
761        assert_eq!(session.dialplan_from.as_deref(), Some("5551234567"));
762        assert_eq!(session.dialplan_to.as_deref(), Some("5559876543"));
763    }
764
765    #[test]
766    fn initial_context_preserved_across_transfers() {
767        let lines = vec![
768            full_line(
769                UUID1,
770                TS1,
771                "Processing 5551234567->5559876543 in context public",
772            ),
773            full_line(
774                UUID1,
775                TS2,
776                "Processing 5551234567->start_recording in context recordings",
777            ),
778        ];
779        let stream = LogStream::new(lines.into_iter());
780        let mut tracker = SessionTracker::new(stream);
781        let entries: Vec<_> = tracker.by_ref().collect();
782
783        let first = entries[0].session.as_ref().unwrap();
784        assert_eq!(
785            first.initial_context.as_deref(),
786            Some("public"),
787            "initial_context set on first Processing line"
788        );
789        assert_eq!(first.dialplan_context.as_deref(), Some("public"));
790
791        let state = tracker.sessions().get(UUID1).unwrap();
792        assert_eq!(
793            state.initial_context.as_deref(),
794            Some("public"),
795            "initial_context keeps the first context seen"
796        );
797        assert_eq!(
798            state.dialplan_context.as_deref(),
799            Some("recordings"),
800            "dialplan_context tracks the current context"
801        );
802        assert_eq!(state.dialplan_to.as_deref(), Some("start_recording"));
803    }
804
805    #[test]
806    fn new_channel_sets_channel_name() {
807        let lines = vec![full_line(
808            UUID1,
809            TS1,
810            "New Channel sofia/internal-v4/sos [a1b2c3d4-e5f6-7890-abcd-ef1234567890]",
811        )];
812        let entries = collect_enriched(lines);
813        let session = entries[0].session.as_ref().unwrap();
814        assert_eq!(
815            session.channel_name.as_deref(),
816            Some("sofia/internal-v4/sos")
817        );
818    }
819
820    #[test]
821    fn originate_success_links_both_legs() {
822        // "Originate Resulted in Success" contains both the A-leg UUID (line prefix)
823        // and B-leg UUID (Peer UUID field). Both legs should learn about each other.
824        let lines = vec![
825            full_line(UUID2, TS1, "New Channel sofia/esinet1-v6-tcp/sip:target.example.com [b2c3d4e5-f6a7-8901-bcde-f12345678901]"),
826            full_line(UUID1, TS2, "Originate Resulted in Success: [sofia/esinet1-v6-tcp/sip:target.example.com] Peer UUID: b2c3d4e5-f6a7-8901-bcde-f12345678901"),
827        ];
828        let stream = LogStream::new(lines.into_iter());
829        let mut tracker = SessionTracker::new(stream);
830        let _: Vec<_> = tracker.by_ref().collect();
831
832        let a_leg = tracker.sessions().get(UUID1).unwrap();
833        assert_eq!(
834            a_leg.other_leg_uuid.as_deref(),
835            Some(UUID2),
836            "A-leg other_leg_uuid set from Originate Resulted in Success"
837        );
838
839        let b_leg = tracker.sessions().get(UUID2).unwrap();
840        assert_eq!(
841            b_leg.other_leg_uuid.as_deref(),
842            Some(UUID1),
843            "B-leg other_leg_uuid points back to A-leg"
844        );
845    }
846
847    #[test]
848    fn originate_success_channel_fallback_links_legs() {
849        // FS 1.10.5-dev and similar omit `Peer UUID:` from "Originate Resulted in Success".
850        // The b-leg's New Channel populates channel_name 3.5 s before originate; the
851        // fallback path matches by channel name when the Peer UUID is absent.
852        let lines = vec![
853            full_line(
854                UUID2,
855                TS1,
856                "New Channel sofia/internal/6244@192.0.2.72:50744 [b2c3d4e5-f6a7-8901-bcde-f12345678901]",
857            ),
858            full_line(
859                UUID1,
860                TS2,
861                "Originate Resulted in Success: [sofia/internal/6244@192.0.2.72:50744]",
862            ),
863        ];
864        let stream = LogStream::new(lines.into_iter());
865        let mut tracker = SessionTracker::new(stream);
866        let _: Vec<_> = tracker.by_ref().collect();
867
868        let a_leg = tracker.sessions().get(UUID1).unwrap();
869        assert_eq!(
870            a_leg.other_leg_uuid.as_deref(),
871            Some(UUID2),
872            "A-leg linked to B-leg via channel-name fallback when Peer UUID absent"
873        );
874
875        let b_leg = tracker.sessions().get(UUID2).unwrap();
876        assert_eq!(
877            b_leg.other_leg_uuid.as_deref(),
878            Some(UUID1),
879            "B-leg linked back to A-leg"
880        );
881    }
882
883    #[test]
884    fn originate_success_peer_uuid_wins_over_channel_fallback() {
885        // When Peer UUID is present, channel-name fallback must not fire — even if
886        // another session shares the channel name. Peer UUID is authoritative.
887        let lines = vec![
888            full_line(
889                UUID2,
890                TS1,
891                "New Channel sofia/internal/6244@192.0.2.72:50744 [b2c3d4e5-f6a7-8901-bcde-f12345678901]",
892            ),
893            full_line(
894                UUID3,
895                TS1,
896                "New Channel sofia/internal/6244@192.0.2.72:50744 [c3d4e5f6-a7b8-9012-cdef-234567890123]",
897            ),
898            full_line(
899                UUID1,
900                TS2,
901                "Originate Resulted in Success: [sofia/internal/6244@192.0.2.72:50744] Peer UUID: b2c3d4e5-f6a7-8901-bcde-f12345678901",
902            ),
903        ];
904        let stream = LogStream::new(lines.into_iter());
905        let mut tracker = SessionTracker::new(stream);
906        let _: Vec<_> = tracker.by_ref().collect();
907
908        let a_leg = tracker.sessions().get(UUID1).unwrap();
909        assert_eq!(
910            a_leg.other_leg_uuid.as_deref(),
911            Some(UUID2),
912            "Peer UUID wins over channel-name match"
913        );
914
915        let decoy = tracker.sessions().get(UUID3).unwrap();
916        assert_eq!(
917            decoy.other_leg_uuid, None,
918            "Decoy session sharing channel name is not touched"
919        );
920    }
921
922    #[test]
923    fn originate_success_channel_fallback_skips_when_ambiguous() {
924        // Two b-leg candidates share the same channel name. The fallback must not
925        // guess — correctness over coverage.
926        let lines = vec![
927            full_line(
928                UUID2,
929                TS1,
930                "New Channel sofia/internal/6244@192.0.2.72:50744 [b2c3d4e5-f6a7-8901-bcde-f12345678901]",
931            ),
932            full_line(
933                UUID3,
934                TS1,
935                "New Channel sofia/internal/6244@192.0.2.72:50744 [c3d4e5f6-a7b8-9012-cdef-234567890123]",
936            ),
937            full_line(
938                UUID1,
939                TS2,
940                "Originate Resulted in Success: [sofia/internal/6244@192.0.2.72:50744]",
941            ),
942        ];
943        let stream = LogStream::new(lines.into_iter());
944        let mut tracker = SessionTracker::new(stream);
945        let _: Vec<_> = tracker.by_ref().collect();
946
947        let a_leg = tracker.sessions().get(UUID1).unwrap();
948        assert_eq!(
949            a_leg.other_leg_uuid, None,
950            "Ambiguous channel name yields no link"
951        );
952        assert_eq!(tracker.sessions().get(UUID2).unwrap().other_leg_uuid, None);
953        assert_eq!(tracker.sessions().get(UUID3).unwrap().other_leg_uuid, None);
954    }
955
956    #[test]
957    fn originate_success_channel_fallback_skips_terminated_candidates() {
958        // Two b-leg sessions share the same channel_name, but one is in
959        // CS_DESTROY (stale prior call on the same registered phone). The
960        // liveness filter must drop the terminated candidate so the live one
961        // becomes the unambiguous match.
962        let lines = vec![
963            full_line(
964                UUID2,
965                TS1,
966                "New Channel sofia/internal/6244@192.0.2.72:50744 [b2c3d4e5-f6a7-8901-bcde-f12345678901]",
967            ),
968            full_line(
969                UUID2,
970                TS1,
971                "(sofia/internal/6244@192.0.2.72:50744) State Change CS_EXECUTE -> CS_DESTROY",
972            ),
973            full_line(
974                UUID3,
975                TS1,
976                "New Channel sofia/internal/6244@192.0.2.72:50744 [c3d4e5f6-a7b8-9012-cdef-234567890123]",
977            ),
978            full_line(
979                UUID1,
980                TS2,
981                "Originate Resulted in Success: [sofia/internal/6244@192.0.2.72:50744]",
982            ),
983        ];
984        let stream = LogStream::new(lines.into_iter());
985        let mut tracker = SessionTracker::new(stream);
986        let _: Vec<_> = tracker.by_ref().collect();
987
988        let a_leg = tracker.sessions().get(UUID1).unwrap();
989        assert_eq!(
990            a_leg.other_leg_uuid.as_deref(),
991            Some(UUID3),
992            "Live b-leg wins over CS_DESTROY straggler"
993        );
994
995        let live_b = tracker.sessions().get(UUID3).unwrap();
996        assert_eq!(
997            live_b.other_leg_uuid.as_deref(),
998            Some(UUID1),
999            "Live b-leg points back to a-leg"
1000        );
1001
1002        let stale_b = tracker.sessions().get(UUID2).unwrap();
1003        assert_eq!(
1004            stale_b.other_leg_uuid, None,
1005            "Terminated b-leg is not touched"
1006        );
1007    }
1008
1009    #[test]
1010    fn originate_success_channel_fallback_skips_when_no_match() {
1011        // a-leg fires Originate with a bracketed channel name no session has.
1012        // Must not panic, must not create a spurious link.
1013        let lines = vec![full_line(
1014            UUID1,
1015            TS2,
1016            "Originate Resulted in Success: [sofia/internal/6244@192.0.2.72:50744]",
1017        )];
1018        let stream = LogStream::new(lines.into_iter());
1019        let mut tracker = SessionTracker::new(stream);
1020        let _: Vec<_> = tracker.by_ref().collect();
1021
1022        let a_leg = tracker.sessions().get(UUID1).unwrap();
1023        assert_eq!(a_leg.other_leg_uuid, None);
1024        assert_eq!(a_leg.pending_bridge_target, None);
1025    }
1026
1027    #[test]
1028    fn bridge_origination_uuid_links_a_leg_immediately() {
1029        // bridge([origination_uuid=BLEG_UUID,...]) guarantees B-leg UUID from execute args alone.
1030        // A-leg knows B-leg immediately, B-leg learns A-leg when New Channel appears.
1031        let lines = vec![
1032            full_line(UUID1, TS1, "EXECUTE [depth=0] sofia/internal-v6/1232@[2001:db8::10] bridge([origination_uuid=b2c3d4e5-f6a7-8901-bcde-f12345678901,leg_timeout=2]sofia/esinet1-v6-tcp/sip:target.example.com)"),
1033            full_line(UUID2, TS1, "New Channel sofia/esinet1-v6-tcp/sip:target.example.com [b2c3d4e5-f6a7-8901-bcde-f12345678901]"),
1034        ];
1035        let stream = LogStream::new(lines.into_iter());
1036        let mut tracker = SessionTracker::new(stream);
1037        let _: Vec<_> = tracker.by_ref().collect();
1038
1039        let a_leg = tracker.sessions().get(UUID1).unwrap();
1040        assert_eq!(
1041            a_leg.other_leg_uuid.as_deref(),
1042            Some(UUID2),
1043            "A-leg knows B-leg UUID from origination_uuid in bridge args"
1044        );
1045
1046        let b_leg = tracker.sessions().get(UUID2).unwrap();
1047        assert_eq!(
1048            b_leg.other_leg_uuid.as_deref(),
1049            Some(UUID1),
1050            "B-leg knows A-leg once New Channel correlates"
1051        );
1052    }
1053
1054    #[test]
1055    fn bridge_target_matches_new_channel() {
1056        // bridge() without origination_uuid — B-leg UUID is auto-generated by FS.
1057        // Match via bridge target channel matching next New Channel with same target.
1058        let lines = vec![
1059            full_line(UUID1, TS1, "EXECUTE [depth=0] sofia/internal/+15550001234@192.0.2.1 bridge(sofia/gateway/carrier/+15559876543)"),
1060            full_line(UUID1, TS1, "Parsing session specific variables"),
1061            full_line(UUID2, TS1, "New Channel sofia/gateway/carrier/+15559876543 [b2c3d4e5-f6a7-8901-bcde-f12345678901]"),
1062        ];
1063        let stream = LogStream::new(lines.into_iter());
1064        let mut tracker = SessionTracker::new(stream);
1065        let _: Vec<_> = tracker.by_ref().collect();
1066
1067        let a_leg = tracker.sessions().get(UUID1).unwrap();
1068        assert_eq!(
1069            a_leg.other_leg_uuid.as_deref(),
1070            Some(UUID2),
1071            "A-leg linked to B-leg via bridge target matching New Channel"
1072        );
1073
1074        let b_leg = tracker.sessions().get(UUID2).unwrap();
1075        assert_eq!(
1076            b_leg.other_leg_uuid.as_deref(),
1077            Some(UUID1),
1078            "B-leg linked back to A-leg"
1079        );
1080    }
1081
1082    #[test]
1083    fn originate_success_corrects_wrong_target_match() {
1084        // Bridge target matching guessed UUID2 as B-leg, but originate success reveals
1085        // the actual B-leg is UUID3. The authoritative success message must override.
1086        let lines = vec![
1087            full_line(UUID1, TS1, "EXECUTE [depth=0] sofia/internal/+15550001234@192.0.2.1 bridge(sofia/gateway/carrier/+15559876543)"),
1088            full_line(UUID2, TS1, "New Channel sofia/gateway/carrier/+15559876543 [b2c3d4e5-f6a7-8901-bcde-f12345678901]"),
1089            full_line(UUID1, TS2, "Originate Resulted in Success: [sofia/gateway/carrier/+15559876543] Peer UUID: c3d4e5f6-a7b8-9012-cdef-234567890123"),
1090        ];
1091        let stream = LogStream::new(lines.into_iter());
1092        let mut tracker = SessionTracker::new(stream);
1093        let _: Vec<_> = tracker.by_ref().collect();
1094
1095        let a_leg = tracker.sessions().get(UUID1).unwrap();
1096        assert_eq!(
1097            a_leg.other_leg_uuid.as_deref(),
1098            Some(UUID3),
1099            "Originate success overrides earlier target-match guess"
1100        );
1101
1102        let real_b_leg = tracker.sessions().get(UUID3).unwrap();
1103        assert_eq!(
1104            real_b_leg.other_leg_uuid.as_deref(),
1105            Some(UUID1),
1106            "Real B-leg points back to A-leg"
1107        );
1108    }
1109
1110    #[test]
1111    fn channel_data_other_leg_uuid() {
1112        // Other-Leg-Unique-ID in CHANNEL_DATA (post-bridge info dump) sets other_leg_uuid
1113        let lines = vec![
1114            full_line(UUID1, TS1, "CHANNEL_DATA:"),
1115            format!("{UUID1} Other-Leg-Unique-ID: [{UUID2}]"),
1116        ];
1117        let stream = LogStream::new(lines.into_iter());
1118        let mut tracker = SessionTracker::new(stream);
1119        let _: Vec<_> = tracker.by_ref().collect();
1120
1121        let state = tracker.sessions().get(UUID1).unwrap();
1122        assert_eq!(
1123            state.other_leg_uuid.as_deref(),
1124            Some(UUID2),
1125            "other_leg_uuid set from Other-Leg-Unique-ID CHANNEL_DATA field"
1126        );
1127    }
1128
1129    #[test]
1130    fn channel_data_populates_session() {
1131        let lines = vec![
1132            full_line(UUID1, TS1, "CHANNEL_DATA:"),
1133            format!("{UUID1} Channel-Name: [sofia/internal/+15550001234@192.0.2.1]"),
1134            format!("{UUID1} Channel-State: [CS_EXECUTE]"),
1135            "variable_sip_call_id: [test123@192.0.2.1]".to_string(),
1136            "variable_direction: [inbound]".to_string(),
1137        ];
1138        let entries = collect_enriched(lines);
1139        assert_eq!(entries.len(), 1);
1140        let session = entries[0].session.as_ref().unwrap();
1141        assert_eq!(
1142            session.channel_name.as_deref(),
1143            Some("sofia/internal/+15550001234@192.0.2.1")
1144        );
1145        assert_eq!(session.channel_state.as_deref(), Some("CS_EXECUTE"));
1146    }
1147
1148    #[test]
1149    fn variables_learned_from_channel_data() {
1150        let lines = vec![
1151            full_line(UUID1, TS1, "CHANNEL_DATA:"),
1152            "variable_sip_call_id: [test123@192.0.2.1]".to_string(),
1153            "variable_direction: [inbound]".to_string(),
1154        ];
1155        let stream = LogStream::new(lines.into_iter());
1156        let mut tracker = SessionTracker::new(stream);
1157        let _: Vec<_> = tracker.by_ref().collect();
1158        let state = tracker.sessions().get(UUID1).unwrap();
1159        assert_eq!(
1160            state.variables.get("sip_call_id").map(|s| s.as_str()),
1161            Some("test123@192.0.2.1")
1162        );
1163        assert_eq!(
1164            state.variables.get("direction").map(|s| s.as_str()),
1165            Some("inbound")
1166        );
1167    }
1168
1169    #[test]
1170    fn variables_learned_from_set_execute() {
1171        let lines = vec![
1172            full_line(UUID1, TS1, "First"),
1173            format!("{UUID1} EXECUTE [depth=0] sofia/internal/+15550001234@192.0.2.1 set(call_direction=inbound)"),
1174            full_line(UUID1, TS2, "After set"),
1175        ];
1176        let stream = LogStream::new(lines.into_iter());
1177        let mut tracker = SessionTracker::new(stream);
1178        let entries: Vec<_> = tracker.by_ref().collect();
1179        assert_eq!(entries.len(), 3);
1180        let state = tracker.sessions().get(UUID1).unwrap();
1181        assert_eq!(
1182            state.variables.get("call_direction").map(|s| s.as_str()),
1183            Some("inbound")
1184        );
1185    }
1186
1187    #[test]
1188    fn variables_learned_from_export_execute() {
1189        let lines = vec![
1190            full_line(UUID1, TS1, "First"),
1191            format!("{UUID1} EXECUTE [depth=0] sofia/internal/+15550001234@192.0.2.1 export(originate_timeout=3600)"),
1192        ];
1193        let stream = LogStream::new(lines.into_iter());
1194        let mut tracker = SessionTracker::new(stream);
1195        let _: Vec<_> = tracker.by_ref().collect();
1196        let state = tracker.sessions().get(UUID1).unwrap();
1197        assert_eq!(
1198            state.variables.get("originate_timeout").map(|s| s.as_str()),
1199            Some("3600")
1200        );
1201    }
1202
1203    #[test]
1204    fn session_isolation_between_uuids() {
1205        let lines = vec![
1206            full_line(
1207                UUID1,
1208                TS1,
1209                "Processing 5551111111->5552222222 in context public",
1210            ),
1211            full_line(
1212                UUID2,
1213                TS2,
1214                "Processing 5553333333->5554444444 in context private",
1215            ),
1216        ];
1217        let stream = LogStream::new(lines.into_iter());
1218        let mut tracker = SessionTracker::new(stream);
1219        let _: Vec<_> = tracker.by_ref().collect();
1220        let s1 = tracker.sessions().get(UUID1).unwrap();
1221        let s2 = tracker.sessions().get(UUID2).unwrap();
1222        assert_eq!(s1.dialplan_context.as_deref(), Some("public"));
1223        assert_eq!(s2.dialplan_context.as_deref(), Some("private"));
1224        assert_eq!(s1.dialplan_from.as_deref(), Some("5551111111"));
1225        assert_eq!(s2.dialplan_from.as_deref(), Some("5553333333"));
1226    }
1227
1228    #[test]
1229    fn processing_line_with_regex_type_and_angle_bracket_caller() {
1230        let lines = vec![full_line(
1231            UUID1,
1232            TS1,
1233            "Processing Emergency S R <5550001234>->start_recording in context recordings",
1234        )];
1235        let entries = collect_enriched(lines);
1236        let session = entries[0].session.as_ref().unwrap();
1237        assert_eq!(session.initial_context.as_deref(), Some("recordings"));
1238        assert_eq!(session.dialplan_context.as_deref(), Some("recordings"));
1239        assert_eq!(
1240            session.dialplan_from.as_deref(),
1241            Some("Emergency S R <5550001234>")
1242        );
1243        assert_eq!(session.dialplan_to.as_deref(), Some("start_recording"));
1244    }
1245
1246    #[test]
1247    fn processing_line_extension_format() {
1248        let lines = vec![full_line(
1249            UUID1,
1250            TS1,
1251            "Processing Extension 1263 <1263>->start_recording in context recordings",
1252        )];
1253        let entries = collect_enriched(lines);
1254        let session = entries[0].session.as_ref().unwrap();
1255        assert_eq!(session.initial_context.as_deref(), Some("recordings"));
1256        assert_eq!(
1257            session.dialplan_from.as_deref(),
1258            Some("Extension 1263 <1263>")
1259        );
1260        assert_eq!(session.dialplan_to.as_deref(), Some("start_recording"));
1261    }
1262
1263    #[test]
1264    fn state_change_updates_channel_state() {
1265        let lines = vec![full_line(UUID1, TS1, "State Change CS_INIT -> CS_ROUTING")];
1266        let entries = collect_enriched(lines);
1267        let session = entries[0].session.as_ref().unwrap();
1268        assert_eq!(session.channel_state.as_deref(), Some("CS_ROUTING"));
1269    }
1270
1271    #[test]
1272    fn callstate_change_updates_channel_state() {
1273        let lines = vec![full_line(
1274            UUID1,
1275            TS1,
1276            "(sofia/internal-v4/sos) Callstate Change DOWN -> RINGING",
1277        )];
1278        let entries = collect_enriched(lines);
1279        let session = entries[0].session.as_ref().unwrap();
1280        assert_eq!(session.channel_state.as_deref(), Some("RINGING"));
1281    }
1282
1283    #[test]
1284    fn state_change_overrides_callstate() {
1285        let lines = vec![
1286            full_line(
1287                UUID1,
1288                TS1,
1289                "(sofia/internal-v4/sos) Callstate Change DOWN -> RINGING",
1290            ),
1291            full_line(
1292                UUID1,
1293                TS2,
1294                "(sofia/internal-v4/sos) State Change CS_CONSUME_MEDIA -> CS_EXCHANGE_MEDIA",
1295            ),
1296        ];
1297        let entries = collect_enriched(lines);
1298        assert_eq!(
1299            entries[0]
1300                .session
1301                .as_ref()
1302                .unwrap()
1303                .channel_state
1304                .as_deref(),
1305            Some("RINGING")
1306        );
1307        assert_eq!(
1308            entries[1]
1309                .session
1310                .as_ref()
1311                .unwrap()
1312                .channel_state
1313                .as_deref(),
1314            Some("CS_EXCHANGE_MEDIA")
1315        );
1316    }
1317
1318    #[test]
1319    fn bleg_lifecycle_extracts_data_from_processing() {
1320        let lines = vec![
1321            full_line(
1322                UUID1,
1323                TS1,
1324                "New Channel sofia/internal-v4/sos [a1b2c3d4-e5f6-7890-abcd-ef1234567890]",
1325            ),
1326            full_line(
1327                UUID1,
1328                TS1,
1329                "(sofia/internal-v4/sos) State Change CS_NEW -> CS_INIT",
1330            ),
1331            full_line(
1332                UUID1,
1333                TS1,
1334                "(sofia/internal-v4/sos) State Change CS_INIT -> CS_ROUTING",
1335            ),
1336            full_line(
1337                UUID1,
1338                TS1,
1339                "(sofia/internal-v4/sos) State Change CS_ROUTING -> CS_CONSUME_MEDIA",
1340            ),
1341            full_line(
1342                UUID1,
1343                TS1,
1344                "(sofia/internal-v4/sos) Callstate Change DOWN -> RINGING",
1345            ),
1346            full_line(
1347                UUID1,
1348                TS2,
1349                "(sofia/internal-v4/sos) State Change CS_CONSUME_MEDIA -> CS_EXCHANGE_MEDIA",
1350            ),
1351            full_line(
1352                UUID1,
1353                TS2,
1354                "Processing Emergency S R <5550001234>->start_recording in context recordings",
1355            ),
1356            full_line(
1357                UUID1,
1358                TS2,
1359                "(sofia/internal-v4/sos) State Change CS_EXCHANGE_MEDIA -> CS_HANGUP",
1360            ),
1361        ];
1362        let entries = collect_enriched(lines);
1363
1364        let after_ringing = entries[4].session.as_ref().unwrap();
1365        assert_eq!(after_ringing.channel_state.as_deref(), Some("RINGING"));
1366        assert!(after_ringing.initial_context.is_none());
1367
1368        let after_processing = entries[6].session.as_ref().unwrap();
1369        assert_eq!(
1370            after_processing.channel_state.as_deref(),
1371            Some("CS_EXCHANGE_MEDIA")
1372        );
1373        assert_eq!(
1374            after_processing.initial_context.as_deref(),
1375            Some("recordings")
1376        );
1377        assert_eq!(
1378            after_processing.dialplan_from.as_deref(),
1379            Some("Emergency S R <5550001234>")
1380        );
1381        assert_eq!(
1382            after_processing.dialplan_to.as_deref(),
1383            Some("start_recording")
1384        );
1385
1386        let after_hangup = entries[7].session.as_ref().unwrap();
1387        assert_eq!(after_hangup.channel_state.as_deref(), Some("CS_HANGUP"));
1388        assert_eq!(after_hangup.initial_context.as_deref(), Some("recordings"));
1389    }
1390
1391    #[test]
1392    fn channel_name_from_new_channel() {
1393        let lines = vec![full_line(
1394            UUID1,
1395            TS1,
1396            "New Channel sofia/internal-v4/sos [a1b2c3d4-e5f6-7890-abcd-ef1234567890]",
1397        )];
1398        let entries = collect_enriched(lines);
1399        let session = entries[0].session.as_ref().unwrap();
1400        assert_eq!(
1401            session.channel_name.as_deref(),
1402            Some("sofia/internal-v4/sos")
1403        );
1404    }
1405
1406    #[test]
1407    fn remove_session() {
1408        let lines = vec![full_line(
1409            UUID1,
1410            TS1,
1411            "Processing 5551111111->5552222222 in context public",
1412        )];
1413        let stream = LogStream::new(lines.into_iter());
1414        let mut tracker = SessionTracker::new(stream);
1415        let _: Vec<_> = tracker.by_ref().collect();
1416        assert!(tracker.sessions().contains_key(UUID1));
1417        let removed = tracker.remove_session(UUID1).unwrap();
1418        assert_eq!(removed.dialplan_context.as_deref(), Some("public"));
1419        assert!(!tracker.sessions().contains_key(UUID1));
1420    }
1421
1422    #[test]
1423    fn stats_delegation() {
1424        let lines = vec![
1425            full_line(UUID1, TS1, "First"),
1426            full_line(UUID1, TS2, "Second"),
1427        ];
1428        let stream = LogStream::new(lines.into_iter());
1429        let mut tracker = SessionTracker::new(stream);
1430        let _: Vec<_> = tracker.by_ref().collect();
1431        assert_eq!(tracker.stats().lines_processed, 2);
1432    }
1433
1434    #[test]
1435    fn snapshot_reflects_cumulative_state() {
1436        let lines = vec![
1437            full_line(UUID1, TS1, "CHANNEL_DATA:"),
1438            format!("{UUID1} Channel-Name: [sofia/internal/+15550001234@192.0.2.1]"),
1439            format!("{UUID1} EXECUTE [depth=0] sofia/internal/+15550001234@192.0.2.1 set(foo=bar)"),
1440            full_line(
1441                UUID1,
1442                TS2,
1443                "Processing 5551111111->5552222222 in context public",
1444            ),
1445        ];
1446        let entries = collect_enriched(lines);
1447        assert_eq!(entries.len(), 3);
1448        let first = entries[0].session.as_ref().unwrap();
1449        assert_eq!(
1450            first.channel_name.as_deref(),
1451            Some("sofia/internal/+15550001234@192.0.2.1"),
1452        );
1453        assert!(first.dialplan_context.is_none());
1454
1455        let last = entries[2].session.as_ref().unwrap();
1456        assert_eq!(
1457            last.channel_name.as_deref(),
1458            Some("sofia/internal/+15550001234@192.0.2.1"),
1459        );
1460        assert_eq!(last.dialplan_context.as_deref(), Some("public"));
1461    }
1462
1463    #[test]
1464    fn post_hook_sets_other_leg_uuid() {
1465        let lines = vec![
1466            full_line(UUID1, TS1, "First entry"),
1467            format!(
1468                "{UUID1} EXECUTE [depth=0] sofia/internal/+15550001234@192.0.2.1 set(api_result=+OK {UUID2} Job-UUID: ...)"
1469            ),
1470        ];
1471        let stream = LogStream::new(lines.into_iter());
1472        let mut tracker = SessionTracker::new(stream).with_post_hook(|entry, state| {
1473            if let MessageKind::Execute {
1474                application,
1475                arguments,
1476                ..
1477            } = &entry.message_kind
1478            {
1479                if application == "set" {
1480                    if let Some(value) = arguments.strip_prefix("api_result=+OK ") {
1481                        let uuid = value.split_whitespace().next().unwrap_or("");
1482                        if uuid.len() == 36 && state.other_leg_uuid.is_none() {
1483                            state.other_leg_uuid = Some(uuid.to_string());
1484                        }
1485                    }
1486                }
1487            }
1488        });
1489
1490        let entries: Vec<_> = tracker.by_ref().collect();
1491        assert_eq!(entries.len(), 2);
1492
1493        let session = entries[1].session.as_ref().unwrap();
1494        assert_eq!(
1495            session.other_leg_uuid.as_deref(),
1496            Some(UUID2),
1497            "post_hook should detect uuid_bridge API result"
1498        );
1499    }
1500
1501    #[test]
1502    fn post_hook_does_not_override_builtin() {
1503        let lines = vec![
1504            full_line(UUID1, TS1, "CHANNEL_DATA:"),
1505            format!("{UUID1} Other-Leg-Unique-ID: [{UUID2}]"),
1506            format!(
1507                "{UUID1} EXECUTE [depth=0] sofia/internal/+15550001234@192.0.2.1 set(api_result=+OK {UUID3} Job-UUID: ...)"
1508            ),
1509        ];
1510        let stream = LogStream::new(lines.into_iter());
1511        let mut tracker = SessionTracker::new(stream).with_post_hook(|entry, state| {
1512            if let MessageKind::Execute {
1513                application,
1514                arguments,
1515                ..
1516            } = &entry.message_kind
1517            {
1518                if application == "set" {
1519                    if let Some(value) = arguments.strip_prefix("api_result=+OK ") {
1520                        let uuid = value.split_whitespace().next().unwrap_or("");
1521                        if uuid.len() == 36 && state.other_leg_uuid.is_none() {
1522                            state.other_leg_uuid = Some(uuid.to_string());
1523                        }
1524                    }
1525                }
1526            }
1527        });
1528
1529        let entries: Vec<_> = tracker.by_ref().collect();
1530        assert_eq!(entries.len(), 2);
1531
1532        let session = entries[1].session.as_ref().unwrap();
1533        assert_eq!(
1534            session.other_leg_uuid.as_deref(),
1535            Some(UUID2),
1536            "built-in Other-Leg-Unique-ID takes precedence over hook"
1537        );
1538    }
1539
1540    #[test]
1541    fn pre_hook_runs_before_builtin() {
1542        let lines = vec![full_line(UUID1, TS1, "State Change CS_INIT -> CS_ROUTING")];
1543        let stream = LogStream::new(lines.into_iter());
1544        let mut tracker = SessionTracker::new(stream).with_pre_hook(|_entry, state| {
1545            state.channel_state = Some("PRE_SET".to_string());
1546        });
1547        let entries: Vec<_> = tracker.by_ref().collect();
1548        assert_eq!(
1549            entries[0]
1550                .session
1551                .as_ref()
1552                .unwrap()
1553                .channel_state
1554                .as_deref(),
1555            Some("CS_ROUTING"),
1556            "built-in overwrites pre_hook value when no guard"
1557        );
1558    }
1559
1560    #[test]
1561    fn post_hook_runs_after_builtin() {
1562        let lines = vec![full_line(UUID1, TS1, "State Change CS_INIT -> CS_ROUTING")];
1563        let stream = LogStream::new(lines.into_iter());
1564        let mut tracker = SessionTracker::new(stream).with_post_hook(|_entry, state| {
1565            if state.channel_state.as_deref() == Some("CS_ROUTING") {
1566                state
1567                    .variables
1568                    .insert("routing_seen".to_string(), "true".to_string());
1569            }
1570        });
1571        let _: Vec<_> = tracker.by_ref().collect();
1572        let state = tracker.sessions().get(UUID1).unwrap();
1573        assert_eq!(
1574            state.variables.get("routing_seen").map(|s| s.as_str()),
1575            Some("true"),
1576            "post_hook can read fields set by built-in"
1577        );
1578    }
1579
1580    #[test]
1581    fn parse_hangup_extracts_cause() {
1582        assert_eq!(
1583            parse_hangup("Hangup sofia/internal/1234 [NORMAL_CLEARING]"),
1584            Some("NORMAL_CLEARING".to_string())
1585        );
1586        assert_eq!(
1587            parse_hangup("Hangup sofia/internal/1234 [USER_BUSY]"),
1588            Some("USER_BUSY".to_string())
1589        );
1590        assert_eq!(parse_hangup("Some other message"), None);
1591        assert_eq!(parse_hangup("New Channel sofia/internal/1234 [uuid]"), None);
1592    }
1593
1594    #[test]
1595    fn is_answered_detects_answer_event() {
1596        assert!(is_answered("sofia/internal/1234 has been answered"));
1597        assert!(!is_answered("sofia/internal/1234 is ringing"));
1598        assert!(!is_answered("New Channel sofia/internal/1234"));
1599    }
1600
1601    #[test]
1602    fn hangup_cause_from_lifecycle() {
1603        let lines = vec![full_line(
1604            UUID1,
1605            TS1,
1606            "Hangup sofia/internal/+15550001234@192.0.2.1 [NORMAL_CLEARING]",
1607        )];
1608        let entries = collect_enriched(lines);
1609        let session = entries[0].session.as_ref().unwrap();
1610        assert_eq!(
1611            session.hangup_cause.as_deref(),
1612            Some("NORMAL_CLEARING"),
1613            "hangup_cause extracted from ChannelLifecycle Hangup"
1614        );
1615    }
1616
1617    #[test]
1618    fn answered_at_from_lifecycle() {
1619        let lines = vec![full_line(
1620            UUID1,
1621            TS1,
1622            "sofia/internal/+15550001234@192.0.2.1 has been answered",
1623        )];
1624        let entries = collect_enriched(lines);
1625        let session = entries[0].session.as_ref().unwrap();
1626        assert_eq!(
1627            session.answered_at.as_deref(),
1628            Some(TS1),
1629            "answered_at captures timestamp when 'has been answered' seen"
1630        );
1631    }
1632
1633    #[test]
1634    fn answered_at_not_overwritten() {
1635        let lines = vec![
1636            full_line(
1637                UUID1,
1638                TS1,
1639                "sofia/internal/+15550001234@192.0.2.1 has been answered",
1640            ),
1641            full_line(
1642                UUID1,
1643                TS2,
1644                "sofia/internal/+15550001234@192.0.2.1 has been answered",
1645            ),
1646        ];
1647        let entries = collect_enriched(lines);
1648        let session = entries[1].session.as_ref().unwrap();
1649        assert_eq!(
1650            session.answered_at.as_deref(),
1651            Some(TS1),
1652            "answered_at preserves first answer timestamp"
1653        );
1654    }
1655
1656    #[test]
1657    fn caller_id_name_from_channel_data() {
1658        let lines = vec![
1659            full_line(UUID1, TS1, "CHANNEL_DATA:"),
1660            format!("{UUID1} Caller-Caller-ID-Name: [Test Caller Name]"),
1661        ];
1662        let entries = collect_enriched(lines);
1663        let session = entries[0].session.as_ref().unwrap();
1664        assert_eq!(
1665            session.caller_id_name.as_deref(),
1666            Some("Test Caller Name"),
1667            "caller_id_name extracted from CHANNEL_DATA"
1668        );
1669    }
1670}