zeal_sdk/
subscription.rs

1//! Webhook subscription management
2
3use crate::errors::{Result, ZealError};
4use crate::events::*;
5use crate::webhooks::WebhooksAPI;
6use futures_util::stream::Stream;
7use serde::{Deserialize, Serialize};
8use std::collections::HashMap;
9use std::pin::Pin;
10use std::sync::{Arc, Mutex};
11use std::task::{Context, Poll};
12use tokio::sync::broadcast;
13
14/// Options for webhook subscriptions
15#[derive(Debug, Clone)]
16pub struct SubscriptionOptions {
17    /// Port for webhook server
18    pub port: Option<u16>,
19    /// Host to bind to (default: '0.0.0.0')
20    pub host: Option<String>,
21    /// Path to listen on (default: '/webhooks')
22    pub path: Option<String>,
23    /// Whether to use HTTPS
24    pub https: Option<bool>,
25    /// SSL key for HTTPS
26    pub key: Option<String>,
27    /// SSL certificate for HTTPS
28    pub cert: Option<String>,
29    /// Whether to automatically register the webhook with Zeal
30    pub auto_register: Option<bool>,
31    /// Namespace for the webhook
32    pub namespace: Option<String>,
33    /// Event types to listen for
34    pub events: Vec<String>,
35    /// Buffer size for event processing
36    pub buffer_size: usize,
37    /// Custom headers to send with webhook registration
38    pub headers: Option<HashMap<String, String>>,
39    /// Whether to verify webhook signatures
40    pub verify_signature: Option<bool>,
41    /// Secret key for signature verification
42    pub secret_key: Option<String>,
43}
44
45impl Default for SubscriptionOptions {
46    fn default() -> Self {
47        Self {
48            port: Some(3001),
49            host: Some("0.0.0.0".to_string()),
50            path: Some("/webhooks".to_string()),
51            https: Some(false),
52            key: None,
53            cert: None,
54            auto_register: Some(true),
55            namespace: Some("default".to_string()),
56            events: vec!["*".to_string()],
57            buffer_size: 1000,
58            headers: None,
59            verify_signature: Some(false),
60            secret_key: None,
61        }
62    }
63}
64
65/// Webhook delivery structure
66#[derive(Debug, Clone, Serialize, Deserialize)]
67pub struct WebhookDelivery {
68    pub webhook_id: String,
69    pub events: Vec<ZipWebhookEvent>,
70    pub metadata: WebhookMetadata,
71}
72
73/// Webhook delivery metadata
74#[derive(Debug, Clone, Serialize, Deserialize)]
75pub struct WebhookMetadata {
76    pub namespace: String,
77    pub delivery_id: String,
78    pub timestamp: String,
79}
80
81/// Event callback type
82pub type WebhookEventCallback = Arc<
83    dyn Fn(ZipWebhookEvent) -> Pin<Box<dyn std::future::Future<Output = ()> + Send>> + Send + Sync,
84>;
85
86/// Delivery callback type
87pub type WebhookDeliveryCallback = Arc<
88    dyn Fn(WebhookDelivery) -> Pin<Box<dyn std::future::Future<Output = ()> + Send>> + Send + Sync,
89>;
90
91/// Error callback type
92pub type WebhookErrorCallback =
93    Arc<dyn Fn(ZealError) -> Pin<Box<dyn std::future::Future<Output = ()> + Send>> + Send + Sync>;
94
95/// Webhook observable stream
96#[pin_project::pin_project]
97pub struct WebhookObservable {
98    #[pin]
99    receiver: broadcast::Receiver<ZipWebhookEvent>,
100}
101
102impl Stream for WebhookObservable {
103    type Item = ZipWebhookEvent;
104
105    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
106        let mut this = self.project();
107        // Use the receiver's poll_recv method directly
108        loop {
109            match this.receiver.try_recv() {
110                Ok(event) => return Poll::Ready(Some(event)),
111                Err(broadcast::error::TryRecvError::Empty) => {
112                    // Register waker and return Pending
113                    cx.waker().wake_by_ref();
114                    return Poll::Pending;
115                }
116                Err(broadcast::error::TryRecvError::Closed) => return Poll::Ready(None),
117                Err(broadcast::error::TryRecvError::Lagged(_)) => {
118                    // Skip lagged events and continue the loop
119                    continue;
120                }
121            }
122        }
123    }
124}
125
126/// Webhook subscription for receiving events
127pub struct WebhookSubscription {
128    webhooks_api: WebhooksAPI,
129    options: SubscriptionOptions,
130    event_sender: broadcast::Sender<ZipWebhookEvent>,
131    event_callbacks: Arc<Mutex<Vec<WebhookEventCallback>>>,
132    delivery_callbacks: Arc<Mutex<Vec<WebhookDeliveryCallback>>>,
133    error_callbacks: Arc<Mutex<Vec<WebhookErrorCallback>>>,
134    webhook_id: Arc<Mutex<Option<String>>>,
135    is_running: Arc<Mutex<bool>>,
136    #[cfg(feature = "webhook-server")]
137    server_handle: Arc<Mutex<Option<tokio::task::JoinHandle<()>>>>,
138}
139
140impl WebhookSubscription {
141    /// Create a new webhook subscription
142    pub fn new(webhooks_api: WebhooksAPI, options: Option<SubscriptionOptions>) -> Self {
143        let options = options.unwrap_or_default();
144        let (event_sender, _) = broadcast::channel(options.buffer_size);
145
146        Self {
147            webhooks_api,
148            options,
149            event_sender,
150            event_callbacks: Arc::new(Mutex::new(Vec::new())),
151            delivery_callbacks: Arc::new(Mutex::new(Vec::new())),
152            error_callbacks: Arc::new(Mutex::new(Vec::new())),
153            webhook_id: Arc::new(Mutex::new(None)),
154            is_running: Arc::new(Mutex::new(false)),
155            #[cfg(feature = "webhook-server")]
156            server_handle: Arc::new(Mutex::new(None)),
157        }
158    }
159
160    /// Subscribe with a callback function
161    pub fn on_event<F, Fut>(&self, callback: F) -> impl Fn() + Send + Sync
162    where
163        F: Fn(ZipWebhookEvent) -> Fut + Send + Sync + 'static,
164        Fut: std::future::Future<Output = ()> + Send + 'static,
165    {
166        let wrapped_callback: WebhookEventCallback =
167            Arc::new(move |event| Box::pin(callback(event)));
168
169        self.event_callbacks.lock().unwrap().push(wrapped_callback);
170        let callbacks = Arc::clone(&self.event_callbacks);
171        let index = callbacks.lock().unwrap().len() - 1;
172
173        move || {
174            callbacks.lock().unwrap().remove(index);
175        }
176    }
177
178    /// Subscribe to full webhook deliveries (multiple events at once)
179    pub fn on_delivery<F, Fut>(&self, callback: F) -> impl Fn() + Send + Sync
180    where
181        F: Fn(WebhookDelivery) -> Fut + Send + Sync + 'static,
182        Fut: std::future::Future<Output = ()> + Send + 'static,
183    {
184        let wrapped_callback: WebhookDeliveryCallback =
185            Arc::new(move |delivery| Box::pin(callback(delivery)));
186
187        self.delivery_callbacks
188            .lock()
189            .unwrap()
190            .push(wrapped_callback);
191        let callbacks = Arc::clone(&self.delivery_callbacks);
192        let index = callbacks.lock().unwrap().len() - 1;
193
194        move || {
195            callbacks.lock().unwrap().remove(index);
196        }
197    }
198
199    /// Subscribe to errors
200    pub fn on_error<F, Fut>(&self, callback: F) -> impl Fn() + Send + Sync
201    where
202        F: Fn(ZealError) -> Fut + Send + Sync + 'static,
203        Fut: std::future::Future<Output = ()> + Send + 'static,
204    {
205        let wrapped_callback: WebhookErrorCallback =
206            Arc::new(move |error| Box::pin(callback(error)));
207
208        self.error_callbacks.lock().unwrap().push(wrapped_callback);
209        let callbacks = Arc::clone(&self.error_callbacks);
210        let index = callbacks.lock().unwrap().len() - 1;
211
212        move || {
213            callbacks.lock().unwrap().remove(index);
214        }
215    }
216
217    /// Get an observable for webhook events
218    pub fn as_observable(&self) -> WebhookObservable {
219        WebhookObservable {
220            receiver: self.event_sender.subscribe(),
221        }
222    }
223
224    /// Start the webhook server
225    pub async fn start(&self) -> Result<()> {
226        {
227            let mut is_running = self.is_running.lock().unwrap();
228            if *is_running {
229                return Err(ZealError::other("Webhook subscription is already running"));
230            }
231            *is_running = true;
232        }
233
234        #[cfg(feature = "webhook-server")]
235        {
236            self.start_webhook_server().await?;
237
238            // Auto-register webhook if enabled
239            if self.options.auto_register.unwrap_or(true) {
240                self.register().await?;
241            }
242            Ok(())
243        }
244
245        #[cfg(not(feature = "webhook-server"))]
246        Err(ZealError::other("Webhook server feature not enabled. Enable 'webhook-server' feature to use this functionality"))
247    }
248
249    /// Stop the webhook server
250    pub async fn stop(&self) -> Result<()> {
251        {
252            let mut is_running = self.is_running.lock().unwrap();
253            if !*is_running {
254                return Ok(());
255            }
256            *is_running = false;
257        }
258
259        // Unregister webhook if it was registered
260        let webhook_id = self.webhook_id.lock().unwrap().take();
261        if let Some(webhook_id) = webhook_id {
262            if let Err(err) = self.webhooks_api.delete(&webhook_id).await {
263                tracing::error!("Failed to unregister webhook {}: {}", webhook_id, err);
264            } else {
265                tracing::info!("Unregistered webhook {}", webhook_id);
266            }
267        }
268
269        #[cfg(feature = "webhook-server")]
270        {
271            if let Some(handle) = self.server_handle.lock().unwrap().take() {
272                handle.abort();
273                let _ = handle.await;
274                tracing::info!("Webhook server stopped");
275            }
276        }
277
278        Ok(())
279    }
280
281    /// Register the webhook with Zeal
282    pub async fn register(&self) -> Result<()> {
283        if !*self.is_running.lock().unwrap() {
284            return Err(ZealError::other(
285                "Webhook server must be running before registration",
286            ));
287        }
288
289        // Determine the public URL for the webhook
290        let protocol = if self.options.https.unwrap_or(false) {
291            "https"
292        } else {
293            "http"
294        };
295        let host = self.options.host.as_deref().unwrap_or("localhost");
296        let host = if host == "0.0.0.0" { "localhost" } else { host };
297        let port = self.options.port.unwrap_or(3001);
298        let path = self.options.path.as_deref().unwrap_or("/webhooks");
299        let url = format!("{}://{}:{}{}", protocol, host, port, path);
300
301        // Register with Zeal
302        let config = crate::types::WebhookConfig {
303            namespace: self
304                .options
305                .namespace
306                .as_deref()
307                .unwrap_or("default")
308                .to_string(),
309            url,
310            events: Some(self.options.events.clone()),
311            headers: self.options.headers.clone(),
312            metadata: None,
313        };
314
315        let result = self.webhooks_api.register(config).await?;
316        *self.webhook_id.lock().unwrap() = Some(result.webhook_id.clone());
317
318        tracing::info!("Registered webhook {} at {}", result.webhook_id, result.url);
319        Ok(())
320    }
321
322    #[cfg(feature = "webhook-server")]
323    /// Process a webhook delivery
324    async fn process_delivery(&self, delivery: WebhookDelivery) {
325        // Call delivery callbacks
326        let delivery_callbacks = self.delivery_callbacks.lock().unwrap().clone();
327        for callback in delivery_callbacks {
328            if let Err(err) = tokio::time::timeout(
329                std::time::Duration::from_secs(30),
330                callback(delivery.clone()),
331            )
332            .await
333            {
334                tracing::error!("Delivery callback timeout: {}", err);
335            }
336        }
337
338        // Process individual events
339        for event in delivery.events {
340            // Send to broadcast channel
341            if let Err(err) = self.event_sender.send(event.clone()) {
342                tracing::error!("Failed to send event to broadcast channel: {}", err);
343            }
344
345            // Call event callbacks
346            let event_callbacks = self.event_callbacks.lock().unwrap().clone();
347            for callback in event_callbacks {
348                if let Err(err) = tokio::time::timeout(
349                    std::time::Duration::from_secs(30),
350                    callback(event.clone()),
351                )
352                .await
353                {
354                    tracing::error!("Event callback timeout: {}", err);
355                }
356            }
357        }
358    }
359
360    #[cfg(feature = "webhook-server")]
361    /// Emit an error to all error callbacks
362    async fn emit_error(&self, error: ZealError) {
363        let error_callbacks = self.error_callbacks.lock().unwrap().clone();
364        for callback in error_callbacks {
365            if let Err(err) =
366                tokio::time::timeout(std::time::Duration::from_secs(30), callback(error.clone()))
367                    .await
368            {
369                tracing::error!("Error callback timeout: {}", err);
370            }
371        }
372    }
373
374    /// Convenience method to create a filtered subscription
375    pub fn filter_events<F>(&self, predicate: F) -> impl Stream<Item = ZipWebhookEvent>
376    where
377        F: Fn(&ZipWebhookEvent) -> bool + Send + Sync + 'static,
378    {
379        use futures_util::StreamExt;
380        StreamExt::filter(self.as_observable(), move |event| {
381            futures_util::future::ready(predicate(event))
382        })
383    }
384
385    /// Subscribe to specific event types
386    pub fn on_event_type<F, Fut>(
387        &self,
388        event_types: Vec<String>,
389        callback: F,
390    ) -> impl Fn() + Send + Sync
391    where
392        F: Fn(ZipWebhookEvent) -> Fut + Send + Sync + 'static,
393        Fut: std::future::Future<Output = ()> + Send + 'static,
394    {
395        let callback = std::sync::Arc::new(callback);
396        self.on_event(move |event| {
397            let event_types = event_types.clone();
398            let callback = callback.clone();
399            async move {
400                let event_type = match &event {
401                    ZipWebhookEvent::Execution(e) => e.event_type(),
402                    ZipWebhookEvent::Workflow(e) => e.event_type(),
403                    ZipWebhookEvent::CRDT(e) => e.event_type(),
404                };
405                if event_types.contains(&event_type.to_string()) {
406                    callback(event).await
407                }
408            }
409        })
410    }
411
412    /// Subscribe to events from a specific source
413    pub fn on_event_source<F, Fut>(
414        &self,
415        sources: Vec<String>,
416        callback: F,
417    ) -> impl Fn() + Send + Sync
418    where
419        F: Fn(ZipWebhookEvent) -> Fut + Send + Sync + 'static,
420        Fut: std::future::Future<Output = ()> + Send + 'static,
421    {
422        let callback = std::sync::Arc::new(callback);
423        self.on_event(move |event| {
424            let sources = sources.clone();
425            let callback = callback.clone();
426            async move {
427                let workflow_id = match &event {
428                    ZipWebhookEvent::Execution(e) => e.workflow_id(),
429                    ZipWebhookEvent::Workflow(e) => e.workflow_id(),
430                    ZipWebhookEvent::CRDT(e) => e.workflow_id(),
431                };
432                if sources.contains(&workflow_id.to_string()) {
433                    callback(event).await
434                }
435            }
436        })
437    }
438
439    /// Get the current webhook ID if registered
440    pub fn webhook_id(&self) -> Option<String> {
441        self.webhook_id.lock().unwrap().clone()
442    }
443
444    /// Check if the subscription is running
445    pub fn is_running(&self) -> bool {
446        *self.is_running.lock().unwrap()
447    }
448
449    #[cfg(feature = "webhook-server")]
450    async fn start_webhook_server(&self) -> Result<()> {
451        use axum::{extract::State, http::StatusCode, response::Json, routing::post, Router};
452        use tower::ServiceBuilder;
453
454        let app_state = WebhookServerState {
455            subscription: self as *const WebhookSubscription,
456        };
457
458        let app = Router::new()
459            .route(
460                self.options.path.as_deref().unwrap_or("/webhooks"),
461                post(webhook_handler),
462            )
463            .layer(ServiceBuilder::new())
464            .with_state(app_state);
465
466        let addr = format!(
467            "{}:{}",
468            self.options.host.as_deref().unwrap_or("0.0.0.0"),
469            self.options.port.unwrap_or(3001)
470        );
471
472        let listener = tokio::net::TcpListener::bind(&addr).await.map_err(|e| {
473            ZealError::other(format!("Failed to bind webhook server to {}: {}", addr, e))
474        })?;
475
476        tracing::info!("Webhook server listening on {}", addr);
477
478        let server_handle = tokio::spawn(async move {
479            if let Err(err) = axum::serve(listener, app).await {
480                tracing::error!("Webhook server error: {}", err);
481            }
482        });
483
484        *self.server_handle.lock().unwrap() = Some(server_handle);
485        Ok(())
486    }
487}
488
489#[cfg(feature = "webhook-server")]
490#[derive(Clone)]
491struct WebhookServerState {
492    subscription: *const WebhookSubscription,
493}
494
495#[cfg(feature = "webhook-server")]
496unsafe impl Send for WebhookServerState {}
497#[cfg(feature = "webhook-server")]
498unsafe impl Sync for WebhookServerState {}
499
500#[cfg(feature = "webhook-server")]
501async fn webhook_handler(
502    State(state): State<WebhookServerState>,
503    Json(delivery): Json<WebhookDelivery>,
504) -> Result<StatusCode, StatusCode> {
505    let subscription = unsafe { &*state.subscription };
506
507    // TODO: Verify signature if enabled
508    if subscription.options.verify_signature.unwrap_or(false) {
509        // Signature verification would be implemented here
510    }
511
512    subscription.process_delivery(delivery).await;
513    Ok(StatusCode::OK)
514}
515
516impl Drop for WebhookSubscription {
517    fn drop(&mut self) {
518        if *self.is_running.lock().unwrap() {
519            tracing::warn!("WebhookSubscription dropped while still running. Consider calling stop() explicitly.");
520        }
521    }
522}
zeal_sdk/subscription.rs

zeal_sdk/
subscription.rs
