subcog/storage/index/
redis.rs

1//! Redis-based index backend using `RediSearch`.
2//!
3//! Provides full-text search using Redis with the `RediSearch` module.
4//!
5//! # Connection Pooling (DB-H6)
6//!
7//! This backend reuses a single connection per instance. For high-concurrency
8//! scenarios, consider using `r2d2-redis` or `deadpool-redis` for connection
9//! pooling. The current implementation is suitable for CLI and single-threaded
10//! MCP server usage.
11//!
12//! # Command Timeout (CHAOS-H2)
13//!
14//! Redis operations use a 5-second response timeout to prevent indefinite
15//! blocking on slow or unresponsive Redis servers. The timeout is configured
16//! via `redis::ConnectionInfo` settings.
17
18#[cfg(feature = "redis")]
19mod implementation {
20    use crate::models::{Memory, MemoryId, MemoryStatus, Namespace, SearchFilter};
21    use crate::storage::resilience::{StorageResilienceConfig, retry_connection};
22    use crate::storage::traits::IndexBackend;
23    use crate::{Error, Result};
24    use redis::{Client, Commands, Connection};
25    use std::sync::Mutex;
26    use std::time::Duration;
27
28    /// Redis-based index backend using `RediSearch`.
29    ///
30    /// # Connection Management (DB-H6)
31    ///
32    /// Maintains a reusable connection via `Mutex<Option<Connection>>`.
33    /// The connection is lazily initialized and reused across operations
34    /// to avoid the overhead of establishing new connections for each command.
35    ///
36    /// # Command Timeout (CHAOS-HIGH-005)
37    ///
38    /// Connections are configured with a configurable response timeout (default 5s)
39    /// to prevent indefinite blocking on unresponsive servers.
40    /// Set `SUBCOG_TIMEOUT_REDIS_MS` to override.
41    pub struct RedisBackend {
42        /// Redis client.
43        client: Client,
44        /// Index name in Redis.
45        index_name: String,
46        /// Cached connection for reuse (DB-H6).
47        connection: Mutex<Option<Connection>>,
48        /// Operation timeout (CHAOS-HIGH-005).
49        timeout: Duration,
50    }
51
52    /// Default timeout for Redis operations (CHAOS-HIGH-005).
53    /// Configurable via `SUBCOG_TIMEOUT_REDIS_MS` environment variable.
54    fn default_redis_timeout() -> Duration {
55        crate::config::OperationTimeoutConfig::from_env().get(crate::config::OperationType::Redis)
56    }
57
58    impl RedisBackend {
59        /// Creates a new Redis backend.
60        ///
61        /// Configures the connection with a 5-second timeout (CHAOS-H2) to prevent
62        /// indefinite blocking on slow or unresponsive Redis servers.
63        ///
64        /// # Errors
65        ///
66        /// Returns an error if the Redis connection fails.
67        pub fn new(connection_url: &str, index_name: impl Into<String>) -> Result<Self> {
68            let client = Client::open(connection_url).map_err(|e| Error::OperationFailed {
69                operation: "redis_connect".to_string(),
70                cause: e.to_string(),
71            })?;
72
73            let backend = Self {
74                client,
75                index_name: index_name.into(),
76                connection: Mutex::new(None),
77                timeout: default_redis_timeout(),
78            };
79
80            // Ensure index exists
81            backend.ensure_index()?;
82
83            Ok(backend)
84        }
85
86        /// Creates a backend with default settings.
87        ///
88        /// # Errors
89        ///
90        /// Returns an error if the Redis connection fails.
91        pub fn with_defaults() -> Result<Self> {
92            Self::new("redis://localhost:6379", "subcog_memories")
93        }
94
95        /// Checks if the Redis connection is healthy (DB-HIGH-001).
96        ///
97        /// Performs a PING command to verify connectivity and responsiveness.
98        ///
99        /// # Returns
100        ///
101        /// `Ok(true)` if the connection is healthy, `Ok(false)` if unhealthy,
102        /// or `Err` if the health check itself fails unexpectedly.
103        ///
104        /// # Example
105        ///
106        /// ```rust,ignore
107        /// let backend = RedisBackend::new("redis://localhost:6379", "subcog")?;
108        /// if backend.health_check()? {
109        ///     println!("Redis is healthy");
110        /// } else {
111        ///     eprintln!("Redis is not responding");
112        /// }
113        /// ```
114        ///
115        /// # Errors
116        ///
117        /// Returns an error if the connection cannot be established.
118        pub fn health_check(&self) -> Result<bool> {
119            let mut conn = match self.get_connection() {
120                Ok(c) => c,
121                Err(_) => return Ok(false),
122            };
123
124            let result: redis::RedisResult<String> = redis::cmd("PING").query(&mut conn);
125
126            let healthy = result.is_ok_and(|response| response == "PONG");
127
128            self.return_connection(conn);
129            Ok(healthy)
130        }
131
132        /// Gets a connection, reusing the cached one if available (DB-H6).
133        ///
134        /// This method reuses an existing connection when possible, falling back
135        /// to creating a new connection if the cache is empty or the connection
136        /// is broken. The connection is stored in a `Mutex` for thread-safety.
137        ///
138        /// # Timeout (CHAOS-H2)
139        ///
140        /// New connections are configured with a 5-second response timeout.
141        ///
142        /// # Connection Retry (CHAOS-HIGH-003)
143        ///
144        /// New connection attempts use exponential backoff with jitter for transient
145        /// failures (Redis starting up, network issues, etc.).
146        fn get_connection(&self) -> Result<Connection> {
147            // Try to reuse existing connection
148            let mut guard = self.connection.lock().map_err(|e| Error::OperationFailed {
149                operation: "redis_lock_connection".to_string(),
150                cause: e.to_string(),
151            })?;
152
153            // Take the existing connection if available
154            if let Some(conn) = guard.take() {
155                // Return the connection - if it fails, caller will get error
156                // and next call will create fresh connection
157                return Ok(conn);
158            }
159            drop(guard); // Release lock before potentially slow retry loop
160
161            // No cached connection, create a new one with retry (CHAOS-HIGH-003)
162            let resilience_config = StorageResilienceConfig::from_env();
163            let timeout = self.timeout;
164
165            retry_connection(&resilience_config, "redis_index", "get_connection", || {
166                let conn = self
167                    .client
168                    .get_connection()
169                    .map_err(|e| Error::OperationFailed {
170                        operation: "redis_get_connection".to_string(),
171                        cause: e.to_string(),
172                    })?;
173
174                // Set response timeout to prevent indefinite blocking (CHAOS-HIGH-005)
175                conn.set_read_timeout(Some(timeout))
176                    .map_err(|e| Error::OperationFailed {
177                        operation: "redis_set_read_timeout".to_string(),
178                        cause: e.to_string(),
179                    })?;
180                conn.set_write_timeout(Some(timeout))
181                    .map_err(|e| Error::OperationFailed {
182                        operation: "redis_set_write_timeout".to_string(),
183                        cause: e.to_string(),
184                    })?;
185
186                Ok(conn)
187            })
188        }
189
190        /// Returns a connection to the cache for reuse (DB-H6).
191        fn return_connection(&self, conn: Connection) {
192            if let Ok(mut guard) = self.connection.lock() {
193                *guard = Some(conn);
194            }
195            // If lock fails, just drop the connection - not critical
196        }
197
198        /// Ensures the `RediSearch` index exists.
199        fn ensure_index(&self) -> Result<()> {
200            let mut conn = self.get_connection()?;
201
202            // Check if index exists
203            let exists = self.index_exists(&mut conn);
204            if exists {
205                self.return_connection(conn);
206                return Ok(());
207            }
208
209            // Create the index with schema
210            let result = self.create_index(&mut conn);
211            self.return_connection(conn);
212            result
213        }
214
215        /// Checks if the index already exists.
216        fn index_exists(&self, conn: &mut Connection) -> bool {
217            let result: redis::RedisResult<Vec<String>> = redis::cmd("FT._LIST").query(conn);
218
219            result
220                .map(|indices| indices.iter().any(|i| i == &self.index_name))
221                .unwrap_or(false)
222        }
223
224        /// Creates the `RediSearch` index.
225        fn create_index(&self, conn: &mut Connection) -> Result<()> {
226            let result: redis::RedisResult<String> = redis::cmd("FT.CREATE")
227                .arg(&self.index_name)
228                .arg("ON")
229                .arg("HASH")
230                .arg("PREFIX")
231                .arg(1)
232                .arg("mem:")
233                .arg("SCHEMA")
234                .arg("content")
235                .arg("TEXT")
236                .arg("WEIGHT")
237                .arg("1.0")
238                .arg("namespace")
239                .arg("TAG")
240                .arg("domain")
241                .arg("TAG")
242                .arg("status")
243                .arg("TAG")
244                .arg("tags")
245                .arg("TAG")
246                .arg("project_id")
247                .arg("TAG")
248                .arg("branch")
249                .arg("TAG")
250                .arg("file_path")
251                .arg("TAG")
252                .arg("created_at")
253                .arg("NUMERIC")
254                .arg("SORTABLE")
255                .arg("updated_at")
256                .arg("NUMERIC")
257                .arg("SORTABLE")
258                .query(conn);
259
260            match result {
261                Ok(_) => Ok(()),
262                Err(e) if e.to_string().contains("Index already exists") => Ok(()),
263                Err(e) => Err(Error::OperationFailed {
264                    operation: "redis_create_index".to_string(),
265                    cause: e.to_string(),
266                }),
267            }
268        }
269
270        /// Builds a filter clause for `RediSearch` queries.
271        fn build_filter_clause(filter: &SearchFilter) -> String {
272            let mut clauses = Vec::new();
273
274            if !filter.namespaces.is_empty() {
275                let ns_tags: Vec<&str> = filter
276                    .namespaces
277                    .iter()
278                    .copied()
279                    .map(namespace_to_tag)
280                    .collect();
281                clauses.push(format!("@namespace:{{{}}}", ns_tags.join("|")));
282            }
283
284            if !filter.domains.is_empty() {
285                let domain_strs: Vec<String> =
286                    filter.domains.iter().map(ToString::to_string).collect();
287                clauses.push(format!("@domain:{{{}}}", domain_strs.join("|")));
288            }
289
290            if !filter.statuses.is_empty() {
291                let status_strs: Vec<&str> = filter
292                    .statuses
293                    .iter()
294                    .map(crate::models::MemoryStatus::as_str)
295                    .collect();
296                clauses.push(format!("@status:{{{}}}", status_strs.join("|")));
297            }
298
299            if let Some(ref project_id) = filter.project_id {
300                clauses.push(format!("@project_id:{{{project_id}}}"));
301            }
302
303            if let Some(ref branch) = filter.branch {
304                clauses.push(format!("@branch:{{{branch}}}"));
305            }
306
307            if let Some(ref file_path) = filter.file_path {
308                clauses.push(format!("@file_path:{{{file_path}}}"));
309            }
310
311            if clauses.is_empty() {
312                String::new()
313            } else {
314                clauses.join(" ")
315            }
316        }
317    }
318
319    /// Converts namespace to tag value.
320    const fn namespace_to_tag(ns: Namespace) -> &'static str {
321        ns.as_str()
322    }
323
324    impl IndexBackend for RedisBackend {
325        fn index(&self, memory: &Memory) -> Result<()> {
326            let mut conn = self.get_connection()?;
327            let key = format!("mem:{}", memory.id.as_str());
328
329            // Store as hash
330            let tags_str = memory.tags.join(",");
331            let domain_str = memory.domain.to_string();
332            let status_str = memory.status.as_str();
333            let namespace_str = memory.namespace.as_str();
334            let project_id = memory.project_id.as_deref().unwrap_or("");
335            let branch = memory.branch.as_deref().unwrap_or("");
336            let file_path = memory.file_path.as_deref().unwrap_or("");
337
338            let result: redis::RedisResult<()> = conn.hset_multiple(
339                &key,
340                &[
341                    ("content", memory.content.as_str()),
342                    ("namespace", namespace_str),
343                    ("domain", &domain_str),
344                    ("status", status_str),
345                    ("tags", &tags_str),
346                    ("project_id", project_id),
347                    ("branch", branch),
348                    ("file_path", file_path),
349                ],
350            );
351
352            if let Err(e) = result {
353                self.return_connection(conn);
354                return Err(Error::OperationFailed {
355                    operation: "redis_index".to_string(),
356                    cause: e.to_string(),
357                });
358            }
359
360            // Set numeric fields separately
361            let result: redis::RedisResult<()> = conn.hset(&key, "created_at", memory.created_at);
362            if let Err(e) = result {
363                self.return_connection(conn);
364                return Err(Error::OperationFailed {
365                    operation: "redis_index_created".to_string(),
366                    cause: e.to_string(),
367                });
368            }
369
370            let result: redis::RedisResult<()> = conn.hset(&key, "updated_at", memory.updated_at);
371            if let Err(e) = result {
372                self.return_connection(conn);
373                return Err(Error::OperationFailed {
374                    operation: "redis_index_updated".to_string(),
375                    cause: e.to_string(),
376                });
377            }
378
379            self.return_connection(conn);
380            Ok(())
381        }
382
383        fn remove(&self, id: &MemoryId) -> Result<bool> {
384            let mut conn = self.get_connection()?;
385            let key = format!("mem:{}", id.as_str());
386
387            let result: redis::RedisResult<i32> = conn.del(&key);
388            match result {
389                Ok(deleted) => {
390                    self.return_connection(conn);
391                    Ok(deleted > 0)
392                },
393                Err(e) => {
394                    self.return_connection(conn);
395                    Err(Error::OperationFailed {
396                        operation: "redis_remove".to_string(),
397                        cause: e.to_string(),
398                    })
399                },
400            }
401        }
402
403        fn search(
404            &self,
405            query: &str,
406            filter: &SearchFilter,
407            limit: usize,
408        ) -> Result<Vec<(MemoryId, f32)>> {
409            let mut conn = self.get_connection()?;
410
411            // Build query with filters
412            let filter_clause = Self::build_filter_clause(filter);
413            let full_query = if filter_clause.is_empty() {
414                query.to_string()
415            } else {
416                format!("{query} {filter_clause}")
417            };
418
419            // FT.SEARCH idx "query" LIMIT 0 limit WITHSCORES
420            let result: redis::RedisResult<Vec<redis::Value>> = redis::cmd("FT.SEARCH")
421                .arg(&self.index_name)
422                .arg(&full_query)
423                .arg("LIMIT")
424                .arg(0)
425                .arg(limit)
426                .arg("WITHSCORES")
427                .query(&mut conn);
428
429            let output = match result {
430                Ok(values) => Ok(parse_search_results(&values)),
431                Err(e) => Err(Error::OperationFailed {
432                    operation: "redis_search".to_string(),
433                    cause: e.to_string(),
434                }),
435            };
436            self.return_connection(conn);
437            output
438        }
439
440        fn list_all(&self, filter: &SearchFilter, limit: usize) -> Result<Vec<(MemoryId, f32)>> {
441            let mut conn = self.get_connection()?;
442
443            // Build query - use * for all, with optional filters
444            let filter_clause = Self::build_filter_clause(filter);
445            let query = if filter_clause.is_empty() {
446                "*".to_string()
447            } else {
448                filter_clause
449            };
450
451            // FT.SEARCH idx "*" LIMIT 0 limit SORTBY updated_at DESC
452            let result: redis::RedisResult<Vec<redis::Value>> = redis::cmd("FT.SEARCH")
453                .arg(&self.index_name)
454                .arg(&query)
455                .arg("LIMIT")
456                .arg(0)
457                .arg(limit)
458                .arg("SORTBY")
459                .arg("updated_at")
460                .arg("DESC")
461                .query(&mut conn);
462
463            let output = match result {
464                Ok(values) => Ok(parse_list_results(&values)),
465                Err(e) => Err(Error::OperationFailed {
466                    operation: "redis_list_all".to_string(),
467                    cause: e.to_string(),
468                }),
469            };
470            self.return_connection(conn);
471            output
472        }
473
474        fn get_memory(&self, id: &MemoryId) -> Result<Option<Memory>> {
475            use crate::models::{Domain, Namespace};
476
477            let mut conn = self.get_connection()?;
478            let key = format!("mem:{}", id.as_str());
479
480            // Get all fields from hash
481            let result: redis::RedisResult<std::collections::HashMap<String, String>> =
482                conn.hgetall(&key);
483
484            let output = match result {
485                Ok(fields) if fields.is_empty() => Ok(None),
486                Ok(fields) => {
487                    let content = fields.get("content").cloned().unwrap_or_default();
488                    let namespace_str = fields.get("namespace").cloned().unwrap_or_default();
489                    let domain_str = fields.get("domain").cloned();
490                    let status_str = fields.get("status").cloned().unwrap_or_default();
491                    let tags_str = fields.get("tags").cloned();
492                    let project_id = fields
493                        .get("project_id")
494                        .cloned()
495                        .filter(|value| !value.is_empty());
496                    let branch = fields
497                        .get("branch")
498                        .cloned()
499                        .filter(|value| !value.is_empty());
500                    let file_path = fields
501                        .get("file_path")
502                        .cloned()
503                        .filter(|value| !value.is_empty());
504                    let created_at: u64 = fields
505                        .get("created_at")
506                        .and_then(|s| s.parse().ok())
507                        .unwrap_or(0);
508                    let updated_at: u64 = fields
509                        .get("updated_at")
510                        .and_then(|s| s.parse().ok())
511                        .unwrap_or(created_at);
512
513                    let namespace = Namespace::parse(&namespace_str).unwrap_or_default();
514                    let domain = domain_str.map_or_else(Domain::new, |_| Domain::new());
515                    let status = parse_memory_status(&status_str);
516                    let tags = parse_tags_string(tags_str);
517
518                    Ok(Some(Memory {
519                        id: id.clone(),
520                        content,
521                        namespace,
522                        domain,
523                        project_id,
524                        branch,
525                        file_path,
526                        status,
527                        created_at,
528                        updated_at,
529                        tombstoned_at: None,
530                        expires_at: None,
531                        embedding: None,
532                        tags,
533                        #[cfg(feature = "group-scope")]
534                        group_id: None,
535                        source: None,
536                        is_summary: false,
537                        source_memory_ids: None,
538                        consolidation_timestamp: None,
539                    }))
540                },
541                Err(e) => Err(Error::OperationFailed {
542                    operation: "redis_get_memory".to_string(),
543                    cause: e.to_string(),
544                }),
545            };
546            self.return_connection(conn);
547            output
548        }
549
550        fn clear(&self) -> Result<()> {
551            let mut conn = self.get_connection()?;
552
553            // Drop and recreate the index
554            let _: redis::RedisResult<String> = redis::cmd("FT.DROPINDEX")
555                .arg(&self.index_name)
556                .arg("DD")
557                .query(&mut conn);
558
559            // Return connection before ensure_index (which gets its own)
560            self.return_connection(conn);
561
562            // Recreate the index
563            self.ensure_index()
564        }
565    }
566
567    /// Parses FT.SEARCH results with scores.
568    fn parse_search_results(values: &[redis::Value]) -> Vec<(MemoryId, f32)> {
569        let mut results = Vec::new();
570
571        if values.is_empty() {
572            return results;
573        }
574
575        let mut i = 1; // Skip count
576        while i + 1 < values.len() {
577            if let redis::Value::BulkString(key_bytes) = &values[i] {
578                let key = String::from_utf8_lossy(key_bytes);
579                let id = key.strip_prefix("mem:").unwrap_or(&key);
580
581                i += 1;
582                let score = parse_score(&values[i]);
583                results.push((MemoryId::new(id), score));
584            }
585
586            i += 1;
587            // Skip fields array if present
588            if i < values.len() && matches!(&values[i], redis::Value::Array(_)) {
589                i += 1;
590            }
591        }
592
593        results
594    }
595
596    /// Parses a score from a Redis value.
597    fn parse_score(value: &redis::Value) -> f32 {
598        match value {
599            redis::Value::BulkString(s) => String::from_utf8_lossy(s).parse::<f32>().unwrap_or(1.0),
600            _ => 1.0,
601        }
602    }
603
604    /// Parses FT.SEARCH results without scores (for `list_all`).
605    fn parse_list_results(values: &[redis::Value]) -> Vec<(MemoryId, f32)> {
606        let mut results = Vec::new();
607
608        if values.is_empty() {
609            return results;
610        }
611
612        let mut i = 1; // Skip count
613        while i < values.len() {
614            if let redis::Value::BulkString(key_bytes) = &values[i] {
615                let key = String::from_utf8_lossy(key_bytes);
616                let id = key.strip_prefix("mem:").unwrap_or(&key);
617                results.push((MemoryId::new(id), 1.0));
618            }
619
620            i += 1;
621            // Skip fields array if present
622            if i < values.len() && matches!(&values[i], redis::Value::Array(_)) {
623                i += 1;
624            }
625        }
626
627        results
628    }
629
630    /// Parses a status string to `MemoryStatus`.
631    fn parse_memory_status(s: &str) -> MemoryStatus {
632        match s.to_lowercase().as_str() {
633            "active" => MemoryStatus::Active,
634            "archived" => MemoryStatus::Archived,
635            "superseded" => MemoryStatus::Superseded,
636            "pending" => MemoryStatus::Pending,
637            "deleted" => MemoryStatus::Deleted,
638            _ => MemoryStatus::Active,
639        }
640    }
641
642    /// Parses a comma-separated tags string.
643    fn parse_tags_string(tags_str: Option<String>) -> Vec<String> {
644        tags_str.map_or_else(Vec::new, |t| {
645            t.split(',')
646                .map(|s| s.trim().to_string())
647                .filter(|s| !s.is_empty())
648                .collect()
649        })
650    }
651}
652
653#[cfg(feature = "redis")]
654pub use implementation::RedisBackend;
655
656#[cfg(not(feature = "redis"))]
657mod stub {
658    use crate::models::{Memory, MemoryId, SearchFilter};
659    use crate::storage::traits::IndexBackend;
660    use crate::{Error, Result};
661
662    /// Stub Redis backend when feature is not enabled.
663    pub struct RedisBackend;
664
665    impl RedisBackend {
666        /// Creates a new Redis backend (stub).
667        ///
668        /// # Errors
669        ///
670        /// Always returns an error because the feature is not enabled.
671        pub fn new(_connection_url: &str, _index_name: impl Into<String>) -> Result<Self> {
672            Err(Error::FeatureNotEnabled("redis".to_string()))
673        }
674
675        /// Creates a backend with default settings (stub).
676        ///
677        /// # Errors
678        ///
679        /// Always returns an error because the feature is not enabled.
680        pub fn with_defaults() -> Result<Self> {
681            Err(Error::FeatureNotEnabled("redis".to_string()))
682        }
683
684        /// Health check stub (DB-HIGH-001).
685        ///
686        /// # Errors
687        ///
688        /// Always returns an error because the feature is not enabled.
689        pub fn health_check(&self) -> Result<bool> {
690            Err(Error::FeatureNotEnabled("redis".to_string()))
691        }
692    }
693
694    impl IndexBackend for RedisBackend {
695        fn index(&self, _memory: &Memory) -> Result<()> {
696            Err(Error::FeatureNotEnabled("redis".to_string()))
697        }
698
699        fn remove(&self, _id: &MemoryId) -> Result<bool> {
700            Err(Error::FeatureNotEnabled("redis".to_string()))
701        }
702
703        fn search(
704            &self,
705            _query: &str,
706            _filter: &SearchFilter,
707            _limit: usize,
708        ) -> Result<Vec<(MemoryId, f32)>> {
709            Err(Error::FeatureNotEnabled("redis".to_string()))
710        }
711
712        fn list_all(&self, _filter: &SearchFilter, _limit: usize) -> Result<Vec<(MemoryId, f32)>> {
713            Err(Error::FeatureNotEnabled("redis".to_string()))
714        }
715
716        fn get_memory(&self, _id: &MemoryId) -> Result<Option<Memory>> {
717            Err(Error::FeatureNotEnabled("redis".to_string()))
718        }
719
720        fn clear(&self) -> Result<()> {
721            Err(Error::FeatureNotEnabled("redis".to_string()))
722        }
723    }
724}
725
726#[cfg(not(feature = "redis"))]
727pub use stub::RedisBackend;
subcog/storage/index/redis.rs

subcog/storage/index/
redis.rs
