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