flippico_cache/providers/
redis.rs

1use futures_util::stream::StreamExt;
2use log::{error, info};
3use redis::{Client, Commands, RedisResult};
4use serde_json;
5use std::future::Future;
6use std::pin::Pin;
7
8use crate::types::{
9    channels::{ChannelMessage, ListChannel, ListMessage, SubscriptionChannel},
10    provider::{
11        AsyncCallback, CacheProvider, Connectable, FifoProvider, ProviderTrait, PubSubProvider,
12        PubSubProviderError, SortedSetProvider,
13    },
14};
15
16pub struct Redis {
17    pub url: String,
18    pub client: Option<Client>,
19}
20
21impl ProviderTrait for Redis {
22    fn new(url: String) -> Self {
23        Redis { url, client: None }
24    }
25}
26
27impl Connectable for Redis {
28    fn set_client(&mut self) {
29        // Open only parses the URL — a failure is a config error; fail fast at startup.
30        let client = redis::Client::open(self.url.clone()).expect("invalid Redis URL");
31        self.client = Some(client);
32    }
33
34    fn get_connection(&self) -> &Client {
35        self.client.as_ref().unwrap()
36    }
37
38    fn get_mut_connection(&mut self) -> &mut Client {
39        self.client.as_mut().unwrap()
40    }
41}
42
43impl PubSubProvider for Redis
44where
45    Redis: Connectable,
46{
47    type Channels = SubscriptionChannel;
48
49    fn subscribe(
50        &self,
51        callback: AsyncCallback<ChannelMessage>,
52        channel: Self::Channels,
53    ) -> Pin<Box<dyn Future<Output = Result<(), PubSubProviderError>> + Send + '_>> {
54        Box::pin(async move {
55            let client = self.get_connection();
56            let mut subscriber = client
57                .get_async_pubsub()
58                .await
59                .map_err(|_| "Failed to get async pubsub")?;
60            subscriber
61                .subscribe(&[channel.get_channel()])
62                .await
63                .map_err(|_| "Failed to subscribe to channel")?;
64            let mut stream = subscriber.into_on_message();
65            while let Some(msg) = stream.next().await {
66                let payload = msg.get_payload::<String>();
67                if let Ok(json_str) = payload {
68                    if let Ok(channel_msg) = serde_json::from_str::<ChannelMessage>(&json_str) {
69                        info!(
70                            "[{}] New message: {}",
71                            channel_msg.channel.get_channel(),
72                            &json_str
73                        );
74                        callback(channel_msg).await;
75                    }
76                }
77            }
78
79            Ok(())
80        })
81    }
82
83    fn publish(&mut self, channel: Self::Channels, message: ChannelMessage) {
84        let client = self.get_mut_connection();
85        if let Ok(json_str) = serde_json::to_string(&message) {
86            let _ = client.publish::<&str, String, String>(channel.get_channel(), json_str);
87        }
88    }
89}
90
91impl FifoProvider for Redis
92where
93    Redis: Connectable,
94{
95    fn pop(&self, channel_name: ListChannel, job_key: &str) -> Result<ListMessage, &'static str> {
96        let list_name = self.get_list_name(channel_name.get_channel(), job_key);
97        let client = self.get_connection();
98        let mut client = client.clone();
99
100        let result = client
101            .brpop(&list_name, 0.0)
102            .map(|mut items: Vec<String>| items.pop().unwrap_or_default());
103
104        match result {
105            Ok(json_str) => serde_json::from_str::<serde_json::Value>(&json_str)
106                .map(|value| ListMessage {
107                    meta: None,
108                    body: Some(value),
109                })
110                .map_err(|_| "Failed to deserialize list message"),
111            Err(_) => Err("Failed to pop from list"),
112        }
113    }
114
115    fn push(&self, channel_name: ListChannel, job_key: &str, job_payload: serde_json::Value) {
116        let client: &Client = self.get_connection();
117        let mut client = client.clone();
118        let Ok(json_str) = serde_json::to_string(&job_payload) else {
119            return;
120        };
121        let list_name = self.get_list_name(channel_name.get_channel(), job_key);
122        if let Err(e) = client.rpush::<_, _, ()>(&list_name, json_str) {
123            error!("Redis RPUSH {list_name} failed: {e}");
124        }
125    }
126}
127
128impl CacheProvider for Redis
129where
130    Redis: Connectable,
131{
132    fn get(
133        &self,
134        cache_space: crate::types::channels::CacheSpace,
135        key: &str,
136        app_name: Option<String>,
137    ) -> String {
138        let client: &Client = self.get_connection();
139        let mut client = client.clone();
140        let redis_key = self.get_key_name(cache_space, key, app_name);
141        let result: RedisResult<String> = client.get(&redis_key);
142        match result {
143            Ok(value) => value,
144            Err(_) => String::new(),
145        }
146    }
147
148    fn set(
149        &self,
150        cache_space: crate::types::channels::CacheSpace,
151        key: &str,
152        value: serde_json::Value,
153        app_name: Option<String>,
154    ) {
155        let client: &Client = self.get_connection();
156        let mut client = client.clone();
157        let redis_key = self.get_key_name(cache_space, key, app_name);
158        let cache_value = self.get_value(None, value);
159        let Ok(json_str) = serde_json::to_string(&cache_value) else {
160            return;
161        };
162        if let Err(e) = client.set::<_, _, ()>(&redis_key, json_str) {
163            error!("Redis SET {redis_key} failed: {e}");
164        }
165    }
166
167    fn delete(
168        &self,
169        cache_space: crate::types::channels::CacheSpace,
170        key: &str,
171        app_name: Option<String>,
172    ) {
173        let client: &Client = self.get_connection();
174        let mut client = client.clone();
175        let redis_key = self.get_key_name(cache_space, key, app_name);
176        if let Err(e) = client.del::<_, ()>(&redis_key) {
177            error!("Redis DEL {redis_key} failed: {e}");
178        }
179    }
180
181    fn set_with_ttl(
182        &self,
183        cache_space: crate::types::channels::CacheSpace,
184        key: &str,
185        value: serde_json::Value,
186        app_name: Option<String>,
187        ttl_secs: u64,
188    ) {
189        let client: &Client = self.get_connection();
190        let mut client = client.clone();
191        let redis_key = self.get_key_name(cache_space, key, app_name);
192        let cache_value = self.get_value(None, value);
193        let Ok(json_str) = serde_json::to_string(&cache_value) else {
194            return;
195        };
196        if let Err(e) = client.set_ex::<_, _, ()>(&redis_key, json_str, ttl_secs) {
197            error!("Redis SETEX {redis_key} failed: {e}");
198        }
199    }
200}
201
202impl SortedSetProvider for Redis
203where
204    Redis: Connectable,
205{
206    fn zadd(&self, key: &str, score: f64, member: &str) -> Result<u32, &'static str> {
207        let client = self.get_connection();
208        let mut client = client.clone();
209        client
210            .zadd(key, member, score)
211            .map_err(|_| "Failed to ZADD")
212    }
213
214    fn zremrangebyscore(&self, key: &str, min: f64, max: f64) -> Result<u32, &'static str> {
215        let client = self.get_connection();
216        let mut client = client.clone();
217        client
218            .zrembyscore(key, min, max)
219            .map_err(|_| "Failed to ZREMRANGEBYSCORE")
220    }
221
222    fn zcard(&self, key: &str) -> Result<u32, &'static str> {
223        let client = self.get_connection();
224        let mut client = client.clone();
225        client.zcard(key).map_err(|_| "Failed to ZCARD")
226    }
227
228    fn zrangebyscore_withscores(
229        &self,
230        key: &str,
231        min: f64,
232        max: f64,
233        limit: Option<usize>,
234    ) -> Result<Vec<(String, f64)>, &'static str> {
235        let client = self.get_connection();
236        let mut client = client.clone();
237        let result: RedisResult<Vec<(String, f64)>> = if let Some(count) = limit {
238            redis::cmd("ZRANGEBYSCORE")
239                .arg(key)
240                .arg(min)
241                .arg(max)
242                .arg("WITHSCORES")
243                .arg("LIMIT")
244                .arg(0)
245                .arg(count)
246                .query(&mut client)
247        } else {
248            redis::cmd("ZRANGEBYSCORE")
249                .arg(key)
250                .arg(min)
251                .arg(max)
252                .arg("WITHSCORES")
253                .query(&mut client)
254        };
255        result.map_err(|_| "Failed to ZRANGEBYSCORE")
256    }
257
258    fn expire(&self, key: &str, ttl_secs: u64) -> Result<(), &'static str> {
259        let client = self.get_connection();
260        let mut client = client.clone();
261        redis::cmd("EXPIRE")
262            .arg(key)
263            .arg(ttl_secs)
264            .query::<()>(&mut client)
265            .map_err(|_| "Failed to EXPIRE")
266    }
267
268    fn eval_script(
269        &self,
270        script: &str,
271        keys: &[&str],
272        args: &[&str],
273    ) -> Result<Vec<i64>, &'static str> {
274        let client = self.get_connection();
275        let mut client = client.clone();
276        let cmd = redis::Script::new(script);
277        let mut invocation = cmd.prepare_invoke();
278        for k in keys {
279            invocation.key(*k);
280        }
281        for a in args {
282            invocation.arg(*a);
283        }
284        invocation
285            .invoke::<Vec<i64>>(&mut client)
286            .map_err(|_| "Failed to evaluate Lua script")
287    }
288}
flippico_cache/providers/redis.rs

flippico_cache/providers/
redis.rs
