refactor: update SNS sink per FEEDBACK.md

- Send only event.payload (not full event structure)
- Make topic_arn optional in config (can come from event.metadata)
- Add dynamic topic ARN resolution from event.metadata
- Use ErrorKind::DestinationError for publish failures
- Use ErrorKind::ConfigError for missing topic ARN
- Remove verbose info! logging on every publish
- Update tests to verify payload-only output

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: psteinroe

src/sink/sns.rs

@@ -1,25 +1,23 @@
 //! AWS SNS sink for publishing events to an Amazon SNS topic.
 //!
-//! Publishes each event as a JSON message to the configured topic ARN.
-//! The sink uses the AWS SDK for async message delivery.
+//! Publishes each event's payload as JSON to a topic ARN determined by:
+//! 1. `topic` key in event metadata (from subscription's metadata/metadata_extensions)
+//! 2. Fallback to `topic_arn` in sink config
 //!
-//! # Payload Extensions
+//! # Dynamic Routing
 //!
-//! This sink does not require any specific payload extensions. However, you can
-//! use payload extensions to add SNS message attributes:
+//! The target topic ARN can be configured per-event using metadata_extensions:
 //!
 //! ```sql
-//! payload_extensions = '[
-//!   {"key": "message_group_id", "value": "orders"},
-//!   {"key": "subject", "value": "New Order"}
+//! metadata_extensions = '[
+//!   {"json_path": "topic", "expression": "''arn:aws:sns:us-east-1:123456789:'' || topic_name"}
 //! ]'
 //! ```
 
 use aws_sdk_sns::Client;
 use etl::error::EtlResult;
 use serde::{Deserialize, Serialize};
 use std::sync::Arc;
-use tracing::info;
 
 use crate::sink::Sink;
 use crate::types::TriggeredEvent;
@@ -30,8 +28,8 @@ use crate::types::TriggeredEvent;
 /// leaking secrets (AWS credentials, endpoint URLs) in serialized forms.
 #[derive(Clone, Debug, Deserialize)]
 pub struct SnsSinkConfig {
-    /// SNS topic ARN to publish messages to.
-    pub topic_arn: String,
+    /// SNS topic ARN to publish messages to. Optional if provided via event metadata.
+    pub topic_arn: Option<String>,
 
     /// AWS region (e.g., "us-east-1").
     pub region: String,
@@ -54,8 +52,8 @@ pub struct SnsSinkConfig {
 /// Safe to serialize and log. Use this for debugging and metrics.
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct SnsSinkConfigWithoutSecrets {
-    /// SNS topic ARN to publish messages to.
-    pub topic_arn: String,
+    /// SNS topic ARN to publish messages to (if configured).
+    pub topic_arn: Option<String>,
 
     /// AWS region.
     pub region: String,
@@ -91,15 +89,15 @@ impl From<&SnsSinkConfig> for SnsSinkConfigWithoutSecrets {
 
 /// Sink that publishes events to an AWS SNS topic.
 ///
-/// Each event is serialized as JSON and published to the configured topic.
+/// Each event's payload is serialized as JSON and published to the configured topic.
 /// The sink uses the AWS SDK with automatic retry handling.
 #[derive(Clone)]
 pub struct SnsSink {
     /// AWS SNS client.
     client: Arc<Client>,
 
-    /// Topic ARN to publish messages to.
-    topic_arn: String,
+    /// Default topic ARN. Can be overridden per-event via metadata.
+    topic_arn: Option<String>,
 }
 
 impl SnsSink {
@@ -144,6 +142,18 @@ impl SnsSink {
             topic_arn: config.topic_arn,
         })
     }
+
+    /// Resolves the topic ARN for an event from metadata or config.
+    fn resolve_topic_arn<'a>(&'a self, event: &'a TriggeredEvent) -> Option<&'a str> {
+        // First check event metadata for dynamic topic
+        if let Some(ref metadata) = event.metadata {
+            if let Some(topic) = metadata.get("topic").and_then(|v| v.as_str()) {
+                return Some(topic);
+            }
+        }
+        // Fall back to config topic ARN
+        self.topic_arn.as_deref()
+    }
 }
 
 impl Sink for SnsSink {
@@ -156,47 +166,35 @@ impl Sink for SnsSink {
             return Ok(());
         }
 
-        info!(
-            "publishing {} events to SNS topic '{}'",
-            events.len(),
-            self.topic_arn
-        );
-
         for event in &events {
-            // Build JSON object manually since TriggeredEvent doesn't implement Serialize.
-            let mut json_obj = serde_json::json!({
-                "id": event.id.id,
-                "created_at": event.id.created_at.to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
-                "payload": event.payload,
-                "stream_id": format!("{:?}", event.stream_id),
-            });
-
-            // Add optional fields.
-            if let Some(ref metadata) = event.metadata {
-                json_obj["metadata"] = metadata.clone();
-            }
-            if let Some(lsn) = event.lsn {
-                json_obj["lsn"] = serde_json::json!(lsn.to_string());
-            }
+            // Resolve topic ARN from event metadata or config.
+            let topic_arn = self.resolve_topic_arn(event).ok_or_else(|| {
+                etl::etl_error!(
+                    etl::error::ErrorKind::ConfigError,
+                    "No topic ARN configured",
+                    "Topic ARN must be provided in sink config or event metadata"
+                )
+            })?;
 
-            let message = serde_json::to_string(&json_obj).map_err(|e| {
+            // Serialize payload to JSON.
+            let message = serde_json::to_string(&event.payload).map_err(|e| {
                 etl::etl_error!(
                     etl::error::ErrorKind::InvalidData,
-                    "Failed to serialize event to JSON",
+                    "Failed to serialize payload to JSON",
                     e.to_string()
                 )
             })?;
 
             // Publish message to SNS topic.
             self.client
                 .publish()
-                .topic_arn(&self.topic_arn)
+                .topic_arn(topic_arn)
                 .message(&message)
                 .send()
                 .await
                 .map_err(|e| {
                     etl::etl_error!(
-                        etl::error::ErrorKind::InvalidData,
+                        etl::error::ErrorKind::DestinationError,
                         "Failed to publish message to SNS",
                         e.to_string()
                     )
@@ -219,7 +217,7 @@ mod tests {
     #[test]
     fn test_config_without_secrets() {
         let config = SnsSinkConfig {
-            topic_arn: "arn:aws:sns:us-east-1:123456789:my-topic".to_string(),
+            topic_arn: Some("arn:aws:sns:us-east-1:123456789:my-topic".to_string()),
             region: "us-east-1".to_string(),
             endpoint_url: Some("http://localhost:4566".to_string()),
             access_key_id: Some("AKIAIOSFODNN7EXAMPLE".to_string()),
@@ -230,7 +228,7 @@ mod tests {
 
         assert_eq!(
             without_secrets.topic_arn,
-            "arn:aws:sns:us-east-1:123456789:my-topic"
+            Some("arn:aws:sns:us-east-1:123456789:my-topic".to_string())
         );
         assert_eq!(without_secrets.region, "us-east-1");
         assert!(without_secrets.has_custom_endpoint);

tests/sns_sink_tests.rs

@@ -203,7 +203,7 @@ async fn test_sns_sink_publishes_events() {
 
     // Create sink.
     let config = SnsSinkConfig {
-        topic_arn: topic_arn.clone(),
+        topic_arn: Some(topic_arn.clone()),
         region: "us-east-1".to_string(),
         endpoint_url: Some(endpoint.clone()),
         access_key_id: Some("test".to_string()),
@@ -221,14 +221,16 @@ async fn test_sns_sink_publishes_events() {
     assert_eq!(messages.len(), 2, "Expected 2 messages");
 
     // SNS wraps messages in an envelope, so we need to parse the outer JSON.
+    // Only payload is sent now.
     for msg in &messages {
         let envelope: serde_json::Value =
             serde_json::from_str(msg).expect("Failed to parse envelope");
         let inner_msg = envelope["Message"].as_str().expect("Missing Message field");
-        let event: serde_json::Value =
-            serde_json::from_str(inner_msg).expect("Failed to parse event");
-        assert!(event["id"].is_string());
-        assert!(event["payload"]["key"].is_string());
+        let payload: serde_json::Value =
+            serde_json::from_str(inner_msg).expect("Failed to parse payload");
+        // Payload fields from make_test_event
+        assert!(payload["key"].is_string());
+        assert!(payload["value"].is_string());
     }
 }
 
@@ -240,7 +242,7 @@ async fn test_sns_sink_handles_empty_batch() {
     let topic_arn = create_sns_topic(&endpoint).await;
 
     let config = SnsSinkConfig {
-        topic_arn,
+        topic_arn: Some(topic_arn),
         region: "us-east-1".to_string(),
         endpoint_url: Some(endpoint),
         access_key_id: Some("test".to_string()),
@@ -256,7 +258,7 @@ async fn test_sns_sink_handles_empty_batch() {
 }
 
 #[tokio::test(flavor = "multi_thread")]
-async fn test_sns_sink_includes_metadata() {
+async fn test_sns_sink_sends_only_payload() {
     let port = ensure_localstack().await;
     let endpoint = format!("http://127.0.0.1:{}", port);
 
@@ -266,7 +268,7 @@ async fn test_sns_sink_includes_metadata() {
     subscribe_sqs_to_sns(&endpoint, &topic_arn, &queue_arn).await;
 
     let config = SnsSinkConfig {
-        topic_arn,
+        topic_arn: Some(topic_arn),
         region: "us-east-1".to_string(),
         endpoint_url: Some(endpoint.clone()),
         access_key_id: Some("test".to_string()),
@@ -275,11 +277,11 @@ async fn test_sns_sink_includes_metadata() {
 
     let sink = SnsSink::new(config).await.expect("Failed to create sink");
 
-    // Create event with metadata.
+    // Create event with metadata (metadata is used for routing, not sent).
     let event = TriggeredEvent {
         id: EventIdentifier::new(Uuid::new_v4().to_string(), Utc::now()),
         stream_id: StreamId::default(),
-        payload: serde_json::json!({ "action": "created" }),
+        payload: serde_json::json!({ "action": "created", "data": "test" }),
         metadata: Some(serde_json::json!({ "user_id": 123, "source": "api" })),
         lsn: Some(PgLsn::from(99999u64)),
     };
@@ -294,12 +296,16 @@ async fn test_sns_sink_includes_metadata() {
     let envelope: serde_json::Value =
         serde_json::from_str(&messages[0]).expect("Failed to parse envelope");
     let inner_msg = envelope["Message"].as_str().expect("Missing Message field");
-    let published: serde_json::Value =
-        serde_json::from_str(inner_msg).expect("Failed to parse event");
-
-    assert_eq!(published["metadata"]["user_id"], 123);
-    assert_eq!(published["metadata"]["source"], "api");
-    assert!(published["lsn"].is_string());
+    let payload: serde_json::Value =
+        serde_json::from_str(inner_msg).expect("Failed to parse payload");
+
+    // Only payload is sent - verify payload fields, not id/metadata/lsn.
+    assert_eq!(payload["action"], "created");
+    assert_eq!(payload["data"], "test");
+    // These should NOT be in the message since we only send payload.
+    assert!(payload.get("id").is_none());
+    assert!(payload.get("metadata").is_none());
+    assert!(payload.get("lsn").is_none());
 }
 
 #[test]