code reviews

Author: gruberb

Cargo.lock

@@ -107,3 +107,6 @@ autoconnect_web = { path = "./autoconnect/autoconnect-web" }
 autoconnect_ws = { path = "./autoconnect/autoconnect-ws" }
 autoconnect_ws_clientsm = { path = "./autoconnect/autoconnect-ws/autoconnect-ws-clientsm" }
 autopush_common = { path = "./autopush-common", features = ["bigtable"] }
+
+[profile.release]
+debug = 1

Cargo.toml

@@ -8,6 +8,7 @@ version.workspace = true
 
 [dependencies]
 actix-web.workspace = true
+bytestring.workspace = true
 cadence.workspace = true
 futures.workspace = true
 futures-locks.workspace = true

autoconnect/autoconnect-common/Cargo.toml

@@ -25,7 +25,7 @@ use autopush_common::notification::Notification;
 ///  use autoconnect_common::protocol::MessageType;
 ///
 /// let message_type = MessageType::Hello;
-/// let message_str = message_type.as_str();  // Returns "hello"
+/// let message_str = message_type.as_ref();  // Returns "hello"
 /// ```
 #[derive(Debug, Clone, Copy, PartialEq, Eq, AsRefStr, Display, EnumString)]
 #[strum(serialize_all = "snake_case")]
@@ -42,14 +42,9 @@ pub enum MessageType {
 }
 
 impl MessageType {
-    /// Converts the enum to its string representation
-    pub fn as_str(&self) -> &str {
-        self.as_ref()
-    }
-
     /// Returns the expected message type string for error messages
     pub fn expected_msg(&self) -> String {
-        format!(r#"Expected messageType="{}""#, self.as_str())
+        format!(r#"Expected messageType="{}""#, self.as_ref())
     }
 }
 
@@ -209,7 +204,8 @@ impl ServerMessage {
 
     pub fn to_json(&self) -> Result<String, serde_json::error::Error> {
         match self {
-            // Traditionally both client/server send the empty object version for ping
+            // Both client and server understand the verbose `{"messageType": "ping"}` and the abbreviated `{}`
+            // as valid ping messages. The server defaults to the shorter `{}` form.
             ServerMessage::Ping => Ok("{}".to_string()),
             _ => serde_json::to_string(self),
         }

autoconnect/autoconnect-common/src/protocol.rs

@@ -1,3 +1,4 @@
+use bytestring::ByteString;
 use uuid::Uuid;
 
 use crate::protocol::MessageType;
@@ -14,19 +15,22 @@ pub const DUMMY_CHID: Uuid = Uuid::from_u128(0xdeadbeef_0000_0000_abad_1dea00000
 
 /// A minimal websocket Push "hello" message, used by an unregistered UA with
 /// no existing channel subscriptions
-pub fn hello_json() -> String {
+pub fn hello_json() -> ByteString {
     format!(
         r#"{{"messageType": "{}", "use_webpush": true}}"#,
-        MessageType::Hello.as_str()
+        MessageType::Hello.as_ref()
     )
+    .into()
 }
 
-pub fn hello_again_json() -> String {
+pub fn hello_again_json() -> ByteString {
     format!(
         r#"{{"messageType": "{}", "use_webpush": true,
-                "uaid": "deadbeef-0000-0000-deca-fbad00000000"}}"#,
-        MessageType::Hello.as_str()
+                "uaid": "{}"}}"#,
+        MessageType::Hello.as_ref(),
+        DUMMY_UAID
     )
+    .into()
 }
 
 pub const CURRENT_MONTH: &str = "message_2018_06";

autoconnect/autoconnect-common/src/test_support.rs

@@ -44,13 +44,10 @@ pub async fn hello_new_user() {
     });
 
     let mut framed = srv.ws().await.unwrap();
-    framed
-        .send(ws::Message::Text(hello_json().into()))
-        .await
-        .unwrap();
+    framed.send(ws::Message::Text(hello_json())).await.unwrap();
 
     let msg = json_msg(&mut framed).await;
-    assert_eq!(msg["messageType"], MessageType::Hello.as_str());
+    assert_eq!(msg["messageType"], MessageType::Hello.as_ref());
     assert_eq!(msg["status"], 200);
     // Ensure that the outbound response to the client includes the
     // `use_webpush` flag set to `true`
@@ -69,12 +66,12 @@ pub async fn hello_again() {
 
     let mut framed = srv.ws().await.unwrap();
     framed
-        .send(ws::Message::Text(hello_again_json().into()))
+        .send(ws::Message::Text(hello_again_json()))
         .await
         .unwrap();
 
     let msg = json_msg(&mut framed).await;
-    assert_eq!(msg["messageType"], MessageType::Hello.as_str());
+    assert_eq!(msg["messageType"], MessageType::Hello.as_ref());
     assert_eq!(msg["uaid"], DUMMY_UAID.as_simple().to_string());
 }
 
@@ -84,7 +81,7 @@ pub async fn unsupported_websocket_message() {
 
     let mut framed = srv.ws().await.unwrap();
     framed
-        .send(ws::Message::Binary(hello_json().into()))
+        .send(ws::Message::Binary(hello_json().into_bytes()))
         .await
         .unwrap();
 
@@ -104,18 +101,12 @@ pub async fn invalid_webpush_message() {
     });
 
     let mut framed = srv.ws().await.unwrap();
-    framed
-        .send(ws::Message::Text(hello_json().into()))
-        .await
-        .unwrap();
+    framed.send(ws::Message::Text(hello_json())).await.unwrap();
 
     let msg = json_msg(&mut framed).await;
     assert_eq!(msg["status"], 200);
 
-    framed
-        .send(ws::Message::Text(hello_json().into()))
-        .await
-        .unwrap();
+    framed.send(ws::Message::Text(hello_json())).await.unwrap();
 
     let item = framed.next().await.unwrap().unwrap();
     let ws::Frame::Close(Some(close_reason)) = item else {
@@ -159,12 +150,12 @@ pub async fn direct_notif() {
 
     let mut framed = srv.ws().await.unwrap();
     framed
-        .send(ws::Message::Text(hello_again_json().into()))
+        .send(ws::Message::Text(hello_again_json()))
         .await
         .unwrap();
 
     let msg = json_msg(&mut framed).await;
-    assert_eq!(msg["messageType"], MessageType::Hello.as_str());
+    assert_eq!(msg["messageType"], MessageType::Hello.as_ref());
 
     app_state
         .clients
@@ -180,7 +171,7 @@ pub async fn direct_notif() {
 
     // Is a small sleep/tick needed here?
     let msg = json_msg(&mut framed).await;
-    assert_eq!(msg["messageType"], MessageType::Notification.as_str());
+    assert_eq!(msg["messageType"], MessageType::Notification.as_ref());
     assert_eq!(msg["data"], "foo");
 }
 
@@ -202,7 +193,7 @@ pub async fn broadcast_after_ping() {
         .add_broadcast(("foo/bar".to_owned(), "v1".to_owned()).into());
     let mut srv = test_server(app_state.clone());
 
-    let hello = json!({"messageType": MessageType::Hello.as_str(), "use_webpush": true,
+    let hello = json!({"messageType": MessageType::Hello.as_ref(), "use_webpush": true,
                        "broadcasts": {"foo/bar": "v1"}});
     let mut framed = srv.ws().await.unwrap();
     framed
@@ -211,7 +202,7 @@ pub async fn broadcast_after_ping() {
         .unwrap();
 
     let msg = json_msg(&mut framed).await;
-    assert_eq!(msg["messageType"], MessageType::Hello.as_str());
+    assert_eq!(msg["messageType"], MessageType::Hello.as_ref());
     let broadcasts = msg["broadcasts"]
         .as_object()
         .expect("!broadcasts.is_object()");
@@ -235,7 +226,7 @@ pub async fn broadcast_after_ping() {
     tokio::time::sleep(Duration::from_secs_f32(0.2)).await;
     let msg = json_msg(&mut framed).await;
     assert_eq!(msg.as_object().map_or(0, |o| o.len()), 2);
-    assert_eq!(msg["messageType"], MessageType::Broadcast.as_str());
+    assert_eq!(msg["messageType"], MessageType::Broadcast.as_ref());
     let broadcasts = msg["broadcasts"]
         .as_object()
         .expect("!broadcasts.is_object()");

autoconnect/autoconnect-web/src/test.rs

@@ -23,8 +23,7 @@ impl WebPushClient {
     ) -> Result<Vec<ServerMessage>, SMError> {
         match msg {
             ClientMessage::Hello { .. } => {
-                // Hello messages should be handled by UnidentifiedClient, not here
-                Err(SMError::expected_message_type(MessageType::Hello))
+                Err(SMError::invalid_message("Already Hello'd".to_owned()))
             }
             ClientMessage::Register { channel_id, key } => {
                 Ok(vec![self.register(channel_id, key).await?])
@@ -55,7 +54,7 @@ impl WebPushClient {
                "uaid" => &self.uaid.to_string(),
                "channel_id" => &channel_id_str,
                "key" => &key,
-               "message_type" => MessageType::Register.as_str(),
+               "message_type" => MessageType::Register.as_ref(),
         );
         let channel_id = Uuid::try_parse(&channel_id_str).map_err(|_| {
             SMError::invalid_message(format!("Invalid channelID: {channel_id_str}"))
@@ -130,7 +129,7 @@ impl WebPushClient {
                "uaid" => &self.uaid.to_string(),
                "channel_id" => &channel_id.to_string(),
                "code" => &code,
-               "message_type" => MessageType::Unregister.as_str(),
+               "message_type" => MessageType::Unregister.as_ref(),
         );
         // TODO: (copied from previous state machine) unregister should check
         // the format of channel_id like register does
@@ -163,7 +162,7 @@ impl WebPushClient {
         &mut self,
         broadcasts: HashMap<String, String>,
     ) -> Result<Option<ServerMessage>, SMError> {
-        trace!("WebPushClient:broadcast_subscribe"; "message_type" => MessageType::BroadcastSubscribe.as_str());
+        trace!("WebPushClient:broadcast_subscribe"; "message_type" => MessageType::BroadcastSubscribe.as_ref());
         let broadcasts = Broadcast::from_hashmap(broadcasts);
         let mut response: HashMap<String, BroadcastValue> = HashMap::new();
 
@@ -186,7 +185,7 @@ impl WebPushClient {
 
     /// Acknowledge receipt of one or more Push Notifications
     async fn ack(&mut self, updates: &[ClientAck]) -> Result<Vec<ServerMessage>, SMError> {
-        trace!("✅ WebPushClient:ack"; "message_type" => MessageType::Ack.as_str());
+        trace!("✅ WebPushClient:ack"; "message_type" => MessageType::Ack.as_ref());
         let _ = self
             .app_state
             .metrics
@@ -220,10 +219,10 @@ impl WebPushClient {
                 .position(|n| n.channel_id == notif.channel_id && n.version == notif.version);
             if let Some(pos) = pos {
                 debug!(
-                    "✅ Ack (Direct)";
+                    "✅ Ack (Stored)";
                        "channel_id" => notif.channel_id.as_hyphenated().to_string(),
                        "version" => &notif.version,
-                       "message_type" => MessageType::Ack.as_str()
+                       "message_type" => MessageType::Ack.as_ref()
                 );
                 // Get the stored notification record.
                 let n = &mut self.ack_state.unacked_stored_notifs[pos];
@@ -272,7 +271,7 @@ impl WebPushClient {
     /// Negative Acknowledgement (a Client error occurred) of one or more Push
     /// Notifications
     fn nack(&mut self, code: Option<i32>) {
-        trace!("WebPushClient:nack"; "message_type" => MessageType::Nack.as_str());
+        trace!("WebPushClient:nack"; "message_type" => MessageType::Nack.as_ref());
         // only metric codes expected from the client (or 0)
         let code = code
             .and_then(|code| (301..=303).contains(&code).then_some(code))
@@ -290,7 +289,7 @@ impl WebPushClient {
     /// Note this is the WebPush Protocol level's Ping: this differs from the
     /// lower level WebSocket Ping frame (handled by the `webpush_ws` handler).
     fn ping(&mut self) -> Result<ServerMessage, SMError> {
-        trace!("WebPushClient:ping"; "message_type" => MessageType::Ping.as_str());
+        trace!("WebPushClient:ping"; "message_type" => MessageType::Ping.as_ref());
         // TODO: why is this 45 vs the comment describing a minute? and 45
         // should be a setting
         // Clients shouldn't ping > than once per minute or we disconnect them
@@ -311,18 +310,18 @@ impl WebPushClient {
     /// method) before proceeding to read the next batch (or potential other
     /// actions such as `reset_uaid`).
     async fn post_process_all_acked(&mut self) -> Result<Vec<ServerMessage>, SMError> {
-        trace!("▶️ WebPushClient:post_process_all_acked"; "message_type" => MessageType::Notification.as_str());
+        trace!("▶️ WebPushClient:post_process_all_acked"; "message_type" => MessageType::Notification.as_ref());
         let flags = &self.flags;
         if flags.check_storage {
             if flags.increment_storage {
                 debug!(
                     "▶️ WebPushClient:post_process_all_acked check_storage && increment_storage";
-                    "message_type" => MessageType::Notification.as_str()
+                    "message_type" => MessageType::Notification.as_ref()
                 );
                 self.increment_storage().await?;
             }
 
-            debug!("▶️ WebPushClient:post_process_all_acked check_storage"; "message_type" => MessageType::Notification.as_str());
+            debug!("▶️ WebPushClient:post_process_all_acked check_storage"; "message_type" => MessageType::Notification.as_ref());
             let smsgs = self.check_storage_loop().await?;
             if !smsgs.is_empty() {
                 debug_assert!(self.flags.check_storage);
@@ -340,7 +339,7 @@ impl WebPushClient {
         debug_assert!(!self.ack_state.unacked_notifs());
         let flags = &self.flags;
         if flags.old_record_version {
-            debug!("▶️ WebPushClient:post_process_all_acked; resetting uaid"; "message_type" => MessageType::Notification.as_str());
+            debug!("▶️ WebPushClient:post_process_all_acked; resetting uaid"; "message_type" => MessageType::Notification.as_ref());
             self.app_state
                 .metrics
                 .incr_with_tags(MetricName::UaExpiration)

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

@@ -70,7 +70,7 @@ impl UnidentifiedClient {
         };
         debug!(
             "👋UnidentifiedClient::on_client_msg {} from uaid?: {:?}",
-            MessageType::Hello.as_str(),
+            MessageType::Hello.as_ref(),
             uaid
         );
 
@@ -85,7 +85,7 @@ impl UnidentifiedClient {
         let uaid = user.uaid;
         debug!(
             "💬UnidentifiedClient::on_client_msg {}! uaid: {} existing_user: {}",
-            MessageType::Hello.as_str(),
+            MessageType::Hello.as_ref(),
             uaid,
             existing_user,
         );
@@ -144,7 +144,7 @@ impl UnidentifiedClient {
     async fn get_or_create_user(&self, uaid: Option<Uuid>) -> Result<GetOrCreateUser, SMError> {
         trace!(
             "❓UnidentifiedClient::get_or_create_user for {}",
-            MessageType::Hello.as_str()
+            MessageType::Hello.as_ref()
         );
         let connected_at = ms_since_epoch();
 
@@ -199,7 +199,7 @@ impl UnidentifiedClient {
     ) -> (BroadcastSubs, HashMap<String, BroadcastValue>) {
         trace!(
             "UnidentifiedClient::broadcast_init for {}",
-            MessageType::Hello.as_str()
+            MessageType::Hello.as_ref()
         );
         let bc = self.app_state.broadcaster.read().await;
         let BroadcastSubsInit(broadcast_subs, delta) = bc.broadcast_delta(broadcasts);
@@ -224,13 +224,12 @@ struct GetOrCreateUser {
 
 #[cfg(test)]
 mod tests {
+    use std::str::FromStr;
     use std::sync::Arc;
 
     use autoconnect_common::{
         protocol::{ClientMessage, MessageType},
-        test_support::{
-            hello_again_db, hello_again_json, hello_db, hello_json, DUMMY_CHID, DUMMY_UAID, UA,
-        },
+        test_support::{hello_again_db, hello_again_json, hello_db, DUMMY_CHID, DUMMY_UAID, UA},
     };
     use autoconnect_settings::AppState;
 
@@ -258,7 +257,7 @@ mod tests {
             .unwrap();
         assert!(matches!(err.kind, SMErrorKind::InvalidMessage(_)));
         // Verify error message contains expected message type
-        assert!(format!("{}", err).contains(MessageType::Hello.as_str()));
+        assert!(format!("{}", err).contains(MessageType::Hello.as_ref()));
 
         // Test with Register message
         let client = uclient(Default::default());
@@ -272,7 +271,7 @@ mod tests {
             .unwrap();
         assert!(matches!(err.kind, SMErrorKind::InvalidMessage(_)));
         // Verify error message contains expected message type
-        assert!(format!("{}", err).contains(MessageType::Hello.as_str()));
+        assert!(format!("{}", err).contains(MessageType::Hello.as_ref()));
     }
 
     #[tokio::test]
@@ -294,9 +293,12 @@ mod tests {
             db: hello_db().into_boxed_arc(),
             ..Default::default()
         });
-        // Using hello_json helper ensures consistent message type strings
-        let raw = hello_json();
-        let msg = raw.parse::<ClientMessage>().unwrap();
+        // Ensure that we do not need to pass the "use_webpush" flag.
+        // (yes, this could just be passing the string, but I want to be
+        // very explicit here.)
+        let json = serde_json::json!({"messageType":"hello"});
+        let raw = json.to_string();
+        let msg = ClientMessage::from_str(&raw).unwrap();
         client.on_client_msg(msg).await.expect("Hello failed");
     }