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