xmtp_mls/groups/
lifecycle.rs

1//! Construction, loading, proposal capability, and insertion.
2
3use super::*;
4
5mod restored;
6use xmtp_db::TransactionalKeyStore;
7use xmtp_events::EventWriter;
8use xmtp_mls_common::app_data::creation::{InitialGroupKind, initial_dictionary};
9
10/// Represents a group, which can contain anywhere from 1 to MAX_GROUP_SIZE inboxes.
11///
12/// This is a wrapper around OpenMLS's `MlsGroup` that handles our application-level configuration
13/// and validations.
14impl<Context> MlsGroup<Context>
15where
16    Context: XmtpSharedContext,
17{
18    // Creates a new group instance. Does not validate that the group exists in the DB
19    pub fn new(
20        context: Context,
21        group_id: GroupId,
22        dm_id: Option<String>,
23        conversation_type: ConversationType,
24        created_at_ns: i64,
25    ) -> Self {
26        Self::new_from_arc(
27            context.clone(),
28            group_id,
29            dm_id,
30            conversation_type,
31            created_at_ns,
32        )
33    }
34
35    /// Creates a new group instance from the database. Validate that the group exists in the DB before constructing
36    /// the group.
37    ///
38    /// # Returns
39    ///
40    /// Returns the Group and the stored group information as a tuple.
41    pub fn new_cached(
42        context: Context,
43        group_id: &GroupId,
44    ) -> Result<(Self, StoredGroup), StorageError> {
45        let conn = context.db();
46        if let Some(group) = conn.find_group(group_id)? {
47            Ok((
48                Self::new_from_arc(
49                    context,
50                    *group_id,
51                    group.dm_id.clone(),
52                    group.conversation_type,
53                    group.created_at_ns,
54                ),
55                group,
56            ))
57        } else {
58            tracing::error!("group {} does not exist", hex::encode(group_id));
59            Err(NotFound::GroupById(*group_id).into())
60        }
61    }
62
63    pub(crate) fn new_from_arc(
64        context: Context,
65        group_id: GroupId,
66        dm_id: Option<String>,
67        conversation_type: ConversationType,
68        created_at_ns: i64,
69    ) -> Self {
70        let mut mutexes = context.mutexes().clone();
71        Self {
72            group_id,
73            dm_id,
74            conversation_type,
75            created_at_ns,
76            mutex: mutexes.get_mutex(group_id),
77            context: context.clone(),
78            #[cfg(test)]
79            mls_commit_lock: Arc::clone(context.mls_commit_lock()),
80        }
81    }
82
83    /// Read a consistent MLS snapshot. Only immutable results leave the transaction.
84    pub(crate) fn with_group_snapshot<R>(
85        &self,
86        operation: impl FnOnce(&OpenMlsGroup) -> Result<R, GroupError>,
87    ) -> Result<R, GroupError> {
88        state_write(self.context.mls_storage(), |tx| {
89            tx.with_group(self.group_id, |group, _| operation(group))
90                .map(Continue)
91        })
92        .map(TransactionOutcome::into_continued)
93    }
94
95    // Test fixtures can deliberately retain an MLS object to construct stale state.
96    #[cfg(test)]
97    #[tracing::instrument(level = "trace", skip_all)]
98    pub(crate) fn load_mls_group_with_lock<F, R>(
99        &self,
100        storage: &impl XmtpMlsStorageProvider,
101        operation: F,
102    ) -> Result<R, GroupError>
103    where
104        F: Fn(OpenMlsGroup) -> Result<R, GroupError>,
105    {
106        // Get the group ID for locking
107        let group_id = self.group_id;
108
109        // Acquire the lock synchronously using blocking_lock
110        let _lock = self.mls_commit_lock.get_lock_sync(group_id);
111        // Load the MLS group
112        let mls_group = OpenMlsGroup::load(storage, &self.group_id.to_openmls())
113            .inspect_err(|e| tracing::error!("openmls error while loading group {e}"))
114            .map_err(|_| NotFound::MlsGroup(self.group_id))?
115            .ok_or(NotFound::MlsGroup(self.group_id))?;
116
117        // Perform the operation with the MLS group
118        operation(mls_group)
119    }
120
121    // Test fixtures can deliberately retain an MLS object across a network wait.
122    #[cfg(test)]
123    #[tracing::instrument(level = "trace", skip(operation))]
124    pub(crate) async fn load_mls_group_with_lock_async<R, E>(
125        &self,
126        operation: impl AsyncFnOnce(OpenMlsGroup) -> Result<R, E>,
127    ) -> Result<R, E>
128    where
129        E: From<crate::StorageError> + From<xmtp_db::sql_key_store::SqlKeyStoreError>,
130    {
131        let mls_storage = self.context.mls_storage();
132        // Get the group ID for locking
133        let group_id = self.group_id;
134
135        // Acquire the lock asynchronously
136        let _lock = self.mls_commit_lock.get_lock_async(group_id).await;
137
138        // Load the MLS group
139        let mls_group = OpenMlsGroup::load(mls_storage, &self.group_id.to_openmls())?
140            .ok_or(StorageError::from(NotFound::GroupById(self.group_id)))?;
141
142        // Perform the operation with the MLS group
143        operation(mls_group).await
144    }
145
146    /// Check if all members in the group support the proposal-by-reference flow.
147    ///
148    /// This checks both:
149    /// 1. Leaf node capabilities in the MLS group (via `check_extension_support`)
150    /// 2. The latest published key packages for all member installations fetched
151    ///    from the network, since leaf nodes may be stale (they aren't updated after
152    ///    the first message is sent).
153    ///
154    /// Returns `true` if all members support proposals, `false` otherwise.
155    pub async fn all_members_support_proposals(
156        &self,
157        mls_group: &OpenMlsGroup,
158    ) -> Result<bool, GroupError> {
159        let (supported, installation_ids) = self.proposal_support_snapshot(mls_group);
160        self.published_members_support_proposals(supported, installation_ids)
161            .await
162    }
163
164    fn proposal_support_snapshot(&self, group: &OpenMlsGroup) -> (bool, Vec<Vec<u8>>) {
165        let supported = group
166            .check_extension_support(&[ExtensionType::AppDataDictionary])
167            .is_ok();
168        let installation_ids = group
169            .members()
170            .map(|member| member.signature_key)
171            .filter(|id| id.as_slice() != self.context.installation_id().as_slice())
172            .collect();
173        (supported, installation_ids)
174    }
175
176    async fn published_members_support_proposals(
177        &self,
178        supported: bool,
179        installation_ids: Vec<Vec<u8>>,
180    ) -> Result<bool, GroupError> {
181        if supported || installation_ids.is_empty() {
182            return Ok(true);
183        }
184
185        let store = crate::mls_store::MlsStore::new(self.context.clone());
186        let key_packages = store
187            .get_key_packages_for_installation_ids(installation_ids)
188            .await?;
189
190        for result in key_packages.values() {
191            match result {
192                Ok(verified_kp) => {
193                    let capabilities = verified_kp.inner.leaf_node().capabilities();
194                    if !capabilities
195                        .extensions()
196                        .contains(&ExtensionType::AppDataDictionary)
197                    {
198                        return Ok(false);
199                    }
200                }
201                Err(_) => {
202                    return Ok(false);
203                }
204            }
205        }
206
207        Ok(true)
208    }
209
210    /// Snapshot this group's membership capabilities: the extension types in
211    /// the group context, plus the extension types each member installation
212    /// advertises.
213    ///
214    /// This reports raw capability facts rather than answers — callers filter
215    /// it to whatever question they care about. For the proposal
216    /// (app-data-dictionary) migration specifically, a caller checks for
217    /// [`MlsExtensionType::AppDataDictionary`] in `context_extensions` (already
218    /// migrated?) and in each installation's `supported_extensions` (eligible /
219    /// who is blocking?).
220    ///
221    /// Every installation's capabilities — the local one included — come from
222    /// its *latest published* key package, not its in-group leaf node: a leaf
223    /// is frozen when the installation joins and is never updated, so it would
224    /// understate a client that upgraded afterward (the same reason
225    /// [`Self::all_members_support_proposals`] falls back to key packages). An
226    /// installation whose key package has not been published or fails
227    /// verification is reported with `capabilities_known == false` and an empty
228    /// extension list, so callers can distinguish "unknown" from "advertises
229    /// nothing". Transient failures (network, auth, db) surface as an `Err`
230    /// rather than masquerading as unknown capabilities.
231    pub async fn membership_capabilities(&self) -> Result<GroupMembershipCapabilities, GroupError> {
232        // Read the group context's extension types under lock. The member list
233        // (below) is read in a separate lock acquisition, so this is a
234        // best-effort snapshot rather than a single atomic view — fine for a
235        // debug surface.
236        let context_extensions = self.with_group_snapshot(|mls_group| {
237            Ok::<_, GroupError>(
238                mls_group
239                    .extensions()
240                    .iter()
241                    .map(|ext| MlsExtensionType::from(ext.extension_type()))
242                    .collect::<Vec<_>>(),
243            )
244        })?;
245
246        let members = self.members().await?;
247        let own_installation_id = self.context.installation_id();
248
249        // Capabilities for every installation come from its latest published
250        // key package. We intentionally include our own installation rather
251        // than reading its in-group leaf, which is frozen at join and would
252        // understate an upgraded local client.
253        let query_ids: Vec<Vec<u8>> = members
254            .iter()
255            .flat_map(|member| member.installation_ids.iter().cloned())
256            .collect();
257
258        let extensions_by_installation = self.installation_extensions(query_ids).await?;
259
260        let installation_capabilities = |installation_id: Vec<u8>| -> InstallationCapabilities {
261            let is_own = installation_id.as_slice() == own_installation_id.as_slice();
262            match extensions_by_installation.get(installation_id.as_slice()) {
263                Some(extensions) => InstallationCapabilities {
264                    installation_id,
265                    is_own,
266                    supported_extensions: extensions.clone(),
267                    capabilities_known: true,
268                },
269                None => InstallationCapabilities {
270                    installation_id,
271                    is_own,
272                    supported_extensions: Vec::new(),
273                    capabilities_known: false,
274                },
275            }
276        };
277
278        let member_caps: Vec<InboxCapabilities> = members
279            .into_iter()
280            .map(|member| InboxCapabilities {
281                inbox_id: member.inbox_id,
282                installations: member
283                    .installation_ids
284                    .into_iter()
285                    .map(installation_capabilities)
286                    .collect(),
287            })
288            .collect();
289
290        Ok(GroupMembershipCapabilities {
291            context_extensions,
292            members: member_caps,
293        })
294    }
295
296    /// Fetch the latest published key package for each given installation and
297    /// return the MLS extension types it advertises, keyed by installation id.
298    ///
299    /// An installation with no published key package (e.g. an old client — what
300    /// this surface exists to flag) or whose key package fails verification is
301    /// simply absent from the returned map; callers treat absence as
302    /// "capabilities unknown". Transient/infrastructure failures (network,
303    /// auth, db) are NOT swallowed — they propagate as `Err`, so a snapshot can
304    /// tell "this installation has nothing published" apart from "we couldn't
305    /// reach the server".
306    ///
307    async fn installation_extensions(
308        &self,
309        query_ids: Vec<Vec<u8>>,
310    ) -> Result<HashMap<Vec<u8>, Vec<MlsExtensionType>>, GroupError> {
311        if query_ids.is_empty() {
312            return Ok(HashMap::new());
313        }
314        let store = crate::mls_store::MlsStore::new(self.context.clone());
315
316        let verified = store
317            .get_key_packages_for_installation_ids(query_ids)
318            .await?;
319
320        Ok(verified
321            .into_iter()
322            .filter_map(|(id, result)| {
323                let extensions = result
324                    .ok()?
325                    .inner
326                    .leaf_node()
327                    .capabilities()
328                    .extensions()
329                    .iter()
330                    .copied()
331                    .map(MlsExtensionType::from)
332                    .collect();
333                Some((id, extensions))
334            })
335            .collect())
336    }
337
338    /// Validate that key packages support the AppData dictionary
339    /// group-context extension. A leaf that advertises
340    /// `ExtensionType::AppDataDictionary` in its standard MLS
341    /// `Capabilities` can join a migrated group and receive standalone
342    /// `AppDataUpdate` proposals.
343    pub fn validate_key_packages_support_proposals(
344        &self,
345        key_packages: &[openmls::key_packages::KeyPackage],
346    ) -> Result<(), GroupError> {
347        let extension_type = ExtensionType::AppDataDictionary;
348
349        for kp in key_packages {
350            let leaf_node = kp.leaf_node();
351            let capabilities = leaf_node.capabilities();
352
353            if !capabilities.extensions().contains(&extension_type) {
354                return Err(GroupError::ProposalsNotSupported(
355                    "Member does not support AppData dictionary: installation cannot receive standalone proposal messages".to_string(),
356                ));
357            }
358        }
359
360        Ok(())
361    }
362
363    // Create a new group and save it to the DB
364    pub(crate) fn create_and_insert(
365        context: Context,
366        conversation_type: ConversationType,
367        permissions_policy_set: PolicySet,
368        opts: GroupMetadataOptions,
369        oneshot_message: Option<OneshotMessage>,
370    ) -> Result<Self, GroupError> {
371        assert!(conversation_type != ConversationType::Dm);
372        let stored_group = Self::insert(
373            &context,
374            None,
375            GroupMembershipState::Allowed,
376            conversation_type,
377            permissions_policy_set,
378            opts,
379            oneshot_message,
380            true,
381        )?;
382        let new_group = Self::new_from_arc(
383            context.clone(),
384            stored_group.id,
385            stored_group.dm_id,
386            conversation_type,
387            stored_group.created_at_ns,
388        );
389
390        Ok(new_group)
391    }
392
393    #[expect(
394        clippy::too_many_arguments,
395        reason = "creation event mode keeps the write atomic"
396    )]
397    pub(crate) fn insert(
398        context: &Context,
399        existing_group_id: Option<&[u8]>,
400        membership_state: GroupMembershipState,
401        conversation_type: ConversationType,
402        permissions_policy_set: PolicySet,
403        opts: GroupMetadataOptions,
404        oneshot_message: Option<OneshotMessage>,
405        emit_created_event: bool,
406    ) -> Result<StoredGroup, GroupError> {
407        assert!(conversation_type != ConversationType::Dm);
408
409        let creator_inbox_id = context.inbox_id();
410        let commit_log_enabled = context.server_configuration().commit_log_enabled();
411        // Deployments without a commit log do not create a signer.
412        let signer =
413            commit_log_enabled.then(xmtp_cryptography::rand::rand_secret::<ED25519_KEY_LENGTH>);
414        let dictionary = initial_dictionary(
415            InitialGroupKind::Group {
416                conversation_type,
417                oneshot_message: oneshot_message.as_ref(),
418            },
419            &permissions_policy_set
420                .to_proto()
421                .map_err(group_permissions::GroupMutablePermissionsError::from)
422                .map_err(MetadataPermissionsError::from)?,
423            &opts,
424            creator_inbox_id,
425            signer.as_ref().map(|key| key.as_slice()),
426            &context
427                .server_configuration()
428                .configuration()
429                .application_components,
430        )
431        .map_err(app_data::migration::BootstrapSynthesisError::from)?;
432        let group_config = build_group_config(dictionary)?;
433
434        if !emit_created_event || conversation_type.is_virtual() {
435            if emit_created_event && conversation_type == ConversationType::Sync {
436                return crate::state_tx::state_write_with_events(
437                    context.mls_storage(),
438                    context.events(),
439                    |tx, events| {
440                        let (stored_group, created) = Self::insert_group_row(
441                            context,
442                            tx,
443                            existing_group_id,
444                            membership_state,
445                            conversation_type,
446                            &opts,
447                            &group_config,
448                            commit_log_enabled,
449                        )?;
450                        if created {
451                            context.task_channels().mark_notification_changed();
452                            events.emit(
453                                None,
454                                Some(crate::subscriptions::internal::InternalEvent::GroupJoined {
455                                    group_id: stored_group.id,
456                                    is_sync: true,
457                                    origin: crate::subscriptions::internal::GroupOrigin::Created,
458                                }),
459                            );
460                        }
461                        Ok::<_, GroupError>(Continue(stored_group))
462                    },
463                )
464                .map(TransactionOutcome::into_continued);
465            }
466            return state_write(context.mls_storage(), |tx| {
467                let (stored_group, _) = Self::insert_group_row(
468                    context,
469                    tx,
470                    existing_group_id,
471                    membership_state,
472                    conversation_type,
473                    &opts,
474                    &group_config,
475                    commit_log_enabled,
476                )?;
477                Ok::<_, GroupError>(Continue(stored_group))
478            })
479            .map(TransactionOutcome::into_continued);
480        }
481
482        let result = crate::state_tx::state_write_with_events(
483            context.mls_storage(),
484            context.events(),
485            |tx, events| {
486                let (stored_group, created) = Self::insert_group_row(
487                    context,
488                    tx,
489                    existing_group_id,
490                    membership_state,
491                    conversation_type,
492                    &opts,
493                    &group_config,
494                    commit_log_enabled,
495                )?;
496                if !created {
497                    return Ok::<_, GroupError>(Continue(stored_group));
498                }
499                let storage = tx.storage();
500                let db = storage.db();
501                let record = StoredConsentRecord::new(
502                    xmtp_db::consent_record::ConsentType::ConversationId,
503                    ConsentState::Allowed,
504                    hex::encode(stored_group.id),
505                );
506                let consent_changes = db.insert_or_replace_consent_records(&[record])?;
507                crate::subscriptions::internal::emit_preference_updates(
508                    events,
509                    consent_changes
510                        .iter()
511                        .cloned()
512                        .map(PreferenceUpdate::Consent)
513                        .collect(),
514                    crate::subscriptions::internal::PreferenceOrigin::Local,
515                    &db,
516                )?;
517                context.task_channels().mark_notification_changed();
518                events.emit(
519                    Some(xmtp_events::ClientEvent::ConversationJoined(
520                        xmtp_events::ConversationJoined {
521                            group_id: stored_group.id.to_vec(),
522                            conversation_type: xmtp_events::ConversationType::Group,
523                            origin: xmtp_events::JoinOrigin::Created,
524                            adder_inbox_id: None,
525                        },
526                    )),
527                    Some(crate::subscriptions::internal::InternalEvent::GroupJoined {
528                        group_id: stored_group.id,
529                        is_sync: false,
530                        origin: crate::subscriptions::internal::GroupOrigin::Created,
531                    }),
532                );
533                Ok::<_, GroupError>(Continue(stored_group))
534            },
535        )?
536        .into_continued();
537        Ok(result)
538    }
539
540    #[expect(
541        clippy::too_many_arguments,
542        reason = "the row stores the creation parameters"
543    )]
544    fn insert_group_row(
545        context: &Context,
546        tx: &mut crate::state_tx::StateTx<'_, impl TransactionalKeyStore>,
547        existing_group_id: Option<&[u8]>,
548        membership_state: GroupMembershipState,
549        conversation_type: ConversationType,
550        opts: &GroupMetadataOptions,
551        group_config: &MlsGroupCreateConfig,
552        commit_log_enabled: bool,
553    ) -> Result<(StoredGroup, bool), GroupError> {
554        let storage = tx.storage();
555        let db = storage.db();
556        if let Some(existing_group_id) = existing_group_id {
557            let group_id = GroupId::try_from(existing_group_id)?;
558            if let Some(existing) = db.find_group(&group_id)? {
559                return Ok((existing, false));
560            }
561        }
562        let provider = XmtpOpenMlsProviderRef::new(&storage);
563        let mls_group = if let Some(existing_group_id) = existing_group_id {
564            // A restored group starts with a stub. A Welcome replaces it.
565            OpenMlsGroup::from_backup_stub_logged(
566                &provider,
567                context.identity(),
568                group_config,
569                GroupId::try_from(existing_group_id)?,
570                commit_log_enabled,
571            )?
572        } else {
573            OpenMlsGroup::from_creation_logged(
574                &provider,
575                context.identity(),
576                group_config,
577                commit_log_enabled,
578            )?
579        };
580        let group_id: GroupId = mls_group.group_id().try_into()?;
581        let stored_group = StoredGroup::builder()
582            .id(group_id)
583            .created_at_ns(now_ns())
584            .membership_state(membership_state)
585            .conversation_type(conversation_type)
586            .added_by_inbox_id(context.inbox_id().to_string())
587            .message_disappear_from_ns(
588                opts.message_disappearing_settings
589                    .as_ref()
590                    .map(|m| m.from_ns),
591            )
592            .message_disappear_in_ns(opts.message_disappearing_settings.as_ref().map(|m| m.in_ns))
593            .should_publish_commit_log(existing_group_id.is_none())
594            .build()?;
595        stored_group.store_or_ignore(&db)?;
596        Ok((stored_group, true))
597    }
598
599    // Create a new DM and save it to the DB
600    pub(crate) fn create_dm_and_insert(
601        context: &Context,
602        membership_state: GroupMembershipState,
603        dm_target_inbox_id: InboxId,
604        opts: GroupMetadataOptions,
605        existing_group_id: Option<&[u8]>,
606    ) -> Result<Self, GroupError> {
607        Self::create_dm_with_members(
608            context,
609            membership_state,
610            DmMembers {
611                member_one_inbox_id: context.inbox_id().to_string(),
612                member_two_inbox_id: dm_target_inbox_id,
613            },
614            opts,
615            existing_group_id,
616        )
617    }
618
619    fn create_dm_with_members(
620        context: &Context,
621        membership_state: GroupMembershipState,
622        dm_members: DmMembers<InboxId>,
623        opts: GroupMetadataOptions,
624        existing_group_id: Option<&[u8]>,
625    ) -> Result<Self, GroupError> {
626        let commit_log_enabled = context.server_configuration().commit_log_enabled();
627        let signer =
628            commit_log_enabled.then(xmtp_cryptography::rand::rand_secret::<ED25519_KEY_LENGTH>);
629        let kind = if membership_state == GroupMembershipState::Restored {
630            InitialGroupKind::RestoredDm {
631                member_one_inbox_id: &dm_members.member_one_inbox_id,
632                member_two_inbox_id: &dm_members.member_two_inbox_id,
633            }
634        } else {
635            InitialGroupKind::Dm {
636                target_inbox_id: &dm_members.member_two_inbox_id,
637            }
638        };
639        let dictionary = initial_dictionary(
640            kind,
641            &PolicySet::new_dm()
642                .to_proto()
643                .map_err(group_permissions::GroupMutablePermissionsError::from)
644                .map_err(MetadataPermissionsError::from)?,
645            &opts,
646            context.inbox_id(),
647            signer.as_ref().map(|key| key.as_slice()),
648            &context
649                .server_configuration()
650                .configuration()
651                .application_components,
652        )
653        .map_err(app_data::migration::BootstrapSynthesisError::from)?;
654        let group_config = build_group_config(dictionary)?;
655
656        let stored_group = if membership_state == GroupMembershipState::Restored {
657            state_write(context.mls_storage(), |tx| {
658                Ok::<_, GroupError>(Continue(Self::insert_dm_row(
659                    context,
660                    tx,
661                    membership_state,
662                    &dm_members,
663                    &opts,
664                    existing_group_id,
665                    &group_config,
666                    commit_log_enabled,
667                )?))
668            })?
669            .into_continued()
670            .0
671        } else {
672            crate::state_tx::state_write_with_events(
673                context.mls_storage(),
674                context.events(),
675                |tx, events| {
676                    let (stored_group, created, consent_changes) = Self::insert_dm_row(
677                        context,
678                        tx,
679                        membership_state,
680                        &dm_members,
681                        &opts,
682                        existing_group_id,
683                        &group_config,
684                        commit_log_enabled,
685                    )?;
686                    if created {
687                        let storage = tx.storage();
688                        let db = storage.db();
689                        crate::subscriptions::internal::emit_preference_updates(
690                            events,
691                            consent_changes
692                                .iter()
693                                .cloned()
694                                .map(PreferenceUpdate::Consent)
695                                .collect(),
696                            crate::subscriptions::internal::PreferenceOrigin::Local,
697                            &db,
698                        )?;
699                        if existing_group_id.is_none() {
700                            context.task_channels().mark_notification_changed();
701                            events.emit(
702                                Some(xmtp_events::ClientEvent::ConversationJoined(
703                                    xmtp_events::ConversationJoined {
704                                        group_id: stored_group.id.to_vec(),
705                                        conversation_type: xmtp_events::ConversationType::Dm,
706                                        origin: xmtp_events::JoinOrigin::Created,
707                                        adder_inbox_id: None,
708                                    },
709                                )),
710                                Some(crate::subscriptions::internal::InternalEvent::GroupJoined {
711                                    group_id: stored_group.id,
712                                    is_sync: false,
713                                    origin: crate::subscriptions::internal::GroupOrigin::Created,
714                                }),
715                            );
716                        }
717                    }
718                    Ok::<_, GroupError>(Continue((stored_group, created, consent_changes)))
719                },
720            )?
721            .into_continued()
722            .0
723        };
724        let new_group = Self::new_from_arc(
725            context.clone(),
726            stored_group.id,
727            stored_group.dm_id,
728            ConversationType::Dm,
729            stored_group.created_at_ns,
730        );
731        Ok(new_group)
732    }
733
734    #[expect(
735        clippy::too_many_arguments,
736        reason = "the row stores the DM creation parameters"
737    )]
738    fn insert_dm_row(
739        context: &Context,
740        tx: &mut crate::state_tx::StateTx<'_, impl TransactionalKeyStore>,
741        membership_state: GroupMembershipState,
742        dm_members: &DmMembers<InboxId>,
743        opts: &GroupMetadataOptions,
744        existing_group_id: Option<&[u8]>,
745        group_config: &MlsGroupCreateConfig,
746        commit_log_enabled: bool,
747    ) -> Result<(StoredGroup, bool, Vec<StoredConsentRecord>), GroupError> {
748        let storage = tx.storage();
749        let db = storage.db();
750        if let Some(group_id) = existing_group_id {
751            let group_id = GroupId::try_from(group_id)?;
752            if let Some(existing) = db.find_group(&group_id)? {
753                return Ok((existing, false, Vec::new()));
754            }
755        }
756        let provider = XmtpOpenMlsProviderRef::new(&storage);
757        let mls_group = if let Some(group_id) = existing_group_id {
758            OpenMlsGroup::from_backup_stub_logged(
759                &provider,
760                context.identity(),
761                group_config,
762                GroupId::try_from(group_id)?,
763                commit_log_enabled,
764            )?
765        } else {
766            OpenMlsGroup::from_creation_logged(
767                &provider,
768                context.identity(),
769                group_config,
770                commit_log_enabled,
771            )?
772        };
773        let group_id: GroupId = mls_group.group_id().try_into()?;
774        let stored_group = StoredGroup::builder()
775            .id(group_id)
776            .created_at_ns(now_ns())
777            .membership_state(membership_state)
778            .added_by_inbox_id(context.inbox_id().to_string())
779            .message_disappear_from_ns(
780                opts.message_disappearing_settings
781                    .as_ref()
782                    .map(|m| m.from_ns),
783            )
784            .message_disappear_in_ns(opts.message_disappearing_settings.as_ref().map(|m| m.in_ns))
785            .dm_id(Some(dm_members.to_string()))
786            .build()?;
787        stored_group.store(&db)?;
788        if membership_state == GroupMembershipState::Restored {
789            return Ok((stored_group, true, Vec::new()));
790        }
791        let record = StoredConsentRecord::new(
792            xmtp_db::consent_record::ConsentType::ConversationId,
793            ConsentState::Allowed,
794            hex::encode(group_id),
795        );
796        let consent_changes = db.insert_or_replace_consent_records(&[record])?;
797        Ok((stored_group, true, consent_changes))
798    }
799
800    // Super admin status is only criteria for whether to publish the commit log for now
801    pub(in crate::groups) fn check_should_publish_commit_log(
802        inbox_id: String,
803        mutable_metadata: Option<GroupMutableMetadata>,
804    ) -> bool {
805        mutable_metadata
806            .as_ref()
807            .map(|metadata| metadata.is_super_admin(&inbox_id))
808            .unwrap_or(false) // Default to false if no mutable metadata
809    }
810}
xmtp_mls/groups/lifecycle.rs

xmtp_mls/groups/
lifecycle.rs
