oxirs_gql/
webhook_support.rs

1//! Webhook Support for GraphQL Events
2//!
3//! This module provides webhook functionality for GraphQL event notifications:
4//! - **Event Types**: Schema changes, query events, subscription updates, errors
5//! - **Delivery Management**: Retry policies, dead letter queue, rate limiting
6//! - **Security**: HMAC signatures, IP allowlists, mTLS support
7//! - **Monitoring**: Delivery tracking, latency metrics, failure analysis
8//! - **Filtering**: Event filtering by type, operation, or custom rules
9
10use serde::{Deserialize, Serialize};
11use std::collections::HashMap;
12use std::sync::Arc;
13use std::time::{Duration, Instant, SystemTime};
14use tokio::sync::RwLock;
15
16/// Webhook event type
17#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
18pub enum WebhookEventType {
19    /// Schema was updated
20    SchemaUpdated,
21    /// Schema validation failed
22    SchemaValidationFailed,
23    /// Query was executed
24    QueryExecuted,
25    /// Query failed
26    QueryFailed,
27    /// Subscription started
28    SubscriptionStarted,
29    /// Subscription update
30    SubscriptionUpdate,
31    /// Subscription ended
32    SubscriptionEnded,
33    /// Rate limit exceeded
34    RateLimitExceeded,
35    /// Authentication failure
36    AuthenticationFailed,
37    /// Cache invalidated
38    CacheInvalidated,
39    /// Health check status changed
40    HealthStatusChanged,
41    /// Custom event
42    Custom(String),
43}
44
45/// A webhook event to be delivered
46#[derive(Debug, Clone, Serialize, Deserialize)]
47pub struct WebhookEvent {
48    /// Unique event ID
49    pub id: String,
50    /// Event type
51    pub event_type: WebhookEventType,
52    /// Timestamp
53    pub timestamp: SystemTime,
54    /// Event payload
55    pub payload: serde_json::Value,
56    /// Source service/component
57    pub source: String,
58    /// Correlation ID for tracing
59    pub correlation_id: Option<String>,
60    /// Event metadata
61    pub metadata: HashMap<String, String>,
62}
63
64impl WebhookEvent {
65    /// Create a new webhook event
66    pub fn new(event_type: WebhookEventType, payload: serde_json::Value, source: &str) -> Self {
67        Self {
68            id: uuid::Uuid::new_v4().to_string(),
69            event_type,
70            timestamp: SystemTime::now(),
71            payload,
72            source: source.to_string(),
73            correlation_id: None,
74            metadata: HashMap::new(),
75        }
76    }
77
78    /// Set correlation ID
79    pub fn with_correlation_id(mut self, id: &str) -> Self {
80        self.correlation_id = Some(id.to_string());
81        self
82    }
83
84    /// Add metadata
85    pub fn with_metadata(mut self, key: &str, value: &str) -> Self {
86        self.metadata.insert(key.to_string(), value.to_string());
87        self
88    }
89}
90
91/// Webhook endpoint configuration
92#[derive(Debug, Clone, Serialize, Deserialize)]
93pub struct WebhookEndpoint {
94    /// Endpoint ID
95    pub id: String,
96    /// Name/description
97    pub name: String,
98    /// Target URL
99    pub url: String,
100    /// Whether endpoint is active
101    pub active: bool,
102    /// Event types to receive
103    pub event_types: Vec<WebhookEventType>,
104    /// HMAC secret for signing
105    pub secret: Option<String>,
106    /// Custom headers to include
107    pub headers: HashMap<String, String>,
108    /// Retry configuration
109    pub retry_config: RetryConfig,
110    /// Rate limit (events per minute)
111    pub rate_limit: Option<u32>,
112    /// IP allowlist (for callback verification)
113    pub ip_allowlist: Vec<String>,
114    /// Created timestamp
115    pub created_at: SystemTime,
116    /// Last successful delivery
117    pub last_success: Option<SystemTime>,
118    /// Failure count
119    pub failure_count: u64,
120}
121
122impl WebhookEndpoint {
123    /// Create a new webhook endpoint
124    pub fn new(name: &str, url: &str, event_types: Vec<WebhookEventType>) -> Self {
125        Self {
126            id: uuid::Uuid::new_v4().to_string(),
127            name: name.to_string(),
128            url: url.to_string(),
129            active: true,
130            event_types,
131            secret: None,
132            headers: HashMap::new(),
133            retry_config: RetryConfig::default(),
134            rate_limit: None,
135            ip_allowlist: Vec::new(),
136            created_at: SystemTime::now(),
137            last_success: None,
138            failure_count: 0,
139        }
140    }
141
142    /// Set HMAC secret
143    pub fn with_secret(mut self, secret: &str) -> Self {
144        self.secret = Some(secret.to_string());
145        self
146    }
147
148    /// Add custom header
149    pub fn with_header(mut self, key: &str, value: &str) -> Self {
150        self.headers.insert(key.to_string(), value.to_string());
151        self
152    }
153
154    /// Set retry configuration
155    pub fn with_retry_config(mut self, config: RetryConfig) -> Self {
156        self.retry_config = config;
157        self
158    }
159}
160
161/// Retry configuration for webhook delivery
162#[derive(Debug, Clone, Serialize, Deserialize)]
163pub struct RetryConfig {
164    /// Maximum retry attempts
165    pub max_retries: u32,
166    /// Initial delay between retries
167    pub initial_delay: Duration,
168    /// Maximum delay between retries
169    pub max_delay: Duration,
170    /// Backoff multiplier
171    pub backoff_multiplier: f64,
172    /// Jitter factor (0.0-1.0)
173    pub jitter: f64,
174}
175
176impl Default for RetryConfig {
177    fn default() -> Self {
178        Self {
179            max_retries: 5,
180            initial_delay: Duration::from_secs(1),
181            max_delay: Duration::from_secs(300),
182            backoff_multiplier: 2.0,
183            jitter: 0.1,
184        }
185    }
186}
187
188impl RetryConfig {
189    /// Calculate delay for a specific retry attempt
190    pub fn delay_for_attempt(&self, attempt: u32) -> Duration {
191        let base_delay =
192            self.initial_delay.as_secs_f64() * self.backoff_multiplier.powi(attempt as i32);
193        let capped_delay = base_delay.min(self.max_delay.as_secs_f64());
194
195        // Add jitter
196        let jitter_range = capped_delay * self.jitter;
197        let jitter = (fastrand::f64() * 2.0 - 1.0) * jitter_range;
198
199        Duration::from_secs_f64((capped_delay + jitter).max(0.0))
200    }
201}
202
203/// Webhook delivery result
204#[derive(Debug, Clone, Serialize, Deserialize)]
205pub struct DeliveryResult {
206    /// Event ID
207    pub event_id: String,
208    /// Endpoint ID
209    pub endpoint_id: String,
210    /// Whether delivery succeeded
211    pub success: bool,
212    /// HTTP status code (if applicable)
213    pub status_code: Option<u16>,
214    /// Response body (truncated)
215    pub response_body: Option<String>,
216    /// Error message (if failed)
217    pub error: Option<String>,
218    /// Delivery attempt number
219    pub attempt: u32,
220    /// Delivery latency
221    pub latency_ms: u64,
222    /// Timestamp
223    pub timestamp: SystemTime,
224}
225
226/// Dead letter entry for failed deliveries
227#[derive(Debug, Clone, Serialize, Deserialize)]
228pub struct DeadLetterEntry {
229    /// Event
230    pub event: WebhookEvent,
231    /// Endpoint ID
232    pub endpoint_id: String,
233    /// Last error
234    pub last_error: String,
235    /// Retry count
236    pub retry_count: u32,
237    /// First failure time
238    pub first_failure: SystemTime,
239    /// Last failure time
240    pub last_failure: SystemTime,
241}
242
243/// Webhook delivery manager
244pub struct WebhookManager {
245    /// Configuration
246    config: WebhookManagerConfig,
247    /// Registered endpoints
248    endpoints: Arc<RwLock<HashMap<String, WebhookEndpoint>>>,
249    /// Event filters
250    filters: Arc<RwLock<Vec<EventFilter>>>,
251    /// Dead letter queue
252    dead_letter_queue: Arc<RwLock<Vec<DeadLetterEntry>>>,
253    /// Delivery statistics
254    stats: Arc<RwLock<WebhookStatistics>>,
255    /// HTTP client
256    client: reqwest::Client,
257    /// Event queue for async delivery
258    event_sender: tokio::sync::mpsc::Sender<(WebhookEvent, String)>,
259}
260
261/// Configuration for webhook manager
262#[derive(Debug, Clone, Serialize, Deserialize)]
263pub struct WebhookManagerConfig {
264    /// Maximum concurrent deliveries
265    pub max_concurrent: usize,
266    /// Default timeout for deliveries
267    pub timeout: Duration,
268    /// Dead letter queue max size
269    pub dlq_max_size: usize,
270    /// Enable async delivery
271    pub async_delivery: bool,
272    /// Event buffer size
273    pub event_buffer_size: usize,
274}
275
276impl Default for WebhookManagerConfig {
277    fn default() -> Self {
278        Self {
279            max_concurrent: 10,
280            timeout: Duration::from_secs(30),
281            dlq_max_size: 1000,
282            async_delivery: true,
283            event_buffer_size: 1000,
284        }
285    }
286}
287
288/// Event filter for webhook routing
289pub struct EventFilter {
290    /// Filter ID
291    pub id: String,
292    /// Filter name
293    pub name: String,
294    /// Filter predicate
295    pub predicate: Arc<dyn Fn(&WebhookEvent) -> bool + Send + Sync>,
296    /// Target endpoint IDs (empty = all)
297    pub target_endpoints: Vec<String>,
298}
299
300impl std::fmt::Debug for EventFilter {
301    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
302        f.debug_struct("EventFilter")
303            .field("id", &self.id)
304            .field("name", &self.name)
305            .field("target_endpoints", &self.target_endpoints)
306            .finish_non_exhaustive()
307    }
308}
309
310/// Webhook statistics
311#[derive(Debug, Clone, Default, Serialize, Deserialize)]
312pub struct WebhookStatistics {
313    /// Total events processed
314    pub total_events: u64,
315    /// Successful deliveries
316    pub successful_deliveries: u64,
317    /// Failed deliveries
318    pub failed_deliveries: u64,
319    /// Events by type
320    pub events_by_type: HashMap<String, u64>,
321    /// Average latency (ms)
322    pub avg_latency_ms: f64,
323    /// P95 latency (ms)
324    pub p95_latency_ms: f64,
325    /// Dead letter queue size
326    pub dlq_size: usize,
327    /// Active endpoints
328    pub active_endpoints: usize,
329}
330
331impl WebhookManager {
332    /// Create a new webhook manager
333    pub fn new(config: WebhookManagerConfig) -> Self {
334        let (event_sender, mut event_receiver) =
335            tokio::sync::mpsc::channel(config.event_buffer_size);
336
337        let endpoints = Arc::new(RwLock::new(HashMap::new()));
338        let stats = Arc::new(RwLock::new(WebhookStatistics::default()));
339        let dead_letter_queue = Arc::new(RwLock::new(Vec::new()));
340
341        let client = reqwest::Client::builder()
342            .timeout(config.timeout)
343            .build()
344            .unwrap_or_else(|_| reqwest::Client::new());
345
346        let manager = Self {
347            config: config.clone(),
348            endpoints: endpoints.clone(),
349            filters: Arc::new(RwLock::new(Vec::new())),
350            dead_letter_queue: dead_letter_queue.clone(),
351            stats: stats.clone(),
352            client: client.clone(),
353            event_sender,
354        };
355
356        // Start async delivery worker
357        if config.async_delivery {
358            let endpoints_clone = endpoints.clone();
359            let stats_clone = stats.clone();
360            let dlq_clone = dead_letter_queue.clone();
361            let config_clone = config.clone();
362
363            tokio::spawn(async move {
364                while let Some((event, endpoint_id)) = event_receiver.recv().await {
365                    let endpoints = endpoints_clone.read().await;
366                    if let Some(endpoint) = endpoints.get(&endpoint_id) {
367                        let result =
368                            Self::deliver_event_internal(&client, &event, endpoint, &config_clone)
369                                .await;
370
371                        // Update statistics
372                        let mut stats = stats_clone.write().await;
373                        stats.total_events += 1;
374
375                        if result.success {
376                            stats.successful_deliveries += 1;
377                        } else {
378                            stats.failed_deliveries += 1;
379
380                            // Add to DLQ if max retries exceeded
381                            if result.attempt >= endpoint.retry_config.max_retries {
382                                let mut dlq = dlq_clone.write().await;
383                                if dlq.len() < config_clone.dlq_max_size {
384                                    dlq.push(DeadLetterEntry {
385                                        event: event.clone(),
386                                        endpoint_id: endpoint_id.clone(),
387                                        last_error: result.error.unwrap_or_default(),
388                                        retry_count: result.attempt,
389                                        first_failure: SystemTime::now(),
390                                        last_failure: SystemTime::now(),
391                                    });
392                                }
393                            }
394                        }
395                    }
396                }
397            });
398        }
399
400        manager
401    }
402
403    /// Register a webhook endpoint
404    pub async fn register_endpoint(&self, endpoint: WebhookEndpoint) -> String {
405        let id = endpoint.id.clone();
406        let mut endpoints = self.endpoints.write().await;
407        endpoints.insert(id.clone(), endpoint);
408
409        // Update stats
410        let mut stats = self.stats.write().await;
411        stats.active_endpoints = endpoints.len();
412
413        id
414    }
415
416    /// Unregister a webhook endpoint
417    pub async fn unregister_endpoint(&self, endpoint_id: &str) -> bool {
418        let mut endpoints = self.endpoints.write().await;
419        let removed = endpoints.remove(endpoint_id).is_some();
420
421        if removed {
422            let mut stats = self.stats.write().await;
423            stats.active_endpoints = endpoints.len();
424        }
425
426        removed
427    }
428
429    /// Get endpoint by ID
430    pub async fn get_endpoint(&self, endpoint_id: &str) -> Option<WebhookEndpoint> {
431        let endpoints = self.endpoints.read().await;
432        endpoints.get(endpoint_id).cloned()
433    }
434
435    /// List all endpoints
436    pub async fn list_endpoints(&self) -> Vec<WebhookEndpoint> {
437        let endpoints = self.endpoints.read().await;
438        endpoints.values().cloned().collect()
439    }
440
441    /// Add an event filter
442    pub async fn add_filter(&self, filter: EventFilter) {
443        let mut filters = self.filters.write().await;
444        filters.push(filter);
445    }
446
447    /// Emit an event to all matching endpoints
448    pub async fn emit(&self, event: WebhookEvent) -> Vec<DeliveryResult> {
449        let endpoints = self.endpoints.read().await;
450        let filters = self.filters.read().await;
451
452        let mut results = Vec::new();
453
454        // Find matching endpoints
455        let matching_endpoints: Vec<_> = endpoints
456            .values()
457            .filter(|ep| {
458                ep.active
459                    && (ep.event_types.is_empty() || ep.event_types.contains(&event.event_type))
460            })
461            .filter(|ep| {
462                // Apply filters
463                filters.iter().all(|f| {
464                    if f.target_endpoints.is_empty() || f.target_endpoints.contains(&ep.id) {
465                        (f.predicate)(&event)
466                    } else {
467                        true
468                    }
469                })
470            })
471            .cloned()
472            .collect();
473
474        drop(endpoints);
475        drop(filters);
476
477        // Deliver to each endpoint
478        for endpoint in matching_endpoints {
479            if self.config.async_delivery {
480                // Queue for async delivery
481                let _ = self
482                    .event_sender
483                    .send((event.clone(), endpoint.id.clone()))
484                    .await;
485                results.push(DeliveryResult {
486                    event_id: event.id.clone(),
487                    endpoint_id: endpoint.id.clone(),
488                    success: true, // Queued successfully
489                    status_code: None,
490                    response_body: None,
491                    error: None,
492                    attempt: 0,
493                    latency_ms: 0,
494                    timestamp: SystemTime::now(),
495                });
496            } else {
497                // Synchronous delivery
498                let result =
499                    Self::deliver_event_internal(&self.client, &event, &endpoint, &self.config)
500                        .await;
501                results.push(result);
502            }
503        }
504
505        // Update statistics
506        {
507            let mut stats = self.stats.write().await;
508            *stats
509                .events_by_type
510                .entry(format!("{:?}", event.event_type))
511                .or_insert(0) += 1;
512        }
513
514        results
515    }
516
517    /// Deliver an event to a specific endpoint (internal implementation)
518    async fn deliver_event_internal(
519        client: &reqwest::Client,
520        event: &WebhookEvent,
521        endpoint: &WebhookEndpoint,
522        _config: &WebhookManagerConfig,
523    ) -> DeliveryResult {
524        let start = Instant::now();
525        let mut attempt = 0;
526        #[allow(unused_assignments)]
527        let mut last_error: Option<String> = None;
528        let mut last_status = None;
529        let mut last_body = None;
530
531        loop {
532            attempt += 1;
533
534            // Build request
535            let mut request = client.post(&endpoint.url);
536
537            // Add custom headers
538            for (key, value) in &endpoint.headers {
539                request = request.header(key, value);
540            }
541
542            // Add standard headers
543            request = request.header("Content-Type", "application/json");
544            request = request.header("X-Webhook-Event", format!("{:?}", event.event_type));
545            request = request.header("X-Webhook-ID", &event.id);
546            request = request.header("X-Webhook-Timestamp", format!("{:?}", event.timestamp));
547
548            // Add HMAC signature if secret is configured
549            if let Some(ref secret) = endpoint.secret {
550                let payload = serde_json::to_string(&event).unwrap_or_default();
551                let signature = Self::compute_hmac(secret, &payload);
552                request = request.header("X-Webhook-Signature", signature);
553            }
554
555            // Send request
556            match request.json(&event).send().await {
557                Ok(response) => {
558                    let status = response.status();
559                    last_status = Some(status.as_u16());
560
561                    if status.is_success() {
562                        return DeliveryResult {
563                            event_id: event.id.clone(),
564                            endpoint_id: endpoint.id.clone(),
565                            success: true,
566                            status_code: Some(status.as_u16()),
567                            response_body: response.text().await.ok(),
568                            error: None,
569                            attempt,
570                            latency_ms: start.elapsed().as_millis() as u64,
571                            timestamp: SystemTime::now(),
572                        };
573                    }
574
575                    last_body = response.text().await.ok();
576                    last_error = Some(format!("HTTP {}", status));
577                }
578                Err(e) => {
579                    last_error = Some(e.to_string());
580                }
581            }
582
583            // Check if we should retry
584            if attempt >= endpoint.retry_config.max_retries {
585                break;
586            }
587
588            // Wait before retry
589            let delay = endpoint.retry_config.delay_for_attempt(attempt);
590            tokio::time::sleep(delay).await;
591        }
592
593        DeliveryResult {
594            event_id: event.id.clone(),
595            endpoint_id: endpoint.id.clone(),
596            success: false,
597            status_code: last_status,
598            response_body: last_body,
599            error: last_error,
600            attempt,
601            latency_ms: start.elapsed().as_millis() as u64,
602            timestamp: SystemTime::now(),
603        }
604    }
605
606    /// Compute HMAC signature
607    fn compute_hmac(secret: &str, payload: &str) -> String {
608        use std::collections::hash_map::DefaultHasher;
609        use std::hash::{Hash, Hasher};
610
611        // Simple hash-based signature (in production, use HMAC-SHA256)
612        let mut hasher = DefaultHasher::new();
613        secret.hash(&mut hasher);
614        payload.hash(&mut hasher);
615        format!("sha256={:x}", hasher.finish())
616    }
617
618    /// Get dead letter queue entries
619    pub async fn get_dead_letters(&self) -> Vec<DeadLetterEntry> {
620        let dlq = self.dead_letter_queue.read().await;
621        dlq.clone()
622    }
623
624    /// Retry a dead letter entry
625    pub async fn retry_dead_letter(&self, event_id: &str) -> Option<DeliveryResult> {
626        let mut dlq = self.dead_letter_queue.write().await;
627
628        if let Some(idx) = dlq.iter().position(|e| e.event.id == event_id) {
629            let entry = dlq.remove(idx);
630            drop(dlq);
631
632            // Get endpoint
633            let endpoints = self.endpoints.read().await;
634            if let Some(endpoint) = endpoints.get(&entry.endpoint_id) {
635                let result = Self::deliver_event_internal(
636                    &self.client,
637                    &entry.event,
638                    endpoint,
639                    &self.config,
640                )
641                .await;
642                return Some(result);
643            }
644        }
645
646        None
647    }
648
649    /// Clear dead letter queue
650    pub async fn clear_dead_letters(&self) {
651        let mut dlq = self.dead_letter_queue.write().await;
652        dlq.clear();
653    }
654
655    /// Get statistics
656    pub async fn get_statistics(&self) -> WebhookStatistics {
657        let mut stats = self.stats.read().await.clone();
658        let dlq = self.dead_letter_queue.read().await;
659        stats.dlq_size = dlq.len();
660        stats
661    }
662
663    /// Test endpoint connectivity
664    pub async fn test_endpoint(&self, endpoint_id: &str) -> Option<DeliveryResult> {
665        let endpoints = self.endpoints.read().await;
666        if let Some(endpoint) = endpoints.get(endpoint_id) {
667            let test_event = WebhookEvent::new(
668                WebhookEventType::Custom("test".to_string()),
669                serde_json::json!({"test": true, "timestamp": SystemTime::now()}),
670                "webhook_manager",
671            );
672
673            return Some(
674                Self::deliver_event_internal(&self.client, &test_event, endpoint, &self.config)
675                    .await,
676            );
677        }
678        None
679    }
680}
681
682/// Builder for webhook manager
683pub struct WebhookManagerBuilder {
684    config: WebhookManagerConfig,
685}
686
687impl WebhookManagerBuilder {
688    /// Create a new builder
689    pub fn new() -> Self {
690        Self {
691            config: WebhookManagerConfig::default(),
692        }
693    }
694
695    /// Set max concurrent deliveries
696    pub fn max_concurrent(mut self, n: usize) -> Self {
697        self.config.max_concurrent = n;
698        self
699    }
700
701    /// Set delivery timeout
702    pub fn timeout(mut self, timeout: Duration) -> Self {
703        self.config.timeout = timeout;
704        self
705    }
706
707    /// Set dead letter queue max size
708    pub fn dlq_max_size(mut self, size: usize) -> Self {
709        self.config.dlq_max_size = size;
710        self
711    }
712
713    /// Enable/disable async delivery
714    pub fn async_delivery(mut self, enabled: bool) -> Self {
715        self.config.async_delivery = enabled;
716        self
717    }
718
719    /// Build the manager
720    pub fn build(self) -> WebhookManager {
721        WebhookManager::new(self.config)
722    }
723}
724
725impl Default for WebhookManagerBuilder {
726    fn default() -> Self {
727        Self::new()
728    }
729}
730
731#[cfg(test)]
732mod tests {
733    use super::*;
734
735    #[test]
736    fn test_webhook_event_creation() {
737        let event = WebhookEvent::new(
738            WebhookEventType::QueryExecuted,
739            serde_json::json!({"query": "test"}),
740            "test_source",
741        );
742
743        assert!(!event.id.is_empty());
744        assert_eq!(event.source, "test_source");
745    }
746
747    #[test]
748    fn test_webhook_endpoint_creation() {
749        let endpoint = WebhookEndpoint::new(
750            "Test Webhook",
751            "https://example.com/webhook",
752            vec![WebhookEventType::QueryExecuted],
753        );
754
755        assert!(!endpoint.id.is_empty());
756        assert!(endpoint.active);
757        assert_eq!(endpoint.url, "https://example.com/webhook");
758    }
759
760    #[test]
761    fn test_retry_config_default() {
762        let config = RetryConfig::default();
763        assert_eq!(config.max_retries, 5);
764        assert_eq!(config.backoff_multiplier, 2.0);
765    }
766
767    #[test]
768    fn test_retry_delay_calculation() {
769        let config = RetryConfig {
770            max_retries: 5,
771            initial_delay: Duration::from_secs(1),
772            max_delay: Duration::from_secs(60),
773            backoff_multiplier: 2.0,
774            jitter: 0.0, // No jitter for deterministic test
775        };
776
777        let delay0 = config.delay_for_attempt(0);
778        let delay1 = config.delay_for_attempt(1);
779        let delay2 = config.delay_for_attempt(2);
780
781        assert!(delay1 > delay0);
782        assert!(delay2 > delay1);
783    }
784
785    #[test]
786    fn test_event_with_metadata() {
787        let event = WebhookEvent::new(
788            WebhookEventType::QueryExecuted,
789            serde_json::json!({}),
790            "test",
791        )
792        .with_metadata("key", "value")
793        .with_correlation_id("corr-123");
794
795        assert_eq!(event.metadata.get("key"), Some(&"value".to_string()));
796        assert_eq!(event.correlation_id, Some("corr-123".to_string()));
797    }
798
799    #[test]
800    fn test_endpoint_with_config() {
801        let endpoint = WebhookEndpoint::new("Test", "https://example.com", vec![])
802            .with_secret("my-secret")
803            .with_header("X-Custom", "value");
804
805        assert_eq!(endpoint.secret, Some("my-secret".to_string()));
806        assert_eq!(endpoint.headers.get("X-Custom"), Some(&"value".to_string()));
807    }
808
809    #[tokio::test]
810    async fn test_manager_creation() {
811        let manager = WebhookManagerBuilder::new()
812            .max_concurrent(5)
813            .timeout(Duration::from_secs(10))
814            .async_delivery(false)
815            .build();
816
817        let stats = manager.get_statistics().await;
818        assert_eq!(stats.total_events, 0);
819    }
820
821    #[tokio::test]
822    async fn test_register_endpoint() {
823        let manager = WebhookManager::new(WebhookManagerConfig {
824            async_delivery: false,
825            ..Default::default()
826        });
827
828        let endpoint = WebhookEndpoint::new(
829            "Test",
830            "https://example.com/webhook",
831            vec![WebhookEventType::QueryExecuted],
832        );
833
834        let id = manager.register_endpoint(endpoint).await;
835        assert!(!id.is_empty());
836
837        let stats = manager.get_statistics().await;
838        assert_eq!(stats.active_endpoints, 1);
839    }
840
841    #[tokio::test]
842    async fn test_unregister_endpoint() {
843        let manager = WebhookManager::new(WebhookManagerConfig {
844            async_delivery: false,
845            ..Default::default()
846        });
847
848        let endpoint = WebhookEndpoint::new("Test", "https://example.com", vec![]);
849        let id = manager.register_endpoint(endpoint).await;
850
851        let removed = manager.unregister_endpoint(&id).await;
852        assert!(removed);
853
854        let stats = manager.get_statistics().await;
855        assert_eq!(stats.active_endpoints, 0);
856    }
857
858    #[tokio::test]
859    async fn test_list_endpoints() {
860        let manager = WebhookManager::new(WebhookManagerConfig {
861            async_delivery: false,
862            ..Default::default()
863        });
864
865        manager
866            .register_endpoint(WebhookEndpoint::new("A", "https://a.com", vec![]))
867            .await;
868        manager
869            .register_endpoint(WebhookEndpoint::new("B", "https://b.com", vec![]))
870            .await;
871
872        let endpoints = manager.list_endpoints().await;
873        assert_eq!(endpoints.len(), 2);
874    }
875
876    #[tokio::test]
877    async fn test_dead_letter_queue() {
878        let manager = WebhookManager::new(WebhookManagerConfig {
879            async_delivery: false,
880            ..Default::default()
881        });
882
883        // Initially empty
884        let dlq = manager.get_dead_letters().await;
885        assert!(dlq.is_empty());
886
887        // Clear (no-op when empty)
888        manager.clear_dead_letters().await;
889    }
890
891    #[test]
892    fn test_hmac_computation() {
893        let sig1 = WebhookManager::compute_hmac("secret", "payload");
894        let sig2 = WebhookManager::compute_hmac("secret", "payload");
895        let sig3 = WebhookManager::compute_hmac("different", "payload");
896
897        assert_eq!(sig1, sig2);
898        assert_ne!(sig1, sig3);
899        assert!(sig1.starts_with("sha256="));
900    }
901
902    #[tokio::test]
903    async fn test_event_filter() {
904        let manager = WebhookManager::new(WebhookManagerConfig {
905            async_delivery: false,
906            ..Default::default()
907        });
908
909        // Add filter that only allows QueryExecuted events
910        manager
911            .add_filter(EventFilter {
912                id: "filter1".to_string(),
913                name: "Query filter".to_string(),
914                predicate: Arc::new(|event| {
915                    matches!(event.event_type, WebhookEventType::QueryExecuted)
916                }),
917                target_endpoints: vec![],
918            })
919            .await;
920
921        // This is a basic test - full filter testing would require mocking HTTP
922    }
923
924    #[test]
925    fn test_webhook_event_types() {
926        let types = vec![
927            WebhookEventType::SchemaUpdated,
928            WebhookEventType::QueryExecuted,
929            WebhookEventType::QueryFailed,
930            WebhookEventType::SubscriptionStarted,
931            WebhookEventType::Custom("custom".to_string()),
932        ];
933
934        for t in types {
935            let json = serde_json::to_string(&t).expect("should succeed");
936            assert!(!json.is_empty());
937        }
938    }
939}
oxirs_gql/webhook_support.rs

oxirs_gql/
webhook_support.rs
