refactor: update Kinesis sink for dynamic routing and payload-only output

- Make stream_name config optional (can come from event metadata)
- Add resolve_stream_name() for dynamic routing from metadata key "stream"
- Send only event.payload instead of full event envelope
- Use DestinationError for publish failures, ConfigError for missing config
- Update tests to verify payload-only content and metadata-based routing

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

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

Author: psteinroe

src/sink/kinesis.rs

@@ -1,27 +1,23 @@
 //! AWS Kinesis sink for publishing events to a Kinesis data stream.
 //!
-//! Publishes each event as a data record to the configured stream.
-//! The sink uses the AWS SDK for async message delivery.
+//! Publishes each event's payload as a data record to a stream determined by:
+//! 1. `stream` key in event metadata (from subscription's metadata/metadata_extensions)
+//! 2. Fallback to `stream_name` in sink config
 //!
-//! # Payload Extensions
+//! # Dynamic Routing
 //!
-//! This sink supports the following optional payload extensions:
+//! The target stream can be configured per-event using metadata_extensions:
 //!
 //! ```sql
-//! payload_extensions = '[
-//!   {"key": "partition_key", "value": "user_123"},
-//!   {"key": "explicit_hash_key", "value": "abc123"}
+//! metadata_extensions = '[
+//!   {"json_path": "stream", "expression": "''my-stream-'' || shard_key"}
 //! ]'
 //! ```
-//!
-//! - `partition_key`: Used for shard distribution. If not provided, the event ID is used.
-//! - `explicit_hash_key`: Optional hash key to override partition key hashing.
 
 use aws_sdk_kinesis::Client;
 use etl::error::EtlResult;
 use serde::{Deserialize, Serialize};
 use std::sync::Arc;
-use tracing::info;
 
 use crate::sink::Sink;
 use crate::types::TriggeredEvent;
@@ -32,8 +28,8 @@ use crate::types::TriggeredEvent;
 /// leaking secrets (AWS credentials, endpoint URLs) in serialized forms.
 #[derive(Clone, Debug, Deserialize)]
 pub struct KinesisSinkConfig {
-    /// Kinesis stream name to publish records to.
-    pub stream_name: String,
+    /// Kinesis stream name. Optional if provided via event metadata.
+    pub stream_name: Option<String>,
 
     /// AWS region (e.g., "us-east-1").
     pub region: String,
@@ -56,8 +52,8 @@ pub struct KinesisSinkConfig {
 /// Safe to serialize and log. Use this for debugging and metrics.
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct KinesisSinkConfigWithoutSecrets {
-    /// Kinesis stream name.
-    pub stream_name: String,
+    /// Kinesis stream name (if configured).
+    pub stream_name: Option<String>,
 
     /// AWS region.
     pub region: String,
@@ -93,15 +89,15 @@ impl From<&KinesisSinkConfig> for KinesisSinkConfigWithoutSecrets {
 
 /// Sink that publishes events to an AWS Kinesis data stream.
 ///
-/// Each event is serialized as JSON and published as a data record.
+/// Each event's payload is serialized as JSON and published as a data record.
 /// The sink uses the AWS SDK with automatic retry handling.
 #[derive(Clone)]
 pub struct KinesisSink {
     /// AWS Kinesis client.
     client: Arc<Client>,
 
-    /// Stream name to publish records to.
-    stream_name: String,
+    /// Default stream name. Can be overridden per-event via metadata.
+    stream_name: Option<String>,
 }
 
 impl KinesisSink {
@@ -146,6 +142,18 @@ impl KinesisSink {
             stream_name: config.stream_name,
         })
     }
+
+    /// Resolves the stream name for an event from metadata or config.
+    fn resolve_stream_name<'a>(&'a self, event: &'a TriggeredEvent) -> Option<&'a str> {
+        // First check event metadata for dynamic stream
+        if let Some(ref metadata) = event.metadata {
+            if let Some(stream) = metadata.get("stream").and_then(|v| v.as_str()) {
+                return Some(stream);
+            }
+        }
+        // Fall back to config stream name
+        self.stream_name.as_deref()
+    }
 }
 
 impl Sink for KinesisSink {
@@ -158,33 +166,21 @@ impl Sink for KinesisSink {
             return Ok(());
         }
 
-        info!(
-            "publishing {} events to Kinesis stream '{}'",
-            events.len(),
-            self.stream_name
-        );
-
         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 stream name from event metadata or config.
+            let stream_name = self.resolve_stream_name(event).ok_or_else(|| {
+                etl::etl_error!(
+                    etl::error::ErrorKind::ConfigError,
+                    "No stream name configured",
+                    "Stream name must be provided in sink config or event metadata"
+                )
+            })?;
 
-            let data = serde_json::to_vec(&json_obj).map_err(|e| {
+            // Serialize payload to JSON.
+            let data = serde_json::to_vec(&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()
                 )
             })?;
@@ -195,14 +191,14 @@ impl Sink for KinesisSink {
             // Publish record to Kinesis stream.
             self.client
                 .put_record()
-                .stream_name(&self.stream_name)
+                .stream_name(stream_name)
                 .partition_key(&partition_key)
                 .data(aws_sdk_kinesis::primitives::Blob::new(data))
                 .send()
                 .await
                 .map_err(|e| {
                     etl::etl_error!(
-                        etl::error::ErrorKind::InvalidData,
+                        etl::error::ErrorKind::DestinationError,
                         "Failed to publish record to Kinesis",
                         e.to_string()
                     )
@@ -225,7 +221,7 @@ mod tests {
     #[test]
     fn test_config_without_secrets() {
         let config = KinesisSinkConfig {
-            stream_name: "my-stream".to_string(),
+            stream_name: Some("my-stream".to_string()),
             region: "us-east-1".to_string(),
             endpoint_url: Some("http://localhost:4566".to_string()),
             access_key_id: Some("AKIAIOSFODNN7EXAMPLE".to_string()),
@@ -234,7 +230,7 @@ mod tests {
 
         let without_secrets: KinesisSinkConfigWithoutSecrets = (&config).into();
 
-        assert_eq!(without_secrets.stream_name, "my-stream");
+        assert_eq!(without_secrets.stream_name, Some("my-stream".to_string()));
         assert_eq!(without_secrets.region, "us-east-1");
         assert!(without_secrets.has_custom_endpoint);
         assert!(without_secrets.has_explicit_credentials);

tests/kinesis_sink_tests.rs

@@ -157,7 +157,7 @@ async fn test_kinesis_sink_publishes_events() {
 
     // Create sink.
     let config = KinesisSinkConfig {
-        stream_name: stream_name.clone(),
+        stream_name: Some(stream_name.clone()),
         region: "us-east-1".to_string(),
         endpoint_url: Some(endpoint.clone()),
         access_key_id: Some("test".to_string()),
@@ -170,15 +170,20 @@ async fn test_kinesis_sink_publishes_events() {
     let events = vec![make_test_event("event1"), make_test_event("event2")];
     sink.publish_events(events).await.expect("Failed to publish");
 
-    // Verify records arrived.
+    // Verify records arrived - only payload is sent.
     let records = get_kinesis_records(&endpoint, &stream_name, 2).await;
     assert_eq!(records.len(), 2, "Expected 2 records");
 
     for record in &records {
-        let event: serde_json::Value =
+        let payload: serde_json::Value =
             serde_json::from_str(record).expect("Failed to parse record");
-        assert!(event["id"].is_string());
-        assert!(event["payload"]["key"].is_string());
+        // Only payload fields should be present.
+        assert!(payload["key"].is_string());
+        assert!(payload["value"].is_string());
+        // No envelope fields.
+        assert!(payload.get("id").is_none());
+        assert!(payload.get("metadata").is_none());
+        assert!(payload.get("lsn").is_none());
     }
 }
 
@@ -191,7 +196,7 @@ async fn test_kinesis_sink_handles_empty_batch() {
     create_kinesis_stream(&endpoint, &stream_name).await;
 
     let config = KinesisSinkConfig {
-        stream_name,
+        stream_name: Some(stream_name),
         region: "us-east-1".to_string(),
         endpoint_url: Some(endpoint),
         access_key_id: Some("test".to_string()),
@@ -207,15 +212,16 @@ async fn test_kinesis_sink_handles_empty_batch() {
 }
 
 #[tokio::test(flavor = "multi_thread")]
-async fn test_kinesis_sink_includes_metadata() {
+async fn test_kinesis_sink_uses_stream_from_metadata() {
     let port = ensure_localstack().await;
     let endpoint = format!("http://127.0.0.1:{}", port);
 
     let stream_name = format!("test-stream-{}", Uuid::new_v4());
     create_kinesis_stream(&endpoint, &stream_name).await;
 
+    // Create sink without stream_name - will get it from metadata.
     let config = KinesisSinkConfig {
-        stream_name: stream_name.clone(),
+        stream_name: None,
         region: "us-east-1".to_string(),
         endpoint_url: Some(endpoint.clone()),
         access_key_id: Some("test".to_string()),
@@ -224,12 +230,12 @@ async fn test_kinesis_sink_includes_metadata() {
 
     let sink = KinesisSink::new(config).await.expect("Failed to create sink");
 
-    // Create event with metadata.
+    // Create event with stream in metadata.
     let event = TriggeredEvent {
         id: EventIdentifier::new(Uuid::new_v4().to_string(), Utc::now()),
         stream_id: StreamId::default(),
         payload: serde_json::json!({ "action": "created" }),
-        metadata: Some(serde_json::json!({ "user_id": 123, "source": "api" })),
+        metadata: Some(serde_json::json!({ "stream": stream_name })),
         lsn: Some(PgLsn::from(99999u64)),
     };
 
@@ -240,12 +246,13 @@ async fn test_kinesis_sink_includes_metadata() {
     let records = get_kinesis_records(&endpoint, &stream_name, 1).await;
     assert_eq!(records.len(), 1);
 
-    let published: serde_json::Value =
+    let payload: serde_json::Value =
         serde_json::from_str(&records[0]).expect("Failed to parse record");
 
-    assert_eq!(published["metadata"]["user_id"], 123);
-    assert_eq!(published["metadata"]["source"], "api");
-    assert!(published["lsn"].is_string());
+    // Only payload is sent.
+    assert_eq!(payload["action"], "created");
+    assert!(payload.get("id").is_none());
+    assert!(payload.get("metadata").is_none());
 }
 
 #[test]