autoendpoint/extractors/
notification.rs

1use crate::error::{ApiError, ApiErrorKind, ApiResult};
2use crate::extractors::{
3    message_id::MessageId, notification_headers::NotificationHeaders, subscription::Subscription,
4};
5use crate::server::AppState;
6use actix_web::{dev::Payload, web, FromRequest, HttpRequest};
7use autopush_common::util::{b64_encode_url, ms_since_epoch, sec_since_epoch};
8use cadence::CountedExt;
9use fernet::MultiFernet;
10use futures::{future, FutureExt};
11use std::collections::HashMap;
12use uuid::Uuid;
13
14/// Extracts notification data from `Subscription` and request data
15#[derive(Clone, Debug)]
16pub struct Notification {
17    /// Unique message_id for this notification
18    pub message_id: String,
19    /// The subscription information block
20    pub subscription: Subscription,
21    /// Set of associated crypto headers
22    pub headers: NotificationHeaders,
23    /// UNIX timestamp in seconds
24    pub timestamp: u64,
25    /// UNIX timestamp in milliseconds
26    pub sort_key_timestamp: u64,
27    /// The encrypted notification body
28    pub data: Option<String>,
29    #[cfg(feature = "reliable_report")]
30    /// The current state the message was in (if tracked)
31    pub reliable_state: Option<autopush_common::reliability::ReliabilityState>,
32    #[cfg(feature = "reliable_report")]
33    pub reliability_id: Option<String>,
34}
35
36impl FromRequest for Notification {
37    type Error = ApiError;
38    type Future = future::LocalBoxFuture<'static, Result<Self, Self::Error>>;
39
40    fn from_request(req: &HttpRequest, payload: &mut Payload) -> Self::Future {
41        let req = req.clone();
42        let mut payload = payload.take();
43
44        async move {
45            let subscription = Subscription::extract(&req).await?;
46            let app_state = web::Data::<AppState>::extract(&req)
47                .await
48                .expect("No server state found");
49
50            let max_notification_ttl_secs = app_state.settings.max_notification_ttl;
51            // Read data
52            let data = web::Bytes::from_request(&req, &mut payload)
53                .await
54                .map_err(|e| {
55                    debug!("▶▶ Request read payload error: {:?}", &e);
56                    ApiErrorKind::PayloadError(e)
57                })?;
58
59            // Convert data to base64
60            let data = if data.is_empty() {
61                None
62            } else {
63                Some(b64_encode_url(&data.to_vec()))
64            };
65
66            let headers =
67                NotificationHeaders::from_request(&req, data.is_some(), max_notification_ttl_secs)?;
68            let timestamp = sec_since_epoch();
69            let sort_key_timestamp = ms_since_epoch();
70            let message_id = Self::generate_message_id(
71                &app_state.fernet,
72                subscription.user.uaid,
73                subscription.channel_id,
74                headers.topic.as_deref(),
75                sort_key_timestamp,
76            );
77
78            #[cfg(feature = "reliable_report")]
79            let reliability_id = subscription.reliability_id.clone();
80
81            #[allow(unused_mut)]
82            let mut notif = Notification {
83                message_id,
84                subscription,
85                headers,
86                timestamp,
87                sort_key_timestamp,
88                data,
89                #[cfg(feature = "reliable_report")]
90                reliable_state: None,
91                #[cfg(feature = "reliable_report")]
92                reliability_id,
93            };
94
95            #[cfg(feature = "reliable_report")]
96            // Brand new notification, so record it as "Received"
97            notif
98                .record_reliability(
99                    &app_state.reliability,
100                    autopush_common::reliability::ReliabilityState::Received,
101                )
102                .await;
103
104            // Record the encoding if we have an encrypted payload
105            if let Some(encoding) = &notif.headers.encoding {
106                if notif.data.is_some() {
107                    app_state
108                        .metrics
109                        .incr(&format!("updates.notification.encoding.{encoding}"))
110                        .ok();
111                }
112            }
113
114            Ok(notif)
115        }
116        .boxed_local()
117    }
118}
119
120impl From<Notification> for autopush_common::notification::Notification {
121    fn from(notification: Notification) -> Self {
122        let topic = notification.headers.topic.clone();
123        let sortkey_timestamp = topic.is_none().then_some(notification.sort_key_timestamp);
124        autopush_common::notification::Notification {
125            channel_id: notification.subscription.channel_id,
126            version: notification.message_id,
127            ttl: notification.headers.ttl as u64,
128            topic,
129            timestamp: notification.timestamp,
130            data: notification.data,
131            sortkey_timestamp,
132            reliability_id: notification.subscription.reliability_id,
133            headers: {
134                let headers: HashMap<String, String> = notification.headers.into();
135                if headers.is_empty() {
136                    None
137                } else {
138                    Some(headers)
139                }
140            },
141            #[cfg(feature = "reliable_report")]
142            reliable_state: notification.reliable_state,
143        }
144    }
145}
146
147impl Notification {
148    /// Generate a message-id suitable for accessing the message
149    ///
150    /// For topic messages, a sort_key version of 01 is used, and the topic
151    /// is included for reference:
152    ///
153    ///     Encrypted('01' : uaid.hex : channel_id.hex : topic)
154    ///
155    /// For non-topic messages, a sort_key version of 02 is used:
156    ///
157    ///     Encrypted('02' : uaid.hex : channel_id.hex : timestamp)
158    fn generate_message_id(
159        fernet: &MultiFernet,
160        uaid: Uuid,
161        channel_id: Uuid,
162        topic: Option<&str>,
163        timestamp: u64,
164    ) -> String {
165        let message_id = if let Some(topic) = topic {
166            MessageId::WithTopic {
167                uaid,
168                channel_id,
169                topic: topic.to_string(),
170            }
171        } else {
172            MessageId::WithoutTopic {
173                uaid,
174                channel_id,
175                timestamp,
176            }
177        };
178
179        message_id.encrypt(fernet)
180    }
181
182    pub fn has_topic(&self) -> bool {
183        self.headers.topic.is_some()
184    }
185
186    /// Serialize the notification for delivery to the connection server. Some
187    /// fields in `autopush_common`'s `Notification` are marked with
188    /// `#[serde(skip_serializing)]` so they are not shown to the UA. These
189    /// fields are still required when delivering to the connection server, so
190    /// we can't simply convert this notification type to that one and serialize
191    /// via serde.
192    pub fn serialize_for_delivery(&self) -> ApiResult<HashMap<&'static str, serde_json::Value>> {
193        let mut map = HashMap::new();
194
195        map.insert(
196            "channelID",
197            serde_json::to_value(self.subscription.channel_id)?,
198        );
199        map.insert("version", serde_json::to_value(&self.message_id)?);
200        map.insert("ttl", serde_json::to_value(self.headers.ttl)?);
201        map.insert("topic", serde_json::to_value(&self.headers.topic)?);
202        map.insert("timestamp", serde_json::to_value(self.timestamp)?);
203        #[cfg(feature = "reliable_report")]
204        {
205            if let Some(reliability_id) = &self.subscription.reliability_id {
206                map.insert("reliability_id", serde_json::to_value(reliability_id)?);
207            }
208            if let Some(reliable_state) = self.reliable_state {
209                map.insert(
210                    "reliable_state",
211                    serde_json::to_value(reliable_state.to_string())?,
212                );
213            }
214        }
215        if let Some(data) = &self.data {
216            map.insert("data", serde_json::to_value(data)?);
217
218            let headers: HashMap<_, _> = self.headers.clone().into();
219            map.insert("headers", serde_json::to_value(headers)?);
220        }
221
222        Ok(map)
223    }
224
225    #[cfg(feature = "reliable_report")]
226    pub async fn record_reliability(
227        &mut self,
228        reliability: &autopush_common::reliability::PushReliability,
229        state: autopush_common::reliability::ReliabilityState,
230    ) {
231        self.reliable_state = reliability
232            .record(
233                &self.reliability_id,
234                state,
235                &self.reliable_state,
236                Some(self.timestamp + self.headers.ttl as u64),
237            )
238            .await
239            .inspect_err(|e| {
240                warn!("🔍⚠️ Unable to record reliability state log: {:?}", e);
241            })
242            .unwrap_or(Some(state))
243    }
244}
autoendpoint/extractors/notification.rs

autoendpoint/extractors/
notification.rs
