faucet_source_redis/
stream.rs

1//! Redis source stream executor.
2
3use crate::config::{RedisSourceConfig, RedisSourceType};
4use async_trait::async_trait;
5use faucet_core::{FaucetError, Stream, StreamPage};
6use redis::AsyncCommands;
7use serde_json::{Value, json};
8use std::pin::Pin;
9
10/// A configured Redis source that reads records from Redis data structures.
11pub struct RedisSource {
12    config: RedisSourceConfig,
13    /// Lazily-opened multiplexed connection, reused across every `fetch_all`
14    /// and `stream_pages` call instead of opening a fresh client + TCP/AUTH
15    /// handshake per call (#78/#22). `MultiplexedConnection` is cheap to clone
16    /// (it shares one underlying socket), so each call clones the cached one.
17    conn: tokio::sync::OnceCell<redis::aio::MultiplexedConnection>,
18}
19
20impl RedisSource {
21    /// Create a new Redis source from the given configuration. The connection
22    /// is opened lazily on first use, so construction stays synchronous and does
23    /// no I/O; it fails only on an invalid config (an out-of-range `batch_size`).
24    pub fn new(config: RedisSourceConfig) -> Result<Self, FaucetError> {
25        faucet_core::validate_batch_size(config.batch_size)?;
26        Ok(Self {
27            config,
28            conn: tokio::sync::OnceCell::new(),
29        })
30    }
31
32    /// Return a clone of the shared multiplexed connection, opening it once on
33    /// first call.
34    async fn connection(&self) -> Result<redis::aio::MultiplexedConnection, FaucetError> {
35        let conn = self
36            .conn
37            .get_or_try_init(|| async {
38                let client = redis::Client::open(self.config.url.as_str())
39                    .map_err(|e| FaucetError::Config(format!("invalid Redis URL: {e}")))?;
40                client
41                    .get_multiplexed_async_connection()
42                    .await
43                    .map_err(|e| FaucetError::Source(format!("Redis connection failed: {e}")))
44            })
45            .await?;
46        Ok(conn.clone())
47    }
48
49    /// Fetch all records from the configured Redis source.
50    pub async fn fetch_all(&self) -> Result<Vec<Value>, FaucetError> {
51        let mut conn = self.connection().await?;
52
53        let mut records = match &self.config.source_type {
54            RedisSourceType::List { key } => self.fetch_list(&mut conn, key).await?,
55            RedisSourceType::Stream {
56                key,
57                group,
58                consumer,
59                count,
60            } => {
61                self.fetch_stream(&mut conn, key, group, consumer, count)
62                    .await?
63            }
64            RedisSourceType::Keys { pattern } => self.fetch_keys(&mut conn, pattern).await?,
65        };
66
67        if let Some(max) = self.config.max_records {
68            records.truncate(max);
69        }
70
71        tracing::info!(records = records.len(), "Redis fetch complete");
72        Ok(records)
73    }
74
75    /// Read all elements from a Redis list.
76    async fn fetch_list(
77        &self,
78        conn: &mut redis::aio::MultiplexedConnection,
79        key: &str,
80    ) -> Result<Vec<Value>, FaucetError> {
81        let values: Vec<String> = conn
82            .lrange(key, 0, -1)
83            .await
84            .map_err(|e| FaucetError::Source(format!("LRANGE failed on '{key}': {e}")))?;
85
86        let records = values
87            .into_iter()
88            .map(|v| serde_json::from_str::<Value>(&v).unwrap_or_else(|_| Value::String(v.clone())))
89            .collect();
90
91        Ok(records)
92    }
93
94    /// Read entries from a Redis stream.
95    async fn fetch_stream(
96        &self,
97        conn: &mut redis::aio::MultiplexedConnection,
98        key: &str,
99        group: &Option<String>,
100        consumer: &Option<String>,
101        count: &Option<usize>,
102    ) -> Result<Vec<Value>, FaucetError> {
103        let mut records = Vec::new();
104        match (group, consumer) {
105            (Some(group_name), Some(consumer_name)) => {
106                // Drain ALL currently-pending new messages for the group, not just
107                // the first `count`. A single XREADGROUP with the default
108                // `count = 100` silently truncated the rest (#146 narrowed); loop,
109                // consuming `>` until a short read (fewer than requested → the
110                // backlog is drained) or the `max_records` cap is hit.
111                let per_read = count.unwrap_or(100).max(1);
112                loop {
113                    let opts = redis::streams::StreamReadOptions::default().count(per_read);
114                    let reply: redis::streams::StreamReadReply = conn
115                        .xread_options(&[key], &[">"], &opts.group(group_name, consumer_name))
116                        .await
117                        .map_err(|e| {
118                            FaucetError::Source(format!("XREADGROUP failed on '{key}': {e}"))
119                        })?;
120                    let mut got = 0usize;
121                    for stream_key in &reply.keys {
122                        for entry in &stream_key.ids {
123                            records.push(stream_entry_to_json(&entry.id, &entry.map));
124                            got += 1;
125                        }
126                    }
127                    // A short read means Redis returned everything currently
128                    // pending — stop (also breaks at `got == 0`). This also avoids
129                    // spinning against a live producer.
130                    if got < per_read {
131                        break;
132                    }
133                    if let Some(max) = self.config.max_records
134                        && records.len() >= max
135                    {
136                        break;
137                    }
138                }
139            }
140            _ => {
141                // No consumer group: XREAD from `0` returns the whole stream when
142                // no `count` is set; an explicit `count` is the caller's own cap.
143                let mut opts = redis::streams::StreamReadOptions::default();
144                if let Some(c) = count {
145                    opts = opts.count(*c);
146                }
147                let reply: redis::streams::StreamReadReply = conn
148                    .xread_options(&[key], &["0"], &opts)
149                    .await
150                    .map_err(|e| FaucetError::Source(format!("XREAD failed on '{key}': {e}")))?;
151                for stream_key in &reply.keys {
152                    for entry in &stream_key.ids {
153                        records.push(stream_entry_to_json(&entry.id, &entry.map));
154                    }
155                }
156            }
157        }
158
159        Ok(records)
160    }
161
162    /// Scan for keys matching a pattern, then MGET all keys in a single round-trip.
163    async fn fetch_keys(
164        &self,
165        conn: &mut redis::aio::MultiplexedConnection,
166        pattern: &str,
167    ) -> Result<Vec<Value>, FaucetError> {
168        let keys: Vec<String> = {
169            let mut collected = Vec::new();
170            let mut iter: redis::AsyncIter<String> =
171                conn.scan_match(pattern).await.map_err(|e| {
172                    FaucetError::Source(format!("SCAN failed with pattern '{pattern}': {e}"))
173                })?;
174
175            while let Some(key) = iter.next_item().await {
176                collected.push(key);
177            }
178            collected
179        };
180
181        if keys.is_empty() {
182            return Ok(Vec::new());
183        }
184
185        let values: Vec<Option<String>> = redis::cmd("MGET")
186            .arg(&keys)
187            .query_async(conn)
188            .await
189            .map_err(|e| FaucetError::Source(format!("MGET failed: {e}")))?;
190
191        let mut records = Vec::new();
192        for (key, value) in keys.iter().zip(values) {
193            if let Some(v) = value {
194                let parsed =
195                    serde_json::from_str::<Value>(&v).unwrap_or_else(|_| Value::String(v.clone()));
196                records.push(json!({
197                    "key": key,
198                    "value": parsed,
199                }));
200            }
201        }
202
203        Ok(records)
204    }
205}
206
207/// Convert a single XRANGE/XREAD stream entry into the JSON record shape used
208/// by both [`RedisSource::fetch_all`] and [`RedisSource::stream_pages`].
209fn stream_entry_to_json(id: &str, map: &std::collections::HashMap<String, redis::Value>) -> Value {
210    let mut fields = serde_json::Map::new();
211    for (field_name, field_value) in map {
212        let val = match field_value {
213            redis::Value::BulkString(bytes) => {
214                let s = String::from_utf8_lossy(bytes);
215                serde_json::from_str::<Value>(&s).unwrap_or_else(|_| Value::String(s.into_owned()))
216            }
217            redis::Value::SimpleString(s) => {
218                serde_json::from_str::<Value>(s).unwrap_or_else(|_| Value::String(s.clone()))
219            }
220            redis::Value::Int(n) => json!(n),
221            redis::Value::Double(n) => json!(n),
222            redis::Value::Boolean(b) => json!(b),
223            redis::Value::Nil => Value::Null,
224            other => Value::String(format!("{other:?}")),
225        };
226        fields.insert(field_name.clone(), val);
227    }
228    json!({
229        "id": id,
230        "fields": Value::Object(fields),
231    })
232}
233
234/// Parse a Redis stream entry ID (`ms-seq`) and return the immediate
235/// successor ID, used to advance the `start` argument of the next `XRANGE`
236/// call without re-emitting the last entry of the previous page.
237fn next_stream_id(id: &str) -> String {
238    // Stream IDs are `<ms>-<seq>`. The "next" ID after `a-b` is `a-(b+1)`,
239    // wrapping to `(a+1)-0` on `u64::MAX` (which we treat as terminal).
240    if let Some((ms, seq)) = id.split_once('-')
241        && let (Ok(ms), Ok(seq)) = (ms.parse::<u64>(), seq.parse::<u64>())
242    {
243        return match seq.checked_add(1) {
244            Some(next_seq) => format!("{ms}-{next_seq}"),
245            None => format!("{}-0", ms.saturating_add(1)),
246        };
247    }
248    // Fall back to appending `\x00` — XRANGE treats this as "just after".
249    // Reachable only if Redis ever returns a malformed ID, which it does not
250    // in practice, but we degrade safely.
251    format!("{id}\u{0}")
252}
253
254#[async_trait]
255impl faucet_core::Source for RedisSource {
256    async fn fetch_with_context(
257        &self,
258        context: &std::collections::HashMap<String, serde_json::Value>,
259    ) -> Result<Vec<Value>, FaucetError> {
260        if context.is_empty() {
261            return RedisSource::fetch_all(self).await;
262        }
263
264        let mut conn = self.connection().await?;
265
266        // Substitute context into the key/pattern of each source type variant.
267        let mut records = match &self.config.source_type {
268            RedisSourceType::List { key } => {
269                let resolved_key = faucet_core::util::substitute_context(key, context);
270                self.fetch_list(&mut conn, &resolved_key).await?
271            }
272            RedisSourceType::Stream {
273                key,
274                group,
275                consumer,
276                count,
277            } => {
278                let resolved_key = faucet_core::util::substitute_context(key, context);
279                self.fetch_stream(&mut conn, &resolved_key, group, consumer, count)
280                    .await?
281            }
282            RedisSourceType::Keys { pattern } => {
283                let resolved_pattern = faucet_core::util::substitute_context(pattern, context);
284                self.fetch_keys(&mut conn, &resolved_pattern).await?
285            }
286        };
287
288        if let Some(max) = self.config.max_records {
289            records.truncate(max);
290        }
291
292        tracing::info!(
293            records = records.len(),
294            "Redis fetch complete (with context)"
295        );
296        Ok(records)
297    }
298
299    /// Stream records page-by-page so the pipeline can write to the sink as
300    /// pages arrive instead of buffering the full result set. Each mode maps
301    /// [`RedisSourceConfig::batch_size`] onto its native paging primitive
302    /// (see the type-level doc on [`RedisSourceConfig::batch_size`]).
303    ///
304    /// The trait-level `batch_size` argument is ignored in favour of the
305    /// config field — the config is the user-facing knob the README
306    /// documents, and routing the pipeline-supplied hint through it would
307    /// silently override an explicit config value.
308    ///
309    /// `batch_size = 0` drains the underlying primitive into a single page.
310    /// The Redis source has no incremental-replication mode today, so every
311    /// emitted page carries `bookmark: None`.
312    fn stream_pages<'a>(
313        &'a self,
314        context: &'a std::collections::HashMap<String, Value>,
315        _batch_size: usize,
316    ) -> Pin<Box<dyn Stream<Item = Result<StreamPage, FaucetError>> + Send + 'a>> {
317        let batch_size = self.config.batch_size;
318        let max_records = self.config.max_records;
319
320        Box::pin(async_stream::try_stream! {
321            let mut conn = self.connection().await?;
322
323            let mut emitted: usize = 0;
324
325            match &self.config.source_type {
326                RedisSourceType::List { key } => {
327                    let resolved = if context.is_empty() {
328                        key.clone()
329                    } else {
330                        faucet_core::util::substitute_context(key, context)
331                    };
332                    let pages = stream_list(&mut conn, &resolved, batch_size, max_records);
333                    futures::pin_mut!(pages);
334                    while let Some(page) = futures::StreamExt::next(&mut pages).await {
335                        let page = page?;
336                        emitted += page.records.len();
337                        yield page;
338                    }
339                }
340                RedisSourceType::Stream { key, .. } => {
341                    // Streaming intentionally uses XRANGE — consumer-group
342                    // semantics (XREADGROUP) don't compose with "drain to a
343                    // bookmarked checkpoint" because acknowledgement state
344                    // would have to be deferred until the sink succeeds, and
345                    // the source has no incremental mode today.
346                    let resolved = if context.is_empty() {
347                        key.clone()
348                    } else {
349                        faucet_core::util::substitute_context(key, context)
350                    };
351                    let pages = stream_xrange(&mut conn, &resolved, batch_size, max_records);
352                    futures::pin_mut!(pages);
353                    while let Some(page) = futures::StreamExt::next(&mut pages).await {
354                        let page = page?;
355                        emitted += page.records.len();
356                        yield page;
357                    }
358                }
359                RedisSourceType::Keys { pattern } => {
360                    let resolved = if context.is_empty() {
361                        pattern.clone()
362                    } else {
363                        faucet_core::util::substitute_context(pattern, context)
364                    };
365                    let pages = stream_keys(&mut conn, &resolved, batch_size, max_records);
366                    futures::pin_mut!(pages);
367                    while let Some(page) = futures::StreamExt::next(&mut pages).await {
368                        let page = page?;
369                        emitted += page.records.len();
370                        yield page;
371                    }
372                }
373            }
374
375            tracing::info!(
376                records = emitted,
377                batch_size,
378                "Redis source stream complete",
379            );
380        })
381    }
382
383    fn config_schema(&self) -> serde_json::Value {
384        serde_json::to_value(faucet_core::schema_for!(RedisSourceConfig))
385            .expect("schema serialization")
386    }
387
388    fn dataset_uri(&self) -> String {
389        use crate::config::RedisSourceType;
390        let base = faucet_core::redact_uri_credentials(&self.config.url);
391        match &self.config.source_type {
392            RedisSourceType::List { key } => format!("{base}?key={key}"),
393            RedisSourceType::Stream { key, .. } => format!("{base}?stream={key}"),
394            RedisSourceType::Keys { pattern } => format!("{base}?key={pattern}"),
395        }
396    }
397}
398
399/// Stream a Redis list via `LRANGE start stop`, sliding the window by
400/// `batch_size`. With `batch_size == 0`, drains the list in a single
401/// `LRANGE 0 -1` round-trip.
402///
403/// **Consistency caveat (#78 LOW):** index-based `LRANGE` paging is only
404/// stable if the list is not mutated mid-scan. A concurrent `LPUSH` / `LPOP`
405/// shifts every element's index, so a writer pushing/popping while this drains
406/// can make the source skip or duplicate elements across page boundaries. For
407/// a queue-style workload where the list is being consumed concurrently,
408/// prefer a Redis Stream (`XRANGE`/consumer groups) over a list.
409fn stream_list<'a>(
410    conn: &'a mut redis::aio::MultiplexedConnection,
411    key: &'a str,
412    batch_size: usize,
413    max_records: Option<usize>,
414) -> impl Stream<Item = Result<StreamPage, FaucetError>> + 'a {
415    async_stream::try_stream! {
416        if batch_size == 0 {
417            let values: Vec<String> = conn
418                .lrange(key, 0, -1)
419                .await
420                .map_err(|e| FaucetError::Source(format!("LRANGE failed on '{key}': {e}")))?;
421            let mut records: Vec<Value> = values
422                .into_iter()
423                .map(|v| serde_json::from_str::<Value>(&v).unwrap_or_else(|_| Value::String(v.clone())))
424                .collect();
425            if let Some(max) = max_records {
426                records.truncate(max);
427            }
428            yield StreamPage { records, bookmark: None };
429            return;
430        }
431
432        let mut start: isize = 0;
433        let mut emitted: usize = 0;
434        loop {
435            let stop: isize = start + batch_size as isize - 1;
436            let values: Vec<String> = conn
437                .lrange(key, start, stop)
438                .await
439                .map_err(|e| FaucetError::Source(format!("LRANGE failed on '{key}': {e}")))?;
440            if values.is_empty() {
441                break;
442            }
443            let mut records: Vec<Value> = values
444                .into_iter()
445                .map(|v| serde_json::from_str::<Value>(&v).unwrap_or_else(|_| Value::String(v.clone())))
446                .collect();
447            let returned = records.len();
448            // Respect max_records — truncate the final page and stop.
449            let mut stop_after_yield = false;
450            if let Some(max) = max_records
451                && emitted + records.len() >= max
452            {
453                records.truncate(max - emitted);
454                stop_after_yield = true;
455            }
456            emitted += records.len();
457            yield StreamPage { records, bookmark: None };
458            if stop_after_yield || returned < batch_size {
459                break;
460            }
461            start += batch_size as isize;
462        }
463    }
464}
465
466/// Stream a Redis stream via `XRANGE start + COUNT batch_size`, advancing the
467/// start ID on each page. With `batch_size == 0`, drains via a single
468/// `XRANGE - +` round-trip.
469fn stream_xrange<'a>(
470    conn: &'a mut redis::aio::MultiplexedConnection,
471    key: &'a str,
472    batch_size: usize,
473    max_records: Option<usize>,
474) -> impl Stream<Item = Result<StreamPage, FaucetError>> + 'a {
475    async_stream::try_stream! {
476        if batch_size == 0 {
477            let reply: redis::streams::StreamRangeReply = conn
478                .xrange_all(key)
479                .await
480                .map_err(|e| FaucetError::Source(format!("XRANGE failed on '{key}': {e}")))?;
481            let mut records: Vec<Value> = reply
482                .ids
483                .iter()
484                .map(|entry| stream_entry_to_json(&entry.id, &entry.map))
485                .collect();
486            if let Some(max) = max_records {
487                records.truncate(max);
488            }
489            yield StreamPage { records, bookmark: None };
490            return;
491        }
492
493        let mut start: String = "-".to_string();
494        let mut emitted: usize = 0;
495        loop {
496            let reply: redis::streams::StreamRangeReply = conn
497                .xrange_count(key, &start, "+", batch_size)
498                .await
499                .map_err(|e| FaucetError::Source(format!("XRANGE failed on '{key}': {e}")))?;
500
501            if reply.ids.is_empty() {
502                break;
503            }
504
505            // Capture the last returned ID before consuming the reply so we
506            // can advance the cursor (`next_stream_id`) without re-emitting it.
507            let last_id = reply
508                .ids
509                .last()
510                .expect("non-empty checked above")
511                .id
512                .clone();
513            let returned = reply.ids.len();
514            let mut records: Vec<Value> = reply
515                .ids
516                .into_iter()
517                .map(|entry| stream_entry_to_json(&entry.id, &entry.map))
518                .collect();
519
520            let mut stop_after_yield = false;
521            if let Some(max) = max_records
522                && emitted + records.len() >= max
523            {
524                records.truncate(max - emitted);
525                stop_after_yield = true;
526            }
527            emitted += records.len();
528            yield StreamPage { records, bookmark: None };
529
530            if stop_after_yield || returned < batch_size {
531                break;
532            }
533            start = next_stream_id(&last_id);
534        }
535    }
536}
537
538/// Stream keys matching `pattern`. The `SCAN` cursor is iterated server-side
539/// (with `COUNT` set to a sensible hint), keys are buffered up to
540/// `batch_size`, then `MGET`'d in one round-trip per page. With
541/// `batch_size == 0`, drains the entire scan and emits one page after a
542/// single `MGET`.
543fn stream_keys<'a>(
544    conn: &'a mut redis::aio::MultiplexedConnection,
545    pattern: &'a str,
546    batch_size: usize,
547    max_records: Option<usize>,
548) -> impl Stream<Item = Result<StreamPage, FaucetError>> + 'a {
549    use faucet_core::DEFAULT_BATCH_SIZE;
550    async_stream::try_stream! {
551        // Drive the SCAN cursor manually (one `SCAN cursor MATCH .. COUNT ..`
552        // round-trip at a time) rather than via the buffering `AsyncIter`, so
553        // we can MGET + yield a page as soon as `batch_size` keys accumulate
554        // instead of materialising the entire matched keyset first (#78 LOW).
555        // SCAN COUNT is only a per-round-trip hint; a call may return more or
556        // fewer keys than the hint, so we still buffer until a full page.
557        let scan_hint = if batch_size == 0 { DEFAULT_BATCH_SIZE } else { batch_size };
558        // `batch_size == 0` is the "no batching" sentinel — accumulate the
559        // whole scan and emit one page (still one MGET).
560        let chunk_size = if batch_size == 0 { usize::MAX } else { batch_size };
561        let cap = max_records.unwrap_or(usize::MAX);
562
563        let mut cursor: u64 = 0;
564        let mut buffer: Vec<String> = Vec::new();
565        let mut emitted: usize = 0;
566
567        'scan: loop {
568            let (next_cursor, keys): (u64, Vec<String>) = redis::cmd("SCAN")
569                .arg(cursor)
570                .arg("MATCH")
571                .arg(pattern)
572                .arg("COUNT")
573                .arg(scan_hint)
574                .query_async(conn)
575                .await
576                .map_err(|e| FaucetError::Source(format!("SCAN failed with pattern '{pattern}': {e}")))?;
577            cursor = next_cursor;
578            buffer.extend(keys);
579
580            // Flush as many full pages as the buffer now holds.
581            while emitted < cap && buffer.len() >= chunk_size {
582                let take = chunk_size.min(cap - emitted);
583                let page_keys: Vec<String> = buffer.drain(..take).collect();
584                let records = mget_records(conn, &page_keys).await?;
585                emitted += records.len();
586                yield StreamPage { records, bookmark: None };
587            }
588
589            if cursor == 0 || emitted >= cap {
590                break 'scan;
591            }
592        }
593
594        // Trailing partial page (and the single page in the batch_size==0 case).
595        if emitted < cap && !buffer.is_empty() {
596            let take = (cap - emitted).min(buffer.len());
597            let page_keys: Vec<String> = buffer.drain(..take).collect();
598            let records = mget_records(conn, &page_keys).await?;
599            yield StreamPage { records, bookmark: None };
600        }
601    }
602}
603
604/// `MGET` a slice of keys and pair them with their values via
605/// [`collect_kv_records`].
606async fn mget_records(
607    conn: &mut redis::aio::MultiplexedConnection,
608    keys: &[String],
609) -> Result<Vec<Value>, FaucetError> {
610    let values: Vec<Option<String>> = redis::cmd("MGET")
611        .arg(keys)
612        .query_async(conn)
613        .await
614        .map_err(|e| FaucetError::Source(format!("MGET failed: {e}")))?;
615    Ok(collect_kv_records(keys, values))
616}
617
618/// Pair `keys` with their `MGET`-returned values into `{ "key", "value" }`
619/// records. Missing values (deleted between `SCAN` and `MGET`) are dropped,
620/// matching [`RedisSource::fetch_keys`].
621fn collect_kv_records(keys: &[String], values: Vec<Option<String>>) -> Vec<Value> {
622    keys.iter()
623        .zip(values)
624        .filter_map(|(key, value)| {
625            value.map(|v| {
626                let parsed =
627                    serde_json::from_str::<Value>(&v).unwrap_or_else(|_| Value::String(v.clone()));
628                json!({ "key": key, "value": parsed })
629            })
630        })
631        .collect()
632}
633
634#[cfg(test)]
635mod tests {
636    use super::*;
637    use crate::config::RedisSourceConfig;
638
639    #[test]
640    fn creates_source() {
641        let config = RedisSourceConfig::new(
642            "redis://localhost",
643            RedisSourceType::List { key: "test".into() },
644        );
645        let _source = RedisSource::new(config).unwrap();
646    }
647
648    #[test]
649    fn dataset_uri_list_source() {
650        use faucet_core::Source;
651        let source = RedisSource::new(RedisSourceConfig::new(
652            "redis://u:p@localhost:6379/0",
653            RedisSourceType::List {
654                key: "my-list".into(),
655            },
656        ))
657        .unwrap();
658        assert_eq!(source.dataset_uri(), "redis://localhost:6379/0?key=my-list");
659    }
660
661    #[test]
662    fn dataset_uri_stream_source() {
663        use faucet_core::Source;
664        let config = RedisSourceConfig::new(
665            "redis://localhost",
666            RedisSourceType::Stream {
667                key: "events".into(),
668                group: None,
669                consumer: None,
670                count: None,
671            },
672        );
673        let source = RedisSource::new(config).unwrap();
674        assert_eq!(source.dataset_uri(), "redis://localhost?stream=events");
675    }
676
677    #[test]
678    fn dataset_uri_keys_source() {
679        use faucet_core::Source;
680        let source = RedisSource::new(RedisSourceConfig::new(
681            "redis://u:p@localhost:6379/0",
682            RedisSourceType::Keys {
683                pattern: "user:*".into(),
684            },
685        ))
686        .unwrap();
687        assert_eq!(
688            source.dataset_uri(),
689            "redis://localhost:6379/0?key=user:*",
690            "keys variant renders the glob pattern as the key, with credentials redacted"
691        );
692    }
693
694    #[test]
695    fn config_schema_describes_redis_source_config() {
696        use faucet_core::Source;
697        let source = RedisSource::new(RedisSourceConfig::new(
698            "redis://localhost",
699            RedisSourceType::List { key: "k".into() },
700        ))
701        .unwrap();
702        let schema = source.config_schema();
703        // The schema must expose the user-facing config fields.
704        let props = &schema["properties"];
705        assert!(props.get("url").is_some(), "schema exposes 'url'");
706        assert!(
707            props.get("source_type").is_some(),
708            "schema exposes 'source_type'"
709        );
710        assert!(
711            props.get("batch_size").is_some(),
712            "schema exposes 'batch_size'"
713        );
714    }
715
716    #[test]
717    fn new_rejects_out_of_range_batch_size() {
718        let mut config = RedisSourceConfig::new(
719            "redis://localhost",
720            RedisSourceType::List { key: "test".into() },
721        );
722        config.batch_size = faucet_core::MAX_BATCH_SIZE + 1;
723        match RedisSource::new(config) {
724            Err(FaucetError::Config(m)) => assert!(m.contains("batch_size"), "got: {m}"),
725            other => panic!(
726                "expected a batch_size Config error, got {:?}",
727                other.is_ok()
728            ),
729        }
730    }
731
732    #[test]
733    fn next_stream_id_increments_sequence() {
734        assert_eq!(next_stream_id("1234-0"), "1234-1");
735        assert_eq!(next_stream_id("1234-99"), "1234-100");
736    }
737
738    #[test]
739    fn next_stream_id_wraps_seq_overflow() {
740        let id = format!("5-{}", u64::MAX);
741        assert_eq!(next_stream_id(&id), "6-0");
742    }
743
744    #[test]
745    fn next_stream_id_falls_back_on_malformed_id() {
746        // Not a real Redis ID — fallback path appends NUL.
747        let next = next_stream_id("not-a-real-id");
748        assert!(next.starts_with("not-a-real-id"));
749        assert!(next.ends_with('\u{0}'));
750    }
751
752    #[test]
753    fn stream_entry_to_json_extracts_id_and_fields() {
754        let mut map = std::collections::HashMap::new();
755        map.insert(
756            "field1".to_string(),
757            redis::Value::BulkString(b"value1".to_vec()),
758        );
759        map.insert("field2".to_string(), redis::Value::Int(42));
760        let json = stream_entry_to_json("100-0", &map);
761        assert_eq!(json["id"], "100-0");
762        assert_eq!(json["fields"]["field1"], "value1");
763        assert_eq!(json["fields"]["field2"], 42);
764    }
765}
faucet_source_redis/stream.rs

faucet_source_redis/
stream.rs
