feat(sink): add NATS sink

Add a new sink implementation that publishes events to NATS subjects.
Features include:
- Publishes JSON-serialized events to configurable subject
- Uses async-nats client with automatic reconnection
- Feature-gated behind `sink-nats` flag
- Adds AnySink enum pattern for runtime sink dispatch

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

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

Author: psteinroe

Cargo.lock

@@ -6,6 +6,7 @@ version = "0.1.0"
 [features]
 default    = []
 test-utils = ["dep:ctor", "dep:testcontainers", "dep:testcontainers-modules"]
+sink-nats  = ["dep:async-nats"]
 
 [dependencies]
 anyhow = { version = "1.0.98", default-features = false, features = ["std"] }
@@ -46,6 +47,9 @@ etl          = { git = "https://github.com/supabase/etl", rev = "dd2987a55efc16a
 etl-postgres = { git = "https://github.com/supabase/etl", rev = "dd2987a55efc16aeb4402e4b06853c7731a6155a" }
 uuid         = { version = "1.19.0", default-features = false, features = ["v4"] }
 
+# Optional sink dependencies.
+async-nats = { version = "0.38", optional = true }
+
 [target.'cfg(not(target_env = "msvc"))'.dependencies]
 tikv-jemalloc-ctl = { version = "0.6.0", default-features = false, features = ["stats"] }
 tikv-jemallocator = { version = "0.6.1", default-features = false, features = [
@@ -56,7 +60,7 @@ tikv-jemallocator = { version = "0.6.1", default-features = false, features = [
 
 ctor                   = { version = "0.4", optional = true }
 testcontainers         = { version = "0.23", optional = true, features = ["blocking"] }
-testcontainers-modules = { version = "0.11", optional = true, features = ["postgres", "blocking"] }
+testcontainers-modules = { version = "0.11", optional = true, features = ["postgres", "nats", "blocking"] }
 
 [dev-dependencies]
 temp-env = "0.3"

Cargo.toml

@@ -1,5 +1,8 @@
 use serde::Deserialize;
 
+#[cfg(feature = "sink-nats")]
+use crate::sink::nats::NatsSinkConfig;
+
 /// Sink destination configuration.
 ///
 /// Determines where replicated events are sent.
@@ -8,4 +11,8 @@ use serde::Deserialize;
 pub enum SinkConfig {
     /// In-memory sink for testing and development.
     Memory,
+
+    /// NATS sink for pub/sub messaging.
+    #[cfg(feature = "sink-nats")]
+    Nats(NatsSinkConfig),
 }

src/config/sink.rs

@@ -5,7 +5,7 @@
 use crate::{
     config::{PipelineConfig, SinkConfig},
     migrations::migrate_etl,
-    sink::memory::MemorySink,
+    sink::{AnySink, memory::MemorySink},
     slot_recovery::{handle_slot_recovery, is_slot_invalidation_error},
     stream::PgStream,
 };
@@ -74,9 +74,22 @@ async fn run_pipeline(config: &PipelineConfig) -> EtlResult<()> {
     // Initialize state store for ETL pipeline state tracking
     let state_store = PostgresStore::new(config.stream.id, config.stream.pg_connection.clone());
 
-    // Create sink based on configuration
+    // Create sink based on configuration.
     let sink = match &config.sink {
-        SinkConfig::Memory => MemorySink::new(),
+        SinkConfig::Memory => AnySink::Memory(MemorySink::new()),
+
+        #[cfg(feature = "sink-nats")]
+        SinkConfig::Nats(cfg) => {
+            use crate::sink::nats::NatsSink;
+            let s = NatsSink::new(cfg.clone()).await.map_err(|e| {
+                etl::etl_error!(
+                    etl::error::ErrorKind::InvalidData,
+                    "Failed to create NATS sink",
+                    e.to_string()
+                )
+            })?;
+            AnySink::Nats(s)
+        }
     };
 
     // Create PgStream as an ETL destination
@@ -122,6 +135,11 @@ fn log_sink_config(config: &SinkConfig) {
         SinkConfig::Memory => {
             debug!("using memory sink");
         }
+
+        #[cfg(feature = "sink-nats")]
+        SinkConfig::Nats(_cfg) => {
+            debug!("using nats sink");
+        }
     }
 }
 

src/core.rs

@@ -1,4 +1,46 @@
+//! Sink implementations for publishing events to external systems.
+
 mod base;
 pub mod memory;
 
+#[cfg(feature = "sink-nats")]
+pub mod nats;
+
 pub use base::Sink;
+
+use etl::error::EtlResult;
+use memory::MemorySink;
+
+#[cfg(feature = "sink-nats")]
+use nats::NatsSink;
+
+use crate::types::TriggeredEvent;
+
+/// Wrapper enum for all supported sink types.
+///
+/// Enables runtime sink selection while maintaining static dispatch.
+/// Each variant wraps a concrete sink implementation gated by its feature flag.
+#[derive(Clone)]
+pub enum AnySink {
+    /// In-memory sink for testing and development.
+    Memory(MemorySink),
+
+    /// NATS sink for pub/sub messaging.
+    #[cfg(feature = "sink-nats")]
+    Nats(NatsSink),
+}
+
+impl Sink for AnySink {
+    fn name() -> &'static str {
+        "any"
+    }
+
+    async fn publish_events(&self, events: Vec<TriggeredEvent>) -> EtlResult<()> {
+        match self {
+            AnySink::Memory(sink) => sink.publish_events(events).await,
+
+            #[cfg(feature = "sink-nats")]
+            AnySink::Nats(sink) => sink.publish_events(events).await,
+        }
+    }
+}

src/sink/mod.rs

@@ -0,0 +1,122 @@
+//! 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.
+
+use async_nats::Client;
+use etl::error::EtlResult;
+use serde::Deserialize;
+use std::sync::Arc;
+use tracing::info;
+
+use crate::sink::Sink;
+use crate::types::TriggeredEvent;
+
+/// Configuration for the NATS sink.
+#[derive(Clone, Debug, Deserialize)]
+pub struct NatsSinkConfig {
+    /// NATS server URL (e.g., "nats://localhost:4222").
+    pub url: String,
+
+    /// Subject to publish messages to.
+    pub subject: String,
+}
+
+/// Sink that publishes events to a NATS subject.
+///
+/// Each event is serialized as JSON and published to the configured 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,
+}
+
+impl NatsSink {
+    /// Creates a new NATS sink from configuration.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if the NATS connection cannot be established.
+    pub async fn new(
+        config: NatsSinkConfig,
+    ) -> Result<Self, Box<dyn std::error::Error + Send + Sync>> {
+        let client = async_nats::connect(&config.url).await?;
+
+        Ok(Self {
+            client: Arc::new(client),
+            subject: config.subject,
+        })
+    }
+}
+
+impl Sink for NatsSink {
+    fn name() -> &'static str {
+        "nats"
+    }
+
+    async fn publish_events(&self, events: Vec<TriggeredEvent>) -> EtlResult<()> {
+        if events.is_empty() {
+            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());
+            }
+
+            let payload = serde_json::to_vec(&json_obj).map_err(|e| {
+                etl::etl_error!(
+                    etl::error::ErrorKind::InvalidData,
+                    "Failed to serialize event to JSON",
+                    e.to_string()
+                )
+            })?;
+
+            // Publish to the configured subject.
+            self.client
+                .publish(self.subject.clone(), payload.into())
+                .await
+                .map_err(|e| {
+                    etl::etl_error!(
+                        etl::error::ErrorKind::InvalidData,
+                        "Failed to publish event to NATS",
+                        e.to_string()
+                    )
+                })?;
+        }
+
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_sink_name() {
+        assert_eq!(NatsSink::name(), "nats");
+    }
+}

src/sink/nats.rs

@@ -2,21 +2,39 @@ use ctor::dtor;
 use etl::config::{PgConnectionConfig, TlsConfig};
 use std::sync::{Mutex, OnceLock};
 use testcontainers::{ContainerRequest, ImageExt, runners::SyncRunner};
+use testcontainers_modules::nats::Nats;
 use testcontainers_modules::postgres::Postgres;
 use uuid::Uuid;
 
 static POSTGRES_PORT: OnceLock<u16> = OnceLock::new();
-// Using Mutex<Option<...>> so we can take ownership for cleanup
+// Using Mutex<Option<...>> so we can take ownership for cleanup.
 static POSTGRES_CONTAINER: OnceLock<Mutex<Option<testcontainers::Container<Postgres>>>> =
     OnceLock::new();
 
-/// Cleanup function that runs at program exit to stop and remove the postgres container
+static NATS_PORT: OnceLock<u16> = OnceLock::new();
+// Using Mutex<Option<...>> so we can take ownership for cleanup.
+static NATS_CONTAINER: OnceLock<Mutex<Option<testcontainers::Container<Nats>>>> = OnceLock::new();
+
+/// Cleanup function that runs at program exit to stop and remove the postgres container.
 #[dtor]
 fn cleanup_postgres_container() {
     if let Some(mutex) = POSTGRES_CONTAINER.get() {
         if let Ok(mut guard) = mutex.lock() {
             if let Some(container) = guard.take() {
-                // rm() stops and removes the container
+                // rm() stops and removes the container.
+                let _ = container.rm();
+            }
+        }
+    }
+}
+
+/// Cleanup function that runs at program exit to stop and remove the NATS container.
+#[dtor]
+fn cleanup_nats_container() {
+    if let Some(mutex) = NATS_CONTAINER.get() {
+        if let Ok(mut guard) = mutex.lock() {
+            if let Some(container) = guard.take() {
+                // rm() stops and removes the container.
                 let _ = container.rm();
             }
         }
@@ -74,3 +92,30 @@ pub async fn test_pg_config() -> PgConnectionConfig {
         keepalive: None,
     }
 }
+
+/// Ensures a NATS container is running and returns its port.
+///
+/// Uses a shared container across all tests. The container is started lazily
+/// on first use and cleaned up automatically when the test process exits.
+pub async fn ensure_nats() -> u16 {
+    // Use get_or_init to handle concurrent initialization attempts.
+    *NATS_PORT.get_or_init(|| {
+        // Run container startup in a separate thread to avoid runtime-in-runtime panic.
+        std::thread::spawn(|| {
+            let container: ContainerRequest<Nats> = Nats::default().into();
+
+            let container = container.start().expect("Failed to start NATS container");
+
+            let port = container
+                .get_host_port_ipv4(4222)
+                .expect("Failed to get NATS container port");
+
+            // Store the container in a Mutex<Option<...>> so we can take ownership later for cleanup.
+            let _ = NATS_CONTAINER.set(Mutex::new(Some(container)));
+
+            port
+        })
+        .join()
+        .expect("Failed to join NATS container startup thread")
+    })
+}