feat: add realtime destination

Cargo.toml

@@ -39,7 +39,7 @@ anyhow = { version = "1.0.98", default-features = false }
 arrow = { version = "57.0", default-features = false }
 async-trait = { version = "0.1.88" }
 aws-lc-rs = { version = "1.13.3", default-features = false }
-base64 = { version = "0.22.1", default-features = false }
+base64 = { version = "0.22.1", default-features = false, features = ["std"] }
 byteorder = { version = "1.5.0", default-features = false }
 bytes = { version = "1.10.1" }
 chrono = { version = "0.4.41", default-features = false }
@@ -82,6 +82,7 @@ tokio = { version = "1.47.0", default-features = false }
 tokio-postgres = { git = "https://github.com/MaterializeInc/rust-postgres", default-features = false, rev = "c4b473b478b3adfbf8667d2fbe895d8423f1290b" }
 tokio-rustls = { version = "0.26.2", default-features = false }
 tokio-stream = { version = "0.1.18", default-features = false, features = ["sync"] }
+tokio-tungstenite = { version = "0.26", default-features = false, features = ["connect", "rustls-tls-native-roots"] }
 tonic = { version = "0.14.2", default-features = false }
 tracing = { version = "0.1.41", default-features = false }
 tracing-actix-web = { version = "0.7.19", default-features = false }

etl-config/src/shared/destination.rs

@@ -46,6 +46,24 @@ pub enum DestinationConfig {
         #[serde(flatten)]
         config: IcebergConfig,
     },
+    #[serde(rename = "realtime")]
+    SupabaseRealtime {
+        /// WebSocket URL, e.g. wss://project.supabase.co/realtime/v1/websocket
+        url: String,
+        /// API key (anon key) for authenticating with Realtime.
+        api_key: SecretString,
+        /// Whether to use private channels (prefixes topic with `"private:"`).
+        /// Defaults to `false`.
+        #[serde(default)]
+        private_channels: bool,
+        /// Maximum number of send retries before giving up. Defaults to `5`.
+        #[serde(default = "default_max_retries")]
+        max_retries: u32,
+    },
+}
+
+fn default_max_retries() -> u32 {
+    5
 }
 
 impl DestinationConfig {
@@ -199,6 +217,12 @@ pub enum DestinationConfigWithoutSecrets {
         #[serde(flatten)]
         config: IcebergConfigWithoutSecrets,
     },
+    #[serde(rename = "realtime")]
+    SupabaseRealtime {
+        url: String,
+        private_channels: bool,
+        max_retries: u32,
+    },
 }
 
 impl From<DestinationConfig> for DestinationConfigWithoutSecrets {
@@ -219,6 +243,16 @@ impl From<DestinationConfig> for DestinationConfigWithoutSecrets {
             DestinationConfig::Iceberg { config } => DestinationConfigWithoutSecrets::Iceberg {
                 config: config.into(),
             },
+            DestinationConfig::SupabaseRealtime {
+                url,
+                api_key: _,
+                private_channels,
+                max_retries,
+            } => DestinationConfigWithoutSecrets::SupabaseRealtime {
+                url,
+                private_channels,
+                max_retries,
+            },
         }
     }
 }

etl-destinations/Cargo.toml

@@ -31,6 +31,16 @@ iceberg = [
     "dep:serde",
     "dep:serde_json",
 ]
+realtime = [
+    "dep:tokio-tungstenite",
+    "dep:tokio",
+    "tokio/time",
+    "dep:tokio-stream",
+    "dep:futures",
+    "dep:tracing",
+    "dep:serde_json",
+    "dep:base64",
+]
 egress = ["etl/egress"]
 # We assume that `test-utils` is always used in conjunction with `bigquery` or `iceberg` thus we only
 # put here the extra dependencies needed.
@@ -43,6 +53,7 @@ arrow = { workspace = true, optional = true }
 async-trait = { workspace = true, optional = true }
 base64 = { workspace = true, optional = true }
 chrono = { workspace = true }
+futures = { workspace = true, optional = true }
 gcp-bigquery-client = { workspace = true, optional = true, features = [
     "rust-tls",
     "aws-lc-rs",
@@ -57,6 +68,8 @@ reqwest = { workspace = true, optional = true, features = ["json"] }
 serde = { workspace = true, optional = true, features = ["derive"] }
 serde_json = { workspace = true, optional = true }
 tokio = { workspace = true, optional = true, features = ["sync"] }
+tokio-stream = { workspace = true, optional = true }
+tokio-tungstenite = { workspace = true, optional = true }
 tonic = { workspace = true, optional = true }
 tracing = { workspace = true, optional = true, default-features = true }
 uuid = { workspace = true, optional = true, features = ["v4"] }

etl-destinations/src/lib.rs

@@ -9,3 +9,5 @@ pub mod bigquery;
 pub mod egress;
 #[cfg(feature = "iceberg")]
 pub mod iceberg;
+#[cfg(feature = "realtime")]
+pub mod realtime;