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        // Refuse before any network work or pending proposal commit.
22        Self::reject_reserved_transcript_content(message)?;
23
24        self.ensure_not_paused().await?;
25        let update_interval_ns = Some(SEND_MESSAGE_UPDATE_INSTALLATIONS_INTERVAL_NS);
26        self.maybe_update_installations(update_interval_ns).await?;
27
28        // Check for pending proposals and commit them first
29        // OpenMLS blocks message creation when there are pending proposals
30        self.commit_pending_proposals_if_any().await?;
31
32        let message_id = self.prepare_message(message, opts)?;
33
34        self.sync_until_last_intent_resolved().await?;
35
36        // implicitly set group consent state to allowed
37        self.update_consent_state(ConsentState::Allowed)?;
38
39        Ok(message_id)
40    }
41
42    /// Checks for pending MLS proposals and commits them if any exist.
43    /// OpenMLS blocks message creation when there are pending proposals,
44    /// so we need to commit them first.
45    async fn commit_pending_proposals_if_any(&self) -> Result<(), GroupError> {
46        let (has_pending, has_pending_add) = self.with_group_snapshot(|openmls_group| {
47            Ok((
48                openmls_group.pending_proposals().next().is_some(),
49                openmls_group.pending_proposals().any(|proposal| {
50                    matches!(proposal.proposal(), openmls::prelude::Proposal::Add(_))
51                }),
52            ))
53        })?;
54
55        if has_pending {
56            // A rejected commit can leave an accepted installation Add without
57            // its membership update. Refresh before committing that proposal,
58            // even when the regular installation-check interval has not elapsed.
59            if has_pending_add {
60                self.add_missing_installations().await?;
61            }
62            tracing::debug!(
63                inbox_id = self.context.inbox_id(),
64                group_id = %self.group_id,
65                "Found pending proposals, committing before sending message"
66            );
67
68            // Queue a CommitPendingProposals intent and wait for it to resolve
69            let intent = intents::QueueIntent::commit_pending_proposals().queue(self)?;
70            self.sync_until_intent_resolved(intent.id).await?;
71        }
72
73        Ok(())
74    }
75
76    /// Publish all unpublished messages. This happens by calling `sync_until_last_intent_resolved`
77    /// which publishes all pending intents and reads them back from the network.
78    #[cfg_attr(any(test, feature = "test-utils"), tracing::instrument(level = "info", fields(inbox_id = self.context.inbox_id()), skip(self)))]
79    #[cfg_attr(not(any(test, feature = "test-utils")), xmtp_common::mls_span)]
80    pub async fn publish_messages(&self) -> Result<(), GroupError> {
81        self.ensure_not_paused().await?;
82        let update_interval_ns = Some(SEND_MESSAGE_UPDATE_INSTALLATIONS_INTERVAL_NS);
83        self.maybe_update_installations(update_interval_ns).await?;
84        self.sync_until_last_intent_resolved().await?;
85
86        // implicitly set group consent state to allowed
87        self.update_consent_state(ConsentState::Allowed)?;
88
89        Ok(())
90    }
91
92    /// Checks the network to see if any group members have identity updates that would cause installations
93    /// to be added or removed from the group.
94    ///
95    /// If so, adds/removes those group members
96    pub async fn update_installations(&self) -> Result<(), GroupError> {
97        self.ensure_not_paused().await?;
98        self.maybe_update_installations(Some(0)).await?;
99        Ok(())
100    }
101
102    /// Send a message, optimistically returning the ID of the message before the result of a message publish.
103    pub fn send_message_optimistic(
104        &self,
105        message: &[u8],
106        opts: send_message_opts::SendMessageOpts,
107    ) -> Result<Vec<u8>, GroupError> {
108        let message_id = self.prepare_message(message, opts)?;
109        Ok(message_id)
110    }
111
112    /// Prepare a message for later publishing.
113    ///
114    /// Stores the message locally with `Unpublished` delivery status but does NOT
115    /// create an intent to publish. Use `publish_stored_message` to publish later.
116    ///
117    /// # Arguments
118    /// * `message` - The message content bytes
119    /// * `should_push` - Whether to send a push notification when publishing
120    /// * `idempotency_key` - Optional caller-supplied key the message id is
121    ///   derived from. Defaults to a random key when `None`.
122    ///
123    /// Returns the message ID.
124    pub fn prepare_message_for_later_publish(
125        &self,
126        message: &[u8],
127        should_push: bool,
128        idempotency_key: Option<String>,
129    ) -> Result<Vec<u8>, GroupError> {
130        state_write(self.context.mls_storage(), |tx| {
131            self.ensure_active_for_send(tx)?;
132            let storage = tx.storage();
133            let message = self.store_message_for_later_publish(
134                &storage.db(),
135                message,
136                should_push,
137                idempotency_key,
138            )?;
139            Ok::<_, GroupError>(Continue(message.id))
140        })
141        .map(TransactionOutcome::into_continued)
142    }
143
144    /// Store an optimistic message using the caller's transaction.
145    fn store_message_for_later_publish(
146        &self,
147        db: &impl DbQuery,
148        message: &[u8],
149        should_push: bool,
150        idempotency_key: Option<String>,
151    ) -> Result<StoredGroupMessage, GroupError> {
152        Self::reject_reserved_transcript_content(message)?;
153        let now = self.context.message_timestamp_ns();
154        // Resolve the key once. Random defaults do not depend on clock resolution.
155        let idempotency_key = idempotency_key.unwrap_or_else(|| {
156            hex::encode(xmtp_common::rand_vec::<DEFAULT_IDEMPOTENCY_KEY_BYTES>())
157        });
158        let queryable_content_fields = Self::extract_queryable_content_fields(message);
159
160        let message_id = calculate_message_id(self.group_id, message, &idempotency_key);
161
162        // Idempotent: a retry with the same key + content resolves to the same id.
163        // Return the existing message rather than failing on the PK conflict, so
164        // crash-recovery retries are at-least-once-with-dedup instead of an error.
165        if let Some(existing) = db.get_group_message(&message_id)? {
166            return Ok(existing);
167        }
168
169        let group_message = StoredGroupMessage {
170            id: message_id.clone(),
171            group_id: self.group_id,
172            decrypted_message_bytes: message.to_vec(),
173            sent_at_ns: now,
174            kind: GroupMessageKind::Application,
175            sender_installation_id: self.context.installation_id().into(),
176            sender_inbox_id: self.context.inbox_id().to_string(),
177            delivery_status: DeliveryStatus::Unpublished,
178            content_type: queryable_content_fields.content_type,
179            version_major: queryable_content_fields.version_major,
180            version_minor: queryable_content_fields.version_minor,
181            authority_id: queryable_content_fields.authority_id,
182            reference_id: queryable_content_fields.reference_id,
183            sequence_id: 0,
184            envelope_hash: None,
185            expiry_ns: None,
186            expire_at_ns: None,
187            inserted_at_ns: 0,
188            should_push,
189            idempotency_key,
190        };
191        group_message.store(db)?;
192        Ok(group_message)
193    }
194
195    /// Publish a previously stored message by ID.
196    ///
197    /// Creates an intent for the message and publishes it to the network.
198    /// Uses the `should_push` value that was stored with the message.
199    /// This is a no-op if the message is already published.
200    ///
201    /// Returns an error if the message is not found.
202    #[xmtp_common::mls_span]
203    pub async fn publish_stored_message(&self, message_id: &[u8]) -> Result<(), GroupError> {
204        if !self.is_active()? {
205            return Err(GroupError::GroupInactive);
206        }
207        self.ensure_not_paused().await?;
208
209        let queued = crate::state_tx::state_write_with_events(
210            self.context.mls_storage(),
211            self.context.events(),
212            |tx, events| {
213                self.ensure_active_for_send(tx)?;
214                let storage = tx.storage();
215                let db = storage.db();
216                let message = db
217                    .get_group_message(message_id)?
218                    .filter(|message| message.group_id == self.group_id)
219                    .ok_or_else(|| {
220                        GroupError::NotFound(NotFound::MessageById(message_id.to_vec()))
221                    })?;
222                // An older build stored this row before the creation check.
223                // It fails here and is never queued; the transaction commits.
224                // A saved prepared attempt may already be on the backend, so
225                // its outcome stays unknown and its saved bytes may still publish.
226                // implements: GMOD-035
227                if message.delivery_status != DeliveryStatus::Published
228                    && Self::is_reserved_transcript_content(&message.decrypted_message_bytes)
229                    && !self.has_saved_attempt(&db, &message.id)?
230                {
231                    if message.delivery_status == DeliveryStatus::Unpublished {
232                        db.set_delivery_status_to_failed(&message.id)?;
233                        self.emit_message_status_changed(
234                            message.id,
235                            xmtp_events::MessageStatus::Unpublished,
236                            xmtp_events::MessageStatus::Failed,
237                            events,
238                        );
239                    }
240                    return Ok(Continue(Err(GroupError::ReservedTranscriptContentType)));
241                }
242                self.queue_stored_message(&db, events, &message)
243                    .map(|queued| Continue(Ok(queued)))
244            },
245        )?
246        .into_continued()?;
247        if !queued {
248            return Ok(());
249        }
250
251        // Publish
252        self.maybe_update_installations(Some(SEND_MESSAGE_UPDATE_INSTALLATIONS_INTERVAL_NS))
253            .await?;
254        self.sync_until_last_intent_resolved().await?;
255
256        // Implicitly set group consent state to allowed
257        self.update_consent_state(ConsentState::Allowed)?;
258
259        Ok(())
260    }
261
262    /// Delete a message by its ID. Returns the ID of the deletion message.
263    ///
264    /// Only the original sender or a super admin can delete a message.
265    ///
266    /// # Wire Protocol
267    /// The `DeleteMessage` protobuf encodes `message_id` as a hex-encoded string for wire
268    /// transmission, while the database stores message IDs as raw bytes. This function handles
269    /// the conversion: it accepts raw bytes, hex-encodes them for the wire protocol, and when
270    /// processing incoming deletions (in `process_delete_message`), the hex string is decoded
271    /// back to bytes for database lookups.
272    ///
273    /// # Arguments
274    /// * `message_id` - The message ID as bytes
275    ///
276    /// # Returns
277    /// The ID of the deletion message
278    pub fn delete_message(&self, message_id: Vec<u8>) -> Result<Vec<u8>, GroupError> {
279        use error::DeleteMessageError;
280
281        let conn = self.context.db();
282
283        // Load the original message
284        let original_msg = conn
285            .get_group_message(&message_id)?
286            .ok_or_else(|| DeleteMessageError::MessageNotFound(hex::encode(&message_id)))?;
287
288        // Validate message belongs to this group (prevent cross-group deletion)
289        if original_msg.group_id.as_slice() != self.group_id.as_slice() {
290            return Err(DeleteMessageError::NotAuthorized.into());
291        }
292
293        // Check if message is already deleted
294        if conn.is_message_deleted(&message_id)? {
295            return Err(DeleteMessageError::MessageAlreadyDeleted.into());
296        }
297
298        let sender_inbox_id = self.context.inbox_id();
299        let is_sender = original_msg.sender_inbox_id == sender_inbox_id;
300        let is_super_admin = self.live_is_super_admin(sender_inbox_id)?;
301
302        if !is_sender && !is_super_admin {
303            return Err(DeleteMessageError::NotAuthorized.into());
304        }
305
306        if !crate::messages::enrichment::is_deletable_stored_message(&original_msg) {
307            return Err(DeleteMessageError::NonDeletableMessage.into());
308        }
309
310        let delete_msg = DeleteMessage {
311            message_id: hex::encode(&message_id),
312        };
313
314        let encoded_delete = DeleteMessageCodec::encode(delete_msg)?;
315        let mut buf = Vec::new();
316        encoded_delete.encode(&mut buf)?;
317
318        let deletion_message_id = self.send_message_optimistic(&buf, SendMessageOpts::default())?;
319
320        let is_super_admin_deletion = !is_sender && is_super_admin;
321
322        let deletion = StoredMessageDeletion {
323            id: deletion_message_id.clone(),
324            group_id: self.group_id,
325            deleted_message_id: message_id,
326            deleted_by_inbox_id: sender_inbox_id.to_string(),
327            is_super_admin_deletion,
328            deleted_at_ns: now_ns(),
329        };
330
331        crate::state_tx::state_write_with_events(
332            self.context.mls_storage(),
333            self.context.events(),
334            |tx, events| {
335                let storage = tx.storage();
336                let db = storage.db();
337                let already_deleted = db
338                    .get_deletions_for_messages(vec![original_msg.id.clone()])?
339                    .iter()
340                    .any(|prior| {
341                        crate::messages::enrichment::is_deletion_valid(
342                            prior,
343                            &original_msg,
344                            &self.group_id,
345                        )
346                    });
347                deletion.store(&db)?;
348                if !already_deleted {
349                    crate::subscriptions::internal::emit_deleted_messages(
350                        events,
351                        vec![original_msg],
352                        xmtp_events::DeletionCause::DeletedLocally,
353                        Some(deletion.deleted_by_inbox_id.clone()),
354                        &db,
355                    )?;
356                }
357                Ok::<_, GroupError>(Continue(()))
358            },
359        )?;
360
361        Ok(deletion_message_id)
362    }
363
364    /// Whether a send intent for this message already has a saved attempt.
365    fn has_saved_attempt(&self, db: &impl DbQuery, message_id: &[u8]) -> Result<bool, GroupError> {
366        for intent in db.find_group_intents(
367            self.group_id,
368            Some(vec![IntentState::Published]),
369            Some(vec![xmtp_db::group_intent::IntentKind::SendMessage]),
370        )? {
371            if crate::utils::id::calculate_message_id_for_intent(&intent)?.as_deref()
372                == Some(message_id)
373                && db.prepared_envelopes(intent.id)?.is_some()
374            {
375                return Ok(true);
376            }
377        }
378        Ok(false)
379    }
380
381    /// Whether the outer content type is a reserved transcript type. Only the
382    /// `xmtp.org` authority reserves these type ids, at any version.
383    // implements: GMOD-035
384    pub(in crate::groups) fn is_reserved_transcript_content(message: &[u8]) -> bool {
385        EncodedContent::decode(message)
386            .ok()
387            .and_then(|content| content.r#type)
388            .is_some_and(|content_type| {
389                content_type.authority_id == "xmtp.org"
390                    && matches!(
391                        content_type.type_id.as_str(),
392                        "group_updated" | "group_membership_change"
393                    )
394            })
395    }
396
397    fn reject_reserved_transcript_content(message: &[u8]) -> Result<(), GroupError> {
398        if Self::is_reserved_transcript_content(message) {
399            Err(GroupError::ReservedTranscriptContentType)
400        } else {
401            Ok(())
402        }
403    }
404
405    /// Helper function to extract queryable content fields from a message
406    pub(in crate::groups) fn extract_queryable_content_fields(
407        message: &[u8],
408    ) -> QueryableContentFields {
409        // Return early with default if decoding fails or type is missing
410        EncodedContent::decode(message)
411            .inspect_err(|_| {
412                tracing::debug!("No queryable content fields, msg not formatted as encoded content")
413            })
414            .and_then(|content| {
415                QueryableContentFields::try_from(content).inspect_err(|e| {
416                    tracing::debug!(
417                        "Failed to convert EncodedContent to QueryableContentFields: {}",
418                        e
419                    )
420                })
421            })
422            .unwrap_or_default()
423    }
424
425    /// Prepare a [`IntentKind::SendMessage`] intent, and [`StoredGroupMessage`] on this users XMTP [`Client`].
426    ///
427    /// # Arguments
428    /// * message: UTF-8 or encoded message bytes
429    /// * opts: Options for sending the message
430    // implements: SEND-021
431    #[tracing::instrument(skip_all, level = "trace")]
432    pub(crate) fn prepare_message(
433        &self,
434        message: &[u8],
435        opts: send_message_opts::SendMessageOpts,
436    ) -> Result<Vec<u8>, GroupError> {
437        crate::state_tx::state_write_with_events(
438            self.context.mls_storage(),
439            self.context.events(),
440            |tx, events| {
441                self.ensure_active_for_send(tx)?;
442                let storage = tx.storage();
443                let db = storage.db();
444                let message = self.store_message_for_later_publish(
445                    &db,
446                    message,
447                    opts.should_push,
448                    opts.idempotency_key,
449                )?;
450                self.queue_stored_message(&db, events, &message)?;
451                Ok::<_, GroupError>(Continue(message.id))
452            },
453        )
454        .map(TransactionOutcome::into_continued)
455    }
456
457    /// Queue the one publish intent of a stored message under the caller's writer.
458    ///
459    /// A published message is never queued again, and an unresolved intent for
460    /// the same envelope is reused. A retry of a failed message returns it to
461    /// `Unpublished` in the same transaction as its new intent. Returns whether
462    /// the message has work to publish.
463    // implements: SEND-003
464    fn queue_stored_message(
465        &self,
466        db: &impl DbQuery,
467        events: &impl xmtp_events::EventWriter<crate::subscriptions::internal::InternalEvent>,
468        message: &StoredGroupMessage,
469    ) -> Result<bool, GroupError> {
470        if message.delivery_status == DeliveryStatus::Published {
471            return Ok(false);
472        }
473        let envelope =
474            Self::into_envelope(&message.decrypted_message_bytes, &message.idempotency_key);
475        QueueIntent::send_message()
476            .data(Vec::<u8>::from(SendMessageIntentData::new(
477                envelope.encode_to_vec(),
478            )))
479            .should_push(message.should_push)
480            .queue_in(db, self)?;
481        if db.set_failed_delivery_status_to_unpublished(&message.id)? > 0 {
482            self.emit_message_status_changed(
483                message.id.clone(),
484                xmtp_events::MessageStatus::Failed,
485                xmtp_events::MessageStatus::Unpublished,
486                events,
487            );
488        }
489        Ok(true)
490    }
491
492    /// Check membership while holding the same writer as the new message or intent.
493    // implements: SEND-005
494    fn ensure_active_for_send<T: xmtp_db::TransactionalKeyStore>(
495        &self,
496        tx: &mut crate::state_tx::StateTx<'_, T>,
497    ) -> Result<(), GroupError> {
498        let stored = tx
499            .storage()
500            .db()
501            .find_group(&self.group_id)?
502            .ok_or(NotFound::GroupById(self.group_id))?;
503        if stored.membership_state == GroupMembershipState::Restored {
504            return Err(GroupError::GroupInactive);
505        }
506        let active = tx.with_group(self.group_id, |mls_group, _| {
507            Ok::<_, GroupError>(mls_group.is_active())
508        })?;
509        if active {
510            Ok(())
511        } else {
512            Err(GroupError::GroupInactive)
513        }
514    }
515
516    fn into_envelope(encoded_msg: &[u8], idempotency_key: &str) -> PlaintextEnvelope {
517        PlaintextEnvelope {
518            content: Some(Content::V1(V1 {
519                content: encoded_msg.to_vec(),
520                idempotency_key: idempotency_key.to_string(),
521            })),
522        }
523    }
524
525    /// Query the database for stored messages. Optionally filtered by time, kind, delivery_status
526    /// and limit
527    pub fn find_messages(
528        &self,
529        args: &MsgQueryArgs,
530    ) -> Result<Vec<StoredGroupMessage>, GroupError> {
531        let conn = self.context.db();
532        let messages = conn.get_group_messages(&self.group_id, args)?;
533        Ok(messages)
534    }
535
536    /// Count the number of stored messages matching the given criteria
537    pub fn count_messages(&self, args: &MsgQueryArgs) -> Result<i64, GroupError> {
538        let conn = self.context.db();
539        let count = conn.count_group_messages(&self.group_id, args)?;
540        Ok(count)
541    }
542
543    pub fn get_last_read_times(&self) -> Result<LatestMessageTimeBySender, GroupError> {
544        let conn = self.context.db();
545        let latest_read_receipt =
546            conn.get_latest_message_times_by_sender(self.group_id, &[ContentType::ReadReceipt])?;
547        Ok(latest_read_receipt)
548    }
549
550    /// Load the group reference stored in the local database
551    pub fn load(&self) -> Result<StoredGroup, StorageError> {
552        let conn = self.context.db();
553        if let Some(group) = conn.find_group(&self.group_id)? {
554            Ok(group)
555        } else {
556            tracing::error!("group {} does not exist", hex::encode(self.group_id));
557            Err(NotFound::GroupById(self.group_id).into())
558        }
559    }
560}
xmtp_mls/groups/messages.rs

xmtp_mls/groups/
messages.rs
