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_group_state::app_data::component_source::ComponentSourceError;
73            let ctx = self.load_group_context()?;
74            let mut metadata = xmtp_mls_group_state::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: std::mem::take(&mut metadata.admin_list),
100                super_admins: std::mem::take(&mut 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_group_state::inbox_id::InboxId::from_hex(&inbox_id).map_err(
154            xmtp_mls_group_state::app_data::component_source::ComponentSourceError::from,
155        )?;
156        let intent_data: Vec<u8> =
157            UpdateAdminListIntentData::new(intent_action_type, inbox_id.to_hex()).into();
158        let intent = QueueIntent::update_admin_list()
159            .data(intent_data)
160            .queue(self)?;
161
162        let _ = self.sync_until_intent_resolved(intent.id).await?;
163        Ok(())
164    }
165
166    /// Find the `inbox_id` of the group member who added the member to the group
167    pub fn added_by_inbox_id(&self) -> Result<String, GroupError> {
168        let conn = self.context.db();
169        let group = conn
170            .find_group(&self.group_id)?
171            .ok_or(NotFound::GroupById(self.group_id))?;
172        Ok(group.added_by_inbox_id)
173    }
174
175    /// The conversation consent of the group: a DM's is its peer's inbox record.
176    // implements: CONS-001, CONS-052
177    pub fn consent_state(&self) -> Result<ConsentState, GroupError> {
178        let db = self.context.db();
179        Ok(conversation_consent(
180            &db,
181            self.context.inbox_id(),
182            &self.group_id,
183        )?)
184    }
185
186    /// Set the consent of this conversation, as a decision.
187    #[tracing::instrument(skip_all, level = "trace")]
188    pub fn update_consent_state(&self, state: ConsentState) -> Result<(), GroupError> {
189        crate::state_tx::state_write_with_events(
190            self.context.mls_storage(),
191            self.context.events(),
192            |tx, events| {
193                let storage = tx.storage();
194                let db = storage.db();
195                let group = db
196                    .find_group(&self.group_id)?
197                    .ok_or(NotFound::GroupById(self.group_id))?;
198                let key = ConsentKey::of_group(&group, self.context.inbox_id())
199                    .ok_or(ConsentActError::InvalidEntity)?;
200                let decision = ConsentWrite::Decide(Decision::Set(state));
201                let changes: Vec<_> = write_consent(&db, &key, decision, Owed::Mark)?
202                    .into_iter()
203                    .collect();
204                crate::subscriptions::internal::emit_consent_changes(
205                    &self.context,
206                    events,
207                    &db,
208                    &changes,
209                )?;
210                Ok::<_, GroupError>(Continue(()))
211            },
212        )?
213        .into_continued();
214
215        Ok(())
216    }
217
218    /// Get the current epoch number of the group.
219    pub async fn epoch(&self) -> Result<u64, GroupError> {
220        self.with_group_snapshot(|mls_group| Ok(mls_group.epoch().as_u64()))
221    }
222
223    /// Get the encryption state of the current epoch. Should match for all installations
224    /// in the same epoch.
225    #[cfg(any(test, feature = "test-utils"))]
226    pub async fn epoch_authenticator(&self) -> Result<Vec<u8>, GroupError> {
227        self.with_group_snapshot(|mls_group| {
228            Ok(mls_group.epoch_authenticator().as_slice().to_vec())
229        })
230    }
231
232    pub async fn cursor(&self) -> Result<Cursor, GroupError> {
233        let db = self.context.db();
234        let msgs = db.get_last_cursor(self.group_id, EntityKind::ApplicationMessage)?;
235        Ok(msgs)
236    }
237
238    pub async fn local_commit_log(&self) -> Result<Vec<LocalCommitLog>, GroupError> {
239        Ok(self.context.db().get_group_logs(&self.group_id)?)
240    }
241
242    pub async fn remote_commit_log(&self) -> Result<Vec<RemoteCommitLog>, GroupError> {
243        Ok(self.context.db().get_remote_commit_log_after_cursor(
244            &self.group_id,
245            0,
246            RemoteCommitLogOrder::AscendingByRowid,
247        )?)
248    }
249
250    pub async fn debug_info(&self) -> Result<ConversationDebugInfo, GroupError> {
251        let epoch = self.epoch().await?;
252        let cursor = self.cursor().await?;
253        let commit_log = self.local_commit_log().await?;
254        let remote_commit_log = self.remote_commit_log().await?;
255        let db = self.context.db();
256
257        let stored_group = match db.find_group(&self.group_id)? {
258            Some(group) => group,
259            None => {
260                return Err(GroupError::NotFound(NotFound::GroupById(self.group_id)));
261            }
262        };
263
264        Ok(ConversationDebugInfo {
265            epoch,
266            maybe_forked: stored_group.maybe_forked,
267            fork_details: stored_group.fork_details,
268            is_commit_log_forked: stored_group.is_commit_log_forked,
269            local_commit_log: format!("{:?}", commit_log),
270            remote_commit_log: format!("{:?}", remote_commit_log),
271            cursor: vec![cursor],
272        })
273    }
274
275    /// Update this installation's leaf key in the group by creating a key update commit
276    #[cfg_attr(any(test, feature = "test-utils"), tracing::instrument(level = "info", fields(inbox_id = %self.context.inbox_id()), skip(self)))]
277    #[cfg_attr(
278        not(any(test, feature = "test-utils")),
279        tracing::instrument(level = "trace", skip(self))
280    )]
281    pub async fn key_update(&self) -> Result<(), GroupError> {
282        let intent = QueueIntent::key_update().queue(self)?;
283        let _ = self.sync_until_intent_resolved(intent.id).await?;
284        Ok(())
285    }
286
287    /// Checks if the current user is active in the group.
288    ///
289    /// If the current user has been kicked out of the group, `is_active` will return `false`
290    #[tracing::instrument(skip_all, level = "trace")]
291    pub fn is_active(&self) -> Result<bool, GroupError> {
292        use openmls::group::MlsGroupState;
293        use openmls_traits::storage::StorageProvider as _;
294
295        // Restored groups that are not yet added are inactive
296        let Some(stored_group) = self.context.db().find_group(&self.group_id)? else {
297            return Err(GroupError::NotFound(NotFound::GroupById(self.group_id)));
298        };
299        if matches!(
300            stored_group.membership_state,
301            GroupMembershipState::Restored
302        ) {
303            return Ok(false);
304        }
305
306        state_write(self.context.mls_storage(), |tx| {
307            let state = tx
308                .storage()
309                .group_state::<MlsGroupState, _>(&self.group_id.to_openmls())?
310                .ok_or_else(|| StorageError::from(NotFound::MlsGroup(self.group_id)))?;
311            Ok::<_, GroupError>(Continue(!matches!(state, MlsGroupState::Inactive)))
312        })
313        .map(TransactionOutcome::into_continued)
314    }
315
316    /// Returns the membership state of the current user in this group.
317    #[tracing::instrument(skip_all, level = "trace")]
318    pub fn membership_state(&self) -> Result<GroupMembershipState, GroupError> {
319        let stored_group = self
320            .context
321            .db()
322            .find_group(&self.group_id)?
323            .ok_or_else(|| GroupError::NotFound(NotFound::GroupById(self.group_id)))?;
324        Ok(stored_group.membership_state)
325    }
326
327    /// Get the `GroupMetadata` of the group.
328    ///
329    /// The AppData dictionary contains CONVERSATION_TYPE,
330    /// CREATOR_INBOX_ID, DM_MEMBERS, and ONESHOT_MESSAGE.
331    /// Restored conversations have an empty creator until activation.
332    pub async fn metadata(&self) -> Result<GroupMetadata, GroupError> {
333        let is_restored = self.membership_state()? == GroupMembershipState::Restored;
334        self.with_group_snapshot(|mls_group| {
335            let seed =
336                xmtp_mls_group_state::app_data::component_source::read_group_metadata_from_dict(
337                    mls_group,
338                )
339                .map_err(MetadataPermissionsError::from)?
340                .ok_or_else(|| {
341                    MetadataPermissionsError::from(GroupMetadataError::MissingExtension)
342                })?;
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_group_state::app_data::component_source::ComponentSourceError;
395        let ctx = self.load_group_context()?;
396        xmtp_mls_group_state::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_group_state::app_data::component_source::ComponentSourceError;
417        self.load_mls_group_with_lock(self.context.mls_storage(), |mls_group| {
418            xmtp_mls_group_state::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_group_state::app_data::typed_facade::MlsGroupAppData`] facade to
443    /// read exactly one [`Component`](xmtp_mls_group_state::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_group_state::app_data::typed::Component,
456    {
457        use xmtp_mls_group_state::app_data::component_source::ComponentSourceError;
458        let ctx = self.load_group_context()?;
459        let facade =
460            xmtp_mls_group_state::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>>, GroupError> {
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}
xmtp_mls/groups/state.rs

xmtp_mls/groups/
state.rs
