feat(cdr): capture caller/callee SIP peer addresses on call records

src/call/cookie.rs

@@ -144,6 +144,14 @@ pub struct MatchedRoute {
     pub name: String,
 }
 
+/// A-leg (caller side) SIP peer address `ip:port`, captured from the inbound
+/// INVITE's transport connection in `handle_invite`. Surfaced into the CDR as
+/// `callerPeer` so consumers (Grafana plugin top-N, IP filtering) need no join
+/// against the signaling table. Rides the cookie because early-failure CDRs
+/// are reported before any SipSession exists.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct CallerPeerContext(pub String);
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct TrunkContext {
     pub id: Option<i64>,

src/call/mod.rs

@@ -36,8 +36,8 @@ pub mod transcription;
 pub mod user;
 pub mod uui;
 pub use cookie::{
-    CalleeDisplayName, CalleeOfflineMarker, MatchedRoute, OutboundTrunkContext, TransactionCookie,
-    TrunkContext,
+    CalleeDisplayName, CalleeOfflineMarker, CallerPeerContext, MatchedRoute, OutboundTrunkContext,
+    TransactionCookie, TrunkContext,
 };
 pub use user::SipUser;
 

src/callrecord/mod.rs

@@ -110,6 +110,13 @@ pub struct CallDetails {
     pub outbound_sip_trunk_id: Option<i64>,
     pub route_id: Option<i64>,
     pub sip_gateway: Option<String>,
+    /// A-leg (caller side) SIP peer `ip:port` — the transport source of the
+    /// inbound INVITE. Serialized as `callerPeer`.
+    pub caller_peer: Option<String>,
+    /// B-leg (callee side) SIP destination `ip:port` — where the outbound
+    /// INVITE was sent. Serialized as `calleePeer`. Not populated for RWI
+    /// originates (PBX-initiated, no inbound leg pairing at report time).
+    pub callee_peer: Option<String>,
     pub recording_url: Option<String>,
     pub recording_duration_secs: Option<i32>,
     pub has_transcript: bool,
@@ -1460,6 +1467,10 @@ impl From<rustpbx_models::call_record::Model> for CallRecord {
             outbound_sip_trunk_id: val.outbound_sip_trunk_id,
             route_id: val.route_id,
             sip_gateway: val.sip_gateway,
+            // The SQL model has no peer columns; peers live in the JSON savers'
+            // payloads only.
+            caller_peer: None,
+            callee_peer: None,
             recording_url: val.recording_url,
             recording_duration_secs: val.recording_duration_secs,
             has_transcript: val.has_transcript,

src/proxy/call.rs

@@ -1783,6 +1783,25 @@ impl CallModule {
             .get_user()
             .ok_or_else(|| anyhow::anyhow!("Missing caller user in transaction cookie"))?;
 
+        // A-leg SIP peer (bare ip:port) for the CDR `callerPeer` field: prefer
+        // the inbound INVITE's transport connection, fall back to the top-Via
+        // address when the connection is unavailable (may be a NAT address).
+        // Rides the cookie so early-failure CDRs (no SipSession yet) keep it.
+        // NB: `SipAddr::to_string()` prefixes the transport ("UDP ip:port");
+        // the bare `addr` (HostWithPort) matches the signaling table's peer
+        // format and the plugin's anchored LIKE matching.
+        let caller_peer = tx
+            .connection
+            .as_ref()
+            .and_then(|conn| conn.get_remote_addr())
+            .map(|addr| addr.addr.to_string())
+            .or_else(|| {
+                crate::proxy::routing::extract_via_ip(&tx.original).map(|ip| ip.to_string())
+            });
+        if let Some(peer) = caller_peer {
+            cookie.insert_extension(crate::call::CallerPeerContext(peer));
+        }
+
         // Immediately acknowledge the INVITE with 100 Trying BEFORE any routing work.
         // Routing (esp. wholesale route_wholesale) may block on CPS locks, DB lookups or
         // semaphores; without an early 100 the upstream retransmits (Timer A: 500ms..16s)

src/proxy/proxy_call.rs

@@ -242,6 +242,9 @@ impl CallSessionBuilder {
             routed_callee: None,
             routed_contact: None,
             routed_destination: None,
+            // Early failure: no B leg was ever dialed. The A-leg peer still
+            // reaches the reporter via the CallerPeerContext cookie extension.
+            callee_peer: None,
             last_queue_name: None,
             callee_call_ids: vec![],
             server_dialog_id: rsipstack::dialog::DialogId {

src/proxy/proxy_call/call_meta.rs

@@ -33,6 +33,11 @@ pub struct CallMeta {
     pub routed_callee: Option<String>,
     pub routed_contact: Option<String>,
     pub routed_destination: Option<String>,
+    /// B-leg (callee side) SIP destination `ip:port` the outbound INVITE was
+    /// sent to. Stashed at dial time (`build_target_invite_option` /
+    /// `initiate_sip_leg`) because `cleanup()` clears the leg dialogs before
+    /// CDR reporting. Last dial wins on re-dial scenarios.
+    pub callee_peer: Option<String>,
     pub queue_name: Option<String>,
     /// Primary skill-group id when the queue dials a `skill-group:{id}` target.
     /// Post-call hooks (CSAT, wrapup, hold-music) resolve skill-group

src/proxy/proxy_call/reporter.rs

@@ -277,6 +277,17 @@ impl CallReporter {
             ..Default::default()
         };
 
+        // Peer addresses for the CDR: A-leg rides the transaction cookie
+        // (`CallerPeerContext`, covers the early-failure path where no
+        // SipSession/snapshot exists); B-leg was stashed into the snapshot at
+        // dial time.
+        details.caller_peer = self
+            .context
+            .cookie
+            .get_extension::<crate::call::CallerPeerContext>()
+            .map(|p| p.0);
+        details.callee_peer = snapshot.callee_peer.clone();
+
         if call_was_accepted
             && details.recording_url.is_none()
             && self.server.recording_policy.load().is_none()
@@ -736,6 +747,7 @@ mod tests {
             connected_callee: None,
             routed_contact: None,
             routed_destination: None,
+            callee_peer: None,
             last_queue_name: None,
             callee_call_ids: vec!["callee-call-id".to_string()],
             server_dialog_id: rsipstack::dialog::DialogId {
@@ -850,6 +862,7 @@ mod tests {
             connected_callee: None,
             routed_contact: None,
             routed_destination: None,
+            callee_peer: None,
             last_queue_name: None,
             callee_call_ids: vec![],
             server_dialog_id: rsipstack::dialog::DialogId {

src/proxy/proxy_call/sip_session/session.rs

@@ -2843,6 +2843,18 @@ impl SipSession {
             ..Default::default()
         };
 
+        // B-leg SIP destination for the CDR `calleePeer` field. Stashed here
+        // at dial time because cleanup() clears the leg dialogs before
+        // reporting. Last dial wins. Bare `ip:port` (SipAddr's Display would
+        // add a transport prefix). When the target carries no explicit
+        // destination (URI-routed leg), the INVITE goes to the request-URI
+        // host — use it (exact for IP-literal hosts like trunks).
+        self.meta.callee_peer = option
+            .destination
+            .as_ref()
+            .map(|d| d.addr.to_string())
+            .or_else(|| Some(callee_uri.host_with_port.to_string()));
+
         Ok((option, callee_uri, callee_call_id))
     }
 
@@ -8691,6 +8703,7 @@ impl SipSession {
             connected_callee: self.meta.connected_callee.clone(),
             routed_contact: self.meta.routed_contact.clone(),
             routed_destination: self.meta.routed_destination.clone(),
+            callee_peer: self.meta.callee_peer.clone(),
             last_queue_name: self.meta.queue_name.clone(),
             callee_call_ids: self.meta.callee_call_ids.iter().cloned().collect(),
             server_dialog_id: self.caller_dialog_id(),
@@ -10431,6 +10444,15 @@ impl SipSession {
             ..Default::default()
         };
 
+        // B-leg SIP destination for the CDR `calleePeer` field (see
+        // build_target_invite_option; last dial wins). Bare `ip:port`;
+        // URI-routed legs fall back to the request-URI host.
+        self.meta.callee_peer = invite_option
+            .destination
+            .as_ref()
+            .map(|d| d.addr.to_string())
+            .or_else(|| Some(callee_uri.host_with_port.to_string()));
+
         // Register the B-leg SIP Call-ID as soon as the INVITE is built so
         // ringing-time CTI (`GET /cc/calls/{call_id}/context`) resolves before
         // the 200 OK / LegConnected notification.

src/proxy/proxy_call/state.rs

@@ -27,6 +27,10 @@ pub struct CallSessionRecordSnapshot {
     pub connected_callee: Option<String>,
     pub routed_contact: Option<String>,
     pub routed_destination: Option<String>,
+    /// B-leg (callee side) SIP destination `ip:port` stashed at dial time
+    /// (`CallMeta::callee_peer`); the A-leg peer rides the transaction cookie
+    /// instead (`CallerPeerContext`), so it is not duplicated here.
+    pub callee_peer: Option<String>,
     pub last_queue_name: Option<String>,
     pub callee_call_ids: Vec<String>,
     pub server_dialog_id: DialogId,

src/proxy/tests/cdr_capture.rs

@@ -355,6 +355,8 @@ mod tests {
                 outbound_sip_trunk_id: None,
                 route_id: None,
                 sip_gateway: None,
+                caller_peer: None,
+                callee_peer: None,
                 recording_url: None,
                 recording_duration_secs: None,
                 has_transcript: false,

tests/common_selftest/cdr_capture_tests.rs

@@ -38,6 +38,8 @@ fn create_test_record() -> CallRecord {
             outbound_sip_trunk_id: None,
             route_id: None,
             sip_gateway: None,
+            caller_peer: None,
+            callee_peer: None,
             recording_url: None,
             recording_duration_secs: None,
             has_transcript: false,