oxigdal_workflow/integrations/
external.rs

1//! External integrations module for workflow orchestration.
2//!
3//! This module provides comprehensive integration capabilities with external systems:
4//! - HTTP/REST API integration
5//! - Message queue integration (trait-based)
6//! - Database integration (trait-based)
7//! - Cloud storage integration
8//! - External service callbacks
9//! - Webhook support
10//! - Event emission to external systems
11
12use crate::engine::state::{TaskState, WorkflowState};
13use crate::error::{Result, WorkflowError};
14use async_trait::async_trait;
15use chrono::{DateTime, Utc};
16use serde::{Deserialize, Serialize};
17use std::collections::HashMap;
18use std::sync::Arc;
19use std::time::Duration;
20use tokio::sync::RwLock;
21#[cfg(feature = "integrations")]
22use tracing::debug;
23use tracing::{error, info, warn};
24
25// =============================================================================
26// HTTP/REST API Integration
27// =============================================================================
28
29/// HTTP client configuration for REST API integration.
30#[derive(Debug, Clone, Serialize, Deserialize)]
31pub struct HttpClientConfig {
32    /// Base URL for API requests.
33    pub base_url: String,
34    /// Request timeout in seconds.
35    pub timeout_secs: u64,
36    /// Maximum number of retries.
37    pub max_retries: u32,
38    /// Retry delay in milliseconds.
39    pub retry_delay_ms: u64,
40    /// Default headers for all requests.
41    pub default_headers: HashMap<String, String>,
42    /// Authentication configuration.
43    pub auth: Option<HttpAuth>,
44    /// Enable request/response logging.
45    pub enable_logging: bool,
46}
47
48impl Default for HttpClientConfig {
49    fn default() -> Self {
50        Self {
51            base_url: String::new(),
52            timeout_secs: 30,
53            max_retries: 3,
54            retry_delay_ms: 1000,
55            default_headers: HashMap::new(),
56            auth: None,
57            enable_logging: false,
58        }
59    }
60}
61
62/// HTTP authentication configuration.
63#[derive(Debug, Clone, Serialize, Deserialize)]
64pub enum HttpAuth {
65    /// Bearer token authentication.
66    Bearer {
67        /// The bearer token value.
68        token: String,
69    },
70    /// Basic authentication.
71    Basic {
72        /// Username for basic auth.
73        username: String,
74        /// Password for basic auth.
75        password: String,
76    },
77    /// API key authentication.
78    ApiKey {
79        /// Header name to use for the API key (e.g. `X-API-Key`).
80        header_name: String,
81        /// API key value.
82        key: String,
83    },
84    /// Custom header authentication.
85    Custom {
86        /// Map of header name → header value pairs.
87        headers: HashMap<String, String>,
88    },
89}
90
91/// HTTP request builder for REST API calls.
92#[derive(Debug, Clone)]
93pub struct HttpRequest {
94    /// HTTP method.
95    pub method: HttpMethod,
96    /// Request path (appended to base URL).
97    pub path: String,
98    /// Query parameters.
99    pub query_params: HashMap<String, String>,
100    /// Request headers.
101    pub headers: HashMap<String, String>,
102    /// Request body (JSON).
103    pub body: Option<serde_json::Value>,
104    /// Request timeout override.
105    pub timeout: Option<Duration>,
106}
107
108/// HTTP method enumeration.
109#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
110pub enum HttpMethod {
111    /// GET request.
112    Get,
113    /// POST request.
114    Post,
115    /// PUT request.
116    Put,
117    /// PATCH request.
118    Patch,
119    /// DELETE request.
120    Delete,
121    /// HEAD request.
122    Head,
123    /// OPTIONS request.
124    Options,
125}
126
127impl HttpMethod {
128    /// Get string representation of HTTP method.
129    pub fn as_str(&self) -> &'static str {
130        match self {
131            Self::Get => "GET",
132            Self::Post => "POST",
133            Self::Put => "PUT",
134            Self::Patch => "PATCH",
135            Self::Delete => "DELETE",
136            Self::Head => "HEAD",
137            Self::Options => "OPTIONS",
138        }
139    }
140}
141
142/// HTTP response from REST API calls.
143#[derive(Debug, Clone, Serialize, Deserialize)]
144pub struct HttpResponse {
145    /// HTTP status code.
146    pub status_code: u16,
147    /// Response headers.
148    pub headers: HashMap<String, String>,
149    /// Response body.
150    pub body: Option<serde_json::Value>,
151    /// Response time in milliseconds.
152    pub response_time_ms: u64,
153    /// Whether the request was successful (2xx status).
154    pub is_success: bool,
155}
156
157impl HttpResponse {
158    /// Check if response indicates success.
159    pub fn is_success(&self) -> bool {
160        (200..300).contains(&self.status_code)
161    }
162
163    /// Check if response indicates client error.
164    pub fn is_client_error(&self) -> bool {
165        (400..500).contains(&self.status_code)
166    }
167
168    /// Check if response indicates server error.
169    pub fn is_server_error(&self) -> bool {
170        (500..600).contains(&self.status_code)
171    }
172}
173
174/// Helper: appends query parameters to a URL string.
175///
176/// Returns the URL with query params appended, or the original URL unchanged if parsing fails.
177/// Defined separately to avoid method-resolution ambiguity with the local
178/// `DatabaseConnection::query` and `DatabaseTransaction::query` traits.
179#[cfg(feature = "integrations")]
180fn url_with_query_params(base_url: &str, params: &HashMap<String, String>) -> String {
181    match reqwest::Url::parse(base_url) {
182        Ok(mut url) => {
183            {
184                let mut pairs = url.query_pairs_mut();
185                for (k, v) in params {
186                    pairs.append_pair(k, v);
187                }
188            }
189            url.to_string()
190        }
191        Err(_) => base_url.to_owned(),
192    }
193}
194
195/// REST API client for external integrations.
196#[cfg(feature = "integrations")]
197pub struct RestApiClient {
198    config: HttpClientConfig,
199    client: reqwest::Client,
200}
201
202#[cfg(feature = "integrations")]
203impl RestApiClient {
204    /// Create a new REST API client.
205    pub fn new(config: HttpClientConfig) -> Result<Self> {
206        let client = reqwest::Client::builder()
207            .timeout(Duration::from_secs(config.timeout_secs))
208            .build()
209            .map_err(|e| {
210                WorkflowError::integration("rest", format!("Failed to create client: {}", e))
211            })?;
212
213        Ok(Self { config, client })
214    }
215
216    /// Execute an HTTP request.
217    pub async fn execute(&self, request: HttpRequest) -> Result<HttpResponse> {
218        let url = if request.path.starts_with("http") {
219            request.path.clone()
220        } else {
221            format!("{}{}", self.config.base_url, request.path)
222        };
223
224        let mut last_error = None;
225
226        for attempt in 0..=self.config.max_retries {
227            if attempt > 0 {
228                debug!("Retrying request (attempt {})", attempt + 1);
229                tokio::time::sleep(Duration::from_millis(
230                    self.config.retry_delay_ms * (1 << attempt.min(5)),
231                ))
232                .await;
233            }
234
235            match self.do_request(&url, &request).await {
236                Ok(response) => {
237                    if self.config.enable_logging {
238                        info!(
239                            "HTTP {} {} -> {} ({}ms)",
240                            request.method.as_str(),
241                            url,
242                            response.status_code,
243                            response.response_time_ms
244                        );
245                    }
246                    return Ok(response);
247                }
248                Err(e) => {
249                    last_error = Some(e);
250                    if attempt < self.config.max_retries {
251                        warn!("Request failed, will retry: {:?}", last_error);
252                    }
253                }
254            }
255        }
256
257        Err(last_error.unwrap_or_else(|| {
258            WorkflowError::integration("rest", "Request failed after all retries")
259        }))
260    }
261
262    async fn do_request(&self, url: &str, request: &HttpRequest) -> Result<HttpResponse> {
263        let start_time = std::time::Instant::now();
264
265        // Incorporate query parameters into the URL before building the reqwest builder,
266        // so we never call `.query()` on `RequestBuilder` (which would be ambiguous with
267        // the local `DatabaseConnection::query` / `DatabaseTransaction::query` traits).
268        let effective_url;
269        let request_url = if request.query_params.is_empty() {
270            url
271        } else {
272            effective_url = url_with_query_params(url, &request.query_params);
273            effective_url.as_str()
274        };
275
276        let mut req_builder = match request.method {
277            HttpMethod::Get => self.client.get(request_url),
278            HttpMethod::Post => self.client.post(request_url),
279            HttpMethod::Put => self.client.put(request_url),
280            HttpMethod::Patch => self.client.patch(request_url),
281            HttpMethod::Delete => self.client.delete(request_url),
282            HttpMethod::Head => self.client.head(request_url),
283            HttpMethod::Options => self.client.request(reqwest::Method::OPTIONS, request_url),
284        };
285
286        // Add default headers
287        for (key, value) in &self.config.default_headers {
288            req_builder = req_builder.header(key, value);
289        }
290
291        // Add request-specific headers
292        for (key, value) in &request.headers {
293            req_builder = req_builder.header(key, value);
294        }
295
296        // Add authentication
297        if let Some(auth) = &self.config.auth {
298            req_builder = match auth {
299                HttpAuth::Bearer { token } => req_builder.bearer_auth(token),
300                HttpAuth::Basic { username, password } => {
301                    req_builder.basic_auth(username, Some(password))
302                }
303                HttpAuth::ApiKey { header_name, key } => req_builder.header(header_name, key),
304                HttpAuth::Custom { headers } => {
305                    for (k, v) in headers {
306                        req_builder = req_builder.header(k, v);
307                    }
308                    req_builder
309                }
310            };
311        }
312
313        // Add body
314        if let Some(body) = &request.body {
315            req_builder = req_builder.json(body);
316        }
317
318        // Set timeout override
319        if let Some(timeout) = request.timeout {
320            req_builder = req_builder.timeout(timeout);
321        }
322
323        // Send request
324        let response = req_builder
325            .send()
326            .await
327            .map_err(|e| WorkflowError::integration("rest", format!("Request failed: {}", e)))?;
328
329        let status_code = response.status().as_u16();
330        let mut headers = HashMap::new();
331        for (key, value) in response.headers() {
332            if let Ok(v) = value.to_str() {
333                headers.insert(key.to_string(), v.to_string());
334            }
335        }
336
337        let body = response.json::<serde_json::Value>().await.ok();
338
339        let response_time_ms = start_time.elapsed().as_millis() as u64;
340
341        Ok(HttpResponse {
342            is_success: (200..300).contains(&status_code),
343            status_code,
344            headers,
345            body,
346            response_time_ms,
347        })
348    }
349}
350
351// =============================================================================
352// Message Queue Integration (Trait-based)
353// =============================================================================
354
355/// Message for queue communication.
356#[derive(Debug, Clone, Serialize, Deserialize)]
357pub struct QueueMessage {
358    /// Unique message ID.
359    pub id: String,
360    /// Message topic/queue name.
361    pub topic: String,
362    /// Message payload.
363    pub payload: serde_json::Value,
364    /// Message headers/metadata.
365    pub headers: HashMap<String, String>,
366    /// Message timestamp.
367    pub timestamp: DateTime<Utc>,
368    /// Message priority (0-9, higher is more important).
369    pub priority: u8,
370    /// Correlation ID for request-response patterns.
371    pub correlation_id: Option<String>,
372    /// Reply-to topic for response.
373    pub reply_to: Option<String>,
374}
375
376impl QueueMessage {
377    /// Create a new queue message.
378    pub fn new(topic: impl Into<String>, payload: serde_json::Value) -> Self {
379        Self {
380            id: uuid::Uuid::new_v4().to_string(),
381            topic: topic.into(),
382            payload,
383            headers: HashMap::new(),
384            timestamp: Utc::now(),
385            priority: 5,
386            correlation_id: None,
387            reply_to: None,
388        }
389    }
390
391    /// Set message correlation ID.
392    pub fn with_correlation_id(mut self, correlation_id: impl Into<String>) -> Self {
393        self.correlation_id = Some(correlation_id.into());
394        self
395    }
396
397    /// Set reply-to topic.
398    pub fn with_reply_to(mut self, reply_to: impl Into<String>) -> Self {
399        self.reply_to = Some(reply_to.into());
400        self
401    }
402
403    /// Set message priority.
404    pub fn with_priority(mut self, priority: u8) -> Self {
405        self.priority = priority.min(9);
406        self
407    }
408
409    /// Add a header to the message.
410    pub fn with_header(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
411        self.headers.insert(key.into(), value.into());
412        self
413    }
414}
415
416/// Message acknowledgment status.
417#[derive(Debug, Clone, Copy, PartialEq, Eq)]
418pub enum AckStatus {
419    /// Message processed successfully.
420    Ack,
421    /// Message processing failed, should be retried.
422    Nack,
423    /// Message should be rejected (dead-lettered).
424    Reject,
425}
426
427/// Message queue producer trait.
428#[async_trait]
429pub trait MessageQueueProducer: Send + Sync {
430    /// Send a message to the queue.
431    async fn send(&self, message: QueueMessage) -> Result<()>;
432
433    /// Send multiple messages in a batch.
434    async fn send_batch(&self, messages: Vec<QueueMessage>) -> Result<Vec<Result<()>>> {
435        let mut results = Vec::with_capacity(messages.len());
436        for msg in messages {
437            results.push(self.send(msg).await);
438        }
439        Ok(results)
440    }
441
442    /// Flush pending messages.
443    async fn flush(&self) -> Result<()>;
444}
445
446/// Message queue consumer trait.
447#[async_trait]
448pub trait MessageQueueConsumer: Send + Sync {
449    /// Subscribe to a topic.
450    async fn subscribe(&self, topic: &str) -> Result<()>;
451
452    /// Unsubscribe from a topic.
453    async fn unsubscribe(&self, topic: &str) -> Result<()>;
454
455    /// Receive a message (blocking with timeout).
456    async fn receive(&self, timeout: Duration) -> Result<Option<QueueMessage>>;
457
458    /// Acknowledge a message.
459    async fn acknowledge(&self, message_id: &str, status: AckStatus) -> Result<()>;
460
461    /// Commit consumer offset.
462    async fn commit(&self) -> Result<()>;
463}
464
465/// Message queue configuration.
466#[derive(Debug, Clone, Serialize, Deserialize)]
467pub struct MessageQueueConfig {
468    /// Queue type.
469    pub queue_type: MessageQueueType,
470    /// Connection endpoints.
471    pub endpoints: Vec<String>,
472    /// Authentication configuration.
473    pub auth: Option<QueueAuth>,
474    /// Consumer group ID.
475    pub consumer_group: Option<String>,
476    /// Message retention period in seconds.
477    pub retention_secs: u64,
478    /// Maximum message size in bytes.
479    pub max_message_size: usize,
480    /// Enable message compression.
481    pub compression: bool,
482}
483
484/// Message queue type.
485#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
486pub enum MessageQueueType {
487    /// Apache Kafka.
488    Kafka,
489    /// RabbitMQ.
490    RabbitMq,
491    /// Amazon SQS.
492    Sqs,
493    /// Google Cloud Pub/Sub.
494    PubSub,
495    /// Azure Service Bus.
496    ServiceBus,
497    /// Redis Streams.
498    Redis,
499}
500
501/// Queue authentication configuration.
502#[derive(Debug, Clone, Serialize, Deserialize)]
503pub enum QueueAuth {
504    /// Username/password authentication.
505    Credentials {
506        /// Username.
507        username: String,
508        /// Password.
509        password: String,
510    },
511    /// API key authentication.
512    ApiKey {
513        /// API key value.
514        key: String,
515    },
516    /// OAuth2 authentication.
517    OAuth2 {
518        /// OAuth2 client identifier.
519        client_id: String,
520        /// OAuth2 client secret.
521        client_secret: String,
522        /// Token endpoint URL.
523        token_url: String,
524    },
525    /// TLS/mTLS authentication.
526    Tls {
527        /// Path to the TLS certificate file.
528        cert_path: String,
529        /// Path to the TLS private key file.
530        key_path: String,
531        /// Optional path to the CA certificate file.
532        ca_path: Option<String>,
533    },
534}
535
536// =============================================================================
537// Database Integration (Trait-based)
538// =============================================================================
539
540/// Database query result.
541#[derive(Debug, Clone, Serialize, Deserialize)]
542pub struct QueryResult {
543    /// Number of rows affected.
544    pub rows_affected: u64,
545    /// Column names.
546    pub columns: Vec<String>,
547    /// Row data.
548    pub rows: Vec<Vec<serde_json::Value>>,
549    /// Query execution time in milliseconds.
550    pub execution_time_ms: u64,
551}
552
553impl QueryResult {
554    /// Create an empty query result.
555    pub fn empty() -> Self {
556        Self {
557            rows_affected: 0,
558            columns: Vec::new(),
559            rows: Vec::new(),
560            execution_time_ms: 0,
561        }
562    }
563
564    /// Get the number of rows.
565    pub fn row_count(&self) -> usize {
566        self.rows.len()
567    }
568
569    /// Check if the result is empty.
570    pub fn is_empty(&self) -> bool {
571        self.rows.is_empty()
572    }
573}
574
575/// Database connection trait.
576#[async_trait]
577pub trait DatabaseConnection: Send + Sync {
578    /// Execute a query and return results.
579    async fn query(&self, sql: &str, params: &[serde_json::Value]) -> Result<QueryResult>;
580
581    /// Execute a statement (INSERT, UPDATE, DELETE).
582    async fn execute(&self, sql: &str, params: &[serde_json::Value]) -> Result<u64>;
583
584    /// Begin a transaction.
585    async fn begin_transaction(&self) -> Result<Box<dyn DatabaseTransaction>>;
586
587    /// Check connection health.
588    async fn ping(&self) -> Result<()>;
589
590    /// Close the connection.
591    async fn close(&self) -> Result<()>;
592}
593
594/// Database transaction trait.
595#[async_trait]
596pub trait DatabaseTransaction: Send + Sync {
597    /// Execute a query within the transaction.
598    async fn query(&self, sql: &str, params: &[serde_json::Value]) -> Result<QueryResult>;
599
600    /// Execute a statement within the transaction.
601    async fn execute(&self, sql: &str, params: &[serde_json::Value]) -> Result<u64>;
602
603    /// Commit the transaction.
604    async fn commit(self: Box<Self>) -> Result<()>;
605
606    /// Rollback the transaction.
607    async fn rollback(self: Box<Self>) -> Result<()>;
608}
609
610/// Database configuration.
611#[derive(Debug, Clone, Serialize, Deserialize)]
612pub struct DatabaseConfig {
613    /// Database type.
614    pub db_type: DatabaseType,
615    /// Connection string or host.
616    pub connection_string: String,
617    /// Database name.
618    pub database: Option<String>,
619    /// Connection pool minimum size.
620    pub pool_min: u32,
621    /// Connection pool maximum size.
622    pub pool_max: u32,
623    /// Connection timeout in seconds.
624    pub connect_timeout_secs: u64,
625    /// Query timeout in seconds.
626    pub query_timeout_secs: u64,
627    /// Enable SSL/TLS.
628    pub ssl: bool,
629}
630
631impl Default for DatabaseConfig {
632    fn default() -> Self {
633        Self {
634            db_type: DatabaseType::PostgreSql,
635            connection_string: String::new(),
636            database: None,
637            pool_min: 1,
638            pool_max: 10,
639            connect_timeout_secs: 30,
640            query_timeout_secs: 60,
641            ssl: false,
642        }
643    }
644}
645
646/// Database type.
647#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
648pub enum DatabaseType {
649    /// PostgreSQL.
650    PostgreSql,
651    /// MySQL.
652    MySql,
653    /// SQLite.
654    Sqlite,
655    /// Microsoft SQL Server.
656    MsSql,
657    /// MongoDB (document store).
658    MongoDb,
659    /// ClickHouse.
660    ClickHouse,
661}
662
663// =============================================================================
664// Cloud Storage Integration
665// =============================================================================
666
667/// Cloud storage object metadata.
668#[derive(Debug, Clone, Serialize, Deserialize)]
669pub struct ObjectMetadata {
670    /// Object key/path.
671    pub key: String,
672    /// Object size in bytes.
673    pub size: u64,
674    /// Content type.
675    pub content_type: Option<String>,
676    /// Last modified timestamp.
677    pub last_modified: Option<DateTime<Utc>>,
678    /// ETag/checksum.
679    pub etag: Option<String>,
680    /// Custom metadata.
681    pub metadata: HashMap<String, String>,
682}
683
684/// Cloud storage upload options.
685#[derive(Debug, Clone, Default, Serialize, Deserialize)]
686pub struct UploadOptions {
687    /// Content type.
688    pub content_type: Option<String>,
689    /// Content encoding.
690    pub content_encoding: Option<String>,
691    /// Cache control header.
692    pub cache_control: Option<String>,
693    /// Custom metadata.
694    pub metadata: HashMap<String, String>,
695    /// Server-side encryption.
696    pub encryption: Option<StorageEncryption>,
697    /// Storage class.
698    pub storage_class: Option<String>,
699}
700
701/// Server-side encryption configuration.
702#[derive(Debug, Clone, Serialize, Deserialize)]
703pub enum StorageEncryption {
704    /// Server-managed keys.
705    ServerManaged,
706    /// Customer-managed keys.
707    CustomerManaged {
708        /// Key identifier for the customer-managed key.
709        key_id: String,
710    },
711}
712
713/// Cloud storage trait.
714#[async_trait]
715pub trait CloudStorage: Send + Sync {
716    /// Upload data to storage.
717    async fn upload(
718        &self,
719        key: &str,
720        data: &[u8],
721        options: UploadOptions,
722    ) -> Result<ObjectMetadata>;
723
724    /// Download data from storage.
725    async fn download(&self, key: &str) -> Result<Vec<u8>>;
726
727    /// Get object metadata.
728    async fn head(&self, key: &str) -> Result<ObjectMetadata>;
729
730    /// Delete an object.
731    async fn delete(&self, key: &str) -> Result<()>;
732
733    /// List objects with prefix.
734    async fn list(&self, prefix: &str, max_keys: Option<usize>) -> Result<Vec<ObjectMetadata>>;
735
736    /// Copy an object.
737    async fn copy(&self, source_key: &str, dest_key: &str) -> Result<ObjectMetadata>;
738
739    /// Generate a presigned URL for download.
740    async fn presigned_download_url(&self, key: &str, expires_in: Duration) -> Result<String>;
741
742    /// Generate a presigned URL for upload.
743    async fn presigned_upload_url(&self, key: &str, expires_in: Duration) -> Result<String>;
744}
745
746/// Cloud storage configuration.
747#[derive(Debug, Clone, Serialize, Deserialize)]
748pub struct CloudStorageConfig {
749    /// Storage provider.
750    pub provider: StorageProvider,
751    /// Bucket/container name.
752    pub bucket: String,
753    /// Region.
754    pub region: Option<String>,
755    /// Endpoint override (for S3-compatible storage).
756    pub endpoint: Option<String>,
757    /// Access credentials.
758    pub credentials: Option<StorageCredentials>,
759    /// Request timeout in seconds.
760    pub timeout_secs: u64,
761}
762
763/// Storage provider.
764#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
765pub enum StorageProvider {
766    /// Amazon S3.
767    S3,
768    /// Google Cloud Storage.
769    Gcs,
770    /// Azure Blob Storage.
771    AzureBlob,
772    /// MinIO (S3-compatible).
773    Minio,
774}
775
776/// Storage credentials.
777#[derive(Debug, Clone, Serialize, Deserialize)]
778pub enum StorageCredentials {
779    /// Access key credentials.
780    AccessKey {
781        /// Access key identifier.
782        access_key: String,
783        /// Secret access key.
784        secret_key: String,
785    },
786    /// Service account credentials.
787    ServiceAccount {
788        /// Path to the service account key file.
789        key_file: String,
790    },
791    /// Instance profile (AWS) or managed identity (Azure).
792    InstanceProfile,
793}
794
795// =============================================================================
796// External Service Callbacks
797// =============================================================================
798
799/// Callback event type.
800#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
801pub enum CallbackEventType {
802    /// Workflow started.
803    WorkflowStarted,
804    /// Workflow completed.
805    WorkflowCompleted,
806    /// Workflow failed.
807    WorkflowFailed,
808    /// Workflow cancelled.
809    WorkflowCancelled,
810    /// Task started.
811    TaskStarted,
812    /// Task completed.
813    TaskCompleted,
814    /// Task failed.
815    TaskFailed,
816    /// Task retry.
817    TaskRetry,
818    /// Custom event.
819    Custom,
820}
821
822impl CallbackEventType {
823    /// Get string representation.
824    pub fn as_str(&self) -> &'static str {
825        match self {
826            Self::WorkflowStarted => "workflow.started",
827            Self::WorkflowCompleted => "workflow.completed",
828            Self::WorkflowFailed => "workflow.failed",
829            Self::WorkflowCancelled => "workflow.cancelled",
830            Self::TaskStarted => "task.started",
831            Self::TaskCompleted => "task.completed",
832            Self::TaskFailed => "task.failed",
833            Self::TaskRetry => "task.retry",
834            Self::Custom => "custom",
835        }
836    }
837}
838
839/// Callback payload.
840#[derive(Debug, Clone, Serialize, Deserialize)]
841pub struct CallbackPayload {
842    /// Event type.
843    pub event_type: CallbackEventType,
844    /// Workflow ID.
845    pub workflow_id: String,
846    /// Execution ID.
847    pub execution_id: String,
848    /// Task ID (if applicable).
849    pub task_id: Option<String>,
850    /// Event timestamp.
851    pub timestamp: DateTime<Utc>,
852    /// Event data.
853    pub data: serde_json::Value,
854    /// Workflow state snapshot (optional).
855    pub state_snapshot: Option<serde_json::Value>,
856}
857
858impl CallbackPayload {
859    /// Create a new callback payload for workflow events.
860    pub fn for_workflow(
861        event_type: CallbackEventType,
862        workflow_id: impl Into<String>,
863        execution_id: impl Into<String>,
864        data: serde_json::Value,
865    ) -> Self {
866        Self {
867            event_type,
868            workflow_id: workflow_id.into(),
869            execution_id: execution_id.into(),
870            task_id: None,
871            timestamp: Utc::now(),
872            data,
873            state_snapshot: None,
874        }
875    }
876
877    /// Create a new callback payload for task events.
878    pub fn for_task(
879        event_type: CallbackEventType,
880        workflow_id: impl Into<String>,
881        execution_id: impl Into<String>,
882        task_id: impl Into<String>,
883        data: serde_json::Value,
884    ) -> Self {
885        Self {
886            event_type,
887            workflow_id: workflow_id.into(),
888            execution_id: execution_id.into(),
889            task_id: Some(task_id.into()),
890            timestamp: Utc::now(),
891            data,
892            state_snapshot: None,
893        }
894    }
895
896    /// Add state snapshot to the payload.
897    pub fn with_state_snapshot(mut self, state: &WorkflowState) -> Self {
898        self.state_snapshot = serde_json::to_value(state).ok();
899        self
900    }
901}
902
903/// Callback handler trait.
904#[async_trait]
905pub trait CallbackHandler: Send + Sync {
906    /// Handle a callback event.
907    async fn handle(&self, payload: CallbackPayload) -> Result<()>;
908
909    /// Get the handler name.
910    fn name(&self) -> &str;
911
912    /// Check if the handler is enabled for an event type.
913    fn is_enabled_for(&self, event_type: CallbackEventType) -> bool;
914}
915
916/// Callback configuration.
917#[derive(Debug, Clone, Serialize, Deserialize)]
918pub struct CallbackConfig {
919    /// Callback URL.
920    pub url: String,
921    /// HTTP method.
922    pub method: HttpMethod,
923    /// Headers to include.
924    pub headers: HashMap<String, String>,
925    /// Authentication.
926    pub auth: Option<HttpAuth>,
927    /// Enabled event types.
928    pub enabled_events: Vec<CallbackEventType>,
929    /// Include state snapshot.
930    pub include_state: bool,
931    /// Retry configuration.
932    pub retry: RetryConfig,
933}
934
935/// Retry configuration for callbacks.
936#[derive(Debug, Clone, Serialize, Deserialize)]
937pub struct RetryConfig {
938    /// Maximum number of retries.
939    pub max_retries: u32,
940    /// Initial delay in milliseconds.
941    pub initial_delay_ms: u64,
942    /// Maximum delay in milliseconds.
943    pub max_delay_ms: u64,
944    /// Backoff multiplier.
945    pub backoff_multiplier: f64,
946}
947
948impl Default for RetryConfig {
949    fn default() -> Self {
950        Self {
951            max_retries: 3,
952            initial_delay_ms: 1000,
953            max_delay_ms: 30000,
954            backoff_multiplier: 2.0,
955        }
956    }
957}
958
959/// HTTP callback handler implementation.
960pub struct HttpCallbackHandler {
961    config: CallbackConfig,
962    #[cfg(feature = "integrations")]
963    client: reqwest::Client,
964}
965
966impl HttpCallbackHandler {
967    /// Create a new HTTP callback handler.
968    #[cfg(feature = "integrations")]
969    pub fn new(config: CallbackConfig) -> Result<Self> {
970        let client = reqwest::Client::builder()
971            .timeout(Duration::from_secs(30))
972            .build()
973            .map_err(|e| {
974                WorkflowError::integration("callback", format!("Failed to create client: {}", e))
975            })?;
976
977        Ok(Self { config, client })
978    }
979
980    /// Get the callback configuration.
981    pub fn config(&self) -> &CallbackConfig {
982        &self.config
983    }
984}
985
986#[cfg(feature = "integrations")]
987#[async_trait]
988impl CallbackHandler for HttpCallbackHandler {
989    async fn handle(&self, payload: CallbackPayload) -> Result<()> {
990        let mut request = match self.config.method {
991            HttpMethod::Get => self.client.get(&self.config.url),
992            HttpMethod::Post => self.client.post(&self.config.url),
993            HttpMethod::Put => self.client.put(&self.config.url),
994            HttpMethod::Patch => self.client.patch(&self.config.url),
995            HttpMethod::Delete => self.client.delete(&self.config.url),
996            HttpMethod::Head => self.client.head(&self.config.url),
997            HttpMethod::Options => self
998                .client
999                .request(reqwest::Method::OPTIONS, &self.config.url),
1000        };
1001
1002        // Add headers
1003        for (key, value) in &self.config.headers {
1004            request = request.header(key, value);
1005        }
1006
1007        // Add authentication
1008        if let Some(auth) = &self.config.auth {
1009            request = match auth {
1010                HttpAuth::Bearer { token } => request.bearer_auth(token),
1011                HttpAuth::Basic { username, password } => {
1012                    request.basic_auth(username, Some(password))
1013                }
1014                HttpAuth::ApiKey { header_name, key } => request.header(header_name, key),
1015                HttpAuth::Custom { headers } => {
1016                    for (k, v) in headers {
1017                        request = request.header(k, v);
1018                    }
1019                    request
1020                }
1021            };
1022        }
1023
1024        let mut last_error = None;
1025        for attempt in 0..=self.config.retry.max_retries {
1026            if attempt > 0 {
1027                let delay = std::cmp::min(
1028                    (self.config.retry.initial_delay_ms as f64
1029                        * self.config.retry.backoff_multiplier.powi(attempt as i32))
1030                        as u64,
1031                    self.config.retry.max_delay_ms,
1032                );
1033                tokio::time::sleep(Duration::from_millis(delay)).await;
1034            }
1035
1036            // Clone request for retry
1037            let req = request
1038                .try_clone()
1039                .ok_or_else(|| WorkflowError::integration("callback", "Failed to clone request"))?;
1040
1041            match req.json(&payload).send().await {
1042                Ok(response) => {
1043                    if response.status().is_success() {
1044                        debug!(
1045                            "Callback delivered successfully: {} -> {}",
1046                            payload.event_type.as_str(),
1047                            self.config.url
1048                        );
1049                        return Ok(());
1050                    } else if response.status().is_server_error() {
1051                        last_error = Some(WorkflowError::integration(
1052                            "callback",
1053                            format!("Server error: {}", response.status()),
1054                        ));
1055                    } else {
1056                        return Err(WorkflowError::integration(
1057                            "callback",
1058                            format!("Client error: {}", response.status()),
1059                        ));
1060                    }
1061                }
1062                Err(e) => {
1063                    last_error = Some(WorkflowError::integration(
1064                        "callback",
1065                        format!("Request failed: {}", e),
1066                    ));
1067                }
1068            }
1069        }
1070
1071        Err(last_error.unwrap_or_else(|| {
1072            WorkflowError::integration("callback", "Callback failed after all retries")
1073        }))
1074    }
1075
1076    fn name(&self) -> &str {
1077        "http_callback"
1078    }
1079
1080    fn is_enabled_for(&self, event_type: CallbackEventType) -> bool {
1081        self.config.enabled_events.contains(&event_type)
1082    }
1083}
1084
1085// =============================================================================
1086// Webhook Support
1087// =============================================================================
1088
1089/// Webhook receiver for incoming workflow triggers.
1090#[derive(Debug, Clone, Serialize, Deserialize)]
1091pub struct WebhookTrigger {
1092    /// Trigger ID.
1093    pub id: String,
1094    /// Workflow ID to trigger.
1095    pub workflow_id: String,
1096    /// Secret for HMAC validation.
1097    pub secret: Option<String>,
1098    /// Allowed source IPs.
1099    pub allowed_ips: Vec<String>,
1100    /// Parameter mapping from webhook payload.
1101    pub parameter_mapping: HashMap<String, String>,
1102    /// Whether the webhook is active.
1103    pub active: bool,
1104}
1105
1106impl WebhookTrigger {
1107    /// Create a new webhook trigger.
1108    pub fn new(workflow_id: impl Into<String>) -> Self {
1109        Self {
1110            id: uuid::Uuid::new_v4().to_string(),
1111            workflow_id: workflow_id.into(),
1112            secret: None,
1113            allowed_ips: Vec::new(),
1114            parameter_mapping: HashMap::new(),
1115            active: true,
1116        }
1117    }
1118
1119    /// Set the webhook secret.
1120    pub fn with_secret(mut self, secret: impl Into<String>) -> Self {
1121        self.secret = Some(secret.into());
1122        self
1123    }
1124
1125    /// Add allowed IP addresses.
1126    pub fn with_allowed_ips(mut self, ips: Vec<String>) -> Self {
1127        self.allowed_ips = ips;
1128        self
1129    }
1130
1131    /// Add parameter mapping.
1132    pub fn with_parameter(
1133        mut self,
1134        webhook_field: impl Into<String>,
1135        workflow_param: impl Into<String>,
1136    ) -> Self {
1137        self.parameter_mapping
1138            .insert(webhook_field.into(), workflow_param.into());
1139        self
1140    }
1141
1142    /// Validate HMAC signature.
1143    pub fn validate_signature(&self, payload: &[u8], signature: &str) -> bool {
1144        if let Some(secret) = &self.secret {
1145            let expected = format!("sha256={}", hmac_sha256_hex(payload, secret.as_bytes()));
1146            // Normalize to lowercase — some webhook providers send uppercase hex
1147            let signature_lower = signature.to_lowercase();
1148            constant_time_compare(&expected, &signature_lower)
1149        } else {
1150            true // No secret configured, skip validation
1151        }
1152    }
1153}
1154
1155/// Constant-time string comparison to prevent timing attacks.
1156fn constant_time_compare(a: &str, b: &str) -> bool {
1157    if a.len() != b.len() {
1158        return false;
1159    }
1160    let result = a
1161        .bytes()
1162        .zip(b.bytes())
1163        .fold(0u8, |acc, (x, y)| acc | (x ^ y));
1164    result == 0
1165}
1166
1167/// HMAC-SHA256 hex digest for webhook signature validation.
1168///
1169/// Returns an empty string in the astronomically unlikely case that `KeyInit::new_from_slice`
1170/// returns an error (HMAC is defined for keys of any length; this branch is unreachable).
1171fn hmac_sha256_hex(data: &[u8], key: &[u8]) -> String {
1172    use hmac::KeyInit;
1173    use hmac::{Hmac, Mac};
1174    use sha2::Sha256;
1175    match <Hmac<Sha256>>::new_from_slice(key) {
1176        Ok(mut mac) => {
1177            mac.update(data);
1178            mac.finalize()
1179                .into_bytes()
1180                .iter()
1181                .map(|b| format!("{b:02x}"))
1182                .collect()
1183        }
1184        Err(_) => String::new(),
1185    }
1186}
1187
1188/// Webhook registry for managing webhook endpoints.
1189pub struct WebhookRegistry {
1190    triggers: Arc<RwLock<HashMap<String, WebhookTrigger>>>,
1191}
1192
1193impl WebhookRegistry {
1194    /// Create a new webhook registry.
1195    pub fn new() -> Self {
1196        Self {
1197            triggers: Arc::new(RwLock::new(HashMap::new())),
1198        }
1199    }
1200
1201    /// Register a webhook trigger.
1202    pub async fn register(&self, trigger: WebhookTrigger) -> String {
1203        let id = trigger.id.clone();
1204        let mut triggers = self.triggers.write().await;
1205        triggers.insert(id.clone(), trigger);
1206        info!("Registered webhook trigger: {}", id);
1207        id
1208    }
1209
1210    /// Unregister a webhook trigger.
1211    pub async fn unregister(&self, trigger_id: &str) -> Option<WebhookTrigger> {
1212        let mut triggers = self.triggers.write().await;
1213        let removed = triggers.remove(trigger_id);
1214        if removed.is_some() {
1215            info!("Unregistered webhook trigger: {}", trigger_id);
1216        }
1217        removed
1218    }
1219
1220    /// Get a webhook trigger by ID.
1221    pub async fn get(&self, trigger_id: &str) -> Option<WebhookTrigger> {
1222        let triggers = self.triggers.read().await;
1223        triggers.get(trigger_id).cloned()
1224    }
1225
1226    /// List all webhook triggers.
1227    pub async fn list(&self) -> Vec<WebhookTrigger> {
1228        let triggers = self.triggers.read().await;
1229        triggers.values().cloned().collect()
1230    }
1231
1232    /// Find triggers for a workflow.
1233    pub async fn find_by_workflow(&self, workflow_id: &str) -> Vec<WebhookTrigger> {
1234        let triggers = self.triggers.read().await;
1235        triggers
1236            .values()
1237            .filter(|t| t.workflow_id == workflow_id && t.active)
1238            .cloned()
1239            .collect()
1240    }
1241}
1242
1243impl Default for WebhookRegistry {
1244    fn default() -> Self {
1245        Self::new()
1246    }
1247}
1248
1249// =============================================================================
1250// Event Emission to External Systems
1251// =============================================================================
1252
1253/// External event for emission.
1254#[derive(Debug, Clone, Serialize, Deserialize)]
1255pub struct ExternalEvent {
1256    /// Event ID.
1257    pub id: String,
1258    /// Event type.
1259    pub event_type: String,
1260    /// Event source.
1261    pub source: String,
1262    /// Event subject.
1263    pub subject: Option<String>,
1264    /// Event timestamp.
1265    pub timestamp: DateTime<Utc>,
1266    /// Event data.
1267    pub data: serde_json::Value,
1268    /// Event metadata.
1269    pub metadata: HashMap<String, String>,
1270}
1271
1272impl ExternalEvent {
1273    /// Create a new external event.
1274    pub fn new(
1275        event_type: impl Into<String>,
1276        source: impl Into<String>,
1277        data: serde_json::Value,
1278    ) -> Self {
1279        Self {
1280            id: uuid::Uuid::new_v4().to_string(),
1281            event_type: event_type.into(),
1282            source: source.into(),
1283            subject: None,
1284            timestamp: Utc::now(),
1285            data,
1286            metadata: HashMap::new(),
1287        }
1288    }
1289
1290    /// Set event subject.
1291    pub fn with_subject(mut self, subject: impl Into<String>) -> Self {
1292        self.subject = Some(subject.into());
1293        self
1294    }
1295
1296    /// Add metadata.
1297    pub fn with_metadata(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
1298        self.metadata.insert(key.into(), value.into());
1299        self
1300    }
1301
1302    /// Create from workflow state.
1303    pub fn from_workflow_state(state: &WorkflowState, event_type: impl Into<String>) -> Self {
1304        Self::new(
1305            event_type,
1306            format!("workflow/{}", state.workflow_id),
1307            serde_json::json!({
1308                "workflow_id": state.workflow_id,
1309                "execution_id": state.execution_id,
1310                "status": format!("{:?}", state.status),
1311            }),
1312        )
1313        .with_subject(state.execution_id.clone())
1314    }
1315
1316    /// Create from task state.
1317    pub fn from_task_state(
1318        workflow_id: &str,
1319        task: &TaskState,
1320        event_type: impl Into<String>,
1321    ) -> Self {
1322        Self::new(
1323            event_type,
1324            format!("workflow/{}/task/{}", workflow_id, task.task_id),
1325            serde_json::json!({
1326                "task_id": task.task_id,
1327                "status": format!("{:?}", task.status),
1328                "attempts": task.attempts,
1329            }),
1330        )
1331        .with_subject(task.task_id.clone())
1332    }
1333}
1334
1335/// Event emitter trait.
1336#[async_trait]
1337pub trait EventEmitter: Send + Sync {
1338    /// Emit an event.
1339    async fn emit(&self, event: ExternalEvent) -> Result<()>;
1340
1341    /// Emit multiple events.
1342    async fn emit_batch(&self, events: Vec<ExternalEvent>) -> Result<Vec<Result<()>>> {
1343        let mut results = Vec::with_capacity(events.len());
1344        for event in events {
1345            results.push(self.emit(event).await);
1346        }
1347        Ok(results)
1348    }
1349
1350    /// Get emitter name.
1351    fn name(&self) -> &str;
1352}
1353
1354/// Event emitter configuration.
1355#[derive(Debug, Clone, Serialize, Deserialize)]
1356pub struct EventEmitterConfig {
1357    /// Emitter type.
1358    pub emitter_type: EventEmitterType,
1359    /// Target endpoint/topic.
1360    pub target: String,
1361    /// Authentication.
1362    pub auth: Option<EmitterAuth>,
1363    /// Batch size for batched emission.
1364    pub batch_size: usize,
1365    /// Flush interval in milliseconds.
1366    pub flush_interval_ms: u64,
1367}
1368
1369/// Event emitter type.
1370#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
1371pub enum EventEmitterType {
1372    /// HTTP webhook.
1373    Webhook,
1374    /// Message queue.
1375    MessageQueue,
1376    /// CloudEvents.
1377    CloudEvents,
1378    /// Custom.
1379    Custom,
1380}
1381
1382/// Emitter authentication.
1383#[derive(Debug, Clone, Serialize, Deserialize)]
1384pub enum EmitterAuth {
1385    /// Bearer token.
1386    Bearer {
1387        /// The bearer token value.
1388        token: String,
1389    },
1390    /// API key.
1391    ApiKey {
1392        /// API key value.
1393        key: String,
1394    },
1395    /// OAuth2.
1396    OAuth2 {
1397        /// OAuth2 client identifier.
1398        client_id: String,
1399        /// OAuth2 client secret.
1400        client_secret: String,
1401    },
1402}
1403
1404/// Multi-emitter that broadcasts events to multiple targets.
1405pub struct MultiEventEmitter {
1406    emitters: Vec<Arc<dyn EventEmitter>>,
1407}
1408
1409impl MultiEventEmitter {
1410    /// Create a new multi-emitter.
1411    pub fn new() -> Self {
1412        Self {
1413            emitters: Vec::new(),
1414        }
1415    }
1416
1417    /// Add an emitter.
1418    pub fn add_emitter(&mut self, emitter: Arc<dyn EventEmitter>) {
1419        self.emitters.push(emitter);
1420    }
1421
1422    /// Remove an emitter by name.
1423    pub fn remove_emitter(&mut self, name: &str) {
1424        self.emitters.retain(|e| e.name() != name);
1425    }
1426
1427    /// Get the number of registered emitters.
1428    pub fn emitter_count(&self) -> usize {
1429        self.emitters.len()
1430    }
1431}
1432
1433impl Default for MultiEventEmitter {
1434    fn default() -> Self {
1435        Self::new()
1436    }
1437}
1438
1439#[async_trait]
1440impl EventEmitter for MultiEventEmitter {
1441    async fn emit(&self, event: ExternalEvent) -> Result<()> {
1442        let mut errors = Vec::new();
1443
1444        for emitter in &self.emitters {
1445            if let Err(e) = emitter.emit(event.clone()).await {
1446                error!("Failed to emit event via {}: {}", emitter.name(), e);
1447                errors.push(e);
1448            }
1449        }
1450
1451        if errors.is_empty() {
1452            Ok(())
1453        } else {
1454            Err(WorkflowError::integration(
1455                "multi_emitter",
1456                format!("Failed to emit to {} targets", errors.len()),
1457            ))
1458        }
1459    }
1460
1461    fn name(&self) -> &str {
1462        "multi_emitter"
1463    }
1464}
1465
1466// =============================================================================
1467// Integration Registry
1468// =============================================================================
1469
1470/// External integration registry.
1471pub struct ExternalIntegrationRegistry {
1472    /// Registered callback handlers.
1473    callbacks: Arc<RwLock<Vec<Arc<dyn CallbackHandler>>>>,
1474    /// Webhook registry.
1475    webhooks: Arc<WebhookRegistry>,
1476    /// Event emitters.
1477    emitters: Arc<RwLock<MultiEventEmitter>>,
1478}
1479
1480impl ExternalIntegrationRegistry {
1481    /// Create a new external integration registry.
1482    pub fn new() -> Self {
1483        Self {
1484            callbacks: Arc::new(RwLock::new(Vec::new())),
1485            webhooks: Arc::new(WebhookRegistry::new()),
1486            emitters: Arc::new(RwLock::new(MultiEventEmitter::new())),
1487        }
1488    }
1489
1490    /// Register a callback handler.
1491    pub async fn register_callback(&self, handler: Arc<dyn CallbackHandler>) {
1492        let mut callbacks = self.callbacks.write().await;
1493        info!("Registering callback handler: {}", handler.name());
1494        callbacks.push(handler);
1495    }
1496
1497    /// Dispatch a callback to all registered handlers.
1498    pub async fn dispatch_callback(&self, payload: CallbackPayload) -> Result<()> {
1499        let callbacks = self.callbacks.read().await;
1500        let mut errors = Vec::new();
1501
1502        for handler in callbacks.iter() {
1503            if handler.is_enabled_for(payload.event_type) {
1504                if let Err(e) = handler.handle(payload.clone()).await {
1505                    error!("Callback handler {} failed: {}", handler.name(), e);
1506                    errors.push(e);
1507                }
1508            }
1509        }
1510
1511        if errors.is_empty() {
1512            Ok(())
1513        } else {
1514            Err(WorkflowError::integration(
1515                "callbacks",
1516                format!("{} callback handlers failed", errors.len()),
1517            ))
1518        }
1519    }
1520
1521    /// Get the webhook registry.
1522    pub fn webhooks(&self) -> &Arc<WebhookRegistry> {
1523        &self.webhooks
1524    }
1525
1526    /// Register an event emitter.
1527    pub async fn register_emitter(&self, emitter: Arc<dyn EventEmitter>) {
1528        let mut emitters = self.emitters.write().await;
1529        info!("Registering event emitter: {}", emitter.name());
1530        emitters.add_emitter(emitter);
1531    }
1532
1533    /// Emit an event to all registered emitters.
1534    pub async fn emit_event(&self, event: ExternalEvent) -> Result<()> {
1535        let emitters = self.emitters.read().await;
1536        emitters.emit(event).await
1537    }
1538
1539    /// Emit workflow started event.
1540    pub async fn emit_workflow_started(&self, state: &WorkflowState) -> Result<()> {
1541        let event = ExternalEvent::from_workflow_state(state, "workflow.started");
1542        let callback = CallbackPayload::for_workflow(
1543            CallbackEventType::WorkflowStarted,
1544            &state.workflow_id,
1545            &state.execution_id,
1546            serde_json::json!({"status": "started"}),
1547        );
1548
1549        let emit_result = self.emit_event(event).await;
1550        let callback_result = self.dispatch_callback(callback).await;
1551
1552        if emit_result.is_err() || callback_result.is_err() {
1553            warn!("Some integrations failed for workflow.started event");
1554        }
1555
1556        Ok(())
1557    }
1558
1559    /// Emit workflow completed event.
1560    pub async fn emit_workflow_completed(&self, state: &WorkflowState) -> Result<()> {
1561        let event = ExternalEvent::from_workflow_state(state, "workflow.completed");
1562        let callback = CallbackPayload::for_workflow(
1563            CallbackEventType::WorkflowCompleted,
1564            &state.workflow_id,
1565            &state.execution_id,
1566            serde_json::json!({"status": "completed"}),
1567        )
1568        .with_state_snapshot(state);
1569
1570        let _ = self.emit_event(event).await;
1571        let _ = self.dispatch_callback(callback).await;
1572
1573        Ok(())
1574    }
1575
1576    /// Emit workflow failed event.
1577    pub async fn emit_workflow_failed(&self, state: &WorkflowState, error: &str) -> Result<()> {
1578        let event = ExternalEvent::from_workflow_state(state, "workflow.failed")
1579            .with_metadata("error", error);
1580        let callback = CallbackPayload::for_workflow(
1581            CallbackEventType::WorkflowFailed,
1582            &state.workflow_id,
1583            &state.execution_id,
1584            serde_json::json!({"status": "failed", "error": error}),
1585        )
1586        .with_state_snapshot(state);
1587
1588        let _ = self.emit_event(event).await;
1589        let _ = self.dispatch_callback(callback).await;
1590
1591        Ok(())
1592    }
1593
1594    /// Emit task started event.
1595    pub async fn emit_task_started(&self, workflow_id: &str, task: &TaskState) -> Result<()> {
1596        let event = ExternalEvent::from_task_state(workflow_id, task, "task.started");
1597        let callback = CallbackPayload::for_task(
1598            CallbackEventType::TaskStarted,
1599            workflow_id,
1600            "",
1601            &task.task_id,
1602            serde_json::json!({"status": "started"}),
1603        );
1604
1605        let _ = self.emit_event(event).await;
1606        let _ = self.dispatch_callback(callback).await;
1607
1608        Ok(())
1609    }
1610
1611    /// Emit task completed event.
1612    pub async fn emit_task_completed(&self, workflow_id: &str, task: &TaskState) -> Result<()> {
1613        let event = ExternalEvent::from_task_state(workflow_id, task, "task.completed");
1614        let callback = CallbackPayload::for_task(
1615            CallbackEventType::TaskCompleted,
1616            workflow_id,
1617            "",
1618            &task.task_id,
1619            serde_json::json!({"status": "completed", "output": task.output}),
1620        );
1621
1622        let _ = self.emit_event(event).await;
1623        let _ = self.dispatch_callback(callback).await;
1624
1625        Ok(())
1626    }
1627
1628    /// Emit task failed event.
1629    pub async fn emit_task_failed(&self, workflow_id: &str, task: &TaskState) -> Result<()> {
1630        let event = ExternalEvent::from_task_state(workflow_id, task, "task.failed")
1631            .with_metadata("error", task.error.as_deref().unwrap_or("unknown"));
1632        let callback = CallbackPayload::for_task(
1633            CallbackEventType::TaskFailed,
1634            workflow_id,
1635            "",
1636            &task.task_id,
1637            serde_json::json!({"status": "failed", "error": task.error}),
1638        );
1639
1640        let _ = self.emit_event(event).await;
1641        let _ = self.dispatch_callback(callback).await;
1642
1643        Ok(())
1644    }
1645}
1646
1647impl Default for ExternalIntegrationRegistry {
1648    fn default() -> Self {
1649        Self::new()
1650    }
1651}
1652
1653// =============================================================================
1654// Tests
1655// =============================================================================
1656
1657#[cfg(test)]
1658mod tests {
1659    use super::*;
1660
1661    #[test]
1662    fn test_http_method_as_str() {
1663        assert_eq!(HttpMethod::Get.as_str(), "GET");
1664        assert_eq!(HttpMethod::Post.as_str(), "POST");
1665        assert_eq!(HttpMethod::Put.as_str(), "PUT");
1666        assert_eq!(HttpMethod::Delete.as_str(), "DELETE");
1667    }
1668
1669    #[test]
1670    fn test_http_response_status_checks() {
1671        let success_response = HttpResponse {
1672            status_code: 200,
1673            headers: HashMap::new(),
1674            body: None,
1675            response_time_ms: 100,
1676            is_success: true,
1677        };
1678        assert!(success_response.is_success());
1679        assert!(!success_response.is_client_error());
1680        assert!(!success_response.is_server_error());
1681
1682        let client_error = HttpResponse {
1683            status_code: 404,
1684            headers: HashMap::new(),
1685            body: None,
1686            response_time_ms: 50,
1687            is_success: false,
1688        };
1689        assert!(!client_error.is_success());
1690        assert!(client_error.is_client_error());
1691
1692        let server_error = HttpResponse {
1693            status_code: 500,
1694            headers: HashMap::new(),
1695            body: None,
1696            response_time_ms: 75,
1697            is_success: false,
1698        };
1699        assert!(!server_error.is_success());
1700        assert!(server_error.is_server_error());
1701    }
1702
1703    #[test]
1704    fn test_queue_message_builder() {
1705        let msg = QueueMessage::new("test-topic", serde_json::json!({"key": "value"}))
1706            .with_correlation_id("corr-123")
1707            .with_reply_to("reply-topic")
1708            .with_priority(8)
1709            .with_header("custom", "header");
1710
1711        assert_eq!(msg.topic, "test-topic");
1712        assert_eq!(msg.correlation_id, Some("corr-123".to_string()));
1713        assert_eq!(msg.reply_to, Some("reply-topic".to_string()));
1714        assert_eq!(msg.priority, 8);
1715        assert_eq!(msg.headers.get("custom"), Some(&"header".to_string()));
1716    }
1717
1718    #[test]
1719    fn test_query_result() {
1720        let result = QueryResult::empty();
1721        assert!(result.is_empty());
1722        assert_eq!(result.row_count(), 0);
1723
1724        let result_with_data = QueryResult {
1725            rows_affected: 1,
1726            columns: vec!["id".to_string(), "name".to_string()],
1727            rows: vec![vec![serde_json::json!(1), serde_json::json!("test")]],
1728            execution_time_ms: 10,
1729        };
1730        assert!(!result_with_data.is_empty());
1731        assert_eq!(result_with_data.row_count(), 1);
1732    }
1733
1734    #[test]
1735    fn test_callback_event_type_as_str() {
1736        assert_eq!(
1737            CallbackEventType::WorkflowStarted.as_str(),
1738            "workflow.started"
1739        );
1740        assert_eq!(CallbackEventType::TaskCompleted.as_str(), "task.completed");
1741        assert_eq!(CallbackEventType::Custom.as_str(), "custom");
1742    }
1743
1744    #[test]
1745    fn test_callback_payload_creation() {
1746        let workflow_payload = CallbackPayload::for_workflow(
1747            CallbackEventType::WorkflowStarted,
1748            "wf-1",
1749            "exec-1",
1750            serde_json::json!({}),
1751        );
1752        assert_eq!(workflow_payload.workflow_id, "wf-1");
1753        assert_eq!(workflow_payload.execution_id, "exec-1");
1754        assert!(workflow_payload.task_id.is_none());
1755
1756        let task_payload = CallbackPayload::for_task(
1757            CallbackEventType::TaskStarted,
1758            "wf-1",
1759            "exec-1",
1760            "task-1",
1761            serde_json::json!({}),
1762        );
1763        assert_eq!(task_payload.task_id, Some("task-1".to_string()));
1764    }
1765
1766    #[test]
1767    fn test_webhook_trigger_creation() {
1768        let trigger = WebhookTrigger::new("workflow-1")
1769            .with_secret("my-secret")
1770            .with_allowed_ips(vec!["192.168.1.1".to_string()])
1771            .with_parameter("payload.id", "workflow_param");
1772
1773        assert_eq!(trigger.workflow_id, "workflow-1");
1774        assert!(trigger.secret.is_some());
1775        assert_eq!(trigger.allowed_ips.len(), 1);
1776        assert!(trigger.parameter_mapping.contains_key("payload.id"));
1777    }
1778
1779    #[tokio::test]
1780    async fn test_webhook_registry() {
1781        let registry = WebhookRegistry::new();
1782
1783        let trigger = WebhookTrigger::new("workflow-1");
1784        let id = registry.register(trigger).await;
1785
1786        assert!(registry.get(&id).await.is_some());
1787
1788        let triggers = registry.list().await;
1789        assert_eq!(triggers.len(), 1);
1790
1791        let workflow_triggers = registry.find_by_workflow("workflow-1").await;
1792        assert_eq!(workflow_triggers.len(), 1);
1793
1794        assert!(registry.unregister(&id).await.is_some());
1795        assert!(registry.get(&id).await.is_none());
1796    }
1797
1798    #[test]
1799    fn test_external_event_creation() {
1800        let event = ExternalEvent::new(
1801            "test.event",
1802            "test-source",
1803            serde_json::json!({"data": "value"}),
1804        )
1805        .with_subject("subject-1")
1806        .with_metadata("key", "value");
1807
1808        assert_eq!(event.event_type, "test.event");
1809        assert_eq!(event.source, "test-source");
1810        assert_eq!(event.subject, Some("subject-1".to_string()));
1811        assert_eq!(event.metadata.get("key"), Some(&"value".to_string()));
1812    }
1813
1814    #[test]
1815    fn test_retry_config_default() {
1816        let config = RetryConfig::default();
1817        assert_eq!(config.max_retries, 3);
1818        assert_eq!(config.initial_delay_ms, 1000);
1819        assert_eq!(config.max_delay_ms, 30000);
1820        assert!((config.backoff_multiplier - 2.0).abs() < f64::EPSILON);
1821    }
1822
1823    #[test]
1824    fn test_http_client_config_default() {
1825        let config = HttpClientConfig::default();
1826        assert_eq!(config.timeout_secs, 30);
1827        assert_eq!(config.max_retries, 3);
1828        assert!(config.base_url.is_empty());
1829    }
1830
1831    #[test]
1832    fn test_database_config_default() {
1833        let config = DatabaseConfig::default();
1834        assert_eq!(config.db_type, DatabaseType::PostgreSql);
1835        assert_eq!(config.pool_min, 1);
1836        assert_eq!(config.pool_max, 10);
1837    }
1838
1839    #[tokio::test]
1840    async fn test_multi_event_emitter() {
1841        let emitter = MultiEventEmitter::new();
1842        assert_eq!(emitter.emitter_count(), 0);
1843    }
1844
1845    #[tokio::test]
1846    async fn test_external_integration_registry() {
1847        let registry = ExternalIntegrationRegistry::new();
1848
1849        let trigger = WebhookTrigger::new("workflow-1");
1850        let _ = registry.webhooks().register(trigger).await;
1851
1852        let triggers = registry.webhooks().list().await;
1853        assert_eq!(triggers.len(), 1);
1854    }
1855
1856    #[test]
1857    fn test_constant_time_compare() {
1858        assert!(constant_time_compare("test", "test"));
1859        assert!(!constant_time_compare("test", "Test"));
1860        assert!(!constant_time_compare("test", "test1"));
1861    }
1862}
oxigdal_workflow/integrations/external.rs

oxigdal_workflow/integrations/
external.rs
