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`).
43pub struct RedisCacheStore {
44 backend: RedisBackend,
45 key_prefix: String,
46 default_ttl_seconds: u64,
47}
48
49impl RedisCacheStore {
50 /// `backend` is the standalone-or-cluster backend (one shared, multiplexed
51 /// connection); `key_prefix` is prepended to every key (blank = none);
52 /// `default_ttl_seconds` applies to writes that do not specify one.
53 pub fn new(
54 backend: RedisBackend,
55 key_prefix: impl Into<String>,
56 default_ttl_seconds: u64,
57 ) -> Self {
58 RedisCacheStore {
59 backend,
60 key_prefix: key_prefix.into(),
61 default_ttl_seconds,
62 }
63 }
64
65 /// The default TTL (seconds) applied when a write omits one.
66 pub fn default_ttl_seconds(&self) -> u64 {
67 self.default_ttl_seconds
68 }
69
70 /// The backend this store runs on (diagnostics: `cluster()`, `endpoint()`).
71 pub fn backend(&self) -> &RedisBackend {
72 &self.backend
73 }
74
75 /// `SETEX key ttl value`.
76 pub async fn put(
77 &self,
78 key: Option<&str>,
79 value: &[u8],
80 ttl_seconds: u64,
81 ) -> Result<(), AppError> {
82 let key = self.prefixed(key)?;
83 self.backend
84 .query::<String>(redis::cmd("SETEX").arg(key).arg(ttl_seconds).arg(value))
85 .await
86 .map(|_| ())
87 }
88
89 /// `GET key` — the value, or `None` on a miss.
90 pub async fn get(&self, key: Option<&str>) -> Result<Option<Vec<u8>>, AppError> {
91 let key = self.prefixed(key)?;
92 self.backend.query(redis::cmd("GET").arg(key)).await
93 }
94
95 /// `MGET k1 k2 …` — misses omitted, request order kept, keys returned
96 /// without the prefix. On a cluster the keys may span slots; the cluster
97 /// client scatter-gathers the request.
98 pub async fn mget(&self, keys: &[String]) -> Result<Vec<(String, Vec<u8>)>, AppError> {
99 if keys.is_empty() {
100 return Ok(Vec::new());
101 }
102 let mut cmd = redis::cmd("MGET");
103 for key in keys {
104 cmd.arg(self.prefixed(Some(key))?);
105 }
106 let values: Vec<Option<Vec<u8>>> = self.backend.query(&cmd).await?;
107 Ok(keys
108 .iter()
109 .zip(values)
110 .filter_map(|(key, value)| value.map(|bytes| (key.clone(), bytes)))
111 .collect())
112 }
113
114 /// Bulk write as a **pipelined** batch of single-key `SETEX` — one round
115 /// trip, each key keeping its TTL (raw `MSET` sets none). Non-atomic
116 /// across the map, and each key routes to its own slot, so the map may
117 /// span cluster slots freely.
118 pub async fn mput(
119 &self,
120 entries: &[(String, Vec<u8>)],
121 ttl_seconds: u64,
122 ) -> Result<(), AppError> {
123 if entries.is_empty() {
124 return Ok(());
125 }
126 let mut pipe = redis::pipe();
127 for (key, value) in entries {
128 pipe.cmd("SETEX")
129 .arg(self.prefixed(Some(key))?)
130 .arg(ttl_seconds)
131 .arg(value.as_slice());
132 }
133 self.backend
134 .query_pipeline::<Vec<String>>(&pipe)
135 .await
136 .map(|_| ())
137 }
138
139 /// `DEL key` — the number of keys removed (0 or 1).
140 pub async fn delete(&self, key: Option<&str>) -> Result<i64, AppError> {
141 let key = self.prefixed(key)?;
142 self.backend.query(redis::cmd("DEL").arg(key)).await
143 }
144
145 /// `SET key value NX EX ttl` — atomic put-if-absent with a TTL in one
146 /// command (not `SETNX` then `EXPIRE`, which leaves a TTL-less key if the
147 /// process dies between them). `true` if stored, `false` if the key existed.
148 pub async fn put_if_absent(
149 &self,
150 key: Option<&str>,
151 value: &[u8],
152 ttl_seconds: u64,
153 ) -> Result<bool, AppError> {
154 let key = self.prefixed(key)?;
155 let reply: Option<String> = self
156 .backend
157 .query(
158 redis::cmd("SET")
159 .arg(key)
160 .arg(value)
161 .arg("NX")
162 .arg("EX")
163 .arg(ttl_seconds),
164 )
165 .await?;
166 Ok(reply.as_deref() == Some("OK"))
167 }
168
169 /// `RPUSH key value` then `EXPIRE key ttl` as one atomic `MULTI`/`EXEC`
170 /// step (so the list key is never left TTL-less). The new list length.
171 pub async fn list_push(
172 &self,
173 key: Option<&str>,
174 value: &[u8],
175 ttl_seconds: u64,
176 ) -> Result<i64, AppError> {
177 let key = self.prefixed(key)?;
178 let (length, _expire_set): (i64, i64) = self
179 .backend
180 .query_pipeline(
181 redis::pipe()
182 .atomic()
183 .cmd("RPUSH")
184 .arg(&key)
185 .arg(value)
186 .cmd("EXPIRE")
187 .arg(&key)
188 .arg(ttl_seconds),
189 )
190 .await?;
191 Ok(length)
192 }
193
194 /// `LPOP key` — destructive: the oldest value, or `None` when the list is
195 /// empty.
196 pub async fn list_pop(&self, key: Option<&str>) -> Result<Option<Vec<u8>>, AppError> {
197 let key = self.prefixed(key)?;
198 self.backend.query(redis::cmd("LPOP").arg(key)).await
199 }
200
201 /// `LLEN key` — the list length (0 for an absent list).
202 pub async fn list_len(&self, key: Option<&str>) -> Result<i64, AppError> {
203 let key = self.prefixed(key)?;
204 self.backend.query(redis::cmd("LLEN").arg(key)).await
205 }
206
207 fn prefixed(&self, key: Option<&str>) -> Result<String, AppError> {
208 match key.map(str::trim) {
209 Some(key) if !key.is_empty() => Ok(format!("{}{key}", self.key_prefix)),
210 _ => Err(AppError::new(400, "Missing 'key'")),
211 }
212 }
213}