xmtp_mls/groups/
lifecycle.rs

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

xmtp_mls/groups/
lifecycle.rs
