siderite_backends/
redis.rs

1//! Redis key, hash and set store.
2//!
3//! [`RedisStore`] wraps a cloneable [`redis::aio::ConnectionManager`]. It is a
4//! specialised data API: relational plans are executed by the SQLite and
5//! PostgreSQL adapters. [`RedisStore::capabilities`] rejects every relational
6//! [`Feature`] before any I/O.
7//!
8//! Keys are strings. A prefix (for example `"cache:"`) is prepended as-is, so
9//! include the separator in the prefix when one is wanted. Values are UTF-8.
10//! Time-to-live is expressed in milliseconds (`PX` / `PEXPIRE` / `PTTL`); a
11//! duration below one millisecond is rejected before a command is sent.
12//!
13//! ```
14//! use siderite_backends::redis::RedisStore;
15//! use siderite_orm::{BackendKind, Feature, OrmError};
16//!
17//! let err = RedisStore::require(Feature::Joins).unwrap_err();
18//! assert!(matches!(
19//!     OrmError::from(err),
20//!     OrmError::Capability(siderite_orm::BackendCapabilityError::Unsupported {
21//!         backend: BackendKind::Redis,
22//!         feature: Feature::Joins,
23//!     })
24//! ));
25//! ```
26
27mod scripts;
28pub use scripts::{MarkerAdmission, MarkerReply};
29
30use std::collections::{HashMap, HashSet};
31use std::time::Duration;
32
33use serde::Serialize;
34use serde::de::DeserializeOwned;
35use siderite_orm::{
36    BackendCapabilities, BackendCapabilityError, BackendError, Feature, OrmError, QueryError,
37};
38use thiserror::Error;
39
40/// Largest TTL accepted, in milliseconds (`i64::MAX / 2`).
41///
42/// Redis rejects a relative TTL whose absolute deadline (now plus the TTL)
43/// overflows a signed 64-bit millisecond count. Half of `i64::MAX` leaves
44/// room for any realistic clock value, so it is validated up front.
45const MAX_TTL_MS: u64 = (i64::MAX / 2) as u64;
46
47/// Failure from [`RedisStore`].
48///
49/// Converts into [`OrmError`]: connection and command failures become
50/// [`BackendError`], JSON and UTF-8 problems become [`QueryError::Decode`],
51/// rejected arguments become [`QueryError::InvalidPlan`], and capability
52/// mismatches stay [`BackendCapabilityError`].
53#[derive(Debug, Error)]
54#[non_exhaustive]
55pub enum RedisError {
56    /// The URL was rejected or the connection could not be established.
57    #[error("redis connection error: {0}")]
58    Connection(String),
59    /// Redis rejected a command, or the reply was not the one this client expects.
60    #[error("redis error: {0}")]
61    Command(String),
62    /// A stored value was not valid UTF-8 or JSON.
63    #[error("cannot decode redis value `{key}`: {reason}")]
64    Decode {
65        /// Logical key (without applying a second prefix).
66        key: String,
67        /// What was wrong with the bytes.
68        reason: String,
69    },
70    /// The call was refused before any command was sent.
71    #[error("{0}")]
72    Invalid(String),
73    /// A relational feature Redis does not provide.
74    #[error(transparent)]
75    Capability(#[from] BackendCapabilityError),
76}
77
78impl From<redis::RedisError> for RedisError {
79    fn from(err: redis::RedisError) -> Self {
80        match err.kind() {
81            redis::ErrorKind::Io
82            | redis::ErrorKind::Client
83            | redis::ErrorKind::AuthenticationFailed
84            | redis::ErrorKind::InvalidClientConfig
85            | redis::ErrorKind::ClusterConnectionNotFound
86            | redis::ErrorKind::MasterNameNotFoundBySentinel
87            | redis::ErrorKind::NoValidReplicasFoundBySentinel
88            | redis::ErrorKind::EmptySentinelList => Self::Connection(err.to_string()),
89            _ => Self::Command(err.to_string()),
90        }
91    }
92}
93
94impl From<RedisError> for OrmError {
95    fn from(err: RedisError) -> Self {
96        match err {
97            RedisError::Connection(message) => Self::Backend(BackendError::Connection(message)),
98            RedisError::Command(message) => Self::Backend(BackendError::Database(message)),
99            RedisError::Decode { key, reason } => Self::Query(QueryError::Decode {
100                column: key,
101                reason,
102            }),
103            RedisError::Invalid(message) => Self::Query(QueryError::InvalidPlan(message)),
104            RedisError::Capability(err) => Self::Capability(err),
105        }
106    }
107}
108
109/// Remaining lifetime of a key, from `PTTL`.
110#[derive(Debug, Clone, Copy, PartialEq, Eq)]
111pub enum Ttl {
112    /// The key does not exist (`PTTL` returned -2).
113    Missing,
114    /// The key exists and has no expiry (`PTTL` returned -1).
115    Persistent,
116    /// Milliseconds Redis still has left on the key.
117    ExpiresIn(Duration),
118}
119
120/// Cloneable Redis client.
121///
122/// Cloning copies the prefix and an `Arc` inside [`redis::aio::ConnectionManager`];
123/// it does not open another TCP connection. Commands issued on any clone share
124/// that multiplexed connection and the database selected by the URL.
125#[derive(Clone)]
126pub struct RedisStore {
127    manager: redis::aio::ConnectionManager,
128    prefix: String,
129}
130
131impl std::fmt::Debug for RedisStore {
132    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
133        f.debug_struct("RedisStore")
134            .field("prefix", &self.prefix)
135            .field("database", &self.database())
136            .finish()
137    }
138}
139
140impl RedisStore {
141    /// Capabilities of this adapter.
142    ///
143    /// The value does not depend on a connection. Checking it never performs I/O.
144    pub fn capabilities() -> BackendCapabilities {
145        BackendCapabilities::redis()
146    }
147
148    /// Reject `feature` when Redis cannot honour it.
149    ///
150    /// No connection is used. Isolation levels, savepoints, joins, row locking
151    /// and the other relational [`Feature`]s all fail here.
152    ///
153    /// # Errors
154    /// [`RedisError::Capability`] when `feature` is not supported.
155    pub fn require(feature: Feature) -> Result<(), RedisError> {
156        Self::capabilities().require(feature).map_err(Into::into)
157    }
158
159    /// Open `url` (for example `redis://127.0.0.1:6379/15`) with an empty prefix.
160    ///
161    /// The path selects the database. Database `15` is the one the live tests use.
162    ///
163    /// # Errors
164    /// [`RedisError::Connection`] when the URL is invalid or the server cannot be reached.
165    pub async fn connect(url: &str) -> Result<Self, RedisError> {
166        let client = redis::Client::open(url)?;
167        Self::connect_with(client, redis::aio::ConnectionManagerConfig::new()).await
168    }
169
170    /// Connect a configured client, including custom TLS trust roots.
171    ///
172    /// Args:
173    ///     client: Driver client with transport, certificate and DB policy.
174    ///     config: Retry, timeout and multiplexed admission limits.
175    ///
176    /// Returns:
177    ///     Connected store with an empty key prefix.
178    ///
179    /// # Errors
180    /// Connection, authentication or certificate verification failure.
181    pub async fn connect_with(
182        client: redis::Client,
183        config: redis::aio::ConnectionManagerConfig,
184    ) -> Result<Self, RedisError> {
185        let manager = redis::aio::ConnectionManager::new_with_config(client, config).await?;
186        Ok(Self {
187            manager,
188            prefix: String::new(),
189        })
190    }
191
192    /// Clone this client with `prefix` replacing the current namespace.
193    ///
194    /// The prefix is concatenated in front of every logical key. It is not
195    /// appended to the previous prefix. `"app:"` turns `"user"` into `"app:user"`.
196    /// An empty prefix stores logical keys unchanged.
197    pub fn with_prefix(&self, prefix: impl Into<String>) -> Self {
198        Self {
199            manager: self.manager.clone(),
200            prefix: prefix.into(),
201        }
202    }
203
204    /// Namespace prepended to logical keys.
205    pub fn prefix(&self) -> &str {
206        &self.prefix
207    }
208
209    /// Physical key stored in Redis for the logical `key`.
210    pub fn key(&self, key: &str) -> String {
211        prefixed_key(&self.prefix, key)
212    }
213
214    /// Database index selected by the connection URL.
215    ///
216    /// This is the index the client was configured with. It is not a live `SELECT`.
217    pub fn database(&self) -> i64 {
218        redis::aio::ConnectionLike::get_db(&self.manager)
219    }
220
221    /// `GET`. `Ok(None)` when the key is absent.
222    ///
223    /// # Errors
224    /// Connection, command, or UTF-8 decode errors. A wrong-type key is a command error.
225    pub async fn get(&self, key: &str) -> Result<Option<String>, RedisError> {
226        let raw: Option<Vec<u8>> = self.query(redis::cmd("GET").arg(self.key(key))).await?;
227        optional_utf8(key, raw)
228    }
229
230    /// `SET`. `ttl` of `None` stores the key with no expiry and clears any previous one.
231    ///
232    /// # Errors
233    /// [`RedisError::Invalid`] when `ttl` is below 1ms or above `i64::MAX / 2` milliseconds,
234    /// before any command is sent. Otherwise connection or command errors.
235    pub async fn set(
236        &self,
237        key: &str,
238        value: &str,
239        ttl: Option<Duration>,
240    ) -> Result<(), RedisError> {
241        let px = optional_ttl(ttl)?;
242        let wrote = self.set_raw(key, value, px, false).await?;
243        if wrote {
244            Ok(())
245        } else {
246            Err(RedisError::Command(format!(
247                "SET `{key}` returned a nil reply"
248            )))
249        }
250    }
251
252    /// `SET key value NX`. `Ok(true)` when this call created the key.
253    ///
254    /// An existing key is left unchanged, including its expiry. This form does
255    /// not set a TTL; use [`set`](Self::set) or [`expire`](Self::expire), or a
256    /// [`pipeline`](Self::pipeline) `SET` with `NX` and `PX`, when the new key
257    /// must expire.
258    ///
259    /// # Errors
260    /// Connection or command errors.
261    pub async fn set_nx(&self, key: &str, value: &str) -> Result<bool, RedisError> {
262        self.set_raw(key, value, None, true).await
263    }
264
265    /// `SET key value NX PX ms` as one command. `Ok(true)` when this call
266    /// created the key; an existing key keeps its value and expiry.
267    ///
268    /// Creation and expiry are one atomic server step: no observer sees the
269    /// key without its TTL, and dropping the returned future either sends the
270    /// whole command or none of it. `ttl` is truncated to whole milliseconds.
271    ///
272    /// # Errors
273    /// [`RedisError::Invalid`] when `ttl` is below 1ms or above
274    /// `i64::MAX / 2` milliseconds, before any command is sent. Otherwise connection or
275    /// command errors.
276    pub async fn set_nx_px(
277        &self,
278        key: &str,
279        value: &str,
280        ttl: Duration,
281    ) -> Result<bool, RedisError> {
282        let px = required_ttl(ttl)?;
283        self.set_raw(key, value, Some(px), true).await
284    }
285
286    /// `DEL` for one logical key. The number is `1` when the key was removed and `0` otherwise.
287    ///
288    /// # Errors
289    /// Connection or command errors.
290    pub async fn del(&self, key: &str) -> Result<u64, RedisError> {
291        let removed: i64 = self.query(redis::cmd("DEL").arg(self.key(key))).await?;
292        as_count(removed)
293    }
294
295    /// `EXISTS` for one logical key.
296    ///
297    /// # Errors
298    /// Connection or command errors.
299    pub async fn exists(&self, key: &str) -> Result<bool, RedisError> {
300        let n: i64 = self.query(redis::cmd("EXISTS").arg(self.key(key))).await?;
301        Ok(n != 0)
302    }
303
304    /// `PEXPIRE`. `Ok(false)` when the key does not exist.
305    ///
306    /// # Errors
307    /// [`RedisError::Invalid`] when `ttl` is below 1ms or too large, before any
308    /// command is sent. Otherwise connection or command errors.
309    pub async fn expire(&self, key: &str, ttl: Duration) -> Result<bool, RedisError> {
310        let ms = required_ttl(ttl)?;
311        let n: i64 = self
312            .query(redis::cmd("PEXPIRE").arg(self.key(key)).arg(ms))
313            .await?;
314        Ok(n != 0)
315    }
316
317    /// `PTTL`.
318    ///
319    /// # Errors
320    /// Connection or command errors, including an unexpected numeric reply.
321    pub async fn ttl(&self, key: &str) -> Result<Ttl, RedisError> {
322        let ms: i64 = self.query(redis::cmd("PTTL").arg(self.key(key))).await?;
323        match ms {
324            -2 => Ok(Ttl::Missing),
325            -1 => Ok(Ttl::Persistent),
326            0.. => Ok(Ttl::ExpiresIn(Duration::from_millis(
327                u64::try_from(ms)
328                    .map_err(|_| RedisError::Command(format!("PTTL `{key}` overflowed: {ms}")))?,
329            ))),
330            other => Err(RedisError::Command(format!(
331                "unexpected PTTL reply for `{key}`: {other}"
332            ))),
333        }
334    }
335
336    /// `INCRBY`. Creates the key at zero first when it is absent. `by` may be negative.
337    ///
338    /// # Errors
339    /// Connection or command errors (the value is not an integer, or it would overflow).
340    pub async fn incr_by(&self, key: &str, by: i64) -> Result<i64, RedisError> {
341        self.query(redis::cmd("INCRBY").arg(self.key(key)).arg(by))
342            .await
343    }
344
345    /// `HGET`. `Ok(None)` when the key or field is absent.
346    ///
347    /// # Errors
348    /// Connection, command, or UTF-8 decode errors.
349    pub async fn hget(&self, key: &str, field: &str) -> Result<Option<String>, RedisError> {
350        let raw: Option<Vec<u8>> = self
351            .query(redis::cmd("HGET").arg(self.key(key)).arg(field))
352            .await?;
353        optional_utf8(key, raw)
354    }
355
356    /// `HSET` of one field. Returns how many fields were newly created (`0` when updated).
357    ///
358    /// # Errors
359    /// Connection or command errors.
360    pub async fn hset(&self, key: &str, field: &str, value: &str) -> Result<u64, RedisError> {
361        let n: i64 = self
362            .query(redis::cmd("HSET").arg(self.key(key)).arg(field).arg(value))
363            .await?;
364        as_count(n)
365    }
366
367    /// `HSET` of many fields in one command. Returns how many fields were newly created.
368    ///
369    /// # Errors
370    /// [`RedisError::Invalid`] when `fields` is empty, before any command is sent.
371    /// Otherwise connection or command errors.
372    pub async fn hset_many<I, F, V>(&self, key: &str, fields: I) -> Result<u64, RedisError>
373    where
374        I: IntoIterator<Item = (F, V)>,
375        F: AsRef<str>,
376        V: AsRef<str>,
377    {
378        let fields = collect_pairs(fields, "hset_many requires at least one field")?;
379        let mut cmd = redis::cmd("HSET");
380        cmd.arg(self.key(key));
381        for (field, value) in &fields {
382            cmd.arg(field).arg(value);
383        }
384        as_count(self.query(&cmd).await?)
385    }
386
387    /// `HGETALL`. An absent key is an empty map, which is how Redis replies.
388    ///
389    /// Field order is not significant.
390    ///
391    /// # Errors
392    /// Connection, command, or UTF-8 decode errors.
393    pub async fn hgetall(&self, key: &str) -> Result<HashMap<String, String>, RedisError> {
394        let raw: Vec<(Vec<u8>, Vec<u8>)> =
395            self.query(redis::cmd("HGETALL").arg(self.key(key))).await?;
396        let mut out = HashMap::with_capacity(raw.len());
397        for (field, value) in raw {
398            out.insert(decode_utf8(key, field)?, decode_utf8(key, value)?);
399        }
400        Ok(out)
401    }
402
403    /// `HDEL`. Returns how many of `fields` were removed.
404    ///
405    /// # Errors
406    /// [`RedisError::Invalid`] when `fields` is empty, before any command is sent.
407    /// Otherwise connection or command errors.
408    pub async fn hdel<I, F>(&self, key: &str, fields: I) -> Result<u64, RedisError>
409    where
410        I: IntoIterator<Item = F>,
411        F: AsRef<str>,
412    {
413        let fields = collect_strings(fields, "hdel requires at least one field")?;
414        let mut cmd = redis::cmd("HDEL");
415        cmd.arg(self.key(key));
416        for field in &fields {
417            cmd.arg(field);
418        }
419        as_count(self.query(&cmd).await?)
420    }
421
422    /// `SADD`. Returns how many of `members` were newly inserted.
423    ///
424    /// # Errors
425    /// [`RedisError::Invalid`] when `members` is empty, before any command is sent.
426    /// Otherwise connection or command errors.
427    pub async fn sadd<I, M>(&self, key: &str, members: I) -> Result<u64, RedisError>
428    where
429        I: IntoIterator<Item = M>,
430        M: AsRef<str>,
431    {
432        self.set_op("SADD", "sadd requires at least one member", key, members)
433            .await
434    }
435
436    /// `SREM`. Returns how many of `members` were removed.
437    ///
438    /// # Errors
439    /// [`RedisError::Invalid`] when `members` is empty, before any command is sent.
440    /// Otherwise connection or command errors.
441    pub async fn srem<I, M>(&self, key: &str, members: I) -> Result<u64, RedisError>
442    where
443        I: IntoIterator<Item = M>,
444        M: AsRef<str>,
445    {
446        self.set_op("SREM", "srem requires at least one member", key, members)
447            .await
448    }
449
450    /// `SMEMBERS`. An absent key is an empty set.
451    ///
452    /// # Errors
453    /// Connection, command, or UTF-8 decode errors.
454    pub async fn smembers(&self, key: &str) -> Result<HashSet<String>, RedisError> {
455        let raw: Vec<Vec<u8>> = self
456            .query(redis::cmd("SMEMBERS").arg(self.key(key)))
457            .await?;
458        raw.into_iter()
459            .map(|bytes| decode_utf8(key, bytes))
460            .collect()
461    }
462
463    /// `SISMEMBER`.
464    ///
465    /// # Errors
466    /// Connection or command errors.
467    pub async fn sismember(&self, key: &str, member: &str) -> Result<bool, RedisError> {
468        let n: i64 = self
469            .query(redis::cmd("SISMEMBER").arg(self.key(key)).arg(member))
470            .await?;
471        Ok(n != 0)
472    }
473
474    /// `GET` decoded as JSON. `Ok(None)` when the key is absent.
475    ///
476    /// A stored JSON `null` is `Ok(Some)` when `T` accepts null, and a decode
477    /// error when it does not.
478    ///
479    /// # Errors
480    /// [`RedisError::Decode`] when the bytes are not UTF-8 or not JSON of type `T`.
481    /// Connection and command errors from [`get`](Self::get).
482    pub async fn get_json<T>(&self, key: &str) -> Result<Option<T>, RedisError>
483    where
484        T: DeserializeOwned,
485    {
486        let Some(raw) = self.get(key).await? else {
487            return Ok(None);
488        };
489        serde_json::from_str(&raw)
490            .map(Some)
491            .map_err(|err| RedisError::Decode {
492                key: key.to_owned(),
493                reason: format!("invalid JSON: {err}"),
494            })
495    }
496
497    /// `SET` of `value` encoded as compact JSON.
498    ///
499    /// # Errors
500    /// [`RedisError::Invalid`] when `value` cannot be serialized or `ttl` is not
501    /// acceptable, before any command is sent. Otherwise the errors of [`set`](Self::set).
502    pub async fn set_json<T>(
503        &self,
504        key: &str,
505        value: &T,
506        ttl: Option<Duration>,
507    ) -> Result<(), RedisError>
508    where
509        T: Serialize,
510    {
511        let px = optional_ttl(ttl)?;
512        let json = serde_json::to_string(value).map_err(|err| {
513            RedisError::Invalid(format!("cannot serialize JSON for `{key}`: {err}"))
514        })?;
515        let wrote = self.set_raw(key, &json, px, false).await?;
516        if wrote {
517            Ok(())
518        } else {
519            Err(RedisError::Command(format!(
520                "SET `{key}` returned a nil reply"
521            )))
522        }
523    }
524
525    /// Run `build` inside one atomic `MULTI`/`EXEC`.
526    ///
527    /// `build` receives a [`redis::Pipeline`] already switched to atomic mode.
528    /// Keys are not prefixed for these raw commands: pass them through [`key`](Self::key).
529    /// Replies decode as `T`, one element per command that was not ignored, in order.
530    ///
531    /// Redis runs the queued commands with no other client's commands interleaved.
532    /// That is not a SQL transaction. A command that fails while `EXEC` is running
533    /// does not roll back earlier commands in the same batch. A command Redis
534    /// rejects while queueing (wrong arity, unknown command) aborts the whole
535    /// batch, and none of it is applied. There are no savepoints and no isolation
536    /// levels; ask [`require`](Self::require) before building a pipeline that
537    /// would need them.
538    ///
539    /// # Errors
540    /// Connection or command errors, including `EXECABORT` and a reply shape `T` cannot decode.
541    pub async fn pipeline<T, F>(&self, build: F) -> Result<T, RedisError>
542    where
543        T: redis::FromRedisValue,
544        F: FnOnce(&mut redis::Pipeline),
545    {
546        let mut pipe = redis::pipe();
547        build(pipe.atomic());
548        let mut conn = self.manager.clone();
549        pipe.query_async(&mut conn).await.map_err(RedisError::from)
550    }
551
552    /// Delete every key in this namespace with `SCAN` and `DEL`.
553    ///
554    /// The match pattern is the prefix with glob metacharacters escaped, then
555    /// `*`. An empty prefix is refused so this cannot wipe a database. This
556    /// method never sends `FLUSHDB` or `FLUSHALL`.
557    ///
558    /// # Errors
559    /// [`RedisError::Invalid`] when the prefix is empty, before any command is
560    /// sent. Otherwise connection or command errors.
561    pub async fn delete_namespace(&self) -> Result<u64, RedisError> {
562        if self.prefix.is_empty() {
563            return Err(RedisError::Invalid(
564                "refusing to delete keys when the prefix is empty; this would match every key"
565                    .into(),
566            ));
567        }
568        let pattern = format!("{}*", glob_literal(&self.prefix));
569        let mut cursor = "0".to_owned();
570        let mut deleted = 0u64;
571        loop {
572            let (next, keys): (String, Vec<Vec<u8>>) = self
573                .query(
574                    redis::cmd("SCAN")
575                        .arg(&cursor)
576                        .arg("MATCH")
577                        .arg(&pattern)
578                        .arg("COUNT")
579                        .arg(200u32),
580                )
581                .await?;
582            if !keys.is_empty() {
583                let mut cmd = redis::cmd("DEL");
584                for key in &keys {
585                    cmd.arg(key);
586                }
587                deleted += as_count(self.query(&cmd).await?)?;
588            }
589            if next == "0" {
590                break;
591            }
592            cursor = next;
593        }
594        Ok(deleted)
595    }
596
597    async fn set_raw(
598        &self,
599        key: &str,
600        value: &str,
601        px: Option<u64>,
602        nx: bool,
603    ) -> Result<bool, RedisError> {
604        let mut cmd = redis::cmd("SET");
605        cmd.arg(self.key(key)).arg(value);
606        if let Some(ms) = px {
607            cmd.arg("PX").arg(ms);
608        }
609        if nx {
610            cmd.arg("NX");
611        }
612        let reply: redis::Value = self.query(&cmd).await?;
613        match reply {
614            redis::Value::Nil => Ok(false),
615            redis::Value::SimpleString(ref status) if status == "OK" => Ok(true),
616            redis::Value::Okay => Ok(true),
617            other => Err(RedisError::Command(format!(
618                "unexpected SET reply for `{key}`: {other:?}"
619            ))),
620        }
621    }
622
623    async fn set_op<I, M>(
624        &self,
625        command: &str,
626        empty: &str,
627        key: &str,
628        members: I,
629    ) -> Result<u64, RedisError>
630    where
631        I: IntoIterator<Item = M>,
632        M: AsRef<str>,
633    {
634        let members = collect_strings(members, empty)?;
635        let mut cmd = redis::cmd(command);
636        cmd.arg(self.key(key));
637        for member in &members {
638            cmd.arg(member);
639        }
640        as_count(self.query(&cmd).await?)
641    }
642
643    async fn query<T>(&self, cmd: &redis::Cmd) -> Result<T, RedisError>
644    where
645        T: redis::FromRedisValue,
646    {
647        let mut conn = self.manager.clone();
648        cmd.query_async(&mut conn).await.map_err(RedisError::from)
649    }
650}
651
652fn prefixed_key(prefix: &str, key: &str) -> String {
653    let mut full = String::with_capacity(prefix.len() + key.len());
654    full.push_str(prefix);
655    full.push_str(key);
656    full
657}
658
659/// Escape `*`, `?`, `[` and `\` so `SCAN MATCH` treats `prefix` as a literal.
660fn glob_literal(prefix: &str) -> String {
661    let mut out = String::with_capacity(prefix.len());
662    for ch in prefix.chars() {
663        if matches!(ch, '*' | '?' | '[' | '\\') {
664            out.push('\\');
665        }
666        out.push(ch);
667    }
668    out
669}
670
671fn optional_ttl(ttl: Option<Duration>) -> Result<Option<u64>, RedisError> {
672    ttl.map(required_ttl).transpose()
673}
674
675fn required_ttl(ttl: Duration) -> Result<u64, RedisError> {
676    let ms = ttl.as_millis();
677    let Ok(ms) = u64::try_from(ms) else {
678        return Err(ttl_rejected());
679    };
680    if ms == 0 || ms > MAX_TTL_MS {
681        return Err(ttl_rejected());
682    }
683    Ok(ms)
684}
685
686fn ttl_rejected() -> RedisError {
687    RedisError::Invalid(
688        "redis TTL must be between 1 millisecond and i64::MAX / 2 milliseconds".into(),
689    )
690}
691
692fn as_count(n: i64) -> Result<u64, RedisError> {
693    u64::try_from(n).map_err(|_| RedisError::Command(format!("negative count from redis: {n}")))
694}
695
696fn decode_utf8(key: &str, bytes: Vec<u8>) -> Result<String, RedisError> {
697    String::from_utf8(bytes).map_err(|err| RedisError::Decode {
698        key: key.to_owned(),
699        reason: format!("value is not UTF-8: {err}"),
700    })
701}
702
703fn optional_utf8(key: &str, bytes: Option<Vec<u8>>) -> Result<Option<String>, RedisError> {
704    bytes.map(|value| decode_utf8(key, value)).transpose()
705}
706
707fn collect_strings<I, S>(items: I, empty: &str) -> Result<Vec<String>, RedisError>
708where
709    I: IntoIterator<Item = S>,
710    S: AsRef<str>,
711{
712    let items: Vec<String> = items
713        .into_iter()
714        .map(|item| item.as_ref().to_owned())
715        .collect();
716    if items.is_empty() {
717        Err(RedisError::Invalid(empty.to_owned()))
718    } else {
719        Ok(items)
720    }
721}
722
723fn collect_pairs<I, F, V>(pairs: I, empty: &str) -> Result<Vec<(String, String)>, RedisError>
724where
725    I: IntoIterator<Item = (F, V)>,
726    F: AsRef<str>,
727    V: AsRef<str>,
728{
729    let pairs: Vec<(String, String)> = pairs
730        .into_iter()
731        .map(|(field, value)| (field.as_ref().to_owned(), value.as_ref().to_owned()))
732        .collect();
733    if pairs.is_empty() {
734        Err(RedisError::Invalid(empty.to_owned()))
735    } else {
736        Ok(pairs)
737    }
738}
739
740#[cfg(test)]
741mod tests {
742    use super::*;
743    use siderite_orm::{BackendKind, IsolationLevel};
744
745    #[test]
746    fn key_prefix_is_prepended_and_empty_prefix_is_identity() {
747        assert_eq!(prefixed_key("", "user:1"), "user:1");
748        assert_eq!(prefixed_key("app:", "user:1"), "app:user:1");
749        assert_eq!(prefixed_key("app:", ""), "app:");
750        assert_eq!(prefixed_key("ns", "user"), "nsuser");
751    }
752
753    #[test]
754    fn glob_metacharacters_in_a_prefix_are_escaped() {
755        assert_eq!(glob_literal("a*b?c[d]\\e"), "a\\*b\\?c\\[d]\\\\e");
756        assert_eq!(glob_literal("plain:"), "plain:");
757    }
758
759    #[test]
760    fn ttl_bounds_are_rejected_before_any_connection() {
761        assert!(matches!(
762            required_ttl(Duration::ZERO),
763            Err(RedisError::Invalid(_))
764        ));
765        assert!(matches!(
766            optional_ttl(Some(Duration::from_nanos(1))),
767            Err(RedisError::Invalid(_))
768        ));
769        assert_eq!(
770            optional_ttl(Some(Duration::from_millis(1500))).unwrap(),
771            Some(1500)
772        );
773        assert_eq!(optional_ttl(None).unwrap(), None);
774        let too_long = Duration::from_millis(MAX_TTL_MS).saturating_add(Duration::from_millis(1));
775        assert!(matches!(
776            required_ttl(too_long),
777            Err(RedisError::Invalid(_))
778        ));
779        let orm = OrmError::from(required_ttl(Duration::ZERO).unwrap_err());
780        assert!(matches!(orm, OrmError::Query(QueryError::InvalidPlan(_))));
781    }
782
783    #[test]
784    fn relational_features_fail_before_any_connection() {
785        assert!(matches!(
786            RedisStore::require(Feature::RowLocking),
787            Err(RedisError::Capability(
788                BackendCapabilityError::RowLockingUnsupported {
789                    backend: BackendKind::Redis
790                }
791            ))
792        ));
793        let err = RedisStore::require(Feature::Joins).unwrap_err();
794        assert!(matches!(
795            OrmError::from(err),
796            OrmError::Capability(BackendCapabilityError::Unsupported {
797                backend: BackendKind::Redis,
798                feature: Feature::Joins,
799            })
800        ));
801        assert!(RedisStore::require(Feature::Savepoints).is_err());
802        assert!(RedisStore::require(Feature::Isolation(IsolationLevel::Serializable)).is_err());
803        assert!(!RedisStore::capabilities().supports(Feature::CaseInsensitiveLike));
804    }
805
806    #[test]
807    fn driver_and_decode_errors_map_into_orm_error() {
808        let connection = RedisError::Connection("refused".into());
809        assert!(matches!(
810            OrmError::from(connection),
811            OrmError::Backend(BackendError::Connection(message)) if message == "refused"
812        ));
813        let command = RedisError::Command("WRONGTYPE".into());
814        assert!(matches!(
815            OrmError::from(command),
816            OrmError::Backend(BackendError::Database(message)) if message == "WRONGTYPE"
817        ));
818        let decode = RedisError::Decode {
819            key: "widget".into(),
820            reason: "invalid JSON".into(),
821        };
822        assert!(matches!(
823            OrmError::from(decode),
824            OrmError::Query(QueryError::Decode { column, reason })
825                if column == "widget" && reason == "invalid JSON"
826        ));
827    }
828
829    #[test]
830    fn empty_argument_lists_are_rejected_without_a_command() {
831        let err = collect_strings(Vec::<&str>::new(), "sadd requires at least one member");
832        assert!(matches!(err, Err(RedisError::Invalid(_))));
833        let err = collect_pairs(
834            Vec::<(&str, &str)>::new(),
835            "hset_many requires at least one field",
836        );
837        assert!(matches!(err, Err(RedisError::Invalid(_))));
838    }
839
840    #[test]
841    fn store_type_is_send_and_sync() {
842        fn assert_send_sync<T: Send + Sync>() {}
843        assert_send_sync::<RedisStore>();
844    }
845
846    #[tokio::test]
847    async fn a_malformed_url_is_a_connection_error() {
848        let err = RedisStore::connect("not a url").await.unwrap_err();
849        assert!(matches!(err, RedisError::Connection(_)));
850        assert!(matches!(
851            OrmError::from(err),
852            OrmError::Backend(BackendError::Connection(_))
853        ));
854    }
855}
siderite_backends/redis.rs

siderite_backends/
redis.rs
