autoendpoint/routers/
common.rs

1use crate::error::{ApiError, ApiResult};
2use crate::extractors::notification::Notification;
3use crate::headers::vapid::VapidHeaderWithKey;
4use crate::routers::RouterError;
5use actix_web::http::StatusCode;
6use autopush_common::db::client::DbClient;
7use autopush_common::metric_name::MetricName;
8use autopush_common::metrics::StatsdClientExt;
9use autopush_common::util::InsertOpt;
10use cadence::{Counted, StatsdClient, Timed};
11use std::collections::HashMap;
12use uuid::Uuid;
13
14use super::fcm::error::FcmError;
15
16/// Convert a notification into a WebPush message
17pub fn build_message_data(notification: &Notification) -> ApiResult<HashMap<&'static str, String>> {
18    let mut message_data = HashMap::new();
19    message_data.insert("chid", notification.subscription.channel_id.to_string());
20
21    // Only add the other headers if there's data
22    if let Some(data) = &notification.data {
23        message_data.insert("body", data.clone());
24        message_data.insert_opt("con", notification.headers.encoding.as_ref());
25        message_data.insert_opt("enc", notification.headers.encryption.as_ref());
26        message_data.insert_opt("cryptokey", notification.headers.crypto_key.as_ref());
27        // Report the data to the UA. How this value is reported back is still a work in progress, but
28        // we do set the state to "accepted" on desktop "ACK" at least.
29        trace!(
30            "🔍 Sending Reliability ID: {:?}",
31            notification.subscription.reliability_id
32        );
33        message_data.insert_opt("rid", notification.subscription.reliability_id.as_ref());
34    }
35
36    Ok(message_data)
37}
38
39/// Check the data against the max data size and return an error if there is too
40/// much data.
41pub fn message_size_check(data: &[u8], max_data: usize) -> Result<(), RouterError> {
42    if data.len() > max_data {
43        trace!("Data is too long by {} bytes", data.len() - max_data);
44        Err(RouterError::TooMuchData(data.len() - max_data))
45    } else {
46        Ok(())
47    }
48}
49
50/// Handle a bridge error by logging, updating metrics, etc
51/// This function uses the standard `slog` recording mechanisms and
52/// optionally calls a generic metric recording function for some
53/// types of errors. The error is returned by this function for later
54/// processing. This can include being called by the sentry middleware,
55/// which uses the `RecordableError` trait to optionally record metrics.
56/// see [autopush_common::middleware::sentry::SentryWrapperMiddleware].`call()` method
57pub async fn handle_error(
58    error: RouterError,
59    metrics: &StatsdClient,
60    db: &dyn DbClient,
61    platform: &str,
62    app_id: &str,
63    uaid: Uuid,
64    vapid: Option<VapidHeaderWithKey>,
65) -> ApiError {
66    match &error {
67        RouterError::Authentication => {
68            error!("Bridge authentication error");
69            incr_error_metric(
70                metrics,
71                platform,
72                app_id,
73                "authentication",
74                error.status(),
75                error.errno(),
76            );
77        }
78        RouterError::RequestTimeout => {
79            // Bridge timeouts are common.
80            info!("Bridge timeout");
81            incr_error_metric(
82                metrics,
83                platform,
84                app_id,
85                "timeout",
86                error.status(),
87                error.errno(),
88            );
89        }
90        RouterError::Connect(e) => {
91            warn!("Bridge unavailable: {}", e);
92            incr_error_metric(
93                metrics,
94                platform,
95                app_id,
96                "connection_unavailable",
97                error.status(),
98                error.errno(),
99            );
100        }
101        RouterError::NotFound => {
102            debug!("Bridge recipient not found, removing user");
103            incr_error_metric(
104                metrics,
105                platform,
106                app_id,
107                "recipient_gone",
108                error.status(),
109                error.errno(),
110            );
111
112            if let Err(e) = db.remove_user(&uaid).await {
113                warn!("Error while removing user due to bridge not_found: {}", e);
114            }
115        }
116        RouterError::TooMuchData(_) => {
117            // Do not log this error since it's fairly common.
118            incr_error_metric(
119                metrics,
120                platform,
121                app_id,
122                "too_much_data",
123                error.status(),
124                error.errno(),
125            );
126        }
127        RouterError::Fcm(FcmError::Upstream {
128            error_code: status, ..
129        }) => incr_error_metric(
130            metrics,
131            platform,
132            app_id,
133            &format!("upstream_{status}"),
134            error.status(),
135            error.errno(),
136        ),
137
138        _ => {
139            warn!("Unknown error while sending bridge request: {error}");
140            incr_error_metric(
141                metrics,
142                platform,
143                app_id,
144                "unknown",
145                error.status(),
146                error.errno(),
147            );
148        }
149    }
150
151    let mut err = ApiError::from(error);
152
153    if let Some(Ok(claims)) = vapid.map(|v| v.vapid.claims()) {
154        let mut extras = err.extras.unwrap_or_default();
155        if let Some(sub) = claims.sub {
156            extras.extend([("sub".to_owned(), sub)]);
157        }
158        err.extras = Some(extras);
159    };
160    err
161}
162
163/// Increment `notification.bridge.error`
164pub fn incr_error_metric(
165    metrics: &StatsdClient,
166    platform: &str,
167    app_id: &str,
168    reason: &str,
169    status: StatusCode,
170    errno: Option<usize>,
171) {
172    // I'd love to extract the status and errno from the passed ApiError, but a2 error handling makes that impossible.
173    metrics
174        .incr_with_tags(MetricName::NotificationBridgeError)
175        .with_tag("platform", platform)
176        .with_tag("app_id", app_id)
177        .with_tag("reason", reason)
178        .with_tag("error", &status.to_string())
179        .with_tag("errno", &errno.unwrap_or(0).to_string())
180        .send();
181}
182
183/// Update metrics after successfully routing the notification
184pub fn incr_success_metrics(
185    metrics: &StatsdClient,
186    platform: &str,
187    app_id: &str,
188    notification: &Notification,
189) {
190    metrics
191        .incr_with_tags(MetricName::NotificationBridgeSent)
192        .with_tag("platform", platform)
193        .with_tag("app_id", app_id)
194        .send();
195    metrics
196        .count_with_tags(
197            MetricName::NotificationMessageData.as_ref(),
198            notification.data.as_ref().map(String::len).unwrap_or(0) as i64,
199        )
200        .with_tag("platform", platform)
201        .with_tag("app_id", app_id)
202        .with_tag("destination", "Direct")
203        .send();
204    metrics
205        .time_with_tags(
206            MetricName::NotificationTotalRequestTime.as_ref(),
207            (autopush_common::util::sec_since_epoch() - notification.timestamp) * 1000,
208        )
209        .with_tag("platform", platform)
210        .with_tag("app_id", app_id)
211        .send();
212}
213
214/// Common router test code
215#[cfg(test)]
216pub mod tests {
217    use crate::extractors::notification::Notification;
218    use crate::extractors::notification_headers::NotificationHeaders;
219    use crate::extractors::routers::RouterType;
220    use crate::extractors::subscription::Subscription;
221    use autopush_common::db::User;
222    use std::collections::HashMap;
223    use std::sync::{Arc, atomic::AtomicUsize};
224    use uuid::Uuid;
225
226    pub const CHANNEL_ID: &str = "deadbeef-13f9-4639-87f9-2ff731824f34";
227
228    /// Get the test channel ID as a Uuid
229    pub fn channel_id() -> Uuid {
230        Uuid::parse_str(CHANNEL_ID).unwrap()
231    }
232
233    /// Create a notification
234    pub fn make_notification(
235        router_data: HashMap<String, serde_json::Value>,
236        data: Option<String>,
237        router_type: RouterType,
238    ) -> Notification {
239        let user = User::builder()
240            .router_data(router_data)
241            .router_type(router_type.to_string())
242            .build()
243            .unwrap();
244        Notification {
245            message_id: "test-message-id".to_string(),
246            subscription: Subscription {
247                user,
248                channel_id: channel_id(),
249                vapid: None,
250                reliability_id: None,
251            },
252            headers: NotificationHeaders {
253                ttl: 0,
254                topic: Some("test-topic".to_string()),
255                encoding: Some("test-encoding".to_string()),
256                encryption: Some("test-encryption".to_string()),
257                crypto_key: Some("test-crypto-key".to_string()),
258            },
259            timestamp: 0,
260            sort_key_timestamp: 0,
261            data,
262            #[cfg(feature = "reliable_report")]
263            reliable_state: None,
264            #[cfg(feature = "reliable_report")]
265            reliability_id: None,
266            in_process_counter: Arc::new(AtomicUsize::new(0)),
267        }
268    }
269}
autoendpoint/routers/common.rs

autoendpoint/routers/
common.rs
