autopush_common/db/redis/
mod.rs

1/// This uses redis as a storage and management
2/// system for Autopush Notifications and Routing information.
3///
4/// Keys for the data are
5/// `autopush/user/{uaid}` String to store the user data
6/// `autopush/co/{uaid}` u64 to store the last time the user has interacted with the server
7/// `autopush/timestamp/{uaid}` u64 to store the last storage timestamp incremented by the server, once messages are delivered
8/// `autopush/channels/{uaid}` List to store the list of the channels of the user
9/// `autopush/msgs/{uaid}` SortedSet to store the list of the pending message ids for the user
10/// `autopush/msgs_exp/{uaid}` SortedSet to store the list of the pending message ids, ordered by expiry date, this is because SortedSet elements can't have independent expiry date
11/// `autopush/msg/{uaid}/{chidmessageid}`, with `{chidmessageid} == {chid}:{version}` String to store
12/// the content of the messages
13///
14mod redis_client;
15
16pub use redis_client::RedisClientImpl;
17
18use std::collections::HashMap;
19use std::time::Duration;
20
21use crate::db::error::DbError;
22use crate::notification::{Notification, default_ttl};
23use crate::util::deserialize_opt_u32_to_duration;
24
25use serde_derive::{Deserialize, Serialize};
26use uuid::Uuid;
27
28/// The settings for accessing the redis contents.
29#[derive(Clone, Debug, Deserialize)]
30#[serde(default)]
31pub struct RedisDbSettings {
32    #[serde(deserialize_with = "deserialize_opt_u32_to_duration")]
33    pub create_timeout: Option<Duration>,
34    #[serde(deserialize_with = "deserialize_opt_u32_to_duration")]
35    // Minimum value is 1 (second), defaults to MAX_ROUTER_TTL_SECS
36    pub router_ttl: Option<Duration>,
37    #[serde(deserialize_with = "deserialize_opt_u32_to_duration")]
38    // Minimum value is 1 (second), defaults to MAX_NOTIFICATION_TTL_SECS
39    pub notification_ttl: Option<Duration>,
40}
41
42#[allow(clippy::derivable_impls)]
43impl Default for RedisDbSettings {
44    fn default() -> Self {
45        Self {
46            create_timeout: Default::default(),
47            router_ttl: Some(Duration::from_secs(crate::MAX_ROUTER_TTL_SECS)),
48            notification_ttl: Some(Duration::from_secs(crate::MAX_NOTIFICATION_TTL_SECS)),
49        }
50    }
51}
52
53impl TryFrom<&str> for RedisDbSettings {
54    type Error = DbError;
55    fn try_from(setting_string: &str) -> Result<Self, Self::Error> {
56        let me: Self = match serde_json::from_str(setting_string) {
57            Ok(me) => me,
58            Err(e) if e.is_eof() => Self::default(),
59            Err(e) => Err(DbError::General(format!(
60                "Could not parse RedisDbSettings: {:?}",
61                e
62            )))?,
63        };
64        if let Some(router_ttl) = me.router_ttl
65            && router_ttl.as_secs() == 0
66        {
67            return Err(DbError::General(
68                "router_ttl must be greater than 0".to_string(),
69            ));
70        }
71        if let Some(notification_ttl) = me.notification_ttl
72            && notification_ttl.as_secs() == 0
73        {
74            return Err(DbError::General(
75                "notification_ttl must be greater than 0".to_string(),
76            ));
77        }
78        // Supply defaults for explicitly null values (deserializer handles missing keys)
79        // Otherwise it defaults to 0 duration, which is not a valid TTL
80        let me = Self {
81            router_ttl: me
82                .router_ttl
83                .or(Some(Duration::from_secs(crate::MAX_ROUTER_TTL_SECS))),
84            notification_ttl: me
85                .notification_ttl
86                .or(Some(Duration::from_secs(crate::MAX_NOTIFICATION_TTL_SECS))),
87            ..me
88        };
89        Ok(me)
90    }
91}
92
93#[derive(Serialize, Default, Deserialize, Clone, Debug)]
94/// A Publishable Notification record. This is a notification that is either
95/// received from a third party or is outbound to a UserAgent.
96///
97pub struct StorableNotification {
98    // Required values
99    #[serde(rename = "channelID")]
100    pub channel_id: Uuid,
101    pub version: String,
102    pub timestamp: u64,
103    // Possibly stored values, provided with a default.
104    // Note: Unlike client-facing `Notification`, these fields
105    // should round-trip faithfully through Redis and not
106    // `skip_serializing` unless truly None.
107    #[serde(default = "default_ttl")]
108    pub ttl: u64,
109    // Optional values, which imply a "None" default.
110    #[serde(skip_serializing_if = "Option::is_none")]
111    pub topic: Option<String>,
112    #[serde(skip_serializing_if = "Option::is_none")]
113    pub data: Option<String>,
114    #[serde(skip_serializing_if = "Option::is_none")]
115    pub sortkey_timestamp: Option<u64>,
116    #[serde(skip_serializing_if = "Option::is_none")]
117    pub headers: Option<HashMap<String, String>>,
118    #[cfg(feature = "reliable_report")]
119    #[serde(skip_serializing_if = "Option::is_none")]
120    pub reliability_id: Option<String>,
121    #[cfg(feature = "reliable_report")]
122    #[serde(skip_serializing_if = "Option::is_none")]
123    pub reliable_state: Option<crate::reliability::ReliabilityState>,
124}
125
126impl From<Notification> for StorableNotification {
127    fn from(notification: Notification) -> Self {
128        Self {
129            channel_id: notification.channel_id,
130            version: notification.version,
131            timestamp: notification.timestamp,
132            ttl: notification.ttl,
133            topic: notification.topic,
134            data: notification.data,
135            sortkey_timestamp: notification.sortkey_timestamp,
136            headers: notification.headers,
137            #[cfg(feature = "reliable_report")]
138            reliability_id: notification.reliability_id,
139            #[cfg(feature = "reliable_report")]
140            reliable_state: notification.reliable_state,
141        }
142    }
143}
144
145impl From<StorableNotification> for Notification {
146    fn from(storable: StorableNotification) -> Self {
147        Self {
148            channel_id: storable.channel_id,
149            version: storable.version,
150            timestamp: storable.timestamp,
151            ttl: storable.ttl,
152            topic: storable.topic,
153            data: storable.data,
154            sortkey_timestamp: storable.sortkey_timestamp,
155            headers: storable.headers,
156            #[cfg(feature = "reliable_report")]
157            reliability_id: storable.reliability_id,
158            #[cfg(feature = "reliable_report")]
159            reliable_state: storable.reliable_state,
160        }
161    }
162}
163
164#[cfg(test)]
165mod tests {
166
167    use std::time::Duration;
168
169    /// A stored notification must round-trip every field through JSON: dropping
170    /// `ttl`, `topic`, or `sortkey_timestamp` makes fetched records look like
171    /// expired legacy messages, breaking deletion and delivery (autopush-rs#1189).
172    #[test]
173    fn test_storable_notification_roundtrip() {
174        use super::StorableNotification;
175        use crate::notification::Notification;
176        use uuid::Uuid;
177
178        // A regular (non-topic) timestamp message.
179        let notif = Notification {
180            channel_id: Uuid::parse_str("DECAFBAD-0000-0000-0000-0123456789AB").unwrap(),
181            version: "gAAAAAdeadbeef".to_owned(),
182            ttl: 300,
183            timestamp: 1_700_000_000,
184            data: Some("encrypted".to_owned()),
185            sortkey_timestamp: Some(1_700_000_000_123),
186            ..Default::default()
187        };
188        let expected_id = notif.chidmessageid();
189
190        let stored: StorableNotification = notif.into();
191        let json = serde_json::to_string(&stored).unwrap();
192        let back: Notification = serde_json::from_str::<StorableNotification>(&json)
193            .unwrap()
194            .into();
195
196        assert_eq!(back.ttl, 300);
197        assert_eq!(back.sortkey_timestamp, Some(1_700_000_000_123));
198        assert_eq!(back.topic, None);
199        // The id used for storage/deletion must survive the round-trip, and must
200        // not degrade into the legacy `{chid}:{version}` form.
201        assert_eq!(back.chidmessageid(), expected_id);
202        assert!(back.chidmessageid().starts_with("02:"));
203
204        // A topic message.
205        let topic_notif = Notification {
206            channel_id: Uuid::parse_str("DECAFBAD-1111-0000-0000-0123456789AB").unwrap(),
207            version: "gAAAAAtopic".to_owned(),
208            ttl: 60,
209            timestamp: 1_700_000_000,
210            topic: Some("mytopic".to_owned()),
211            data: Some("encrypted".to_owned()),
212            ..Default::default()
213        };
214        let expected_topic_id = topic_notif.chidmessageid();
215        let stored: StorableNotification = topic_notif.into();
216        let json = serde_json::to_string(&stored).unwrap();
217        let back: Notification = serde_json::from_str::<StorableNotification>(&json)
218            .unwrap()
219            .into();
220        assert_eq!(back.topic, Some("mytopic".to_owned()));
221        assert_eq!(back.chidmessageid(), expected_topic_id);
222        assert!(back.chidmessageid().starts_with("01:"));
223    }
224
225    #[test]
226    fn test_settings_parse() -> Result<(), crate::db::error::DbError> {
227        let settings = super::RedisDbSettings::try_from("{\"create_timeout\": 123}")?;
228        assert_eq!(
229            settings.create_timeout,
230            Some(std::time::Duration::from_secs(123))
231        );
232        let settings = super::RedisDbSettings::try_from("{}")?;
233        assert_ne!(settings.router_ttl, Some(Duration::from_secs(0)));
234        assert_ne!(settings.notification_ttl, Some(Duration::from_secs(0)));
235        let settings = super::RedisDbSettings::try_from("{\"router_ttl\":0}");
236        assert!(settings.is_err());
237        let settings =
238            super::RedisDbSettings::try_from("{\"notification_ttl\": null, \"router_ttl\": null}")?;
239        assert_eq!(
240            settings.notification_ttl,
241            Some(std::time::Duration::from_secs(
242                crate::MAX_NOTIFICATION_TTL_SECS
243            ))
244        );
245        assert_eq!(
246            settings.router_ttl,
247            Some(std::time::Duration::from_secs(crate::MAX_ROUTER_TTL_SECS))
248        );
249        Ok(())
250    }
251}
autopush_common/db/redis/mod.rs

autopush_common/db/redis/
mod.rs
