systemprompt_agent/services/external_integrations/webhook/service/
delivery.rs

1//! Webhook delivery with retry policy application.
2//!
3//! Copyright (c) systemprompt.io — Business Source License 1.1.
4//! See <https://systemprompt.io> for licensing details.
5
6use super::WebhookService;
7use super::types::{WebhookConfig, WebhookDeliveryResult, WebhookStats, WebhookTestResult};
8use crate::models::external_integrations::{IntegrationError, IntegrationResult};
9use serde_json::Value;
10use std::collections::HashMap;
11use systemprompt_identifiers::WebhookEndpointId;
12use systemprompt_models::net::validate_outbound_url;
13
14impl WebhookService {
15    pub async fn send_webhook(
16        &self,
17        url: &str,
18        payload: Value,
19        config: Option<WebhookConfig>,
20    ) -> IntegrationResult<WebhookDeliveryResult> {
21        validate_outbound_url(url)
22            .map_err(|e| IntegrationError::Webhook(format!("invalid webhook url: {e}")))?;
23        let config = config.unwrap_or_else(WebhookConfig::default);
24
25        let mut request_builder = self
26            .http_client
27            .post(url)
28            .json(&payload)
29            .header("Content-Type", "application/json")
30            .header(
31                "User-Agent",
32                concat!("systemprompt.io-Webhook/", env!("CARGO_PKG_VERSION")),
33            );
34
35        for (key, value) in &config.headers {
36            request_builder = request_builder.header(key, value);
37        }
38
39        if let Some(secret) = &config.secret {
40            let signature = Self::generate_signature(secret, &payload)?;
41            request_builder = request_builder.header("X-Webhook-Signature", signature);
42        }
43
44        if let Some(timeout) = config.timeout {
45            request_builder = request_builder.timeout(timeout);
46        }
47
48        let start_time = std::time::Instant::now();
49
50        match request_builder.send().await {
51            Ok(response) => {
52                let status = response.status().as_u16();
53                let headers: HashMap<String, String> = response
54                    .headers()
55                    .iter()
56                    .map(|(k, v)| (k.to_string(), v.to_str().unwrap_or("").to_owned()))
57                    .collect();
58
59                let body = response
60                    .text()
61                    .await
62                    .unwrap_or_else(|e| format!("<error reading response: {}>", e));
63                let duration = start_time.elapsed();
64
65                Ok(WebhookDeliveryResult {
66                    success: (200..300).contains(&status),
67                    status_code: status,
68                    response_body: body,
69                    response_headers: headers,
70                    duration_ms: duration.as_millis() as u64,
71                    error: None,
72                })
73            },
74            Err(e) => {
75                let duration = start_time.elapsed();
76                Ok(WebhookDeliveryResult {
77                    success: false,
78                    status_code: 0,
79                    response_body: String::new(),
80                    response_headers: HashMap::new(),
81                    duration_ms: duration.as_millis() as u64,
82                    error: Some(e.to_string()),
83                })
84            },
85        }
86    }
87
88    pub async fn get_endpoint_stats(
89        &self,
90        endpoint_id: &WebhookEndpointId,
91    ) -> IntegrationResult<WebhookStats> {
92        let endpoint = {
93            let endpoints = self.endpoints.read().await;
94            endpoints.get(endpoint_id).cloned().ok_or_else(|| {
95                IntegrationError::Webhook(format!("Endpoint not found: {endpoint_id}"))
96            })?
97        };
98
99        Ok(WebhookStats {
100            endpoint_id: endpoint.id,
101            total_requests: 0,
102            successful_requests: 0,
103            failed_requests: 0,
104            last_request_at: None,
105            average_response_time_ms: 0,
106        })
107    }
108
109    pub async fn test_endpoint(
110        &self,
111        endpoint_id: &WebhookEndpointId,
112    ) -> IntegrationResult<WebhookTestResult> {
113        let endpoint = {
114            let endpoints = self.endpoints.read().await;
115            endpoints.get(endpoint_id).cloned().ok_or_else(|| {
116                IntegrationError::Webhook(format!("Endpoint not found: {endpoint_id}"))
117            })?
118        };
119
120        let test_payload = serde_json::json!({
121            "test": true,
122            "timestamp": chrono::Utc::now().to_rfc3339(),
123            "endpoint_id": endpoint_id
124        });
125
126        let config = WebhookConfig {
127            secret: endpoint.secret.clone(),
128            headers: endpoint.headers.clone(),
129            timeout: Some(std::time::Duration::from_secs(10)),
130        };
131
132        let result = self
133            .send_webhook(&endpoint.url, test_payload, Some(config))
134            .await?;
135
136        Ok(WebhookTestResult {
137            endpoint_id: endpoint.id,
138            success: result.success,
139            status_code: result.status_code,
140            response_time_ms: result.duration_ms,
141            error: result.error,
142        })
143    }
144}
systemprompt_agent/services/external_integrations/webhook/service/delivery.rs

systemprompt_agent/services/external_integrations/webhook/service/
delivery.rs
