Merge pull request #15 from rohas-dev/feat/adapter-sqs

feat(adapter): introduce AWS adapter support for SQS and EventBridge, refactor existing SQS adapter, and update configuration handling

Author: sophatvathana

Cargo.lock

@@ -13,7 +13,7 @@ members = [
     "crates/rohas-adapters/adapter-nats",
     "crates/rohas-adapters/adapter-kafka",
     "crates/rohas-adapters/adapter-rabbitmq",
-    "crates/rohas-adapters/adapter-sqs",
+    "crates/rohas-adapters/adapter-aws",
     "crates/rohas-adapters/adapter-rocksdb",
 ]
 
@@ -70,6 +70,7 @@ async-nats = "0.45.0"
 rdkafka = { version = "0.38.0", features = ["cmake-build"] }
 lapin = "3.7.2"
 aws-sdk-sqs = "1.9"
+aws-sdk-eventbridge = "1.9"
 
 # Watching / Hot reload
 notify = "8.2.0"
@@ -95,7 +96,7 @@ adapter-memory = { path = "crates/rohas-adapters/adapter-memory" }
 adapter-nats = { path = "crates/rohas-adapters/adapter-nats" }
 adapter-kafka = { path = "crates/rohas-adapters/adapter-kafka" }
 adapter-rabbitmq = { path = "crates/rohas-adapters/adapter-rabbitmq" }
-adapter-sqs = { path = "crates/rohas-adapters/adapter-sqs" }
+adapter-aws = { path = "crates/rohas-adapters/adapter-aws" }
 adapter-rocksdb = { path = "crates/rohas-adapters/adapter-rocksdb" }
 rohas-telemetry = { path = "crates/rohas-telemetry" }
 

Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "adapter-sqs"
+name = "adapter-aws"
 version = { workspace = true }
 edition = { workspace = true }
 authors = { workspace = true }
@@ -8,6 +8,8 @@ license = { workspace = true }
 [dependencies]
 tokio = { workspace = true }
 aws-sdk-sqs = { workspace = true }
+aws-sdk-eventbridge = "1.9"
+aws-config = "1.1"
 serde = { workspace = true }
 serde_json = { workspace = true }
 thiserror = { workspace = true }

…es/rohas-adapters/adapter-sqs/Cargo.toml …es/rohas-adapters/adapter-aws/Cargo.tomlcrates/rohas-adapters/adapter-sqs/Cargo.toml renamed to crates/rohas-adapters/adapter-aws/Cargo.toml crates/rohas-adapters/adapter-sqs/Cargo.toml renamed to crates/rohas-adapters/adapter-aws/Cargo.toml

@@ -0,0 +1,87 @@
+use async_trait::async_trait;
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use thiserror::Error;
+
+pub type Result<T> = std::result::Result<T, AdapterError>;
+
+#[derive(Error, Debug)]
+pub enum AdapterError {
+    #[error("AWS SQS error: {0}")]
+    AwsSqs(String),
+
+    #[error("AWS EventBridge error: {0}")]
+    AwsEventBridge(String),
+
+    #[error("Queue not found: {0}")]
+    QueueNotFound(String),
+
+    #[error("Serialization error: {0}")]
+    Serialization(#[from] serde_json::Error),
+
+    #[error("Invalid message format: {0}")]
+    InvalidMessage(String),
+
+    #[error("Configuration error: {0}")]
+    Configuration(String),
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Message {
+    pub topic: String,
+    pub payload: serde_json::Value,
+    pub timestamp: String,
+    pub metadata: HashMap<String, String>,
+}
+
+impl Message {
+    pub fn new(topic: impl Into<String>, payload: serde_json::Value) -> Self {
+        use std::time::SystemTime;
+        Self {
+            topic: topic.into(),
+            payload,
+            timestamp: SystemTime::now()
+                .duration_since(SystemTime::UNIX_EPOCH)
+                .unwrap()
+                .as_secs()
+                .to_string(),
+            metadata: HashMap::new(),
+        }
+    }
+
+    pub fn with_metadata(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
+        self.metadata.insert(key.into(), value.into());
+        self
+    }
+}
+
+#[async_trait]
+pub trait MessageHandler: Send + Sync {
+    async fn handle(&self, message: Message) -> Result<()>;
+}
+
+#[derive(Debug, Clone)]
+pub struct AwsConfig {
+    pub region: String,
+    pub queue_prefix: Option<String>, // For SQS
+    pub event_bus_name: Option<String>, // For EventBridge (default: "default")
+    pub source: Option<String>, // For EventBridge (default: "rohas")
+    pub visibility_timeout_seconds: Option<i32>, // For SQS
+    pub message_retention_seconds: Option<i32>, // For SQS
+    pub receive_wait_time_seconds: Option<i32>, // For SQS (long polling)
+}
+
+impl Default for AwsConfig {
+    fn default() -> Self {
+        Self {
+            region: "us-east-1".to_string(),
+            queue_prefix: Some("rohas-".to_string()),
+            event_bus_name: None, // Use default event bus
+            source: Some("rohas".to_string()),
+            visibility_timeout_seconds: Some(30),
+            message_retention_seconds: Some(345600), // 4 days
+            receive_wait_time_seconds: Some(20), // Long polling
+        }
+    }
+}
+