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

- Make subject config optional (can come from event metadata key "topic")
- Add resolve_subject() for dynamic routing from metadata
- Send only event.payload instead of full event envelope
- Remove info! logging
- 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/nats.rs

@@ -1,24 +1,23 @@
 //! NATS sink for publishing events to a NATS subject.
 //!
-//! Publishes each event as a JSON message to the configured subject.
-//! The sink maintains a persistent connection with automatic reconnection.
+//! Publishes each event's payload as a JSON message to a subject determined by:
+//! 1. `topic` key in event metadata (from subscription's metadata/metadata_extensions)
+//! 2. Fallback to `subject` in sink config
 //!
-//! # Payload Extensions
+//! # Dynamic Routing
 //!
-//! This sink does not require any specific payload extensions. However, you can
-//! use payload extensions to add routing metadata for NATS subject hierarchies:
+//! The target subject can be configured per-event using metadata_extensions:
 //!
 //! ```sql
-//! payload_extensions = '[
-//!   {"key": "subject_suffix", "value": "orders.created"}
+//! metadata_extensions = '[
+//!   {"json_path": "topic", "expression": "''events.'' || table_name"}
 //! ]'
 //! ```
 
 use async_nats::Client;
 use etl::error::EtlResult;
 use serde::{Deserialize, Serialize};
 use std::sync::Arc;
-use tracing::info;
 
 use crate::sink::Sink;
 use crate::types::TriggeredEvent;
@@ -33,17 +32,18 @@ pub struct NatsSinkConfig {
     /// Contains credentials and should be treated as sensitive.
     pub url: String,
 
-    /// Subject to publish messages to.
-    pub subject: String,
+    /// Subject to publish messages to. Optional if provided via event metadata.
+    #[serde(default)]
+    pub subject: Option<String>,
 }
 
 /// Configuration for the NATS sink without sensitive data.
 ///
 /// Safe to serialize and log. Use this for debugging and metrics.
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct NatsSinkConfigWithoutSecrets {
-    /// Subject to publish messages to.
-    pub subject: String,
+    /// Subject to publish messages to (if configured).
+    pub subject: Option<String>,
 }
 
 impl From<NatsSinkConfig> for NatsSinkConfigWithoutSecrets {
@@ -64,15 +64,15 @@ impl From<&NatsSinkConfig> for NatsSinkConfigWithoutSecrets {
 
 /// Sink that publishes events to a NATS subject.
 ///
-/// Each event is serialized as JSON and published to the configured subject.
+/// Each event's payload is serialized as JSON and published to the subject.
 /// The NATS client handles connection pooling and automatic reconnection.
 #[derive(Clone)]
 pub struct NatsSink {
     /// Shared NATS client connection.
     client: Arc<Client>,
 
-    /// Subject to publish messages to.
-    subject: String,
+    /// Default subject to publish messages to. Can be overridden per-event via metadata.
+    subject: Option<String>,
 }
 
 impl NatsSink {
@@ -91,6 +91,18 @@ impl NatsSink {
             subject: config.subject,
         })
     }
+
+    /// Resolves the subject for an event from metadata or config.
+    fn resolve_subject<'a>(&'a self, event: &'a TriggeredEvent) -> Option<&'a str> {
+        // First check event metadata for dynamic subject (using generic "topic" key).
+        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 subject.
+        self.subject.as_deref()
+    }
 }
 
 impl Sink for NatsSink {
@@ -103,44 +115,32 @@ impl Sink for NatsSink {
             return Ok(());
         }
 
-        info!(
-            "publishing {} events to NATS subject '{}'",
-            events.len(),
-            self.subject
-        );
-
         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 subject from event metadata or config.
+            let subject = self.resolve_subject(event).ok_or_else(|| {
+                etl::etl_error!(
+                    etl::error::ErrorKind::ConfigError,
+                    "No subject configured",
+                    "Subject must be provided in sink config or event metadata (topic key)"
+                )
+            })?;
 
-            let payload = serde_json::to_vec(&json_obj).map_err(|e| {
+            // Serialize payload to JSON.
+            let payload = 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()
                 )
             })?;
 
-            // Publish to the configured subject.
+            // Publish to the resolved subject.
             self.client
-                .publish(self.subject.clone(), payload.into())
+                .publish(subject.to_string(), payload.into())
                 .await
                 .map_err(|e| {
                     etl::etl_error!(
-                        etl::error::ErrorKind::InvalidData,
+                        etl::error::ErrorKind::DestinationError,
                         "Failed to publish event to NATS",
                         e.to_string()
                     )

tests/nats_sink_tests.rs

@@ -32,7 +32,7 @@ async fn test_nats_sink_publishes_events() {
 
     let config = NatsSinkConfig {
         url: nats_url.clone(),
-        subject: subject.to_string(),
+        subject: Some(subject.to_string()),
     };
 
     let sink = NatsSink::new(config)
@@ -66,16 +66,21 @@ async fn test_nats_sink_publishes_events() {
         .expect("Timeout waiting for second message")
         .expect("Expected second message");
 
-    // Parse and verify message content.
-    let event1: serde_json::Value =
+    // Parse and verify message content - only payload is sent.
+    let payload1: serde_json::Value =
         serde_json::from_slice(&msg1.payload).expect("Failed to parse first message");
-    let event2: serde_json::Value =
+    let payload2: serde_json::Value =
         serde_json::from_slice(&msg2.payload).expect("Failed to parse second message");
 
-    assert!(event1.get("id").is_some());
-    assert!(event1.get("payload").is_some());
-    assert!(event2.get("id").is_some());
-    assert!(event2.get("payload").is_some());
+    // Only payload fields should be present.
+    assert!(payload1.get("test_id").is_some());
+    assert!(payload1.get("message").is_some());
+    assert!(payload2.get("test_id").is_some());
+    assert!(payload2.get("message").is_some());
+    // No envelope fields.
+    assert!(payload1.get("id").is_none());
+    assert!(payload1.get("metadata").is_none());
+    assert!(payload1.get("lsn").is_none());
 }
 
 #[tokio::test(flavor = "multi_thread")]
@@ -85,7 +90,7 @@ async fn test_nats_sink_empty_batch() {
 
     let config = NatsSinkConfig {
         url: nats_url,
-        subject: "pgstream.empty-test".to_string(),
+        subject: Some("pgstream.empty-test".to_string()),
     };
 
     let sink = NatsSink::new(config)
@@ -98,6 +103,63 @@ async fn test_nats_sink_empty_batch() {
         .expect("Empty batch should succeed");
 }
 
+#[tokio::test(flavor = "multi_thread")]
+async fn test_nats_sink_uses_topic_from_metadata() {
+    let nats_port = ensure_nats().await;
+    let nats_url = format!("nats://127.0.0.1:{}", nats_port);
+    let subject = "pgstream.metadata-topic";
+
+    // Create sink without subject - will get it from metadata.
+    let config = NatsSinkConfig {
+        url: nats_url.clone(),
+        subject: None,
+    };
+
+    let sink = NatsSink::new(config)
+        .await
+        .expect("Failed to create NATS sink");
+
+    // Subscribe before publishing to verify messages.
+    let client = async_nats::connect(&nats_url)
+        .await
+        .expect("Failed to connect to NATS");
+
+    let mut subscriber = client
+        .subscribe(subject.to_string())
+        .await
+        .expect("Failed to subscribe");
+
+    // Create event with topic in metadata.
+    let event = TriggeredEvent {
+        id: EventIdentifier::new("nats-metadata-event".to_string(), Utc::now()),
+        payload: serde_json::json!({
+            "test_id": "metadata-event",
+            "message": "Test event with metadata topic",
+        }),
+        metadata: Some(serde_json::json!({ "topic": subject })),
+        stream_id: StreamId::from(1u64),
+        lsn: Some("0/16B3748".parse().unwrap()),
+    };
+
+    sink.publish_events(vec![event])
+        .await
+        .expect("Failed to publish events");
+
+    // Verify event was received.
+    let msg = tokio::time::timeout(std::time::Duration::from_secs(5), subscriber.next())
+        .await
+        .expect("Timeout waiting for message")
+        .expect("Expected message");
+
+    // Parse and verify message content - only payload is sent.
+    let payload: serde_json::Value =
+        serde_json::from_slice(&msg.payload).expect("Failed to parse message");
+
+    assert_eq!(payload["test_id"], "metadata-event");
+    assert!(payload.get("id").is_none());
+    assert!(payload.get("metadata").is_none());
+}
+
 #[test]
 fn test_sink_name() {
     assert_eq!(NatsSink::name(), "nats");