Correctly set shouldPush for messages (#1707)

* generate the protos and the database changes

* store should push

* update places to set the should push

* Update schema_gen.rs

* Update schema_gen.rs

* dont pull in the payer proto changes

* undo the toml change

* fix test utils

* add should push mapping for content types (#1715)

* add should push mapping for content types

* add comment

---------

Co-authored-by: cameronvoell <cameronvoell@users.noreply.github.com>

* fix the borrow

* cargo fmt

* add should_push

* set to false for other content types

* update the defaults

* should push stored on inten

* should push stored on inten (#1717)

* add tests for it

* fix the linte

---------

Co-authored-by: Cameron Voell <1103838+cameronvoell@users.noreply.github.com>
Co-authored-by: cameronvoell <cameronvoell@users.noreply.github.com>
Co-authored-by: Dakota Brink <git@kota.is>
Co-authored-by: Dakota Brink <779390+codabrink@users.noreply.github.com>

Author: 3 people

xmtp_api/src/test_utils.rs

@@ -43,6 +43,7 @@ pub fn build_group_messages(num_messages: usize, group_id: Vec<u8>) -> Vec<Group
                 group_id: group_id.clone(),
                 data: vec![i as u8],
                 sender_hmac: vec![],
+                should_push: Some(true),
             })),
         })
     }

xmtp_api_grpc/src/replication_client.rs

@@ -335,6 +335,7 @@ impl XmtpMlsClient for ClientV4 {
                             group_id: req.group_id.clone(),
                             data: v1_group_message.data,
                             sender_hmac: v1_group_message.sender_hmac,
+                            should_push: v1_group_message.should_push,
                         })),
                     })
                 })

xmtp_content_types/src/lib.rs

@@ -21,6 +21,66 @@ pub enum CodecError {
     Decode(String),
 }
 
+pub enum ContentType {
+    Text,
+    GroupMembershipChange,
+    GroupUpdated,
+    Reaction,
+    ReadReceipt,
+    Reply,
+    Attachment,
+    RemoteAttachment,
+    MultiRemoteAttachment,
+    TransactionReference,
+}
+
+impl TryFrom<&str> for ContentType {
+    type Error = String;
+
+    fn try_from(type_id: &str) -> Result<Self, Self::Error> {
+        match type_id {
+            text::TextCodec::TYPE_ID => Ok(Self::Text),
+            membership_change::GroupMembershipChangeCodec::TYPE_ID => {
+                Ok(Self::GroupMembershipChange)
+            }
+            group_updated::GroupUpdatedCodec::TYPE_ID => Ok(Self::GroupUpdated),
+            reaction::ReactionCodec::TYPE_ID => Ok(Self::Reaction),
+            read_receipt::ReadReceiptCodec::TYPE_ID => Ok(Self::ReadReceipt),
+            reply::ReplyCodec::TYPE_ID => Ok(Self::Reply),
+            attachment::AttachmentCodec::TYPE_ID => Ok(Self::Attachment),
+            remote_attachment::RemoteAttachmentCodec::TYPE_ID => Ok(Self::RemoteAttachment),
+            multi_remote_attachment::MultiRemoteAttachmentCodec::TYPE_ID => {
+                Ok(Self::MultiRemoteAttachment)
+            }
+            transaction_reference::TransactionReferenceCodec::TYPE_ID => {
+                Ok(Self::TransactionReference)
+            }
+            _ => Err(format!("Unknown content type ID: {}", type_id)),
+        }
+    }
+}
+
+// Represents whether this message type should send pushed notification when received by a user
+pub fn should_push(content_type_id: String) -> bool {
+    let content_type = ContentType::try_from(content_type_id.as_str()).ok();
+    if let Some(content_type) = content_type {
+        match content_type {
+            ContentType::Text => true,
+            ContentType::GroupMembershipChange => false,
+            ContentType::GroupUpdated => false,
+            ContentType::Reaction => false,
+            ContentType::ReadReceipt => false,
+            ContentType::Reply => true,
+            ContentType::Attachment => true,
+            ContentType::RemoteAttachment => true,
+            ContentType::MultiRemoteAttachment => true,
+            ContentType::TransactionReference => true,
+        }
+    } else {
+        false
+    }
+}
+
 pub trait ContentCodec<T> {
     fn content_type() -> ContentTypeId;
     fn encode(content: T) -> Result<EncodedContent, CodecError>;

xmtp_mls/migrations/2025-03-05-045206_add_should_push_to_messages/down.sql

@@ -0,0 +1,2 @@
+ALTER TABLE group_messages
+    DROP COLUMN should_push;

xmtp_mls/migrations/2025-03-05-045206_add_should_push_to_messages/up.sql

@@ -0,0 +1,2 @@
+ALTER TABLE group_intents
+ADD COLUMN should_push BOOLEAN NOT NULL DEFAULT TRUE;

xmtp_mls/src/groups/device_sync.rs

@@ -438,15 +438,20 @@ where
         let content = DeviceSyncContent::Request(request.clone());
         let content_bytes = serde_json::to_vec(&content)?;
 
-        let _message_id = sync_group.prepare_message(&content_bytes, provider, {
-            let request = request.clone();
-            move |now| PlaintextEnvelope {
-                content: Some(Content::V2(V2 {
-                    message_type: Some(MessageType::DeviceSyncRequest(request)),
-                    idempotency_key: now.to_string(),
-                })),
-            }
-        })?;
+        let _message_id = sync_group.prepare_message(
+            &content_bytes,
+            provider,
+            {
+                let request = request.clone();
+                move |now| PlaintextEnvelope {
+                    content: Some(Content::V2(V2 {
+                        message_type: Some(MessageType::DeviceSyncRequest(request)),
+                        idempotency_key: now.to_string(),
+                    })),
+                }
+            },
+            false,
+        )?;
 
         // publish the intent
         sync_group.publish_intents(provider).await?;
@@ -507,12 +512,17 @@ where
             (content_bytes, contents)
         };
 
-        sync_group.prepare_message(&content_bytes, provider, |now| PlaintextEnvelope {
-            content: Some(Content::V2(V2 {
-                message_type: Some(MessageType::DeviceSyncReply(contents)),
-                idempotency_key: now.to_string(),
-            })),
-        })?;
+        sync_group.prepare_message(
+            &content_bytes,
+            provider,
+            |now| PlaintextEnvelope {
+                content: Some(Content::V2(V2 {
+                    message_type: Some(MessageType::DeviceSyncReply(contents)),
+                    idempotency_key: now.to_string(),
+                })),
+            },
+            false,
+        )?;
 
         sync_group.publish_intents(provider).await?;
 

xmtp_mls/src/groups/device_sync/preference_sync.rs

@@ -31,12 +31,17 @@ impl UserPreferenceUpdate {
             .collect::<Result<Vec<_>, _>>()?;
         let update_proto = UserPreferenceUpdateProto { contents: updates };
         let content_bytes = serde_json::to_vec(&update_proto)?;
-        sync_group.prepare_message(&content_bytes, &provider, |now| PlaintextEnvelope {
-            content: Some(Content::V2(V2 {
-                message_type: Some(MessageType::UserPreferenceUpdate(update_proto)),
-                idempotency_key: now.to_string(),
-            })),
-        })?;
+        sync_group.prepare_message(
+            &content_bytes,
+            &provider,
+            |now| PlaintextEnvelope {
+                content: Some(Content::V2(V2 {
+                    message_type: Some(MessageType::UserPreferenceUpdate(update_proto)),
+                    idempotency_key: now.to_string(),
+                })),
+            },
+            false,
+        )?;
 
         sync_group.publish_intents(&provider).await?;
 

xmtp_mls/src/groups/intents.rs

@@ -61,10 +61,11 @@ impl<ScopedClient: ScopedGroupClient> MlsGroup<ScopedClient> {
         provider: &XmtpOpenMlsProvider,
         intent_kind: IntentKind,
         intent_data: Vec<u8>,
+        should_push: bool,
     ) -> Result<StoredGroupIntent, GroupError> {
         let res = provider.transaction(|provider| {
             let conn = provider.conn_ref();
-            self.queue_intent_with_conn(conn, intent_kind, intent_data)
+            self.queue_intent_with_conn(conn, intent_kind, intent_data, should_push)
         });
 
         res
@@ -75,6 +76,7 @@ impl<ScopedClient: ScopedGroupClient> MlsGroup<ScopedClient> {
         conn: &DbConnection,
         intent_kind: IntentKind,
         intent_data: Vec<u8>,
+        should_push: bool,
     ) -> Result<StoredGroupIntent, GroupError> {
         if intent_kind == IntentKind::SendMessage {
             self.maybe_insert_key_update_intent(conn)?;
@@ -84,6 +86,7 @@ impl<ScopedClient: ScopedGroupClient> MlsGroup<ScopedClient> {
             intent_kind,
             self.group_id.clone(),
             intent_data,
+            should_push,
         ))?;
 
         if intent_kind != IntentKind::SendMessage {
@@ -99,7 +102,7 @@ impl<ScopedClient: ScopedGroupClient> MlsGroup<ScopedClient> {
         let now_ns = xmtp_common::time::now_ns();
         let elapsed_ns = now_ns - last_rotated_at_ns;
         if elapsed_ns > GROUP_KEY_ROTATION_INTERVAL_NS {
-            self.queue_intent_with_conn(conn, IntentKind::KeyUpdate, vec![])?;
+            self.queue_intent_with_conn(conn, IntentKind::KeyUpdate, vec![], false)?;
         }
         Ok(())
     }

xmtp_mls/src/groups/mls_sync.rs

@@ -175,6 +175,7 @@ struct PublishIntentData {
     staged_commit: Option<Vec<u8>>,
     post_commit_action: Option<Vec<u8>>,
     payload_to_publish: Vec<u8>,
+    should_send_push_notification: bool,
 }
 
 impl<ScopedClient> MlsGroup<ScopedClient>
@@ -1234,6 +1235,7 @@ where
                                 payload_to_publish,
                                 post_commit_action,
                                 staged_commit,
+                                should_send_push_notification
                             })) => {
                         let payload_slice = payload_to_publish.as_slice();
                         let has_staged_commit = staged_commit.is_some();
@@ -1255,7 +1257,7 @@ where
                             intent.id
                         );
 
-                        let messages = self.prepare_group_messages(vec![payload_slice])?;
+                        let messages = self.prepare_group_messages(vec![(payload_slice, should_send_push_notification)])?;
                         self.client
                             .api()
                             .send_group_messages(messages)
@@ -1330,6 +1332,7 @@ where
                     payload_to_publish: msg.tls_serialize_detached()?,
                     post_commit_action: None,
                     staged_commit: None,
+                    should_send_push_notification: intent.should_push,
                 }))
             }
             IntentKind::KeyUpdate => {
@@ -1343,6 +1346,7 @@ where
                     payload_to_publish: commit.tls_serialize_detached()?,
                     staged_commit: get_and_clear_pending_commit(openmls_group, provider)?,
                     post_commit_action: None,
+                    should_send_push_notification: intent.should_push,
                 }))
             }
             IntentKind::MetadataUpdate => {
@@ -1365,6 +1369,7 @@ where
                     payload_to_publish: commit_bytes,
                     staged_commit: get_and_clear_pending_commit(openmls_group, provider)?,
                     post_commit_action: None,
+                    should_send_push_notification: intent.should_push,
                 }))
             }
             IntentKind::UpdateAdminList => {
@@ -1386,6 +1391,7 @@ where
                     payload_to_publish: commit_bytes,
                     staged_commit: get_and_clear_pending_commit(openmls_group, provider)?,
                     post_commit_action: None,
+                    should_send_push_notification: intent.should_push,
                 }))
             }
             IntentKind::UpdatePermission => {
@@ -1405,6 +1411,7 @@ where
                     payload_to_publish: commit_bytes,
                     staged_commit: get_and_clear_pending_commit(openmls_group, provider)?,
                     post_commit_action: None,
+                    should_send_push_notification: intent.should_push,
                 }))
             }
         }
@@ -1496,6 +1503,7 @@ where
             provider,
             IntentKind::UpdateGroupMembership,
             intent_data.into(),
+            false,
         )?;
 
         self.sync_until_intent_resolved(provider, intent.id).await
@@ -1689,7 +1697,7 @@ where
     #[tracing::instrument(level = "trace", skip_all)]
     pub(super) fn prepare_group_messages(
         &self,
-        payloads: Vec<&[u8]>,
+        payloads: Vec<(&[u8], bool)>,
     ) -> Result<Vec<GroupMessageInput>, GroupError> {
         let hmac_key = self
             .hmac_keys(0..=0)?
@@ -1699,7 +1707,7 @@ where
             Hmac::<Sha256>::new_from_slice(&hmac_key.key).expect("HMAC can take key of any size");
 
         let mut result = vec![];
-        for payload in payloads {
+        for (payload, should_push) in payloads {
             let mut sender_hmac = sender_hmac.clone();
             sender_hmac.update(payload);
             let sender_hmac = sender_hmac.finalize();
@@ -1708,6 +1716,7 @@ where
                 version: Some(GroupMessageInputVersion::V1(GroupMessageInputV1 {
                     data: payload.to_vec(),
                     sender_hmac: sender_hmac.into_bytes().to_vec(),
+                    should_push: Some(should_push),
                 })),
             });
         }
@@ -1907,6 +1916,7 @@ async fn apply_update_group_membership_intent(
         payload_to_publish: commit.tls_serialize_detached()?,
         post_commit_action: post_commit_action.map(|action| action.to_bytes()),
         staged_commit: Some(staged_commit),
+        should_send_push_notification: false,
     }))
 }