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#[derive(Debug, Clone, Default)]
18pub struct SessionState {
19 pub channel_name: Option<String>,
21 pub channel_state: Option<String>,
23 pub initial_context: Option<String>,
25 pub dialplan_context: Option<String>,
27 pub dialplan_from: Option<String>,
29 pub dialplan_to: Option<String>,
31 pub call_direction: Option<CallDirection>,
33 pub caller_id_number: Option<String>,
35 pub destination_number: Option<String>,
37 pub other_leg_uuid: Option<String>,
40 pub(crate) pending_bridge_target: Option<String>,
42 pub variables: HashMap<String, String>,
44}
45
46#[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#[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
300fn 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
324fn 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
336fn 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
350fn 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#[derive(Debug)]
366pub struct EnrichedEntry {
367 pub entry: LogEntry,
368 pub session: Option<SessionSnapshot>,
370}
371
372pub 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 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 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 pub fn sessions(&self) -> &HashMap<String, SessionState> {
431 &self.sessions
432 }
433
434 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 pub fn stats(&self) -> &ParseStats {
457 self.inner.stats()
458 }
459
460 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 fn link_legs(&mut self, uuid: &str, entry: &LogEntry) {
504 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 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 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 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 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 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 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 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 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 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 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 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 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 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}