diff --git a/Cargo.lock b/Cargo.lock index 780a4ae..13992a1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -44,6 +44,15 @@ version = "1.0.100" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a23eb6b1614318a8071c9b2521f36b424b2c83db5eb3a0fead4a6c0809af6e61" +[[package]] +name = "arc-swap" +version = "1.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "51d03449bb8ca2cc2ef70869af31463d1ae5ccc8fa3e334b307203fbf815207e" +dependencies = [ + "rustversion", +] + [[package]] name = "arraydeque" version = "0.5.1" @@ -104,6 +113,15 @@ dependencies = [ "fs_extra", ] +[[package]] +name = "backon" +version = "1.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cffb0e931875b666fc4fcb20fee52e9bbd1ef836fd9e9e04ec21555f9f85f7ef" +dependencies = [ + "fastrand", +] + [[package]] name = "base64" version = "0.21.7" @@ -253,6 +271,20 @@ dependencies = [ "cc", ] +[[package]] +name = "combine" +version = "4.6.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba5a308b75df32fe02788e748662718f03fde005016435c444eea572398219fd" +dependencies = [ + "bytes", + "futures-core", + "memchr", + "pin-project-lite", + "tokio", + "tokio-util", +] + [[package]] name = "concurrent-queue" version = "2.5.0" @@ -1255,6 +1287,15 @@ version = "2.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "469fb0b9cefa57e3ef31275ee7cacb78f2fdca44e4765491884a2b119d4eb130" +[[package]] +name = "itertools" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "413ee7dfc52ee1a4949ceeb7dbc8a33f2d6c088194d9f922fb8318faf1f01186" +dependencies = [ + "either", +] + [[package]] name = "itoa" version = "1.0.15" @@ -1448,6 +1489,16 @@ dependencies = [ "minimal-lexical", ] +[[package]] +name = "num-bigint" +version = "0.4.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a5e44f723f1133c9deac646763579fdb3ac745e418f2a7af9cd0c431da1f20b9" +dependencies = [ + "num-integer", + "num-traits", +] + [[package]] name = "num-bigint-dig" version = "0.8.6" @@ -1736,6 +1787,7 @@ dependencies = [ "metrics", "metrics-exporter-prometheus", "pin-project-lite", + "redis", "ring", "rustls", "rustls-pemfile", @@ -1914,6 +1966,30 @@ dependencies = [ "bitflags 2.10.0", ] +[[package]] +name = "redis" +version = "0.27.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09d8f99a4090c89cc489a94833c901ead69bfbf3877b4867d5482e321ee875bc" +dependencies = [ + "arc-swap", + "async-trait", + "backon", + "bytes", + "combine", + "futures", + "futures-util", + "itertools", + "itoa", + "num-bigint", + "percent-encoding", + "pin-project-lite", + "ryu", + "tokio", + "tokio-util", + "url", +] + [[package]] name = "redox_syscall" version = "0.3.5" diff --git a/Cargo.toml b/Cargo.toml index b53db85..38565b8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,8 +4,9 @@ name = "postgres-stream" version = "0.1.0" [features] -default = [] -test-utils = ["dep:ctor", "dep:testcontainers", "dep:testcontainers-modules"] +default = [] +sink-redis-strings = ["dep:redis"] +test-utils = ["dep:ctor", "dep:testcontainers", "dep:testcontainers-modules"] [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. +redis = { version = "0.27", default-features = false, features = ["tokio-comp", "connection-manager"], 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", "redis", "blocking"] } [dev-dependencies] temp-env = "0.3" diff --git a/src/config/sink.rs b/src/config/sink.rs index 3f16033..dcdf216 100644 --- a/src/config/sink.rs +++ b/src/config/sink.rs @@ -1,5 +1,8 @@ use serde::Deserialize; +#[cfg(feature = "sink-redis-strings")] +use crate::sink::redis_strings::RedisStringsSinkConfig; + /// Sink destination configuration. /// /// Determines where replicated events are sent. @@ -8,4 +11,9 @@ use serde::Deserialize; pub enum SinkConfig { /// In-memory sink for testing and development. Memory, + + /// Redis strings sink for key-value storage. + #[cfg(feature = "sink-redis-strings")] + #[serde(rename = "redis-strings")] + RedisStrings(RedisStringsSinkConfig), } diff --git a/src/core.rs b/src/core.rs index 1304429..e6670b5 100644 --- a/src/core.rs +++ b/src/core.rs @@ -77,6 +77,19 @@ async fn run_pipeline(config: &PipelineConfig) -> EtlResult<()> { // Create sink based on configuration. let sink = match &config.sink { SinkConfig::Memory => AnySink::Memory(MemorySink::new()), + + #[cfg(feature = "sink-redis-strings")] + SinkConfig::RedisStrings(cfg) => { + use crate::sink::redis_strings::RedisStringsSink; + let s = RedisStringsSink::new(cfg.clone()).await.map_err(|e| { + etl::etl_error!( + etl::error::ErrorKind::InvalidData, + "Failed to create Redis Strings sink", + e.to_string() + ) + })?; + AnySink::RedisStrings(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-redis-strings")] + SinkConfig::RedisStrings(_cfg) => { + debug!("using redis-strings sink"); + } } } diff --git a/src/sink/mod.rs b/src/sink/mod.rs index 4f8ab01..c968b83 100644 --- a/src/sink/mod.rs +++ b/src/sink/mod.rs @@ -1,11 +1,17 @@ mod base; pub mod memory; +#[cfg(feature = "sink-redis-strings")] +pub mod redis_strings; + pub use base::Sink; use etl::error::EtlResult; use memory::MemorySink; +#[cfg(feature = "sink-redis-strings")] +use redis_strings::RedisStringsSink; + use crate::types::TriggeredEvent; /// Wrapper enum for all supported sink types. @@ -16,6 +22,10 @@ use crate::types::TriggeredEvent; pub enum AnySink { /// In-memory sink for testing and development. Memory(MemorySink), + + /// Redis strings sink for key-value storage. + #[cfg(feature = "sink-redis-strings")] + RedisStrings(RedisStringsSink), } impl Sink for AnySink { @@ -26,6 +36,9 @@ impl Sink for AnySink { async fn publish_events(&self, events: Vec) -> EtlResult<()> { match self { AnySink::Memory(sink) => sink.publish_events(events).await, + + #[cfg(feature = "sink-redis-strings")] + AnySink::RedisStrings(sink) => sink.publish_events(events).await, } } } diff --git a/src/sink/redis_strings.rs b/src/sink/redis_strings.rs new file mode 100644 index 0000000..4c1ae4b --- /dev/null +++ b/src/sink/redis_strings.rs @@ -0,0 +1,153 @@ +//! Redis Strings sink for publishing events as key-value pairs. +//! +//! Stores each event's payload as a Redis string. The key is determined by: +//! 1. `key` in event metadata (from subscription's metadata/metadata_extensions) +//! 2. Fallback to event ID (optionally with key_prefix from config) +//! +//! # Dynamic Routing +//! +//! The Redis key can be configured per-event using metadata_extensions: +//! +//! ```sql +//! metadata_extensions = '[ +//! {"json_path": "key", "expression": "''user:'' || (payload->''user_id'')::text"} +//! ]' +//! ``` + +use etl::error::EtlResult; +use redis::aio::ConnectionManager; +use serde::{Deserialize, Serialize}; +use std::sync::Arc; +use tokio::sync::Mutex; + +use crate::sink::Sink; +use crate::types::TriggeredEvent; + +/// Configuration for the Redis Strings sink. +/// +/// This intentionally does not implement [`Serialize`] to avoid accidentally +/// leaking secrets (URL credentials) in serialized forms. +#[derive(Clone, Debug, Deserialize)] +pub struct RedisStringsSinkConfig { + /// Redis connection URL (e.g., "redis://localhost:6379"). + /// Contains credentials and should be treated as sensitive. + pub url: String, + + /// Optional prefix for all keys. + #[serde(default)] + pub key_prefix: Option, +} + +/// Configuration for the Redis Strings sink without sensitive data. +/// +/// Safe to serialize and log. Use this for debugging and metrics. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct RedisStringsSinkConfigWithoutSecrets { + /// Optional prefix for all keys. + pub key_prefix: Option, +} + +impl From for RedisStringsSinkConfigWithoutSecrets { + fn from(config: RedisStringsSinkConfig) -> Self { + Self { + key_prefix: config.key_prefix, + } + } +} + +impl From<&RedisStringsSinkConfig> for RedisStringsSinkConfigWithoutSecrets { + fn from(config: &RedisStringsSinkConfig) -> Self { + Self { + key_prefix: config.key_prefix.clone(), + } + } +} + +/// Sink that stores events as Redis string key-value pairs. +/// +/// Each event's payload is stored with a dynamic or default key. +/// The sink uses a connection manager for automatic reconnection handling. +#[derive(Clone)] +pub struct RedisStringsSink { + /// Shared Redis connection manager. + connection: Arc>, + + /// Optional prefix prepended to all keys. + key_prefix: Option, +} + +impl RedisStringsSink { + /// Creates a new Redis Strings sink from configuration. + /// + /// # Errors + /// + /// Returns an error if the Redis connection cannot be established. + pub async fn new( + config: RedisStringsSinkConfig, + ) -> Result> { + let client = redis::Client::open(config.url)?; + let connection = ConnectionManager::new(client).await?; + + Ok(Self { + connection: Arc::new(Mutex::new(connection)), + key_prefix: config.key_prefix, + }) + } + + /// Resolves the Redis key for an event from metadata or default (event ID). + fn resolve_key(&self, event: &TriggeredEvent) -> String { + // First check event metadata for dynamic key. + if let Some(ref metadata) = event.metadata { + if let Some(key) = metadata.get("key").and_then(|v| v.as_str()) { + return key.to_string(); + } + } + // Fall back to event ID with optional prefix. + match &self.key_prefix { + Some(prefix) => format!("{}:{}", prefix, event.id.id), + None => event.id.id.clone(), + } + } +} + +impl Sink for RedisStringsSink { + fn name() -> &'static str { + "redis-strings" + } + + async fn publish_events(&self, events: Vec) -> EtlResult<()> { + if events.is_empty() { + return Ok(()); + } + + let mut conn = self.connection.lock().await; + + // Use pipeline for batch efficiency. + let mut pipe = redis::pipe(); + + for event in events { + let key = self.resolve_key(&event); + pipe.set(&key, event.payload.to_string()); + } + + pipe.query_async::<()>(&mut *conn).await.map_err(|e| { + etl::etl_error!( + etl::error::ErrorKind::DestinationError, + "Failed to publish events to Redis", + e.to_string() + ) + })?; + + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_sink_name() { + assert_eq!(RedisStringsSink::name(), "redis-strings"); + } +} diff --git a/src/test_utils/container.rs b/src/test_utils/container.rs index 82625ea..ff9bbed 100644 --- a/src/test_utils/container.rs +++ b/src/test_utils/container.rs @@ -3,20 +3,35 @@ use etl::config::{PgConnectionConfig, TlsConfig}; use std::sync::{Mutex, OnceLock}; use testcontainers::{ContainerRequest, ImageExt, runners::SyncRunner}; use testcontainers_modules::postgres::Postgres; +use testcontainers_modules::redis::Redis; use uuid::Uuid; static POSTGRES_PORT: OnceLock = OnceLock::new(); -// Using Mutex> so we can take ownership for cleanup +static REDIS_PORT: OnceLock = OnceLock::new(); + +// Using Mutex> so we can take ownership for cleanup. static POSTGRES_CONTAINER: OnceLock>>> = OnceLock::new(); +static REDIS_CONTAINER: OnceLock>>> = OnceLock::new(); -/// Cleanup function that runs at program exit to stop and remove the postgres container +/// 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 + let _ = container.rm(); + } + } + } +} + +/// Cleanup function that runs at program exit to stop and remove the redis container. +#[dtor] +fn cleanup_redis_container() { + if let Some(mutex) = REDIS_CONTAINER.get() { + if let Ok(mut guard) = mutex.lock() { + if let Some(container) = guard.take() { let _ = container.rm(); } } @@ -74,3 +89,26 @@ pub async fn test_pg_config() -> PgConnectionConfig { keepalive: None, } } + +/// Ensures a Redis container is running and returns its port. +/// +/// Uses singleton pattern to reuse the same container across tests. +pub async fn ensure_redis() -> u16 { + *REDIS_PORT.get_or_init(|| { + std::thread::spawn(|| { + let container: ContainerRequest = Redis::default().with_tag("7-alpine"); + + let container = container.start().expect("Failed to start redis container"); + + let port = container + .get_host_port_ipv4(6379) + .expect("Failed to get redis container port"); + + let _ = REDIS_CONTAINER.set(Mutex::new(Some(container))); + + port + }) + .join() + .expect("Failed to join redis container startup thread") + }) +} diff --git a/tests/redis_strings_sink_tests.rs b/tests/redis_strings_sink_tests.rs new file mode 100644 index 0000000..55db1da --- /dev/null +++ b/tests/redis_strings_sink_tests.rs @@ -0,0 +1,177 @@ +//! Integration tests for the Redis Strings sink. + +#![cfg(feature = "sink-redis-strings")] + +use postgres_stream::sink::Sink; +use postgres_stream::sink::redis_strings::{RedisStringsSink, RedisStringsSinkConfig}; +use postgres_stream::test_utils::ensure_redis; +use postgres_stream::types::{EventIdentifier, StreamId, TriggeredEvent}; + +use chrono::Utc; +use redis::AsyncCommands; + +/// Creates a test event with the given ID. +fn make_test_event(id: &str) -> TriggeredEvent { + TriggeredEvent { + id: EventIdentifier::new(id.to_string(), Utc::now()), + payload: serde_json::json!({ + "test_id": id, + "message": format!("Test event {}", id), + }), + metadata: None, + stream_id: StreamId::from(1u64), + lsn: Some("0/16B3748".parse().unwrap()), + } +} + +#[tokio::test(flavor = "multi_thread")] +async fn test_redis_strings_sink_publishes_events() { + let redis_port = ensure_redis().await; + let redis_url = format!("redis://127.0.0.1:{redis_port}"); + + let config = RedisStringsSinkConfig { + url: redis_url.clone(), + key_prefix: None, + }; + + let sink = RedisStringsSink::new(config) + .await + .expect("Failed to create Redis Strings sink"); + + // Publish test events. + let events = vec![make_test_event("event-1"), make_test_event("event-2")]; + sink.publish_events(events) + .await + .expect("Failed to publish events"); + + // Verify events are stored in Redis. + let client = redis::Client::open(redis_url).expect("Failed to open redis client"); + let mut conn = client + .get_multiplexed_async_connection() + .await + .expect("Failed to get connection"); + + let value1: String = conn + .get("event-1") + .await + .expect("Failed to get event-1 from Redis"); + let value2: String = conn + .get("event-2") + .await + .expect("Failed to get event-2 from Redis"); + + // Verify payload contains expected data. + let parsed1: serde_json::Value = serde_json::from_str(&value1).expect("Failed to parse JSON"); + let parsed2: serde_json::Value = serde_json::from_str(&value2).expect("Failed to parse JSON"); + + assert_eq!(parsed1["test_id"], "event-1"); + assert_eq!(parsed2["test_id"], "event-2"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn test_redis_strings_sink_with_key_prefix() { + let redis_port = ensure_redis().await; + let redis_url = format!("redis://127.0.0.1:{redis_port}"); + + let config = RedisStringsSinkConfig { + url: redis_url.clone(), + key_prefix: Some("pgstream".to_string()), + }; + + let sink = RedisStringsSink::new(config) + .await + .expect("Failed to create Redis Strings sink"); + + // Publish test event. + let events = vec![make_test_event("prefixed-event")]; + sink.publish_events(events) + .await + .expect("Failed to publish events"); + + // Verify event is stored with prefix. + let client = redis::Client::open(redis_url).expect("Failed to open redis client"); + let mut conn = client + .get_multiplexed_async_connection() + .await + .expect("Failed to get connection"); + + let value: String = conn + .get("pgstream:prefixed-event") + .await + .expect("Failed to get prefixed event from Redis"); + + let parsed: serde_json::Value = serde_json::from_str(&value).expect("Failed to parse JSON"); + assert_eq!(parsed["test_id"], "prefixed-event"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn test_redis_strings_sink_empty_batch() { + let redis_port = ensure_redis().await; + let redis_url = format!("redis://127.0.0.1:{redis_port}"); + + let config = RedisStringsSinkConfig { + url: redis_url, + key_prefix: None, + }; + + let sink = RedisStringsSink::new(config) + .await + .expect("Failed to create Redis Strings sink"); + + // Empty batch should succeed without error. + sink.publish_events(vec![]) + .await + .expect("Empty batch should succeed"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn test_redis_strings_sink_uses_key_from_metadata() { + let redis_port = ensure_redis().await; + let redis_url = format!("redis://127.0.0.1:{redis_port}"); + + let config = RedisStringsSinkConfig { + url: redis_url.clone(), + key_prefix: None, + }; + + let sink = RedisStringsSink::new(config) + .await + .expect("Failed to create Redis Strings sink"); + + // Create event with custom key in metadata. + let custom_key = "user:12345:profile"; + let event = TriggeredEvent { + id: EventIdentifier::new("event-with-metadata-key".to_string(), Utc::now()), + payload: serde_json::json!({ + "test_id": "metadata-key-event", + "message": "Test event with metadata key", + }), + metadata: Some(serde_json::json!({ "key": custom_key })), + stream_id: StreamId::from(1u64), + lsn: Some("0/16B3748".parse().unwrap()), + }; + + sink.publish_events(vec![event]) + .await + .expect("Failed to publish events"); + + // Verify event is stored with metadata key, not event ID. + let client = redis::Client::open(redis_url).expect("Failed to open redis client"); + let mut conn = client + .get_multiplexed_async_connection() + .await + .expect("Failed to get connection"); + + let value: String = conn + .get(custom_key) + .await + .expect("Failed to get event from Redis"); + + let parsed: serde_json::Value = serde_json::from_str(&value).expect("Failed to parse JSON"); + assert_eq!(parsed["test_id"], "metadata-key-event"); +} + +#[test] +fn test_sink_name() { + assert_eq!(RedisStringsSink::name(), "redis-strings"); +}