distributed_cache/store.rs
1//
2// Copyright 2018-2026 Accenture Technology
3//
4// Licensed under the Apache License, Version 2.0 (the "License");
5// you may not use this file except in compliance with the License.
6// You may obtain a copy of the License at
7//
8// http://www.apache.org/licenses/LICENSE-2.0
9//
10// Unless required by applicable law or agreed to in writing, software
11// distributed under the License is distributed on an "AS IS" BASIS,
12// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13// See the License for the specific language governing permissions and
14// limitations under the License.
15//
16
17//! The cache operations over a [`RedisBackend`] whose values are opaque bytes
18//! — Rust port of the Java `RedisCacheStore` (design spec §4.4, Q3). The
19//! caller owns serialisation, which maximises cross-layer and cross-language
20//! interop. Keys are strings, each transparently namespaced by an optional
21//! application key-prefix (Q6) so several apps can share one Redis without
22//! colliding; the prefix is stripped again on the way out of `MGET`.
23//!
24//! Every operation is **cluster-safe by construction** (spec §4.5): all
25//! single-key ops route to their slot as-is; `MGET` keys may span slots and
26//! the cluster client scatter-gathers them; `MPUT` is a pipelined batch of
27//! single-key `SETEX` (each routes to its own slot, non-atomic across the map
28//! — correct for a cache, and unavoidable on a cluster). Every stored key
29//! carries a **TTL from creation**: `SETEX` for values, atomic `SET NX EX` for
30//! put-if-absent, and `RPUSH` + `EXPIRE` as ONE atomic `MULTI`/`EXEC` step for
31//! list push — never a two-command sequence that could leave a TTL-less key
32//! if the client died between them (the discipline the sync-over-async return
33//! route follows; the Java module uses an `EVAL` for the same step — the
34//! transaction is this port's ruled equivalent, port spec §4).
35//!
36//! Thread-safe: the backend multiplexes over the one shared connection, so
37//! every worker instance uses this store concurrently.
38
39use platform_core::AppError;
40use redis_connection::RedisBackend;
41
42/// The cache operations (Java `RedisCacheStore`).
43/// Commands that are safe to run twice (`SETEX`, `GET`, `MGET`, `MPUT`, `DEL`,
44/// `LLEN`) go through the backend's idempotent path and are retried once when
45/// the connection was lost to a restart; `SET NX`, `RPUSH` and `LPOP` are never
46/// replayed (`redis_connection::backend`, *Lifecycle*).
47pub struct RedisCacheStore {
48 backend: RedisBackend,
49 key_prefix: String,
50 default_ttl_seconds: u64,
51}
52
53impl RedisCacheStore {
54 /// `backend` is the standalone-or-cluster backend (one shared, multiplexed
55 /// connection); `key_prefix` is prepended to every key (blank = none);
56 /// `default_ttl_seconds` applies to writes that do not specify one.
57 pub fn new(
58 backend: RedisBackend,
59 key_prefix: impl Into<String>,
60 default_ttl_seconds: u64,
61 ) -> Self {
62 RedisCacheStore {
63 backend,
64 key_prefix: key_prefix.into(),
65 default_ttl_seconds,
66 }
67 }
68
69 /// The default TTL (seconds) applied when a write omits one.
70 pub fn default_ttl_seconds(&self) -> u64 {
71 self.default_ttl_seconds
72 }
73
74 /// The backend this store runs on (diagnostics: `cluster()`, `endpoint()`).
75 pub fn backend(&self) -> &RedisBackend {
76 &self.backend
77 }
78
79 /// `SETEX key ttl value`.
80 pub async fn put(
81 &self,
82 key: Option<&str>,
83 value: &[u8],
84 ttl_seconds: u64,
85 ) -> Result<(), AppError> {
86 let key = self.prefixed(key)?;
87 self.backend
88 .query_idempotent::<String>(redis::cmd("SETEX").arg(key).arg(ttl_seconds).arg(value))
89 .await
90 .map(|_| ())
91 }
92
93 /// `GET key` — the value, or `None` on a miss.
94 pub async fn get(&self, key: Option<&str>) -> Result<Option<Vec<u8>>, AppError> {
95 let key = self.prefixed(key)?;
96 self.backend
97 .query_idempotent(redis::cmd("GET").arg(key))
98 .await
99 }
100
101 /// `MGET k1 k2 …` — misses omitted, request order kept, keys returned
102 /// without the prefix. On a cluster the keys may span slots; the cluster
103 /// client scatter-gathers the request.
104 pub async fn mget(&self, keys: &[String]) -> Result<Vec<(String, Vec<u8>)>, AppError> {
105 if keys.is_empty() {
106 return Ok(Vec::new());
107 }
108 let mut cmd = redis::cmd("MGET");
109 for key in keys {
110 cmd.arg(self.prefixed(Some(key))?);
111 }
112 let values: Vec<Option<Vec<u8>>> = self.backend.query_idempotent(&cmd).await?;
113 Ok(keys
114 .iter()
115 .zip(values)
116 .filter_map(|(key, value)| value.map(|bytes| (key.clone(), bytes)))
117 .collect())
118 }
119
120 /// Bulk write as a **pipelined** batch of single-key `SETEX` — one round
121 /// trip, each key keeping its TTL (raw `MSET` sets none). Non-atomic
122 /// across the map, and each key routes to its own slot, so the map may
123 /// span cluster slots freely.
124 pub async fn mput(
125 &self,
126 entries: &[(String, Vec<u8>)],
127 ttl_seconds: u64,
128 ) -> Result<(), AppError> {
129 if entries.is_empty() {
130 return Ok(());
131 }
132 let mut pipe = redis::pipe();
133 for (key, value) in entries {
134 pipe.cmd("SETEX")
135 .arg(self.prefixed(Some(key))?)
136 .arg(ttl_seconds)
137 .arg(value.as_slice());
138 }
139 self.backend
140 .query_pipeline_idempotent::<Vec<String>>(&pipe)
141 .await
142 .map(|_| ())
143 }
144
145 /// `DEL key` — the number of keys removed (0 or 1).
146 pub async fn delete(&self, key: Option<&str>) -> Result<i64, AppError> {
147 let key = self.prefixed(key)?;
148 self.backend
149 .query_idempotent(redis::cmd("DEL").arg(key))
150 .await
151 }
152
153 /// `SET key value NX EX ttl` — atomic put-if-absent with a TTL in one
154 /// command (not `SETNX` then `EXPIRE`, which leaves a TTL-less key if the
155 /// process dies between them). `true` if stored, `false` if the key existed.
156 pub async fn put_if_absent(
157 &self,
158 key: Option<&str>,
159 value: &[u8],
160 ttl_seconds: u64,
161 ) -> Result<bool, AppError> {
162 let key = self.prefixed(key)?;
163 let reply: Option<String> = self
164 .backend
165 .query(
166 redis::cmd("SET")
167 .arg(key)
168 .arg(value)
169 .arg("NX")
170 .arg("EX")
171 .arg(ttl_seconds),
172 )
173 .await?;
174 Ok(reply.as_deref() == Some("OK"))
175 }
176
177 /// `RPUSH key value` then `EXPIRE key ttl` as one atomic `MULTI`/`EXEC`
178 /// step (so the list key is never left TTL-less). The new list length.
179 pub async fn list_push(
180 &self,
181 key: Option<&str>,
182 value: &[u8],
183 ttl_seconds: u64,
184 ) -> Result<i64, AppError> {
185 let key = self.prefixed(key)?;
186 let (length, _expire_set): (i64, i64) = self
187 .backend
188 .query_pipeline(
189 redis::pipe()
190 .atomic()
191 .cmd("RPUSH")
192 .arg(&key)
193 .arg(value)
194 .cmd("EXPIRE")
195 .arg(&key)
196 .arg(ttl_seconds),
197 )
198 .await?;
199 Ok(length)
200 }
201
202 /// `LPOP key` — destructive: the oldest value, or `None` when the list is
203 /// empty.
204 pub async fn list_pop(&self, key: Option<&str>) -> Result<Option<Vec<u8>>, AppError> {
205 let key = self.prefixed(key)?;
206 self.backend.query(redis::cmd("LPOP").arg(key)).await
207 }
208
209 /// `LLEN key` — the list length (0 for an absent list).
210 pub async fn list_len(&self, key: Option<&str>) -> Result<i64, AppError> {
211 let key = self.prefixed(key)?;
212 self.backend
213 .query_idempotent(redis::cmd("LLEN").arg(key))
214 .await
215 }
216
217 fn prefixed(&self, key: Option<&str>) -> Result<String, AppError> {
218 match key.map(str::trim) {
219 Some(key) if !key.is_empty() => Ok(format!("{}{key}", self.key_prefix)),
220 _ => Err(AppError::new(400, "Missing 'key'")),
221 }
222 }
223}