xmtp_mls/groups/
messages.rs

1//! Sending, preparing, and querying messages.
2
3use super::mls_sync::publish::OutgoingPreparationError;
4use super::*;
5use crate::groups::deletion::may_delete;
6use crate::messages::decoded_message::carries_reserved_transcript;
7
8/// An installation refresh before a send applies its membership change even
9/// when the Welcomes for the installations it adds cannot be sent. That
10/// outcome is logged and dropped here, so the send goes on; the installations
11/// stay without a Welcome, as the SEND Known limitations describe.
12fn unsent_welcomes_do_not_stop_a_send(result: Result<(), GroupError>) -> Result<(), GroupError> {
13    match result {
14        Err(GroupError::OutgoingPreparation(OutgoingPreparationError::WelcomesUnsendable)) => {
15            tracing::warn!("installations added before a send have no Welcome");
16            Ok(())
17        }
18        other => other,
19    }
20}
21
22impl<Context> MlsGroup<Context>
23where
24    Context: XmtpSharedContext,
25{
26    /// Send a message on this users XMTP [`Client`](crate::client::Client).
27    #[xmtp_common::mls_span]
28    pub async fn send_message(
29        &self,
30        message: &[u8],
31        opts: send_message_opts::SendMessageOpts,
32    ) -> Result<Vec<u8>, GroupError> {
33        if !self.is_active()? {
34            tracing::warn!("Unable to send a message on an inactive group.");
35            return Err(GroupError::GroupInactive);
36        }
37
38        // Refuse before any network work or pending proposal commit.
39        Self::reject_reserved_transcript_content(message)?;
40
41        self.ensure_not_paused().await?;
42        // implements: DMS-020
43        self.ensure_dm_peer().await?;
44        let update_interval_ns = Some(SEND_MESSAGE_UPDATE_INSTALLATIONS_INTERVAL_NS);
45        unsent_welcomes_do_not_stop_a_send(
46            self.maybe_update_installations(update_interval_ns).await,
47        )?;
48
49        // Check for pending proposals and commit them first
50        // OpenMLS blocks message creation when there are pending proposals
51        self.commit_pending_proposals_if_any().await?;
52
53        let message_id = self.prepare_message(message, opts)?;
54
55        self.sync_until_last_intent_resolved().await?;
56
57        Ok(message_id)
58    }
59
60    /// Checks for pending MLS proposals and commits them if any exist.
61    /// OpenMLS blocks message creation when there are pending proposals,
62    /// so we need to commit them first.
63    async fn commit_pending_proposals_if_any(&self) -> Result<(), GroupError> {
64        let (has_pending, has_pending_add) = self.with_group_snapshot(|openmls_group| {
65            Ok((
66                openmls_group.pending_proposals().next().is_some(),
67                openmls_group.pending_proposals().any(|proposal| {
68                    matches!(proposal.proposal(), openmls::prelude::Proposal::Add(_))
69                }),
70            ))
71        })?;
72
73        if has_pending {
74            // A rejected commit can leave an accepted installation Add without
75            // its membership update. Refresh before committing that proposal,
76            // even when the regular installation-check interval has not elapsed.
77            if has_pending_add {
78                unsent_welcomes_do_not_stop_a_send(self.add_missing_installations().await)?;
79            }
80            tracing::debug!(
81                inbox_id = self.context.inbox_id(),
82                group_id = %self.group_id,
83                "Found pending proposals, committing before sending message"
84            );
85
86            // Queue a CommitPendingProposals intent and wait for it to resolve
87            let intent = intents::QueueIntent::commit_pending_proposals().queue(self)?;
88            self.sync_until_intent_resolved(intent.id).await?;
89        }
90
91        Ok(())
92    }
93
94    /// Publish all unpublished messages. This happens by calling `sync_until_last_intent_resolved`
95    /// which publishes all pending intents and reads them back from the network.
96    #[cfg_attr(any(test, feature = "test-utils"), tracing::instrument(level = "info", fields(inbox_id = self.context.inbox_id()), skip(self)))]
97    #[cfg_attr(not(any(test, feature = "test-utils")), xmtp_common::mls_span)]
98    pub async fn publish_messages(&self) -> Result<(), GroupError> {
99        self.ensure_not_paused().await?;
100        let update_interval_ns = Some(SEND_MESSAGE_UPDATE_INSTALLATIONS_INTERVAL_NS);
101        unsent_welcomes_do_not_stop_a_send(
102            self.maybe_update_installations(update_interval_ns).await,
103        )?;
104        self.sync_until_last_intent_resolved().await?;
105
106        Ok(())
107    }
108
109    /// Checks the network to see if any group members have identity updates that would cause installations
110    /// to be added or removed from the group.
111    ///
112    /// If so, adds/removes those group members
113    pub async fn update_installations(&self) -> Result<(), GroupError> {
114        self.ensure_not_paused().await?;
115        self.maybe_update_installations(Some(0)).await?;
116        Ok(())
117    }
118
119    /// Send a message, optimistically returning the ID of the message before the result of a message publish.
120    ///
121    /// In a DM, await [`Self::ensure_dm_peer`] first. This method does not
122    /// add a missing peer, so the peer could never read the message.
123    pub fn send_message_optimistic(
124        &self,
125        message: &[u8],
126        opts: send_message_opts::SendMessageOpts,
127    ) -> Result<Vec<u8>, GroupError> {
128        let message_id = self.prepare_message(message, opts)?;
129        Ok(message_id)
130    }
131
132    /// Prepare a message for later publishing.
133    ///
134    /// Stores the message locally with `Unpublished` delivery status but does NOT
135    /// create an intent to publish. Use `publish_stored_message` to publish later.
136    ///
137    /// # Arguments
138    /// * `message` - The message content bytes
139    /// * `should_push` - Whether to send a push notification when publishing
140    /// * `idempotency_key` - Optional caller-supplied key the message id is
141    ///   derived from. Defaults to a random key when `None`.
142    ///
143    /// Returns the message ID.
144    ///
145    /// In a DM, await [`Self::ensure_dm_peer`] first. This method does not
146    /// add a missing peer, so the peer could never read the message.
147    pub fn prepare_message_for_later_publish(
148        &self,
149        message: &[u8],
150        should_push: bool,
151        idempotency_key: Option<String>,
152    ) -> Result<Vec<u8>, GroupError> {
153        state_write(self.context.mls_storage(), |tx| {
154            self.ensure_active_for_send(tx)?;
155            let storage = tx.storage();
156            let mut message = self.store_message_for_later_publish(
157                &storage.db(),
158                message,
159                should_push,
160                idempotency_key,
161            )?;
162            message.decrypted_message_bytes.zeroize();
163            Ok::<_, GroupError>(Continue(message.id))
164        })
165        .map(TransactionOutcome::into_continued)
166    }
167
168    /// Store an optimistic message using the caller's transaction.
169    fn store_message_for_later_publish(
170        &self,
171        db: &impl DbQuery,
172        message: &[u8],
173        should_push: bool,
174        idempotency_key: Option<String>,
175    ) -> Result<StoredGroupMessage, GroupError> {
176        Self::reject_reserved_transcript_content(message)?;
177        let now = self.context.message_timestamp_ns();
178        // Resolve the key once. Random defaults do not depend on clock resolution.
179        let idempotency_key = idempotency_key.unwrap_or_else(|| {
180            hex::encode(xmtp_common::rand_vec::<DEFAULT_IDEMPOTENCY_KEY_BYTES>())
181        });
182        let queryable_content_fields = Self::extract_queryable_content_fields(message);
183
184        let message_id = calculate_message_id(self.group_id, message, &idempotency_key);
185
186        // Idempotent: a retry with the same key + content resolves to the same id.
187        // Return the existing message rather than failing on the PK conflict, so
188        // crash-recovery retries are at-least-once-with-dedup instead of an error.
189        if let Some(mut existing) = db.get_group_message(&message_id)? {
190            if existing.sender_inbox_id != self.context.inbox_id() {
191                existing.decrypted_message_bytes.zeroize();
192                return Err(GroupError::MessageSenderConflict);
193            }
194            return Ok(existing);
195        }
196
197        let mut group_message = StoredGroupMessage {
198            id: message_id.clone(),
199            group_id: self.group_id,
200            decrypted_message_bytes: message.to_vec(),
201            sent_at_ns: now,
202            kind: GroupMessageKind::Application,
203            sender_installation_id: self.context.installation_id().into(),
204            sender_inbox_id: self.context.inbox_id().to_string(),
205            delivery_status: DeliveryStatus::Unpublished,
206            content_type: queryable_content_fields.content_type,
207            version_major: queryable_content_fields.version_major,
208            version_minor: queryable_content_fields.version_minor,
209            authority_id: queryable_content_fields.authority_id,
210            reference_id: queryable_content_fields.reference_id,
211            sequence_id: 0,
212            envelope_hash: None,
213            expiry_ns: None,
214            expiry_unknown: false,
215            deleted_by_inbox_id: None,
216            expire_at_ns: None,
217            inserted_at_ns: 0,
218            should_push,
219            idempotency_key,
220        };
221        // The content can hold a device-sync HMAC root key; a failed insert
222        // must not drop it unwiped.
223        if let Err(error) = group_message.store(db) {
224            group_message.decrypted_message_bytes.zeroize();
225            return Err(error.into());
226        }
227        Ok(group_message)
228    }
229
230    /// Publish a previously stored message by ID.
231    ///
232    /// Creates an intent for the message and publishes it to the network.
233    /// Uses the `should_push` value that was stored with the message.
234    /// This is a no-op if the message is already published.
235    ///
236    /// Returns an error if the message is not found.
237    #[xmtp_common::mls_span]
238    pub async fn publish_stored_message(&self, message_id: &[u8]) -> Result<(), GroupError> {
239        if !self.is_active()? {
240            return Err(GroupError::GroupInactive);
241        }
242        self.ensure_not_paused().await?;
243
244        let queued = crate::state_tx::state_write_with_events(
245            self.context.mls_storage(),
246            self.context.events(),
247            |tx, events| {
248                self.ensure_active_for_send(tx)?;
249                let storage = tx.storage();
250                let db = storage.db();
251                let mut message = db.get_group_message(message_id)?.ok_or_else(|| {
252                    GroupError::NotFound(NotFound::MessageById(message_id.to_vec()))
253                })?;
254                // The content can hold a device-sync HMAC root key, or a
255                // commit-log signer an older build recorded, so it is wiped
256                // on every exit.
257                let outcome = (|| {
258                    if self.reject_reserved_stored_message(&db, &message, events)? {
259                        return Ok(Continue(Err(GroupError::ReservedTranscriptContentType)));
260                    }
261                    self.queue_stored_message(&db, events, &message)
262                        .map(|queued| Continue(Ok(queued)))
263                })();
264                message.decrypted_message_bytes.zeroize();
265                outcome
266            },
267        )?
268        .into_continued()?;
269        let Some(intent_id) = queued else {
270            return Ok(());
271        };
272
273        // Publish
274        unsent_welcomes_do_not_stop_a_send(
275            self.maybe_update_installations(Some(SEND_MESSAGE_UPDATE_INSTALLATIONS_INTERVAL_NS))
276                .await,
277        )?;
278        self.sync_until_intent_resolved(intent_id).await?;
279
280        Ok(())
281    }
282
283    /// Refuse reserved stored content before work that can publish, such as DM repair.
284    /// Publication rechecks the row under its own transaction.
285    pub fn reject_reserved_stored_transcript_content(
286        &self,
287        message_id: &[u8],
288    ) -> Result<(), GroupError> {
289        if !self.is_active()? {
290            return Err(GroupError::GroupInactive);
291        }
292        crate::state_tx::state_write_with_events(
293            self.context.mls_storage(),
294            self.context.events(),
295            |tx, events| {
296                self.ensure_active_for_send(tx)?;
297                let storage = tx.storage();
298                let db = storage.db();
299                let mut message = db.get_group_message(message_id)?.ok_or_else(|| {
300                    GroupError::NotFound(NotFound::MessageById(message_id.to_vec()))
301                })?;
302                let outcome = self
303                    .reject_reserved_stored_message(&db, &message, events)
304                    .map(|refused| {
305                        Continue(if refused {
306                            Err(GroupError::ReservedTranscriptContentType)
307                        } else {
308                            Ok(())
309                        })
310                    });
311                // Wipe even allowed, saved-attempt and wrong-group payloads.
312                message.decrypted_message_bytes.zeroize();
313                outcome
314            },
315        )?
316        .into_continued()
317    }
318
319    /// Delete a message for everyone.
320    ///
321    /// Only the original sender or a super admin can delete a message. An
322    /// unsent own message is cancelled: it and its queued sends are removed,
323    /// nothing is published, and its own id is returned. Otherwise a
324    /// deletion message is queued and its id returned; the target changes when
325    /// the deletion's echo is processed.
326    ///
327    /// In a DM, await [`Self::ensure_dm_peer`] first unless the target is an
328    /// unsent own message. This method does not add a missing peer, so the
329    /// peer could never read the deletion.
330    ///
331    /// # Wire Protocol
332    /// The `DeleteMessage` protobuf encodes `message_id` as a hex-encoded string for wire
333    /// transmission, while the database stores message IDs as raw bytes.
334    // implements: PROC-056
335    pub fn delete_message(&self, message_id: Vec<u8>) -> Result<Vec<u8>, GroupError> {
336        if self.cancel_unsent(&message_id)? {
337            return Ok(message_id);
338        }
339        let mut buf = Vec::new();
340        DeleteMessageCodec::encode(DeleteMessage {
341            message_id: hex::encode(&message_id),
342        })?
343        .encode(&mut buf)?;
344        self.send_message_optimistic(&buf, SendMessageOpts::default())
345    }
346
347    /// Check that this installation may delete `message_id`, then cancel it
348    /// when it is this installation's unsent message. Returns false
349    /// when the message is not unsent here; [`Self::delete_message`] then
350    /// queues a deletion. A DM caller repairs the peer between the two.
351    pub fn cancel_unsent(&self, message_id: &[u8]) -> Result<bool, GroupError> {
352        use error::DeleteMessageError;
353
354        let target = self
355            .context
356            .db()
357            .get_group_message(message_id)?
358            .ok_or_else(|| DeleteMessageError::MessageNotFound(hex::encode(message_id)))?;
359        let own = self.context.inbox_id();
360        // The sender needs no super-admin status, so no metadata read.
361        let super_admin = target.sender_inbox_id != own && self.live_is_super_admin(own)?;
362        may_delete(&self.group_id, own, super_admin, &target)?;
363        self.cancel_unsent_message(message_id)
364    }
365
366    /// Refuse an unsaved reserved row and persist its terminal failure in this write.
367    // implements: GMOD-054
368    fn reject_reserved_stored_message(
369        &self,
370        db: &impl DbQuery,
371        message: &StoredGroupMessage,
372        events: &impl xmtp_events::EventWriter<crate::subscriptions::internal::InternalEvent>,
373    ) -> Result<bool, GroupError> {
374        if message.group_id != self.group_id {
375            return Err(GroupError::NotFound(NotFound::MessageById(
376                message.id.clone(),
377            )));
378        }
379        // An older build may have stored this content before the creation check.
380        // Saved prepared bytes can already be on the backend and remain publishable.
381        if message.delivery_status == DeliveryStatus::Published
382            || !Self::is_reserved_transcript_content(&message.decrypted_message_bytes)
383            || self.has_saved_attempt(db, &message.id)?
384        {
385            return Ok(false);
386        }
387        if message.delivery_status == DeliveryStatus::Unpublished {
388            db.set_delivery_status_to_failed(&message.id)?;
389            self.emit_message_status_changed(
390                message.id.clone(),
391                xmtp_events::MessageStatus::Unpublished,
392                xmtp_events::MessageStatus::Failed,
393                events,
394            );
395        }
396        Ok(true)
397    }
398
399    /// Whether a send intent for this message already has a saved attempt.
400    fn has_saved_attempt(&self, db: &impl DbQuery, message_id: &[u8]) -> Result<bool, GroupError> {
401        for intent in db.find_group_intents(
402            self.group_id,
403            Some(vec![IntentState::Published]),
404            Some(vec![xmtp_db::group_intent::IntentKind::SendMessage]),
405        )? {
406            if crate::groups::message_id::calculate_message_id_for_intent(&intent)?.as_deref()
407                == Some(message_id)
408                && db.prepared_envelopes(intent.id)?.is_some()
409            {
410                return Ok(true);
411            }
412        }
413        Ok(false)
414    }
415
416    /// Whether content or any nested reply carries a reserved transcript type.
417    /// Only the `xmtp.org` authority reserves these type ids, at any version.
418    // implements: GMOD-054
419    pub(in crate::groups) fn is_reserved_transcript_content(message: &[u8]) -> bool {
420        carries_reserved_transcript(message)
421    }
422
423    /// Refuse reserved transcript types at any reply depth and replies that exceed
424    /// the bounded scan. Call this before network work such as [`Self::ensure_dm_peer`].
425    pub fn reject_reserved_transcript_content(message: &[u8]) -> Result<(), GroupError> {
426        if Self::is_reserved_transcript_content(message) {
427            Err(GroupError::ReservedTranscriptContentType)
428        } else {
429            Ok(())
430        }
431    }
432
433    /// Helper function to extract queryable content fields from a message
434    pub(in crate::groups) fn extract_queryable_content_fields(
435        message: &[u8],
436    ) -> QueryableContentFields {
437        // Return early with default if decoding fails or type is missing
438        let Ok(content_type_id) =
439            crate::messages::encoded_content_type(message).inspect_err(|_| {
440                tracing::debug!("No queryable content fields, msg not formatted as encoded content")
441            })
442        else {
443            return QueryableContentFields::default();
444        };
445        let content_type_id = content_type_id.unwrap_or_default();
446        // Only a type with a reference id needs the payload. Other content,
447        // such as a device-sync HMAC update, is never copied.
448        if !QueryableContentFields::has_reference(&content_type_id) {
449            return QueryableContentFields::from_type(&content_type_id);
450        }
451        EncodedContent::decode(message)
452            .and_then(|content| {
453                QueryableContentFields::try_from(content).inspect_err(|e| {
454                    tracing::debug!(
455                        "Failed to convert EncodedContent to QueryableContentFields: {}",
456                        e
457                    )
458                })
459            })
460            .unwrap_or_default()
461    }
462
463    /// Prepare a [`IntentKind::SendMessage`] intent, and [`StoredGroupMessage`] on this users XMTP [`Client`].
464    ///
465    /// # Arguments
466    /// * message: UTF-8 or encoded message bytes
467    /// * opts: Options for sending the message
468    // implements: SEND-021
469    #[tracing::instrument(skip_all, level = "trace")]
470    pub(crate) fn prepare_message(
471        &self,
472        message: &[u8],
473        opts: send_message_opts::SendMessageOpts,
474    ) -> Result<Vec<u8>, GroupError> {
475        crate::state_tx::state_write_with_events_durable(
476            self.context.mls_storage(),
477            self.context.events(),
478            |tx, events| {
479                self.ensure_active_for_send(tx)?;
480                let storage = tx.storage();
481                let db = storage.db();
482                let mut message = self.store_message_for_later_publish(
483                    &db,
484                    message,
485                    opts.should_push,
486                    opts.idempotency_key,
487                )?;
488                // The content can hold a device-sync HMAC root key.
489                let queued = self.queue_stored_message(&db, events, &message);
490                message.decrypted_message_bytes.zeroize();
491                queued?;
492                Ok::<_, GroupError>(Continue(message.id))
493            },
494        )
495        .map(TransactionOutcome::into_continued)
496    }
497
498    /// Queue the one publish intent of a stored message under the caller's writer.
499    ///
500    /// A published message is never queued again, and an unresolved intent for
501    /// the same envelope is reused. A retry of a failed message returns it to
502    /// `Unpublished` in the same transaction as its new intent. Returns that
503    /// exact intent's ID, or None when the message is already published.
504    // implements: SEND-003
505    fn queue_stored_message(
506        &self,
507        db: &impl DbQuery,
508        events: &impl xmtp_events::EventWriter<crate::subscriptions::internal::InternalEvent>,
509        message: &StoredGroupMessage,
510    ) -> Result<Option<i32>, GroupError> {
511        if message.delivery_status == DeliveryStatus::Published {
512            return Ok(None);
513        }
514        // Its bytes are gone, so a retry would publish empty content.
515        // implements: PROC-037
516        if message.deleted_by_inbox_id.is_some() {
517            return Err(error::DeleteMessageError::MessageAlreadyDeleted.into());
518        }
519        let envelope =
520            Self::encode_envelope(&message.decrypted_message_bytes, &message.idempotency_key);
521        let (intent, reused) = QueueIntent::send_message()
522            .data(Vec::<u8>::from(SendMessageIntentData::new(envelope)))
523            .should_push(message.should_push)
524            .queue_in_reporting_reuse(db, self)?;
525        // A retry that reuses the queued intent sends nothing new.
526        if !reused {
527            self.allow_after_send(db, events, message)?;
528        }
529        if db.set_failed_delivery_status_to_unpublished(&message.id)? > 0 {
530            self.emit_message_status_changed(
531                message.id.clone(),
532                xmtp_events::MessageStatus::Failed,
533                xmtp_events::MessageStatus::Unpublished,
534                events,
535            );
536        }
537        Ok(Some(intent.id))
538    }
539
540    /// Store allowed for a conversation that does not read allowed when the
541    /// app queues a message other than a deletion or a leave request.
542    // implements: CONS-057
543    fn allow_after_send(
544        &self,
545        db: &impl DbQuery,
546        events: &impl xmtp_events::EventWriter<crate::subscriptions::internal::InternalEvent>,
547        message: &StoredGroupMessage,
548    ) -> Result<(), GroupError> {
549        let Some(decision) = Decision::on_send(self.conversation_type, message.content_type) else {
550            return Ok(());
551        };
552        let Some(group) = db.find_group(&self.group_id)? else {
553            return Ok(());
554        };
555        let Some(key) = ConsentKey::of_group(&group, self.context.inbox_id()) else {
556            return Ok(());
557        };
558        let changes: Vec<_> = write_consent(db, &key, ConsentWrite::Decide(decision), Owed::Mark)?
559            .into_iter()
560            .collect();
561        crate::subscriptions::internal::emit_consent_changes(&self.context, events, db, &changes)?;
562        Ok(())
563    }
564
565    /// Check membership while holding the same writer as the new message or intent.
566    // implements: SEND-005
567    fn ensure_active_for_send<T: xmtp_db::TransactionalKeyStore>(
568        &self,
569        tx: &mut crate::state_tx::StateTx<'_, T>,
570    ) -> Result<(), GroupError> {
571        let stored = tx
572            .storage()
573            .db()
574            .find_group(&self.group_id)?
575            .ok_or(NotFound::GroupById(self.group_id))?;
576        if stored.membership_state == GroupMembershipState::Restored {
577            return Err(GroupError::GroupInactive);
578        }
579        let active = tx.with_group(self.group_id, |mls_group, _| {
580            Ok::<_, GroupError>(mls_group.is_active())
581        })?;
582        if active {
583            Ok(())
584        } else {
585            Err(GroupError::GroupInactive)
586        }
587    }
588
589    /// Encode the plaintext envelope of a message. The content can hold a
590    /// device-sync HMAC root key, so the proto temporary is wiped after
591    /// encoding.
592    fn encode_envelope(encoded_msg: &[u8], idempotency_key: &str) -> Vec<u8> {
593        let mut envelope = PlaintextEnvelope {
594            content: Some(Content::V1(V1 {
595                content: encoded_msg.to_vec(),
596                idempotency_key: idempotency_key.to_string(),
597            })),
598        };
599        let bytes = envelope.encode_to_vec();
600        if let Some(Content::V1(v1)) = &mut envelope.content {
601            v1.content.zeroize();
602        }
603        bytes
604    }
605
606    /// Query the database for stored messages. Optionally filtered by time, kind, delivery_status
607    /// and limit
608    pub fn find_messages(
609        &self,
610        args: &MsgQueryArgs,
611    ) -> Result<Vec<StoredGroupMessage>, GroupError> {
612        let conn = self.context.db();
613        let messages = conn.get_group_messages(&self.group_id, args)?;
614        Ok(messages)
615    }
616
617    /// Count the number of stored messages matching the given criteria
618    pub fn count_messages(&self, args: &MsgQueryArgs) -> Result<i64, GroupError> {
619        let conn = self.context.db();
620        let count = conn.count_group_messages(&self.group_id, args)?;
621        Ok(count)
622    }
623
624    pub fn get_last_read_times(&self) -> Result<LatestMessageTimeBySender, GroupError> {
625        let conn = self.context.db();
626        let latest_read_receipt =
627            conn.get_latest_message_times_by_sender(self.group_id, &[ContentType::ReadReceipt])?;
628        Ok(latest_read_receipt)
629    }
630
631    /// Load the group reference stored in the local database
632    pub fn load(&self) -> Result<StoredGroup, StorageError> {
633        let conn = self.context.db();
634        if let Some(group) = conn.find_group(&self.group_id)? {
635            Ok(group)
636        } else {
637            tracing::error!("group {} does not exist", hex::encode(self.group_id));
638            Err(NotFound::GroupById(self.group_id).into())
639        }
640    }
641}
xmtp_mls/groups/messages.rs

xmtp_mls/groups/
messages.rs
