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 <[email protected]>

* 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 <[email protected]>
Co-authored-by: cameronvoell <[email protected]>
Co-authored-by: Dakota Brink <[email protected]>
Co-authored-by: Dakota Brink <[email protected]>

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,
     }))
 }
 

xmtp_mls/src/groups/mod.rs

@@ -97,6 +97,7 @@ use tokio::sync::Mutex;
 use xmtp_common::retry::RetryableError;
 use xmtp_common::time::now_ns;
 use xmtp_content_types::reaction::{LegacyReaction, ReactionCodec};
+use xmtp_content_types::should_push;
 use xmtp_cryptography::signature::{sanitize_evm_addresses, AddressValidationError};
 use xmtp_id::{InboxId, InboxIdRef};
 use xmtp_proto::xmtp::mls::{
@@ -337,6 +338,7 @@ pub struct QueryableContentFields {
     pub version_minor: i32,
     pub authority_id: String,
     pub reference_id: Option<Vec<u8>>,
+    pub should_push: bool,
 }
 
 impl Default for QueryableContentFields {
@@ -347,6 +349,7 @@ impl Default for QueryableContentFields {
             version_minor: 0,
             authority_id: String::new(),
             reference_id: None,
+            should_push: false,
         }
     }
 }
@@ -356,10 +359,10 @@ impl TryFrom<EncodedContent> for QueryableContentFields {
 
     fn try_from(content: EncodedContent) -> Result<Self, Self::Error> {
         let content_type_id = content.r#type.unwrap_or_default();
-        let reference_id = match (
-            content_type_id.type_id.as_str(),
-            content_type_id.version_major,
-        ) {
+
+        let type_id_str = content_type_id.type_id.clone();
+
+        let reference_id = match (type_id_str.as_str(), content_type_id.version_major) {
             (ReactionCodec::TYPE_ID, major) if major >= 2 => {
                 ReactionV2::decode(content.content.as_slice())
                     .ok()
@@ -376,6 +379,7 @@ impl TryFrom<EncodedContent> for QueryableContentFields {
             version_minor: content_type_id.version_minor as i32,
             authority_id: content_type_id.authority_id.to_string(),
             reference_id,
+            should_push: should_push(type_id_str),
         })
     }
 }
@@ -791,8 +795,12 @@ impl<ScopedClient: ScopedGroupClient> MlsGroup<ScopedClient> {
         self.maybe_update_installations(provider, update_interval_ns)
             .await?;
 
-        let message_id =
-            self.prepare_message(message, provider, |now| Self::into_envelope(message, now))?;
+        let message_id = self.prepare_message(
+            message,
+            provider,
+            |now| Self::into_envelope(message, now),
+            true,
+        )?;
 
         self.sync_until_last_intent_resolved(provider).await?;
         // implicitly set group consent state to allowed
@@ -830,8 +838,12 @@ impl<ScopedClient: ScopedGroupClient> MlsGroup<ScopedClient> {
     /// Send a message, optimistically returning the ID of the message before the result of a message publish.
     pub fn send_message_optimistic(&self, message: &[u8]) -> Result<Vec<u8>, GroupError> {
         let provider = self.mls_provider()?;
-        let message_id =
-            self.prepare_message(message, &provider, |now| Self::into_envelope(message, now))?;
+        let message_id = self.prepare_message(
+            message,
+            &provider,
+            |now| Self::into_envelope(message, now),
+            true,
+        )?;
         Ok(message_id)
     }
 
@@ -865,6 +877,7 @@ impl<ScopedClient: ScopedGroupClient> MlsGroup<ScopedClient> {
         message: &[u8],
         provider: &XmtpOpenMlsProvider,
         envelope: F,
+        should_push: bool,
     ) -> Result<Vec<u8>, GroupError>
     where
         F: FnOnce(i64) -> PlaintextEnvelope,
@@ -877,7 +890,7 @@ impl<ScopedClient: ScopedGroupClient> MlsGroup<ScopedClient> {
             .map_err(GroupError::EncodeError)?;
 
         let intent_data: Vec<u8> = SendMessageIntentData::new(encoded_envelope).into();
-        self.queue_intent(provider, IntentKind::SendMessage, intent_data)?;
+        self.queue_intent(provider, IntentKind::SendMessage, intent_data, should_push)?;
 
         // store this unpublished message locally before sending
         let message_id = calculate_message_id(&self.group_id, message, &now.to_string());
@@ -1013,6 +1026,7 @@ impl<ScopedClient: ScopedGroupClient> MlsGroup<ScopedClient> {
             provider,
             IntentKind::UpdateGroupMembership,
             intent_data.into(),
+            false,
         )?;
 
         self.sync_until_intent_resolved(provider, intent.id).await?;
@@ -1063,6 +1077,7 @@ impl<ScopedClient: ScopedGroupClient> MlsGroup<ScopedClient> {
             &provider,
             IntentKind::UpdateGroupMembership,
             intent_data.into(),
+            false,
         )?;
 
         self.sync_until_intent_resolved(&provider, intent.id).await
@@ -1082,7 +1097,8 @@ impl<ScopedClient: ScopedGroupClient> MlsGroup<ScopedClient> {
         }
         let intent_data: Vec<u8> =
             UpdateMetadataIntentData::new_update_group_name(group_name).into();
-        let intent = self.queue_intent(&provider, IntentKind::MetadataUpdate, intent_data)?;
+        let intent =
+            self.queue_intent(&provider, IntentKind::MetadataUpdate, intent_data, false)?;
 
         self.sync_until_intent_resolved(&provider, intent.id).await
     }
@@ -1111,7 +1127,8 @@ impl<ScopedClient: ScopedGroupClient> MlsGroup<ScopedClient> {
         )
         .into();
 
-        let intent = self.queue_intent(&provider, IntentKind::UpdatePermission, intent_data)?;
+        let intent =
+            self.queue_intent(&provider, IntentKind::UpdatePermission, intent_data, false)?;
 
         self.sync_until_intent_resolved(&provider, intent.id).await
     }
@@ -1147,7 +1164,8 @@ impl<ScopedClient: ScopedGroupClient> MlsGroup<ScopedClient> {
         }
         let intent_data: Vec<u8> =
             UpdateMetadataIntentData::new_update_group_description(group_description).into();
-        let intent = self.queue_intent(&provider, IntentKind::MetadataUpdate, intent_data)?;
+        let intent =
+            self.queue_intent(&provider, IntentKind::MetadataUpdate, intent_data, false)?;
 
         self.sync_until_intent_resolved(&provider, intent.id).await
     }
@@ -1183,7 +1201,8 @@ impl<ScopedClient: ScopedGroupClient> MlsGroup<ScopedClient> {
         let intent_data: Vec<u8> =
             UpdateMetadataIntentData::new_update_group_image_url_square(group_image_url_square)
                 .into();
-        let intent = self.queue_intent(&provider, IntentKind::MetadataUpdate, intent_data)?;
+        let intent =
+            self.queue_intent(&provider, IntentKind::MetadataUpdate, intent_data, false)?;
 
         self.sync_until_intent_resolved(&provider, intent.id).await
     }
@@ -1236,7 +1255,7 @@ impl<ScopedClient: ScopedGroupClient> MlsGroup<ScopedClient> {
                 expire_from_ms,
             )
             .into();
-        let intent = self.queue_intent(provider, IntentKind::MetadataUpdate, intent_data)?;
+        let intent = self.queue_intent(provider, IntentKind::MetadataUpdate, intent_data, false)?;
         self.sync_until_intent_resolved(provider, intent.id).await
     }
 
@@ -1248,7 +1267,7 @@ impl<ScopedClient: ScopedGroupClient> MlsGroup<ScopedClient> {
         let intent_data: Vec<u8> =
             UpdateMetadataIntentData::new_update_conversation_message_disappear_in_ns(expire_in_ms)
                 .into();
-        let intent = self.queue_intent(provider, IntentKind::MetadataUpdate, intent_data)?;
+        let intent = self.queue_intent(provider, IntentKind::MetadataUpdate, intent_data, false)?;
         self.sync_until_intent_resolved(provider, intent.id).await
     }
 
@@ -1348,7 +1367,8 @@ impl<ScopedClient: ScopedGroupClient> MlsGroup<ScopedClient> {
         };
         let intent_data: Vec<u8> =
             UpdateAdminListIntentData::new(intent_action_type, inbox_id).into();
-        let intent = self.queue_intent(&provider, IntentKind::UpdateAdminList, intent_data)?;
+        let intent =
+            self.queue_intent(&provider, IntentKind::UpdateAdminList, intent_data, false)?;
 
         self.sync_until_intent_resolved(&provider, intent.id).await
     }
@@ -1425,7 +1445,7 @@ impl<ScopedClient: ScopedGroupClient> MlsGroup<ScopedClient> {
     /// Update this installation's leaf key in the group by creating a key update commit
     pub async fn key_update(&self) -> Result<(), GroupError> {
         let provider = self.client.mls_provider()?;
-        let intent = self.queue_intent(&provider, IntentKind::KeyUpdate, vec![])?;
+        let intent = self.queue_intent(&provider, IntentKind::KeyUpdate, vec![], false)?;
         self.sync_until_intent_resolved(&provider, intent.id).await
     }
 
@@ -2034,7 +2054,7 @@ pub(crate) mod tests {
             serialized_welcome,
         );
         let messages = sender_group
-            .prepare_group_messages(vec![serialized_commit.as_slice()])
+            .prepare_group_messages(vec![(serialized_commit.as_slice(), false)])
             .unwrap();
         sender_client
             .api_client
@@ -4405,6 +4425,7 @@ pub(crate) mod tests {
                 provider,
                 IntentKind::UpdateGroupMembership,
                 intent_data.into(),
+                false,
             )
             .unwrap();
     }

xmtp_mls/src/storage/encrypted_store/group_intent.rs

@@ -78,6 +78,7 @@ pub struct StoredGroupIntent {
     pub publish_attempts: i32,
     pub staged_commit: Option<Vec<u8>>,
     pub published_in_epoch: Option<i64>,
+    pub should_push: bool,
 }
 
 impl std::fmt::Debug for StoredGroupIntent {
@@ -176,17 +177,19 @@ pub struct NewGroupIntent {
     pub group_id: Vec<u8>,
     pub data: Vec<u8>,
     pub state: IntentState,
+    pub should_push: bool,
 }
 
 impl_store!(NewGroupIntent, group_intents);
 
 impl NewGroupIntent {
-    pub fn new(kind: IntentKind, group_id: Vec<u8>, data: Vec<u8>) -> Self {
+    pub fn new(kind: IntentKind, group_id: Vec<u8>, data: Vec<u8>, should_push: bool) -> Self {
         Self {
             kind,
             group_id,
             data,
             state: IntentState::ToPublish,
+            should_push,
         }
     }
 }
@@ -469,6 +472,7 @@ pub(crate) mod tests {
                 group_id,
                 data,
                 state,
+                should_push: false,
             }
         }
     }
@@ -609,6 +613,7 @@ pub(crate) mod tests {
                 IntentKind::UpdateGroupMembership,
                 group_id.clone(),
                 rand_vec::<24>(),
+                false,
             )
             .store(conn)
             .unwrap();
@@ -652,6 +657,7 @@ pub(crate) mod tests {
                 IntentKind::UpdateGroupMembership,
                 group_id.clone(),
                 rand_vec::<24>(),
+                false,
             )
             .store(conn)
             .unwrap();
@@ -698,6 +704,7 @@ pub(crate) mod tests {
                 IntentKind::UpdateGroupMembership,
                 group_id.clone(),
                 rand_vec::<24>(),
+                false,
             )
             .store(conn)
             .unwrap();
@@ -743,6 +750,7 @@ pub(crate) mod tests {
                 IntentKind::UpdateGroupMembership,
                 group_id.clone(),
                 rand_vec::<24>(),
+                false,
             )
             .store(conn)
             .unwrap();
@@ -776,6 +784,7 @@ pub(crate) mod tests {
                 IntentKind::UpdateGroupMembership,
                 group_id.clone(),
                 rand_vec::<24>(),
+                false,
             )
             .store(conn)
             .unwrap();

xmtp_mls/src/storage/encrypted_store/group_message.rs

@@ -488,6 +488,7 @@ pub(crate) mod tests {
     };
     use wasm_bindgen_test::wasm_bindgen_test;
     use xmtp_common::{assert_err, assert_ok, rand_time, rand_vec};
+    use xmtp_content_types::should_push;
 
     pub(crate) fn generate_message(
         kind: Option<GroupMessageKind>,
@@ -818,6 +819,8 @@ pub(crate) mod tests {
                 .unwrap();
             assert_eq!(text_messages.len(), 1);
             assert_eq!(text_messages[0].content_type, ContentType::Text);
+            assert!(should_push(text_messages[0].content_type.to_string()));
+
             assert_eq!(text_messages[0].sent_at_ns, 1_000);
 
             // Query for membership change messages
@@ -849,6 +852,7 @@ pub(crate) mod tests {
                 .unwrap();
             assert_eq!(updated_messages.len(), 1);
             assert_eq!(updated_messages[0].content_type, ContentType::GroupUpdated);
+            assert!(!should_push(updated_messages[0].content_type.to_string()));
             assert_eq!(updated_messages[0].sent_at_ns, 3_000);
         })
         .await

xmtp_mls/src/storage/encrypted_store/schema_gen.rs

@@ -1,6 +1,6 @@
 // @generated automatically by Diesel CLI.
 
-use crate::storage::schema::conversation_list;
+use super::schema::conversation_list;
 
 diesel::table! {
     association_state (inbox_id, sequence_id) {
@@ -30,6 +30,7 @@ diesel::table! {
         publish_attempts -> Integer,
         staged_commit -> Nullable<Binary>,
         published_in_epoch -> Nullable<BigInt>,
+        should_push -> Bool,
     }
 }
 

xmtp_proto/src/gen/xmtp.mls.api.v1.rs

@@ -1044,6 +1044,9 @@ impl serde::Serialize for group_message::V1 {
         if !self.sender_hmac.is_empty() {
             len += 1;
         }
+        if self.should_push.is_some() {
+            len += 1;
+        }
         let mut struct_ser = serializer.serialize_struct("xmtp.mls.api.v1.GroupMessage.V1", len)?;
         if self.id != 0 {
             #[allow(clippy::needless_borrow)]
@@ -1070,6 +1073,9 @@ impl serde::Serialize for group_message::V1 {
             #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("senderHmac", pbjson::private::base64::encode(&self.sender_hmac).as_str())?;
         }
+        if let Some(v) = self.should_push.as_ref() {
+            struct_ser.serialize_field("shouldPush", v)?;
+        }
         struct_ser.end()
     }
 }
@@ -1088,6 +1094,8 @@ impl<'de> serde::Deserialize<'de> for group_message::V1 {
             "data",
             "sender_hmac",
             "senderHmac",
+            "should_push",
+            "shouldPush",
         ];
 
         #[allow(clippy::enum_variant_names)]
@@ -1097,6 +1105,7 @@ impl<'de> serde::Deserialize<'de> for group_message::V1 {
             GroupId,
             Data,
             SenderHmac,
+            ShouldPush,
         }
         impl<'de> serde::Deserialize<'de> for GeneratedField {
             fn deserialize<D>(deserializer: D) -> std::result::Result<GeneratedField, D::Error>
@@ -1123,6 +1132,7 @@ impl<'de> serde::Deserialize<'de> for group_message::V1 {
                             "groupId" | "group_id" => Ok(GeneratedField::GroupId),
                             "data" => Ok(GeneratedField::Data),
                             "senderHmac" | "sender_hmac" => Ok(GeneratedField::SenderHmac),
+                            "shouldPush" | "should_push" => Ok(GeneratedField::ShouldPush),
                             _ => Err(serde::de::Error::unknown_field(value, FIELDS)),
                         }
                     }
@@ -1147,6 +1157,7 @@ impl<'de> serde::Deserialize<'de> for group_message::V1 {
                 let mut group_id__ = None;
                 let mut data__ = None;
                 let mut sender_hmac__ = None;
+                let mut should_push__ = None;
                 while let Some(k) = map_.next_key()? {
                     match k {
                         GeneratedField::Id => {
@@ -1189,6 +1200,12 @@ impl<'de> serde::Deserialize<'de> for group_message::V1 {
                                 Some(map_.next_value::<::pbjson::private::BytesDeserialize<_>>()?.0)
                             ;
                         }
+                        GeneratedField::ShouldPush => {
+                            if should_push__.is_some() {
+                                return Err(serde::de::Error::duplicate_field("shouldPush"));
+                            }
+                            should_push__ = map_.next_value()?;
+                        }
                     }
                 }
                 Ok(group_message::V1 {
@@ -1197,6 +1214,7 @@ impl<'de> serde::Deserialize<'de> for group_message::V1 {
                     group_id: group_id__.unwrap_or_default(),
                     data: data__.unwrap_or_default(),
                     sender_hmac: sender_hmac__.unwrap_or_default(),
+                    should_push: should_push__,
                 })
             }
         }
@@ -1313,6 +1331,9 @@ impl serde::Serialize for group_message_input::V1 {
         if !self.sender_hmac.is_empty() {
             len += 1;
         }
+        if self.should_push.is_some() {
+            len += 1;
+        }
         let mut struct_ser = serializer.serialize_struct("xmtp.mls.api.v1.GroupMessageInput.V1", len)?;
         if !self.data.is_empty() {
             #[allow(clippy::needless_borrow)]
@@ -1324,6 +1345,9 @@ impl serde::Serialize for group_message_input::V1 {
             #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("senderHmac", pbjson::private::base64::encode(&self.sender_hmac).as_str())?;
         }
+        if let Some(v) = self.should_push.as_ref() {
+            struct_ser.serialize_field("shouldPush", v)?;
+        }
         struct_ser.end()
     }
 }
@@ -1337,12 +1361,15 @@ impl<'de> serde::Deserialize<'de> for group_message_input::V1 {
             "data",
             "sender_hmac",
             "senderHmac",
+            "should_push",
+            "shouldPush",
         ];
 
         #[allow(clippy::enum_variant_names)]
         enum GeneratedField {
             Data,
             SenderHmac,
+            ShouldPush,
         }
         impl<'de> serde::Deserialize<'de> for GeneratedField {
             fn deserialize<D>(deserializer: D) -> std::result::Result<GeneratedField, D::Error>
@@ -1366,6 +1393,7 @@ impl<'de> serde::Deserialize<'de> for group_message_input::V1 {
                         match value {
                             "data" => Ok(GeneratedField::Data),
                             "senderHmac" | "sender_hmac" => Ok(GeneratedField::SenderHmac),
+                            "shouldPush" | "should_push" => Ok(GeneratedField::ShouldPush),
                             _ => Err(serde::de::Error::unknown_field(value, FIELDS)),
                         }
                     }
@@ -1387,6 +1415,7 @@ impl<'de> serde::Deserialize<'de> for group_message_input::V1 {
             {
                 let mut data__ = None;
                 let mut sender_hmac__ = None;
+                let mut should_push__ = None;
                 while let Some(k) = map_.next_key()? {
                     match k {
                         GeneratedField::Data => {
@@ -1405,11 +1434,18 @@ impl<'de> serde::Deserialize<'de> for group_message_input::V1 {
                                 Some(map_.next_value::<::pbjson::private::BytesDeserialize<_>>()?.0)
                             ;
                         }
+                        GeneratedField::ShouldPush => {
+                            if should_push__.is_some() {
+                                return Err(serde::de::Error::duplicate_field("shouldPush"));
+                            }
+                            should_push__ = map_.next_value()?;
+                        }
                     }
                 }
                 Ok(group_message_input::V1 {
                     data: data__.unwrap_or_default(),
                     sender_hmac: sender_hmac__.unwrap_or_default(),
+                    should_push: should_push__,
                 })
             }
         }