xmtp_mls/groups/
state.rs

1//! Admin lists, consent, epoch state, and group context.
2
3use super::*;
4/// Locally stored group fields read without an MLS group load.
5#[derive(Debug, Clone, PartialEq)]
6pub struct GroupMetadataSnapshot {
7    pub name: String,
8    pub image_url: String,
9    pub description: String,
10    pub app_data: String,
11    pub membership_state: GroupMembershipState,
12    pub admins: Vec<String>,
13    pub super_admins: Vec<String>,
14    pub permissions: GroupMutablePermissions,
15    pub policy_type: Option<PreconfiguredPolicies>,
16}
17
18/// State for one conversation at the time of the local reads.
19#[derive(Debug, Clone, PartialEq)]
20pub struct ConversationStateSnapshot {
21    pub is_active: bool,
22    pub consent_state: ConsentState,
23    pub paused_for_version: Option<String>,
24    pub is_disappearing_enabled: bool,
25    pub disappearing_settings: Option<MessageDisappearingSettings>,
26    pub notifications_enabled: bool,
27    pub commit_log_fork_status: Option<bool>,
28    pub group: Option<GroupMetadataSnapshot>,
29}
30
31impl<Context> MlsGroup<Context>
32where
33    Context: XmtpSharedContext,
34{
35    /// Read conversation state with one group and consent query, one notification
36    /// query, and two OpenMLS key reads.
37    pub fn state_snapshot(&self) -> Result<ConversationStateSnapshot, GroupError> {
38        use openmls::group::MlsGroupState;
39        use openmls_traits::storage::StorageProvider as _;
40
41        let db = self.context.db();
42        let (stored, consent) = db
43            .conversation_state_row(&self.group_id)?
44            .ok_or(NotFound::GroupById(self.group_id))?;
45        let consent_state = consent.unwrap_or(ConsentState::Unknown);
46        let notification = db.notification_record()?;
47        let notifications_enabled = crate::client::notifications::enabled_from_record(
48            &notification,
49            &stored,
50            consent_state,
51        )?;
52        let disappearing_settings = match (
53            stored.message_disappear_from_ns,
54            stored.message_disappear_in_ns,
55        ) {
56            (Some(from_ns), Some(in_ns)) => Some(MessageDisappearingSettings { from_ns, in_ns }),
57            _ => None,
58        };
59        let is_disappearing_enabled = disappearing_settings
60            .is_some_and(|settings| settings.from_ns > 0 && settings.in_ns > 0);
61        let is_active = if stored.membership_state == GroupMembershipState::Restored {
62            false
63        } else {
64            let state = self
65                .context
66                .mls_storage()
67                .group_state::<MlsGroupState, _>(&self.group_id.to_openmls())?
68                .ok_or(NotFound::GroupById(self.group_id))?;
69            !matches!(state, MlsGroupState::Inactive)
70        };
71        let group = if stored.conversation_type == ConversationType::Group {
72            use xmtp_mls_common::app_data::component_source::ComponentSourceError;
73            let ctx = self.load_group_context()?;
74            let mut metadata = xmtp_mls_common::app_data::component_source::extract_group_mutable_metadata_capability_aware_from_extensions(ctx.extensions())
75                .map_err(|error| match error {
76                    ComponentSourceError::GroupMutableMetadata(inner) => GroupError::MetadataPermissionsError(MetadataPermissionsError::Mutable(inner)),
77                    other => GroupError::MetadataPermissionsError(MetadataPermissionsError::ComponentSource(other)),
78                })?;
79            if stored.membership_state == GroupMembershipState::Restored {
80                metadata.admin_list.clear();
81                metadata.super_admin_list.clear();
82            }
83            let permissions = group_permissions::policy_set_from_dictionary(ctx.extensions())
84                .map_err(|error| GroupError::MetadataPermissionsError(error.into()))?;
85            let field = |key: MetadataField| {
86                metadata
87                    .attributes
88                    .get(key.as_str())
89                    .cloned()
90                    .unwrap_or_default()
91            };
92            let policy_type = permissions.preconfigured_policy().ok();
93            Some(GroupMetadataSnapshot {
94                name: field(MetadataField::GroupName),
95                image_url: field(MetadataField::GroupImageUrlSquare),
96                description: field(MetadataField::Description),
97                app_data: field(MetadataField::AppData),
98                membership_state: stored.membership_state,
99                admins: metadata.admin_list,
100                super_admins: metadata.super_admin_list,
101                permissions,
102                policy_type,
103            })
104        } else {
105            None
106        };
107        Ok(ConversationStateSnapshot {
108            is_active,
109            consent_state,
110            paused_for_version: stored.paused_for_version,
111            is_disappearing_enabled,
112            disappearing_settings,
113            notifications_enabled,
114            commit_log_fork_status: stored.is_commit_log_forked,
115            group,
116        })
117    }
118
119    /// Get the last application message time for the selected content types.
120    pub fn last_activity_ns(
121        &self,
122        content_types: Option<&[ContentType]>,
123    ) -> Result<i64, GroupError> {
124        let content_types =
125            content_types.unwrap_or(xmtp_db::conversation_list::CONVERSATION_LIST_CONTENT_TYPES);
126        self.context
127            .db()
128            .last_activity_ns(&self.group_id, content_types)?
129            .ok_or_else(|| NotFound::GroupById(self.group_id).into())
130    }
131    /// Updates the admin list of the group and syncs the changes to the network.
132    #[cfg_attr(any(test, feature = "test-utils"), tracing::instrument(level = "info", fields(inbox_id = %self.context.inbox_id()), skip(self)))]
133    #[cfg_attr(
134        not(any(test, feature = "test-utils")),
135        tracing::instrument(level = "trace", skip(self))
136    )]
137    pub async fn update_admin_list(
138        &self,
139        action_type: UpdateAdminListType,
140        inbox_id: String,
141    ) -> Result<(), GroupError> {
142        if self.metadata().await?.conversation_type == ConversationType::Dm {
143            return Err(MetadataPermissionsError::DmGroupMetadataForbidden.into());
144        }
145        let intent_action_type = match action_type {
146            UpdateAdminListType::Add => AdminListActionType::Add,
147            UpdateAdminListType::Remove => AdminListActionType::Remove,
148            UpdateAdminListType::AddSuper => AdminListActionType::AddSuper,
149            UpdateAdminListType::RemoveSuper => AdminListActionType::RemoveSuper,
150        };
151        // Parse before queueing. Publication parses the id again, and an
152        // intent that can never publish would block every later intent.
153        let inbox_id = xmtp_mls_common::inbox_id::InboxId::from_hex(&inbox_id)
154            .map_err(xmtp_mls_common::app_data::component_source::ComponentSourceError::from)?;
155        let intent_data: Vec<u8> =
156            UpdateAdminListIntentData::new(intent_action_type, inbox_id.to_hex()).into();
157        let intent = QueueIntent::update_admin_list()
158            .data(intent_data)
159            .queue(self)?;
160
161        let _ = self.sync_until_intent_resolved(intent.id).await?;
162        Ok(())
163    }
164
165    /// Find the `inbox_id` of the group member who added the member to the group
166    pub fn added_by_inbox_id(&self) -> Result<String, GroupError> {
167        let conn = self.context.db();
168        let group = conn
169            .find_group(&self.group_id)?
170            .ok_or(NotFound::GroupById(self.group_id))?;
171        Ok(group.added_by_inbox_id)
172    }
173
174    /// Find the `consent_state` of the group
175    pub fn consent_state(&self) -> Result<ConsentState, GroupError> {
176        let conn = self.context.db();
177        let record =
178            conn.get_consent_record(hex::encode(self.group_id), ConsentType::ConversationId)?;
179
180        match record {
181            Some(rec) => Ok(rec.state),
182            None => Ok(ConsentState::Unknown),
183        }
184    }
185
186    // Returns new consent records. Does not broadcast changes.
187    pub fn quietly_update_consent_state(
188        &self,
189        state: ConsentState,
190        db: &impl DbQuery,
191    ) -> Result<Vec<StoredConsentRecord>, GroupError> {
192        let consent_record = StoredConsentRecord::new(
193            ConsentType::ConversationId,
194            state,
195            hex::encode(self.group_id),
196        );
197
198        Ok(db.insert_or_replace_consent_records(std::slice::from_ref(&consent_record))?)
199    }
200
201    #[tracing::instrument(skip_all, level = "trace")]
202    pub fn update_consent_state(&self, state: ConsentState) -> Result<(), GroupError> {
203        crate::state_tx::state_write_with_events(
204            self.context.mls_storage(),
205            self.context.events(),
206            |tx, events| {
207                let storage = tx.storage();
208                let db = storage.db();
209                let updates = self
210                    .quietly_update_consent_state(state, &db)?
211                    .into_iter()
212                    .map(PreferenceUpdate::Consent)
213                    .collect::<Vec<_>>();
214                if !updates.is_empty() {
215                    self.context.task_channels().mark_notification_changed();
216                }
217                crate::subscriptions::internal::emit_preference_updates(
218                    events,
219                    updates,
220                    crate::subscriptions::internal::PreferenceOrigin::Local,
221                    &db,
222                )?;
223                Ok::<_, GroupError>(Continue(()))
224            },
225        )?
226        .into_continued();
227
228        Ok(())
229    }
230
231    /// Get the current epoch number of the group.
232    pub async fn epoch(&self) -> Result<u64, GroupError> {
233        self.with_group_snapshot(|mls_group| Ok(mls_group.epoch().as_u64()))
234    }
235
236    /// Get the encryption state of the current epoch. Should match for all installations
237    /// in the same epoch.
238    #[cfg(any(test, feature = "test-utils"))]
239    pub async fn epoch_authenticator(&self) -> Result<Vec<u8>, GroupError> {
240        self.with_group_snapshot(|mls_group| {
241            Ok(mls_group.epoch_authenticator().as_slice().to_vec())
242        })
243    }
244
245    pub async fn cursor(&self) -> Result<Cursor, GroupError> {
246        let db = self.context.db();
247        let msgs = db.get_last_cursor(self.group_id, EntityKind::ApplicationMessage)?;
248        Ok(msgs)
249    }
250
251    pub async fn local_commit_log(&self) -> Result<Vec<LocalCommitLog>, GroupError> {
252        Ok(self.context.db().get_group_logs(&self.group_id)?)
253    }
254
255    pub async fn remote_commit_log(&self) -> Result<Vec<RemoteCommitLog>, GroupError> {
256        Ok(self.context.db().get_remote_commit_log_after_cursor(
257            &self.group_id,
258            0,
259            RemoteCommitLogOrder::AscendingByRowid,
260        )?)
261    }
262
263    pub async fn debug_info(&self) -> Result<ConversationDebugInfo, GroupError> {
264        let epoch = self.epoch().await?;
265        let cursor = self.cursor().await?;
266        let commit_log = self.local_commit_log().await?;
267        let remote_commit_log = self.remote_commit_log().await?;
268        let db = self.context.db();
269
270        let stored_group = match db.find_group(&self.group_id)? {
271            Some(group) => group,
272            None => {
273                return Err(GroupError::NotFound(NotFound::GroupById(self.group_id)));
274            }
275        };
276
277        Ok(ConversationDebugInfo {
278            epoch,
279            maybe_forked: stored_group.maybe_forked,
280            fork_details: stored_group.fork_details,
281            is_commit_log_forked: stored_group.is_commit_log_forked,
282            local_commit_log: format!("{:?}", commit_log),
283            remote_commit_log: format!("{:?}", remote_commit_log),
284            cursor: vec![cursor],
285        })
286    }
287
288    /// Update this installation's leaf key in the group by creating a key update commit
289    #[cfg_attr(any(test, feature = "test-utils"), tracing::instrument(level = "info", fields(inbox_id = %self.context.inbox_id()), skip(self)))]
290    #[cfg_attr(
291        not(any(test, feature = "test-utils")),
292        tracing::instrument(level = "trace", skip(self))
293    )]
294    pub async fn key_update(&self) -> Result<(), GroupError> {
295        let intent = QueueIntent::key_update().queue(self)?;
296        let _ = self.sync_until_intent_resolved(intent.id).await?;
297        Ok(())
298    }
299
300    /// Checks if the current user is active in the group.
301    ///
302    /// If the current user has been kicked out of the group, `is_active` will return `false`
303    #[tracing::instrument(skip_all, level = "trace")]
304    pub fn is_active(&self) -> Result<bool, GroupError> {
305        // Restored groups that are not yet added are inactive
306        let Some(stored_group) = self.context.db().find_group(&self.group_id)? else {
307            return Err(GroupError::NotFound(NotFound::GroupById(self.group_id)));
308        };
309        if matches!(
310            stored_group.membership_state,
311            GroupMembershipState::Restored
312        ) {
313            return Ok(false);
314        }
315
316        self.with_group_snapshot(|mls_group| Ok(mls_group.is_active()))
317    }
318
319    /// Returns the membership state of the current user in this group.
320    #[tracing::instrument(skip_all, level = "trace")]
321    pub fn membership_state(&self) -> Result<GroupMembershipState, GroupError> {
322        let stored_group = self
323            .context
324            .db()
325            .find_group(&self.group_id)?
326            .ok_or_else(|| GroupError::NotFound(NotFound::GroupById(self.group_id)))?;
327        Ok(stored_group.membership_state)
328    }
329
330    /// Get the `GroupMetadata` of the group.
331    ///
332    /// The AppData dictionary contains CONVERSATION_TYPE,
333    /// CREATOR_INBOX_ID, DM_MEMBERS, and ONESHOT_MESSAGE.
334    /// Restored conversations have an empty creator until activation.
335    pub async fn metadata(&self) -> Result<GroupMetadata, GroupError> {
336        let is_restored = self.membership_state()? == GroupMembershipState::Restored;
337        self.with_group_snapshot(|mls_group| {
338            let seed = xmtp_mls_common::app_data::component_source::read_group_metadata_from_dict(
339                mls_group,
340            )
341            .map_err(MetadataPermissionsError::from)?
342            .ok_or_else(|| MetadataPermissionsError::from(GroupMetadataError::MissingExtension))?;
343            use xmtp_proto::xmtp::mls::message_contents::GroupMetadataV1 as GroupMetadataProto;
344            let proto = GroupMetadataProto {
345                conversation_type: seed.conversation_type,
346                creator_inbox_id: if is_restored {
347                    String::new()
348                } else {
349                    seed.creator_inbox_id
350                },
351                creator_account_address: String::new(),
352                dm_members: seed.dm_members,
353                oneshot_message: seed.oneshot,
354            };
355            Ok(GroupMetadata::try_from(proto).map_err(MetadataPermissionsError::from)?)
356        })
357    }
358
359    /// Read the group's `GroupContext` from storage — a single KV round-trip,
360    /// no ratchet tree, no secrets, no commit lock. All group metadata lives in
361    /// the context extensions, so metadata reads go through this rather than a
362    /// full `OpenMlsGroup::load`. The context key is written atomically on
363    /// commit, so a single-key read is metadata-consistent.
364    ///
365    /// (The pre-refactor sync `load_mls_group_with_lock` used only an *advisory*
366    /// lock for these reads — a failed `get_lock_sync` was ignored and the read
367    /// proceeded anyway — so dropping it changes nothing for reads.)
368    pub(crate) fn load_group_context(&self) -> Result<openmls::group::GroupContext, GroupError> {
369        use openmls_traits::storage::StorageProvider as _;
370        self.context
371            .mls_storage()
372            .group_context::<_, openmls::group::GroupContext>(&self.group_id.to_openmls())
373            .map_err(GroupError::from)?
374            .ok_or_else(|| GroupError::from(StorageError::from(NotFound::GroupById(self.group_id))))
375    }
376
377    /// Get the `GroupMutableMetadata` of the group.
378    ///
379    /// The AppData dictionary contains all mutable metadata.
380    /// Restored conversations have no public admin roles until activation.
381    pub fn mutable_metadata(&self) -> Result<GroupMutableMetadata, GroupError> {
382        let is_restored = self.membership_state()? == GroupMembershipState::Restored;
383        let mut metadata = self.live_mutable_metadata()?;
384        if is_restored {
385            metadata.admin_list.clear();
386            metadata.super_admin_list.clear();
387        }
388        Ok(metadata)
389    }
390
391    /// The current MLS dictionary for display and permission checks.
392    // implements: ARCH-015
393    pub(crate) fn live_mutable_metadata(&self) -> Result<GroupMutableMetadata, GroupError> {
394        use xmtp_mls_common::app_data::component_source::ComponentSourceError;
395        let ctx = self.load_group_context()?;
396        xmtp_mls_common::app_data::component_source::extract_group_mutable_metadata_capability_aware_from_extensions(
397            ctx.extensions(),
398        )
399        .map_err(|e| match e {
400            ComponentSourceError::GroupMutableMetadata(inner) => {
401                GroupError::MetadataPermissionsError(MetadataPermissionsError::Mutable(inner))
402            }
403            other => GroupError::MetadataPermissionsError(
404                MetadataPermissionsError::ComponentSource(other),
405            ),
406        })
407    }
408
409    /// Pre-L implementation of [`Self::mutable_metadata`]: a full
410    /// `OpenMlsGroup::load`. Retained only as the baseline the
411    /// read-amplification benchmark measures the context-read path against.
412    #[cfg(test)]
413    pub(crate) fn mutable_metadata_via_full_load(
414        &self,
415    ) -> Result<GroupMutableMetadata, GroupError> {
416        use xmtp_mls_common::app_data::component_source::ComponentSourceError;
417        self.load_mls_group_with_lock(self.context.mls_storage(), |mls_group| {
418            xmtp_mls_common::app_data::component_source::extract_group_mutable_metadata_capability_aware(
419                &mls_group,
420            )
421            .map_err(|e| match e {
422                ComponentSourceError::GroupMutableMetadata(inner) => {
423                    GroupError::MetadataPermissionsError(MetadataPermissionsError::Mutable(inner))
424                }
425                other => GroupError::MetadataPermissionsError(
426                    MetadataPermissionsError::ComponentSource(other),
427                ),
428            })
429        })
430    }
431
432    /// Returns the permissions projection for this group.
433    pub fn permissions(&self) -> Result<GroupMutablePermissions, GroupError> {
434        let ctx = self.load_group_context()?;
435        self::group_permissions::policy_set_from_dictionary(ctx.extensions())
436            .map_err(|error| MetadataPermissionsError::from(error).into())
437    }
438
439    /// Capability-aware single-component read.
440    ///
441    /// Reads the group's `GroupContext` (via [`Self::load_group_context`]),
442    /// then uses the [`xmtp_mls_common::app_data::typed_facade::MlsGroupAppData`] facade to
443    /// read exactly one [`Component`](xmtp_mls_common::app_data::typed::Component)
444    /// out of its extensions — avoiding both the full `OpenMlsGroup::load` and
445    /// the full `GroupMutableMetadata` composite parse that a naive read would
446    /// run on every call.
447    ///
448    /// Returns `Ok(None)` when the dictionary has no stored value.
449    ///
450    /// `ComponentSourceError::GroupMutableMetadata` maps to
451    /// `MetadataPermissionsError::Mutable`. Other component-source errors map
452    /// to `MetadataPermissionsError::ComponentSource`.
453    pub(in crate::groups) fn read_single_component<C>(&self) -> Result<Option<C::Value>, GroupError>
454    where
455        C: xmtp_mls_common::app_data::typed::Component,
456    {
457        use xmtp_mls_common::app_data::component_source::ComponentSourceError;
458        let ctx = self.load_group_context()?;
459        let facade =
460            xmtp_mls_common::app_data::typed_facade::MlsGroupAppData::new(ctx.extensions());
461        facade.get::<C>().map_err(|e| match e {
462            ComponentSourceError::GroupMutableMetadata(inner) => {
463                GroupError::MetadataPermissionsError(MetadataPermissionsError::Mutable(inner))
464            }
465            other => GroupError::MetadataPermissionsError(
466                MetadataPermissionsError::ComponentSource(other),
467            ),
468        })
469    }
470
471    /// Fetches the message disappearing settings for a given group ID.
472    ///
473    /// Returns `Some(MessageDisappearingSettings)` if the group exists and has valid settings,
474    /// `None` if the group or settings are missing, or `Err(ClientError)` on a database error.
475    pub fn disappearing_settings(&self) -> Result<Option<MessageDisappearingSettings>, GroupError> {
476        let conn = self.context.db();
477        let stored_group: Option<StoredGroup> = conn.fetch(&self.group_id)?;
478
479        let settings = stored_group.and_then(|group| {
480            let from_ns = group.message_disappear_from_ns?;
481            let in_ns = group.message_disappear_in_ns?;
482
483            Some(MessageDisappearingSettings { from_ns, in_ns })
484        });
485
486        Ok(settings)
487    }
488
489    /// Find all the duplicate dms for this group
490    pub fn find_duplicate_dms(&self) -> Result<Vec<MlsGroup<Context>>, ClientError> {
491        let duplicates = self.context.db().other_dms(&self.group_id)?;
492
493        let mls_groups = duplicates
494            .into_iter()
495            .map(|g| {
496                MlsGroup::new(
497                    self.context.clone(),
498                    g.id,
499                    g.dm_id,
500                    g.conversation_type,
501                    g.created_at_ns,
502                )
503            })
504            .collect();
505
506        Ok(mls_groups)
507    }
508
509    /// Used for testing that dm group validation works as expected.
510    ///
511    /// See the `test_validate_dm_group` test function for more details.
512    #[cfg(test)]
513    pub fn create_test_dm_group(
514        context: Context,
515        dm_target_inbox_id: InboxId,
516        custom_protected_metadata: Option<Extension>,
517        custom_mutable_metadata: Option<Extension>,
518        custom_group_membership: Option<Extension>,
519        custom_mutable_permissions: Option<PolicySet>,
520        opts: Option<DMMetadataOptions>,
521    ) -> Result<Self, GroupError> {
522        let provider = context.mls_provider();
523        let commit_log_enabled = context.server_configuration().commit_log_enabled();
524
525        let protected_metadata = custom_protected_metadata.unwrap_or_else(|| {
526            build_dm_protected_metadata_extension(context.inbox_id(), dm_target_inbox_id.clone())
527                .unwrap()
528        });
529        let mutable_metadata = custom_mutable_metadata.unwrap_or_else(|| {
530            build_dm_mutable_metadata_extension_default(
531                context.inbox_id(),
532                &dm_target_inbox_id,
533                opts.unwrap_or_default(),
534                commit_log_enabled,
535            )
536            .unwrap()
537        });
538        let group_membership = custom_group_membership
539            .unwrap_or_else(|| build_starting_group_membership_extension(context.inbox_id(), 0));
540        let mutable_permissions = custom_mutable_permissions.unwrap_or_else(PolicySet::new_dm);
541        let mutable_permission_extension =
542            build_mutable_permissions_extension(mutable_permissions)?;
543
544        let group_config = build_legacy_test_group_config(
545            protected_metadata,
546            mutable_metadata,
547            group_membership,
548            mutable_permission_extension,
549        )?;
550
551        let mls_group = OpenMlsGroup::from_creation_logged(
552            &provider,
553            context.identity(),
554            &group_config,
555            commit_log_enabled,
556        )?;
557        let group_id: GroupId = mls_group.group_id().try_into()?;
558        let stored_group = StoredGroup::builder()
559            .id(group_id)
560            .created_at_ns(now_ns())
561            .membership_state(GroupMembershipState::Allowed)
562            .added_by_inbox_id(context.inbox_id().to_string())
563            .dm_id(Some(
564                DmMembers {
565                    member_one_inbox_id: context.inbox_id().to_string(),
566                    member_two_inbox_id: dm_target_inbox_id,
567                }
568                .to_string(),
569            ))
570            .build()?;
571
572        stored_group.store(&context.db())?;
573        Ok(Self::new_from_arc(
574            context,
575            group_id,
576            stored_group.dm_id.clone(),
577            ConversationType::Dm,
578            stored_group.created_at_ns,
579        ))
580    }
581}
xmtp_mls/groups/state.rs

xmtp_mls/groups/
state.rs
