xmtp_mls/groups/

metadata_fields.rs

//! Typed metadata fields and user data.
//!
//! These methods are the developer-facing view of a group's AppData
//! dictionary. Each call reads one committed dictionary snapshot through
//! [`FieldSnapshot`]; the committed registry decides which fields exist and
//! their policies, the protocol fixes a well-known field's type, and the backend
//! catalogue only labels application fields. Writes are checked against the
//! committed registry's types and policies before they are queued as
//! [`AppDataUpdateIntentData::Fields`], and again when the commit is built,
//! so a write every receiver would reject is never published.

use std::collections::BTreeMap;

use openmls::group::MlsGroup as OpenMlsGroup;
use xmtp_db::{
    Fetch,
    group_intent::{ID, IntentState, StoredGroupIntent},
};
use xmtp_mls_common::{
    app_data::fields::{
        ComponentMutation, FieldError, FieldKey, FieldSnapshot, FieldValue, FieldWrite,
        MetadataFieldDescriptor, MetadataFieldRef, MetadataFieldValue, MetadataValue,
        UserFieldUpdate, UserFieldValue,
    },
    inbox_id::InboxId,
};

use super::{
    GroupError, MlsGroup,
    app_data::sender_intents::field_writes_commit,
    intents::{AppDataUpdateIntentData, QueueIntent},
};
use crate::context::XmtpSharedContext;

impl<Context> MlsGroup<Context>
where
    Context: XmtpSharedContext,
{
    /// The group's metadata fields in component ID order.
    // implements: META-069
    pub fn metadata_fields(&self) -> Result<Vec<MetadataFieldDescriptor>, GroupError> {
        self.read_fields(|fields| Ok(fields.fields().to_vec()))
    }

    /// The field named `name`, preferring a well-known field to an
    /// application field of the same name.
    // implements: META-069
    pub fn metadata_field(
        &self,
        name: &str,
    ) -> Result<Option<MetadataFieldDescriptor>, GroupError> {
        self.read_fields(|fields| Ok(fields.field(name).cloned()))
    }

    /// The value of `field`, `None` when absent.
    // implements: META-070
    pub fn metadata_value(
        &self,
        field: &MetadataFieldRef,
    ) -> Result<Option<MetadataValue>, GroupError> {
        self.read_fields(|fields| fields.value(field))
    }

    /// The values of `fields` in request order, read from one snapshot.
    // implements: META-070
    pub fn metadata_values(
        &self,
        fields: &[MetadataFieldRef],
    ) -> Result<Vec<MetadataFieldValue>, GroupError> {
        self.read_fields(|snapshot| snapshot.values(fields))
    }

    /// The value under `key` of the map `field`.
    // implements: META-070
    pub fn map_value(
        &self,
        field: &MetadataFieldRef,
        key: &FieldKey,
    ) -> Result<Option<FieldValue>, GroupError> {
        self.read_fields(|fields| fields.map_value(field, key))
    }

    /// Per-inbox user field values, read from one snapshot. See
    /// [`FieldSnapshot::user_data`] for the `None` selections.
    // implements: META-072
    pub fn user_data(
        &self,
        fields: Option<&[MetadataFieldRef]>,
        inbox_ids: Option<&[InboxId]>,
    ) -> Result<BTreeMap<InboxId, Vec<UserFieldValue>>, GroupError> {
        self.read_fields(|snapshot| snapshot.user_data(fields, inbox_ids))
    }

    /// Apply `mutation` to `field` in one commit, subject to the committed
    /// registry's type and policies.
    // implements: META-071
    pub async fn update_metadata_field(
        &self,
        field: &MetadataFieldRef,
        mutation: &ComponentMutation,
    ) -> Result<(), GroupError> {
        self.write_fields(|fields| Ok(vec![fields.field_write(field, mutation)?]))
            .await
    }

    /// Set or clear this inbox's own entries of user fields in one commit.
    /// Nothing is committed when `values` is empty or only clears absent
    /// entries.
    // implements: META-073
    pub async fn update_user_data(&self, values: &[UserFieldUpdate]) -> Result<(), GroupError> {
        self.write_fields(|fields| fields.user_data_writes(values))
            .await
    }

    fn read_fields<T>(
        &self,
        read: impl FnOnce(&FieldSnapshot<'_>) -> Result<T, FieldError>,
    ) -> Result<T, GroupError> {
        let context = self.load_group_context()?;
        let dictionary = context
            .extensions()
            .app_data_dictionary()
            .map(|extension| extension.dictionary());
        let catalogue = &self
            .context
            .server_configuration()
            .configuration()
            .application_components;
        Ok(read(&FieldSnapshot::new(dictionary, catalogue)?)?)
    }

    /// Encode writes with `plan` and commit them. The writes are resolved
    /// and authorized once here, so a type error, a denied policy, or a
    /// payload the group would reject is reported to the caller rather than
    /// as a failed intent.
    ///
    /// A write that changes nothing commits nothing. It is judged after a
    /// sync, against the group's committed state at a point during this
    /// call, so success is true at that point, and a commit that lands
    /// later is a later write, as it would be after a commit of ours. The
    /// publisher cannot make this call instead: it publishes before it
    /// receives, so it sees the same local state. A write that pending
    /// proposals carry out but the committed values do not still commits
    /// them, here and when it is published, so success never rests on an
    /// uncommitted value.
    async fn write_fields(
        &self,
        plan: impl FnOnce(&FieldSnapshot<'_>) -> Result<Vec<FieldWrite>, FieldError>,
    ) -> Result<(), GroupError> {
        self.ensure_not_paused().await?;
        self.sync().await?;
        let own = InboxId::from_hex(self.context.inbox_id())
            .map_err(|e| GroupError::ComponentSource(e.into()))?;
        let writes = self.with_group_snapshot(|group: &OpenMlsGroup| {
            let committed = group
                .extensions()
                .app_data_dictionary()
                .map(|extension| extension.dictionary());
            let writes = plan(&FieldSnapshot::new(committed, &[])?)?;
            Ok(field_writes_commit(group, own, &writes)?.map(|_| writes))
        })?;
        let Some(writes) = writes else {
            return Ok(());
        };
        let intent = QueueIntent::app_data_update()
            .data(Vec::<u8>::from(AppDataUpdateIntentData::Fields(
                writes.clone(),
            )))
            .queue(self)?;
        self.publish_field_writes(intent.id, own, &writes).await
    }

    /// Publish `writes`, which `own` queued as `intent_id`. The publisher
    /// resolves them again when it builds their commit, so a commit that
    /// lands first and re-types a field or tightens its policy fails the
    /// intent before it is published. That failure is stored without
    /// details, and another sync or another queued write may be the one
    /// that reports it, so a failed intent is resolved again here and its
    /// own [`FieldError`] returned in place of the sync error. A write that
    /// is still queued keeps the sync error: it has not failed yet.
    pub(super) async fn publish_field_writes(
        &self,
        intent_id: ID,
        own: InboxId,
        writes: &[FieldWrite],
    ) -> Result<(), GroupError> {
        let Err(error) = self.sync_until_intent_resolved(intent_id).await else {
            return Ok(());
        };
        let failed = matches!(
            self.context.db().fetch(&intent_id),
            Ok(Some(StoredGroupIntent {
                state: IntentState::Error,
                ..
            }))
        );
        if failed
            && let Err(field @ GroupError::MetadataField(_)) =
                self.with_group_snapshot(|group| field_writes_commit(group, own, writes))
        {
            return Err(field);
        }
        Err(error)
    }
}