active_call/useragent/
webhook.rs

1use crate::{call::RoutingState, useragent::invitation::InvitationHandler};
2use anyhow::Result;
3use async_trait::async_trait;
4use chrono::Utc;
5use reqwest::Client;
6use rsipstack::dialog::invite_dialog::InviteDialog;
7use rsipstack::rsip::prelude::{HasHeaders, HeadersExt};
8use serde_json::json;
9use std::{sync::Arc, time::Instant};
10use tokio_util::sync::CancellationToken;
11use tracing::info;
12
13pub struct WebhookInvitationHandler {
14    urls: Vec<String>,
15    method: Option<String>,
16    headers: Option<Vec<(String, String)>>,
17}
18
19impl WebhookInvitationHandler {
20    pub fn new(
21        urls: Vec<String>,
22        method: Option<String>,
23        headers: Option<Vec<(String, String)>>,
24    ) -> Self {
25        Self {
26            urls,
27            method,
28            headers,
29        }
30    }
31}
32
33#[async_trait]
34impl InvitationHandler for WebhookInvitationHandler {
35    async fn on_invite(
36        &self,
37        session_id: String,
38        _cancel_token: CancellationToken,
39        dialog: InviteDialog,
40        routing_state: Arc<RoutingState>,
41    ) -> Result<()> {
42        let client = Client::new();
43        let create_time = Utc::now().to_rfc3339();
44
45        let invite_request = dialog.initial_request();
46        let caller = invite_request.from_header()?.uri()?.to_string();
47        let callee = invite_request.to_header()?.uri()?.to_string();
48        let sip_call_id = invite_request.call_id_header()?.value().to_string();
49        let headers = invite_request
50            .headers()
51            .clone()
52            .into_iter()
53            .map(|h| h.to_string())
54            .collect::<Vec<_>>();
55
56        let payload = json!({
57            "dialogId": session_id,
58            "sipCallId": sip_call_id,
59            "createdAt": create_time,
60            "caller": caller,
61            "callee": callee,
62            "event": "invite",
63            "headers": headers,
64            "offer": String::from_utf8_lossy(invite_request.body()),
65        });
66        // TODO: better load balancing strategy
67        // just use round-robin for now
68        let idx = routing_state.next_round_robin_index("useragent_webhook", self.urls.len());
69        let url = match self.urls.get(idx) {
70            Some(u) => u,
71            None => {
72                return Err(anyhow::anyhow!("no webhook URL configured"));
73            }
74        };
75
76        let method = self.method.as_deref().unwrap_or("POST");
77        let mut request = client.request(reqwest::Method::from_bytes(method.as_bytes())?, url);
78
79        if let Some(headers) = &self.headers {
80            for (key, value) in headers {
81                request = request.header(key, value);
82            }
83        }
84
85        let start_time = Instant::now();
86        match request.json(&payload).send().await {
87            Ok(response) => {
88                info!(
89                    session_id,
90                    sip_call_id,
91                    url,
92                    caller,
93                    callee,
94                    elapsed = start_time.elapsed().as_millis(),
95                    status = ?response.status(),
96                    "invite to webhook"
97                );
98                if !response.status().is_success() {
99                    return Err(anyhow::anyhow!("failed to send invite to webhook"));
100                }
101            }
102            Err(e) => {
103                return Err(anyhow::anyhow!("failed to send invite to webhook: {}", e));
104            }
105        }
106        Ok(())
107    }
108}
active_call/useragent/webhook.rs

active_call/useragent/
webhook.rs
