Feat: Add urgency support

Allow clients to define the minimum urgency of messages they wish to receive. For example, a client with low battery can request a high urgency message only until their battery is good again.

Note: this is a resubmit of
#815.

Author: jrconlin

autoconnect/Cargo.toml

@@ -56,6 +56,7 @@ docopt = "1.1"
 default = ["bigtable", "reliable_report"]
 bigtable = ["autopush_common/bigtable", "autoconnect_settings/bigtable"]
 emulator = ["bigtable"]
+urgency = ["autoconnect_ws/urgency"]
 log_vapid = []
 reliable_report = [
     "autoconnect_settings/reliable_report",

autoconnect/autoconnect-common/Cargo.toml

@@ -27,3 +27,4 @@ autopush_common.workspace = true
 
 [features]
 test-support = []
+urgency = []

autoconnect/autoconnect-common/src/protocol.rs

@@ -14,6 +14,9 @@ use uuid::Uuid;
 
 use autopush_common::notification::Notification;
 
+#[cfg(feature = "urgency")]
+use autopush_common::db::Urgency;
+
 #[derive(Debug, Eq, PartialEq, Serialize)]
 #[serde(untagged)]
 pub enum BroadcastValue {
@@ -66,6 +69,11 @@ pub enum ClientMessage {
         version: String,
     },
 
+    #[cfg(feature = "urgency")]
+    Urgency {
+        min: Urgency,
+    },
+
     Ping,
 }
 
@@ -123,6 +131,11 @@ pub enum ServerMessage {
 
     Notification(Notification),
 
+    #[cfg(feature = "urgency")]
+    Urgency {
+        status: u32,
+    },
+
     Ping,
 }
 

autoconnect/autoconnect-ws/Cargo.toml

@@ -39,3 +39,4 @@ reliable_report = [
     "autopush_common/reliable_report",
     "autoconnect_ws_sm/reliable_report",
 ]
+urgency = ["autoconnect_ws_sm/urgency"]

autoconnect/autoconnect-ws/autoconnect-ws-sm/Cargo.toml

@@ -33,3 +33,4 @@ autoconnect_common = { workspace = true, features = ["test-support"] }
 
 [features]
 reliable_report = []
+urgency = ["autoconnect_common/urgency"]

autoconnect/autoconnect-ws/autoconnect-ws-sm/src/identified/mod.rs

@@ -12,7 +12,7 @@ use autoconnect_common::{
 
 use autoconnect_settings::{AppState, Settings};
 use autopush_common::{
-    db::User,
+    db::{Urgency, User},
     notification::Notification,
     util::{ms_since_epoch, user_agent::UserAgentInfo},
 };
@@ -297,6 +297,8 @@ pub struct ClientFlags {
     pub old_record_version: bool,
     /// First time a user has connected "today"
     pub emit_channel_metrics: bool,
+    /// Minimum urgency
+    pub min_urgency: Urgency,
 }
 
 impl Default for ClientFlags {
@@ -307,6 +309,7 @@ impl Default for ClientFlags {
             check_storage: false,
             old_record_version: false,
             emit_channel_metrics: false,
+            min_urgency: Urgency::VeryLow,
         }
     }
 }

autoconnect/autoconnect-ws/autoconnect-ws-sm/src/identified/on_client_msg.rs

@@ -9,6 +9,9 @@ use autoconnect_common::{
 };
 use autopush_common::{endpoint::make_endpoint, util::sec_since_epoch};
 
+#[cfg(feature = "urgency")]
+use autopush_common::{db::Urgency, util::ms_since_epoch};
+
 use super::WebPushClient;
 use crate::error::{SMError, SMErrorKind};
 
@@ -38,6 +41,8 @@ impl WebPushClient {
                 self.nack(code);
                 Ok(vec![])
             }
+            #[cfg(feature = "urgency")]
+            ClientMessage::Urgency { min } => Ok(self.change_min_urgency(min).await?),
             ClientMessage::Ping => Ok(vec![self.ping()?]),
         }
     }
@@ -330,4 +335,40 @@ impl WebPushClient {
             Ok(vec![])
         }
     }
+
+    /// Update minimum urgency for the user and the flag
+    ///
+    /// If the new urgency is lower than the previous one,
+    /// We check pending messages, to send messages that were
+    /// retained because of their urgency
+    #[cfg(feature = "urgency")]
+    async fn change_min_urgency(
+        &mut self,
+        new_min: Urgency,
+    ) -> Result<Vec<ServerMessage>, SMError> {
+        // Change the min urgency
+        self.flags.min_urgency = new_min;
+
+        if let Some(mut user) = self.app_state.db.get_user(&self.uaid).await? {
+            // If the user hasn't set a minimum urgency yet, they receive all messages,
+            // which is equivalent to setting very-low as a minimum
+            let current_urgency = user.urgency.unwrap_or(Urgency::VeryLow);
+
+            // We update the user
+            user.urgency = Some(new_min);
+            user.connected_at = ms_since_epoch();
+            self.app_state.db.update_user(&mut user).await?;
+
+            let mut res = vec![ServerMessage::Urgency { status: 200 }];
+            // if new urgency < previous: fetch pending messages
+            if new_min < current_urgency {
+                self.ack_state.unacked_stored_highest = None;
+                self.current_timestamp = None;
+                res.append(&mut self.check_storage().await?);
+            }
+            Ok(res)
+        } else {
+            Ok(vec![ServerMessage::Urgency { status: 404 }])
+        }
+    }
 }

autoconnect/autoconnect-ws/autoconnect-ws-sm/src/identified/on_server_notif.rs

@@ -6,7 +6,10 @@ use autopush_common::{
 };
 
 use super::WebPushClient;
-use crate::error::{SMError, SMErrorKind};
+use crate::{
+    error::{SMError, SMErrorKind},
+    identified::Urgency,
+};
 
 impl WebPushClient {
     /// Handle a `ServerNotification` for this user
@@ -119,7 +122,11 @@ impl WebPushClient {
         let mut expired_topic_sort_keys = vec![];
         messages.retain(|msg| {
             if !msg.expired(now_sec) {
-                return true;
+                if let Some(headers) = msg.headers.as_ref() {
+                    return Urgency::from(headers.get("urgency")) >= self.flags.min_urgency;
+                } else {
+                    return true;
+                }
             }
             if msg.sortkey_timestamp.is_none() {
                 expired_topic_sort_keys.push(msg.chidmessageid());

autoconnect/autoconnect-ws/autoconnect-ws-sm/src/unidentified.rs

@@ -9,7 +9,7 @@ use autoconnect_common::{
 };
 use autoconnect_settings::{AppState, Settings};
 use autopush_common::{
-    db::{User, USER_RECORD_VERSION},
+    db::{Urgency, User, USER_RECORD_VERSION},
     util::{ms_since_epoch, ms_utc_midnight},
 };
 
@@ -145,6 +145,7 @@ impl UnidentifiedClient {
                         .record_version
                         .map_or(true, |rec_ver| rec_ver < USER_RECORD_VERSION),
                     emit_channel_metrics: user.connected_at < ms_utc_midnight(),
+                    min_urgency: user.urgency.unwrap_or(Urgency::VeryLow),
                     ..Default::default()
                 };
                 user.node_id = Some(self.app_state.router_url.to_owned());

autoendpoint/src/extractors/notification_headers.rs

@@ -7,7 +7,7 @@ use lazy_static::lazy_static;
 use regex::Regex;
 use std::cmp::min;
 use std::collections::HashMap;
-use validator::Validate;
+use validator::{Validate, ValidationError};
 use validator_derive::Validate;
 
 lazy_static! {
@@ -37,6 +37,9 @@ pub struct NotificationHeaders {
     )]
     pub topic: Option<String>,
 
+    #[validate(custom(function = "validate_urgency"))]
+    pub urgency: Option<String>,
+
     // These fields are validated separately, because the validation is complex
     // and based upon the content encoding
     pub encoding: Option<String>,
@@ -45,10 +48,21 @@ pub struct NotificationHeaders {
     pub crypto_key: Option<String>,
 }
 
+fn validate_urgency(value: &str) -> Result<(), ValidationError> {
+    if ["very-low", "low", "normal", "high"].contains(&value) {
+        Ok(())
+    } else {
+        Err(ValidationError::new(
+            "Value not equal to \"very-low\", \"low\", \"normal\" or \"high\"",
+        ))
+    }
+}
+
 impl From<NotificationHeaders> for HashMap<String, String> {
     fn from(headers: NotificationHeaders) -> Self {
         let mut map = HashMap::new();
 
+        map.insert_opt("urgency", headers.urgency);
         map.insert_opt("encoding", headers.encoding);
         map.insert_opt("encryption", headers.encryption);
         map.insert_opt("encryption_key", headers.encryption_key);
@@ -73,11 +87,13 @@ impl NotificationHeaders {
             .map(|ttl| min(ttl, MAX_NOTIFICATION_TTL as i64))
             .ok_or(ApiErrorKind::NoTTL)?;
         let topic = get_owned_header(req, "topic");
+        let urgency = get_owned_header(req, "urgency");
 
         let headers = if has_data {
             NotificationHeaders {
                 ttl,
                 topic,
+                urgency,
                 encoding: get_owned_header(req, "content-encoding"),
                 encryption: get_owned_header(req, "encryption").map(Self::strip_header),
                 encryption_key: get_owned_header(req, "encryption-key"),
@@ -88,6 +104,7 @@ impl NotificationHeaders {
             NotificationHeaders {
                 ttl,
                 topic,
+                urgency,
                 encoding: None,
                 encryption: None,
                 encryption_key: None,
@@ -359,6 +376,7 @@ mod tests {
             NotificationHeaders {
                 ttl: 10,
                 topic: None,
+                urgency: None,
                 encoding: Some("aesgcm".to_string()),
                 encryption: Some("salt=foo".to_string()),
                 encryption_key: None,
@@ -384,6 +402,7 @@ mod tests {
             NotificationHeaders {
                 ttl: 10,
                 topic: None,
+                urgency: None,
                 encoding: Some("aes128gcm".to_string()),
                 encryption: Some("notsalt=foo".to_string()),
                 encryption_key: None,
@@ -410,6 +429,7 @@ mod tests {
             NotificationHeaders {
                 ttl: 10,
                 topic: None,
+                urgency: None,
                 encoding: Some("aesgcm".to_string()),
                 encryption: Some("salt=foo".to_string()),
                 encryption_key: None,