xmtp_mls/groups/
messages.rs

1//! Sending, preparing, and querying messages.
2
3use super::*;
4
5impl<Context> MlsGroup<Context>
6where
7    Context: XmtpSharedContext,
8{
9    /// Send a message on this users XMTP [`Client`](crate::client::Client).
10    #[xmtp_common::mls_span]
11    pub async fn send_message(
12        &self,
13        message: &[u8],
14        opts: send_message_opts::SendMessageOpts,
15    ) -> Result<Vec<u8>, GroupError> {
16        if !self.is_active()? {
17            tracing::warn!("Unable to send a message on an inactive group.");
18            return Err(GroupError::GroupInactive);
19        }
20
21        self.ensure_not_paused().await?;
22        let update_interval_ns = Some(SEND_MESSAGE_UPDATE_INSTALLATIONS_INTERVAL_NS);
23        self.maybe_update_installations(update_interval_ns).await?;
24
25        // Check for pending proposals and commit them first
26        // OpenMLS blocks message creation when there are pending proposals
27        self.commit_pending_proposals_if_any().await?;
28
29        let message_id =
30            self.prepare_message(message, opts, |key| Self::into_envelope(message, key))?;
31
32        self.sync_until_last_intent_resolved().await?;
33
34        // implicitly set group consent state to allowed
35        self.update_consent_state(ConsentState::Allowed)?;
36
37        Ok(message_id)
38    }
39
40    /// Checks for pending MLS proposals and commits them if any exist.
41    /// OpenMLS blocks message creation when there are pending proposals,
42    /// so we need to commit them first.
43    async fn commit_pending_proposals_if_any(&self) -> Result<(), GroupError> {
44        let has_pending = self.with_group_snapshot(|openmls_group| {
45            Ok::<bool, GroupError>(openmls_group.pending_proposals().next().is_some())
46        })?;
47
48        if has_pending {
49            tracing::debug!(
50                inbox_id = self.context.inbox_id(),
51                group_id = %self.group_id,
52                "Found pending proposals, committing before sending message"
53            );
54
55            // Queue a CommitPendingProposals intent and wait for it to resolve
56            let intent = intents::QueueIntent::commit_pending_proposals().queue(self)?;
57            self.sync_until_intent_resolved(intent.id).await?;
58        }
59
60        Ok(())
61    }
62
63    /// Publish all unpublished messages. This happens by calling `sync_until_last_intent_resolved`
64    /// which publishes all pending intents and reads them back from the network.
65    #[cfg_attr(any(test, feature = "test-utils"), tracing::instrument(level = "info", fields(inbox_id = self.context.inbox_id()), skip(self)))]
66    #[cfg_attr(not(any(test, feature = "test-utils")), xmtp_common::mls_span)]
67    pub async fn publish_messages(&self) -> Result<(), GroupError> {
68        self.ensure_not_paused().await?;
69        let update_interval_ns = Some(SEND_MESSAGE_UPDATE_INSTALLATIONS_INTERVAL_NS);
70        self.maybe_update_installations(update_interval_ns).await?;
71        self.sync_until_last_intent_resolved().await?;
72
73        // implicitly set group consent state to allowed
74        self.update_consent_state(ConsentState::Allowed)?;
75
76        Ok(())
77    }
78
79    /// Checks the network to see if any group members have identity updates that would cause installations
80    /// to be added or removed from the group.
81    ///
82    /// If so, adds/removes those group members
83    pub async fn update_installations(&self) -> Result<(), GroupError> {
84        self.ensure_not_paused().await?;
85        self.maybe_update_installations(Some(0)).await?;
86        Ok(())
87    }
88
89    /// Send a message, optimistically returning the ID of the message before the result of a message publish.
90    pub fn send_message_optimistic(
91        &self,
92        message: &[u8],
93        opts: send_message_opts::SendMessageOpts,
94    ) -> Result<Vec<u8>, GroupError> {
95        let message_id =
96            self.prepare_message(message, opts, |key| Self::into_envelope(message, key))?;
97        Ok(message_id)
98    }
99
100    /// Prepare a message for later publishing.
101    ///
102    /// Stores the message locally with `Unpublished` delivery status but does NOT
103    /// create an intent to publish. Use `publish_stored_message` to publish later.
104    ///
105    /// # Arguments
106    /// * `message` - The message content bytes
107    /// * `should_push` - Whether to send a push notification when publishing
108    /// * `idempotency_key` - Optional caller-supplied key the message id is
109    ///   derived from. Defaults to a random key when `None`.
110    ///
111    /// Returns the message ID.
112    pub fn prepare_message_for_later_publish(
113        &self,
114        message: &[u8],
115        should_push: bool,
116        idempotency_key: Option<String>,
117    ) -> Result<Vec<u8>, GroupError> {
118        state_write(self.context.mls_storage(), |tx| {
119            let storage = tx.storage();
120            let message = self.store_message_for_later_publish(
121                &storage.db(),
122                message,
123                should_push,
124                idempotency_key,
125            )?;
126            Ok::<_, GroupError>(Continue(message.id))
127        })
128        .map(TransactionOutcome::into_continued)
129    }
130
131    /// Store an optimistic message using the caller's transaction.
132    fn store_message_for_later_publish(
133        &self,
134        db: &impl DbQuery,
135        message: &[u8],
136        should_push: bool,
137        idempotency_key: Option<String>,
138    ) -> Result<StoredGroupMessage, GroupError> {
139        let now = now_ns();
140        // Resolve the key once. Random defaults do not depend on clock resolution.
141        let idempotency_key = idempotency_key.unwrap_or_else(|| {
142            hex::encode(xmtp_common::rand_vec::<DEFAULT_IDEMPOTENCY_KEY_BYTES>())
143        });
144        let queryable_content_fields = Self::extract_queryable_content_fields(message);
145
146        let message_id = calculate_message_id(self.group_id, message, &idempotency_key);
147
148        // Idempotent: a retry with the same key + content resolves to the same id.
149        // Return the existing message rather than failing on the PK conflict, so
150        // crash-recovery retries are at-least-once-with-dedup instead of an error.
151        if let Some(existing) = db.get_group_message(&message_id)? {
152            return Ok(existing);
153        }
154
155        let group_message = StoredGroupMessage {
156            id: message_id.clone(),
157            group_id: self.group_id,
158            decrypted_message_bytes: message.to_vec(),
159            sent_at_ns: now,
160            kind: GroupMessageKind::Application,
161            sender_installation_id: self.context.installation_id().into(),
162            sender_inbox_id: self.context.inbox_id().to_string(),
163            delivery_status: DeliveryStatus::Unpublished,
164            content_type: queryable_content_fields.content_type,
165            version_major: queryable_content_fields.version_major,
166            version_minor: queryable_content_fields.version_minor,
167            authority_id: queryable_content_fields.authority_id,
168            reference_id: queryable_content_fields.reference_id,
169            sequence_id: 0,
170            envelope_hash: None,
171            expiry_ns: None,
172            expire_at_ns: None,
173            inserted_at_ns: 0,
174            should_push,
175            idempotency_key,
176        };
177        group_message.store(db)?;
178        Ok(group_message)
179    }
180
181    /// Publish a previously stored message by ID.
182    ///
183    /// Creates an intent for the message and publishes it to the network.
184    /// Uses the `should_push` value that was stored with the message.
185    /// This is a no-op if the message is already published.
186    ///
187    /// Returns an error if the message is not found.
188    #[xmtp_common::mls_span]
189    pub async fn publish_stored_message(&self, message_id: &[u8]) -> Result<(), GroupError> {
190        if !self.is_active()? {
191            return Err(GroupError::GroupInactive);
192        }
193        self.ensure_not_paused().await?;
194
195        let queued = state_write(self.context.mls_storage(), |tx| {
196            let storage = tx.storage();
197            let db = storage.db();
198            let message = db
199                .get_group_message(message_id)?
200                .filter(|message| message.group_id == self.group_id)
201                .ok_or_else(|| GroupError::NotFound(NotFound::MessageById(message_id.to_vec())))?;
202            if message.delivery_status == DeliveryStatus::Published {
203                return Ok(Continue(false));
204            }
205            let envelope =
206                Self::into_envelope(&message.decrypted_message_bytes, &message.idempotency_key);
207            let intent_data: Vec<u8> = SendMessageIntentData::new(envelope.encode_to_vec()).into();
208            QueueIntent::send_message()
209                .data(intent_data)
210                .should_push(message.should_push)
211                .queue_in(&db, self)?;
212            Ok::<_, GroupError>(Continue(true))
213        })?
214        .into_continued();
215        if !queued {
216            return Ok(());
217        }
218
219        // Publish
220        self.maybe_update_installations(Some(SEND_MESSAGE_UPDATE_INSTALLATIONS_INTERVAL_NS))
221            .await?;
222        self.sync_until_last_intent_resolved().await?;
223
224        // Implicitly set group consent state to allowed
225        self.update_consent_state(ConsentState::Allowed)?;
226
227        Ok(())
228    }
229
230    /// Delete a message by its ID. Returns the ID of the deletion message.
231    ///
232    /// Only the original sender or a super admin can delete a message.
233    ///
234    /// # Wire Protocol
235    /// The `DeleteMessage` protobuf encodes `message_id` as a hex-encoded string for wire
236    /// transmission, while the database stores message IDs as raw bytes. This function handles
237    /// the conversion: it accepts raw bytes, hex-encodes them for the wire protocol, and when
238    /// processing incoming deletions (in `process_delete_message`), the hex string is decoded
239    /// back to bytes for database lookups.
240    ///
241    /// # Arguments
242    /// * `message_id` - The message ID as bytes
243    ///
244    /// # Returns
245    /// The ID of the deletion message
246    pub fn delete_message(&self, message_id: Vec<u8>) -> Result<Vec<u8>, GroupError> {
247        use error::DeleteMessageError;
248
249        let conn = self.context.db();
250
251        // Load the original message
252        let original_msg = conn
253            .get_group_message(&message_id)?
254            .ok_or_else(|| DeleteMessageError::MessageNotFound(hex::encode(&message_id)))?;
255
256        // Validate message belongs to this group (prevent cross-group deletion)
257        if original_msg.group_id.as_slice() != self.group_id.as_slice() {
258            return Err(DeleteMessageError::NotAuthorized.into());
259        }
260
261        // Check if message is already deleted
262        if conn.is_message_deleted(&message_id)? {
263            return Err(DeleteMessageError::MessageAlreadyDeleted.into());
264        }
265
266        let sender_inbox_id = self.context.inbox_id();
267        let is_sender = original_msg.sender_inbox_id == sender_inbox_id;
268        let is_super_admin = self.is_super_admin(sender_inbox_id.to_string())?;
269
270        if !is_sender && !is_super_admin {
271            return Err(DeleteMessageError::NotAuthorized.into());
272        }
273
274        if !original_msg.kind.is_deletable() || !original_msg.content_type.is_deletable() {
275            return Err(DeleteMessageError::NonDeletableMessage.into());
276        }
277
278        let delete_msg = DeleteMessage {
279            message_id: hex::encode(&message_id),
280        };
281
282        let encoded_delete = DeleteMessageCodec::encode(delete_msg)?;
283        let mut buf = Vec::new();
284        encoded_delete.encode(&mut buf)?;
285
286        let deletion_message_id = self.send_message_optimistic(&buf, SendMessageOpts::default())?;
287
288        let is_super_admin_deletion = !is_sender && is_super_admin;
289
290        let deletion = StoredMessageDeletion {
291            id: deletion_message_id.clone(),
292            group_id: self.group_id,
293            deleted_message_id: message_id,
294            deleted_by_inbox_id: sender_inbox_id.to_string(),
295            is_super_admin_deletion,
296            deleted_at_ns: now_ns(),
297        };
298
299        deletion.store(&conn)?;
300
301        Ok(deletion_message_id)
302    }
303
304    /// Helper function to extract queryable content fields from a message
305    pub(in crate::groups) fn extract_queryable_content_fields(
306        message: &[u8],
307    ) -> QueryableContentFields {
308        // Return early with default if decoding fails or type is missing
309        EncodedContent::decode(message)
310            .inspect_err(|_| {
311                tracing::debug!("No queryable content fields, msg not formatted as encoded content")
312            })
313            .and_then(|content| {
314                QueryableContentFields::try_from(content).inspect_err(|e| {
315                    tracing::debug!(
316                        "Failed to convert EncodedContent to QueryableContentFields: {}",
317                        e
318                    )
319                })
320            })
321            .unwrap_or_default()
322    }
323
324    /// Prepare a [`IntentKind::SendMessage`] intent, and [`StoredGroupMessage`] on this users XMTP [`Client`].
325    ///
326    /// # Arguments
327    /// * message: UTF-8 or encoded message bytes
328    /// * opts: Options for sending the message
329    /// * envelope: closure that returns context-specific [`PlaintextEnvelope`]. Closure accepts
330    ///   timestamp attached to intent & stored message.
331    #[tracing::instrument(skip_all, level = "trace")]
332    pub(crate) fn prepare_message<F>(
333        &self,
334        message: &[u8],
335        opts: send_message_opts::SendMessageOpts,
336        envelope: F,
337    ) -> Result<Vec<u8>, GroupError>
338    where
339        F: FnOnce(&str) -> PlaintextEnvelope,
340    {
341        state_write(self.context.mls_storage(), |tx| {
342            let storage = tx.storage();
343            let db = storage.db();
344            let stored_message = self.store_message_for_later_publish(
345                &db,
346                message,
347                opts.should_push,
348                opts.idempotency_key,
349            )?;
350            if stored_message.delivery_status == DeliveryStatus::Published {
351                return Ok(Continue(stored_message.id));
352            }
353            // Create envelope using the stored idempotency key so the id stays consistent
354            let plain_envelope = envelope(&stored_message.idempotency_key);
355            let mut encoded_envelope = vec![];
356            plain_envelope.encode(&mut encoded_envelope)?;
357
358            // Queue the intent (use should_push from stored message)
359            let intent_data: Vec<u8> = SendMessageIntentData::new(encoded_envelope).into();
360            QueueIntent::send_message()
361                .data(intent_data)
362                .should_push(stored_message.should_push)
363                .queue_in(&db, self)?;
364
365            Ok::<_, GroupError>(Continue(stored_message.id))
366        })
367        .map(TransactionOutcome::into_continued)
368    }
369
370    fn into_envelope(encoded_msg: &[u8], idempotency_key: &str) -> PlaintextEnvelope {
371        PlaintextEnvelope {
372            content: Some(Content::V1(V1 {
373                content: encoded_msg.to_vec(),
374                idempotency_key: idempotency_key.to_string(),
375            })),
376        }
377    }
378
379    /// Query the database for stored messages. Optionally filtered by time, kind, delivery_status
380    /// and limit
381    pub fn find_messages(
382        &self,
383        args: &MsgQueryArgs,
384    ) -> Result<Vec<StoredGroupMessage>, GroupError> {
385        let conn = self.context.db();
386        let messages = conn.get_group_messages(&self.group_id, args)?;
387        Ok(messages)
388    }
389
390    /// Count the number of stored messages matching the given criteria
391    pub fn count_messages(&self, args: &MsgQueryArgs) -> Result<i64, GroupError> {
392        let conn = self.context.db();
393        let count = conn.count_group_messages(&self.group_id, args)?;
394        Ok(count)
395    }
396
397    /// Query the database for stored messages. Optionally filtered by time, kind, delivery_status
398    /// and limit
399    pub fn find_messages_with_reactions(
400        &self,
401        args: &MsgQueryArgs,
402    ) -> Result<Vec<StoredGroupMessageWithReactions>, GroupError> {
403        let conn = self.context.db();
404        let messages = conn.get_group_messages_with_reactions(&self.group_id, args)?;
405        Ok(messages)
406    }
407
408    /// Query for enriched messages (with reactions, replies, and deletion status)
409    #[xmtp_common::mls_span]
410    pub fn find_enriched_messages(
411        &self,
412        args: &MsgQueryArgs,
413    ) -> Result<Vec<crate::messages::decoded_message::DecodedMessage>, EnrichMessageError> {
414        let conn = self.context.db();
415        let messages = conn.get_group_messages(&self.group_id, args)?;
416        let enriched =
417            crate::messages::enrichment::enrich_messages(conn, &self.group_id, messages)?;
418        Ok(enriched)
419    }
420
421    pub fn get_last_read_times(&self) -> Result<LatestMessageTimeBySender, GroupError> {
422        let conn = self.context.db();
423        let latest_read_receipt =
424            conn.get_latest_message_times_by_sender(self.group_id, &[ContentType::ReadReceipt])?;
425        Ok(latest_read_receipt)
426    }
427
428    /// Load the group reference stored in the local database
429    pub fn load(&self) -> Result<StoredGroup, StorageError> {
430        let conn = self.context.db();
431        if let Some(group) = conn.find_group(&self.group_id)? {
432            Ok(group)
433        } else {
434            tracing::error!("group {} does not exist", hex::encode(self.group_id));
435            Err(NotFound::GroupById(self.group_id).into())
436        }
437    }
438}
xmtp_mls/groups/messages.rs

xmtp_mls/groups/
messages.rs
