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

- Make exchange and routing_key config optional (can come from event metadata)
- Add resolve_exchange() and resolve_routing_key() for dynamic routing
- 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/rabbitmq.rs

@@ -1,17 +1,21 @@
 //! RabbitMQ sink for publishing events to a message queue.
 //!
-//! Publishes each event as a JSON message to the configured exchange
-//! with the specified routing key.
+//! Publishes each event's payload as a JSON message to an exchange determined by:
+//! 1. `exchange` key in event metadata (from subscription's metadata/metadata_extensions)
+//! 2. Fallback to `exchange` in sink config
 //!
-//! # Payload Extensions
+//! The routing key is determined by:
+//! 1. `routing_key` key in event metadata
+//! 2. Fallback to `routing_key` in sink config
 //!
-//! This sink does not require any specific payload extensions. However, you can
-//! use payload extensions to add routing metadata that can be used with RabbitMQ's
-//! topic exchange routing:
+//! # Dynamic Routing
+//!
+//! The target exchange and routing key can be configured per-event using metadata_extensions:
 //!
 //! ```sql
-//! payload_extensions = '[
-//!   {"key": "routing_key", "value": "orders.created"}
+//! metadata_extensions = '[
+//!   {"json_path": "exchange", "expression": "''events''"},
+//!   {"json_path": "routing_key", "expression": "table_name || ''.'' || operation"}
 //! ]'
 //! ```
 
@@ -24,7 +28,6 @@ use lapin::{
 use serde::{Deserialize, Serialize};
 use std::sync::Arc;
 use tokio::sync::Mutex;
-use tracing::info;
 
 use crate::sink::Sink;
 use crate::types::TriggeredEvent;
@@ -39,11 +42,13 @@ pub struct RabbitmqSinkConfig {
     /// Contains credentials and should be treated as sensitive.
     pub url: String,
 
-    /// Exchange name to publish messages to.
-    pub exchange: String,
+    /// Exchange name to publish messages to. Optional if provided via event metadata.
+    #[serde(default)]
+    pub exchange: Option<String>,
 
-    /// Routing key for message routing.
-    pub routing_key: String,
+    /// Routing key for message routing. Optional if provided via event metadata.
+    #[serde(default)]
+    pub routing_key: Option<String>,
 
     /// Optional queue name to bind to the exchange.
     /// If provided, the queue will be declared and bound.
@@ -56,11 +61,11 @@ pub struct RabbitmqSinkConfig {
 /// Safe to serialize and log. Use this for debugging and metrics.
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct RabbitmqSinkConfigWithoutSecrets {
-    /// Exchange name to publish messages to.
-    pub exchange: String,
+    /// Exchange name to publish messages to (if configured).
+    pub exchange: Option<String>,
 
-    /// Routing key for message routing.
-    pub routing_key: String,
+    /// Routing key for message routing (if configured).
+    pub routing_key: Option<String>,
 
     /// Optional queue name to bind to the exchange.
     pub queue: Option<String>,
@@ -88,19 +93,18 @@ impl From<&RabbitmqSinkConfig> for RabbitmqSinkConfigWithoutSecrets {
 
 /// Sink that publishes events to a RabbitMQ exchange.
 ///
-/// Each event is serialized as JSON and published to the configured exchange
-/// with the specified routing key. The sink maintains a persistent connection
-/// and channel for publishing.
+/// Each event's payload is serialized as JSON and published to the exchange.
+/// The sink maintains a persistent connection and channel for publishing.
 #[derive(Clone)]
 pub struct RabbitmqSink {
     /// Shared channel for publishing messages.
     channel: Arc<Mutex<Channel>>,
 
-    /// Exchange name to publish messages to.
-    exchange: String,
+    /// Default exchange name. Can be overridden per-event via metadata.
+    exchange: Option<String>,
 
-    /// Routing key for message routing.
-    routing_key: String,
+    /// Default routing key. Can be overridden per-event via metadata.
+    routing_key: Option<String>,
 }
 
 impl RabbitmqSink {
@@ -120,41 +124,45 @@ impl RabbitmqSink {
 
         let channel = connection.create_channel().await?;
 
-        // Declare the exchange as a topic exchange.
-        channel
-            .exchange_declare(
-                &config.exchange,
-                lapin::ExchangeKind::Topic,
-                ExchangeDeclareOptions {
-                    durable: true,
-                    ..Default::default()
-                },
-                FieldTable::default(),
-            )
-            .await?;
-
-        // If a queue is specified, declare and bind it.
-        if let Some(ref queue_name) = config.queue {
+        // Declare the exchange as a topic exchange if provided.
+        if let Some(ref exchange) = config.exchange {
             channel
-                .queue_declare(
-                    queue_name,
-                    QueueDeclareOptions {
+                .exchange_declare(
+                    exchange,
+                    lapin::ExchangeKind::Topic,
+                    ExchangeDeclareOptions {
                         durable: true,
                         ..Default::default()
                     },
                     FieldTable::default(),
                 )
                 .await?;
 
-            channel
-                .queue_bind(
-                    queue_name,
-                    &config.exchange,
-                    &config.routing_key,
-                    QueueBindOptions::default(),
-                    FieldTable::default(),
-                )
-                .await?;
+            // If a queue is specified, declare and bind it.
+            if let Some(ref queue_name) = config.queue {
+                if let Some(ref routing_key) = config.routing_key {
+                    channel
+                        .queue_declare(
+                            queue_name,
+                            QueueDeclareOptions {
+                                durable: true,
+                                ..Default::default()
+                            },
+                            FieldTable::default(),
+                        )
+                        .await?;
+
+                    channel
+                        .queue_bind(
+                            queue_name,
+                            exchange,
+                            routing_key,
+                            QueueBindOptions::default(),
+                            FieldTable::default(),
+                        )
+                        .await?;
+                }
+            }
         }
 
         Ok(Self {
@@ -163,6 +171,30 @@ impl RabbitmqSink {
             routing_key: config.routing_key,
         })
     }
+
+    /// Resolves the exchange name for an event from metadata or config.
+    fn resolve_exchange<'a>(&'a self, event: &'a TriggeredEvent) -> Option<&'a str> {
+        // First check event metadata for dynamic exchange.
+        if let Some(ref metadata) = event.metadata {
+            if let Some(exchange) = metadata.get("exchange").and_then(|v| v.as_str()) {
+                return Some(exchange);
+            }
+        }
+        // Fall back to config exchange.
+        self.exchange.as_deref()
+    }
+
+    /// Resolves the routing key for an event from metadata or config.
+    fn resolve_routing_key<'a>(&'a self, event: &'a TriggeredEvent) -> Option<&'a str> {
+        // First check event metadata for dynamic routing key.
+        if let Some(ref metadata) = event.metadata {
+            if let Some(routing_key) = metadata.get("routing_key").and_then(|v| v.as_str()) {
+                return Some(routing_key);
+            }
+        }
+        // Fall back to config routing key.
+        self.routing_key.as_deref()
+    }
 }
 
 impl Sink for RabbitmqSink {
@@ -175,44 +207,41 @@ impl Sink for RabbitmqSink {
             return Ok(());
         }
 
-        info!(
-            "publishing {} events to RabbitMQ exchange '{}'",
-            events.len(),
-            self.exchange
-        );
-
         let channel = self.channel.lock().await;
 
         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 exchange from event metadata or config.
+            let exchange = self.resolve_exchange(event).ok_or_else(|| {
+                etl::etl_error!(
+                    etl::error::ErrorKind::ConfigError,
+                    "No exchange configured",
+                    "Exchange must be provided in sink config or event metadata"
+                )
+            })?;
+
+            // Resolve routing key from event metadata or config.
+            let routing_key = self.resolve_routing_key(event).ok_or_else(|| {
+                etl::etl_error!(
+                    etl::error::ErrorKind::ConfigError,
+                    "No routing_key configured",
+                    "Routing key must be provided in sink config or event metadata"
+                )
+            })?;
 
-            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 exchange with routing key.
             channel
                 .basic_publish(
-                    &self.exchange,
-                    &self.routing_key,
+                    exchange,
+                    routing_key,
                     BasicPublishOptions::default(),
                     &payload,
                     BasicProperties::default()
@@ -222,15 +251,15 @@ impl Sink for RabbitmqSink {
                 .await
                 .map_err(|e| {
                     etl::etl_error!(
-                        etl::error::ErrorKind::InvalidData,
+                        etl::error::ErrorKind::DestinationError,
                         "Failed to publish event to RabbitMQ",
                         e.to_string()
                     )
                 })?
                 .await
                 .map_err(|e| {
                     etl::etl_error!(
-                        etl::error::ErrorKind::InvalidData,
+                        etl::error::ErrorKind::DestinationError,
                         "Failed to confirm RabbitMQ publish",
                         e.to_string()
                     )