feat(sink): add AWS SQS sink

Cargo.toml

@@ -10,6 +10,7 @@ sink-nats          = ["dep:async-nats"]
 sink-rabbitmq      = ["dep:lapin"]
 sink-redis-streams = ["dep:redis"]
 sink-redis-strings = ["dep:redis"]
+sink-sqs           = ["dep:aws-sdk-sqs", "dep:aws-config"]
 sink-webhook       = ["dep:reqwest"]
 test-utils         = ["dep:ctor", "dep:testcontainers", "dep:testcontainers-modules"]
 
@@ -54,6 +55,8 @@ uuid         = { version = "1.19.0", default-features = false, features = ["v4"]
 
 # Optional sink dependencies.
 async-nats = { version = "0.38", optional = true }
+aws-config = { version = "1.5", optional = true }
+aws-sdk-sqs = { version = "1.60", optional = true }
 lapin = { version = "2.5", optional = true }
 rdkafka = { version = "0.36", optional = true, default-features = false, features = ["tokio", "cmake-build"] }
 redis = { version = "0.27", default-features = false, features = [
@@ -79,6 +82,7 @@ testcontainers-modules = { version = "0.11", optional = true, features = [
   "nats",
   "rabbitmq",
   "kafka",
+  "elasticmq",
   "blocking",
 ] }
 

src/config/sink.rs

@@ -18,6 +18,9 @@ use crate::sink::rabbitmq::RabbitmqSinkConfig;
 #[cfg(feature = "sink-webhook")]
 use crate::sink::webhook::WebhookSinkConfig;
 
+#[cfg(feature = "sink-sqs")]
+use crate::sink::sqs::SqsSinkConfig;
+
 /// Sink destination configuration.
 ///
 /// Determines where replicated events are sent.
@@ -56,4 +59,9 @@ pub enum SinkConfig {
     #[cfg(feature = "sink-kafka")]
     #[serde(rename = "kafka")]
     Kafka(KafkaSinkConfig),
+
+    /// AWS SQS sink for queue messaging.
+    #[cfg(feature = "sink-sqs")]
+    #[serde(rename = "sqs")]
+    Sqs(SqsSinkConfig),
 }

src/core.rs

@@ -155,6 +155,19 @@ async fn run_pipeline(config: &PipelineConfig) -> EtlResult<()> {
             })?;
             AnySink::Kafka(s)
         }
+
+        #[cfg(feature = "sink-sqs")]
+        SinkConfig::Sqs(cfg) => {
+            use crate::sink::sqs::SqsSink;
+            let s = SqsSink::new(cfg.clone()).await.map_err(|e| {
+                etl::etl_error!(
+                    etl::error::ErrorKind::InvalidData,
+                    "Failed to create SQS sink",
+                    e.to_string()
+                )
+            })?;
+            AnySink::Sqs(s)
+        }
     };
 
     // Create PgStream as an ETL destination
@@ -230,6 +243,11 @@ fn log_sink_config(config: &SinkConfig) {
         SinkConfig::Kafka(_cfg) => {
             debug!("using kafka sink");
         }
+
+        #[cfg(feature = "sink-sqs")]
+        SinkConfig::Sqs(_cfg) => {
+            debug!("using sqs sink");
+        }
     }
 }
 

src/sink/mod.rs

@@ -19,6 +19,9 @@ pub mod webhook;
 #[cfg(feature = "sink-kafka")]
 pub mod kafka;
 
+#[cfg(feature = "sink-sqs")]
+pub mod sqs;
+
 pub use base::Sink;
 
 use etl::error::EtlResult;
@@ -42,6 +45,9 @@ use webhook::WebhookSink;
 #[cfg(feature = "sink-kafka")]
 use kafka::KafkaSink;
 
+#[cfg(feature = "sink-sqs")]
+use sqs::SqsSink;
+
 use crate::types::TriggeredEvent;
 
 /// Wrapper enum for all supported sink types.
@@ -76,6 +82,10 @@ pub enum AnySink {
     /// Kafka sink for Apache Kafka messaging.
     #[cfg(feature = "sink-kafka")]
     Kafka(KafkaSink),
+
+    /// AWS SQS sink for queue messaging.
+    #[cfg(feature = "sink-sqs")]
+    Sqs(SqsSink),
 }
 
 impl Sink for AnySink {
@@ -104,6 +114,9 @@ impl Sink for AnySink {
 
             #[cfg(feature = "sink-kafka")]
             AnySink::Kafka(sink) => sink.publish_events(events).await,
+
+            #[cfg(feature = "sink-sqs")]
+            AnySink::Sqs(sink) => sink.publish_events(events).await,
         }
     }
 }