xmtp_mls_common/app_data/

component_source.rs

//! Single source-of-truth for the per-`ComponentId` read/encode/apply logic.
//!
//! Centralizes everything the rest of the MLS pipeline needs to know about
//! a well-known `ComponentId`: its logical [`ComponentType`], where its
//! current bytes live (OpenMLS AppData dictionary vs. legacy group context
//! extensions), and how to encode/apply `AppDataUpdate` payloads.
//!
//! The helpers are `pub` so `xmtp_mls` can call them. They read and write
//! raw component bytes; `xmtp_mls` owns the intent and commit paths that
//! decide when a write is allowed.
//!
//! ## Field bounds
//!
//! [`apply_app_data_update_payload`] rejects a new value of an application
//! component, `USER_DISPLAY_NAME`, or `GROUP_IMAGE` whose scalar,
//! collection key, or collection value exceeds [`MAX_FIELD_ELEMENT_BYTES`],
//! or whose serialized collection exceeds [`MAX_FIELD_SNAPSHOT_BYTES`].
//! [`check_update_payload_bounds`] applies the element bound to every
//! element an update payload names, without the state it applies to.
//!
//! ## Inbox-id encoding
//!
//! The legacy GMM extension stores inbox ids as 64-character hex strings.
//! Anything serialized through the new `AppDataUpdate` path uses the
//! versioned [`InboxId`] newtype.
//!
//! See [`crate::inbox_id`] for the full wire-format contract;
//! the short version is `varint(version) || 32-byte payload`, with
//! version 0 producing a 33-byte encoding.

use crate::{
    app_data::{
        component_id::ComponentId,
        component_registry::ComponentRegistry,
        components::type_dispatch::{apply_update_payload_for_type, expand_to_changes_for_type},
        registry_table::lookup_component,
        typed::ComponentTypedError,
    },
    group_mutable_metadata::{GroupMutableMetadata, GroupMutableMetadataError, MetadataField},
    inbox_id::{InboxId, InboxIdError},
    tls_map::{TlsMap, TlsMapDelta, TlsMapError, TlsMapMutation},
    tls_set::{TlsSet, TlsSetDelta, TlsSetError, TlsSetMutation},
};
use openmls::{
    extensions::Extensions,
    group::{GroupContext, MlsGroup as OpenMlsGroup, StagedCommit},
    messages::proposals::AppDataUpdateOperation,
};
use tls_codec::{Deserialize, Serialize, VLBytes};
use xmtp_proto::xmtp::mls::message_contents::ComponentType;

/// Longest scalar, collection key, or collection value that an application
/// component, `USER_DISPLAY_NAME`, `GROUP_IMAGE`, or `COMPONENT_REGISTRY` may
/// hold.
pub const MAX_FIELD_ELEMENT_BYTES: usize = 8192;

/// Longest serialized collection snapshot that an application component,
/// `USER_DISPLAY_NAME`, `GROUP_IMAGE`, or `COMPONENT_REGISTRY` may hold.
pub const MAX_FIELD_SNAPSHOT_BYTES: usize = 65536;

/// Errors surfaced by the component_source layer.
///
/// `xmtp_mls` embeds this in `GroupError` via `#[from]` for the
/// AppDataUpdate path.
#[derive(Debug, thiserror::Error)]
pub enum ComponentSourceError {
    /// The component id is outside the well-known XMTP range.
    #[error("unknown component {0}")]
    UnknownComponent(ComponentId),

    /// The component is known but its wiring hasn't been built yet.
    #[error("component {0} wiring is not yet implemented")]
    NotImplemented(ComponentId),

    /// An `AppDataUpdate::Update` write was attempted against an immutable
    /// component. Insert-once writes should be expressed as `Insert`, not
    /// caught here.
    #[error("component {0} is immutable and cannot be updated via AppDataUpdate")]
    ImmutableUpdate(ComponentId),

    /// A mutation does not match the component type of the component it
    /// targets (e.g. a scalar write to `ADMIN_LIST`).
    #[error("mutation shape does not match component {0}")]
    MismatchedMutation(ComponentId),

    /// Failed to convert an inbox id string or byte slice into an
    /// [`InboxId`]. Wraps [`InboxIdError`] — callers that need to
    /// distinguish "not hex" from "wrong length" can match the inner
    /// variant.
    #[error("invalid inbox id: {0}")]
    InvalidInboxId(#[from] InboxIdError),

    /// A wire-format violation on a component value: the bytes stored in
    /// the AppData dictionary for a known component don't decode under the
    /// expected encoding (e.g. non-UTF-8 bytes for a `Bytes`-typed
    /// metadata attribute, malformed `TlsSet` for a collection component).
    #[error("malformed value for component {component_id}: {reason}")]
    MalformedComponentValue {
        /// The component whose stored bytes failed to decode.
        component_id: ComponentId,
        /// Human-readable reason — surface to logs, not user-facing.
        reason: String,
    },

    /// A `MetadataUpdate` intent referenced a metadata field name that has
    /// no corresponding `ComponentId`. Most commonly fires when a future
    /// metadata field is added to one of the senders without also being
    /// added to [`metadata_field_to_component_id`].
    #[error("unknown metadata field name: {0}")]
    UnknownMetadataField(String),

    /// Failed to read, decode, or encode the legacy group mutable metadata
    /// extension while servicing a component-source request.
    #[error(transparent)]
    GroupMutableMetadata(#[from] GroupMutableMetadataError),

    /// A TLS-codec operation on a delta or stored collection value failed.
    #[error("tls codec error: {0}")]
    TlsCodec(#[from] tls_codec::Error),

    /// A `TlsSet::apply_delta` call failed while synthesizing the new full
    /// value of a collection component from an incoming delta.
    #[error("tls set apply error: {0}")]
    TlsSetApply(#[from] TlsSetError),

    /// A `TlsMap::apply_delta` call failed while synthesizing the new full
    /// value of a map component from an incoming delta.
    #[error("tls map apply error: {0}")]
    TlsMapApply(#[from] TlsMapError),

    /// An update would leave a bounded component holding a scalar,
    /// collection element, or collection snapshot longer than its limit
    /// ([`MAX_FIELD_ELEMENT_BYTES`] or [`MAX_FIELD_SNAPSHOT_BYTES`]).
    #[error("component {component_id} holds {len} bytes, over the {max}-byte bound")]
    FieldBoundExceeded {
        /// The component whose new value is too large.
        component_id: ComponentId,
        /// Length of the offending scalar, element, or snapshot.
        len: usize,
        /// The bound it exceeds.
        max: usize,
    },
}

impl ComponentSourceError {
    /// Best-effort `ComponentId` extraction for the variants that carry
    /// one — so error-mapping shims can preserve structured context
    /// across the crate boundary into
    /// [`GroupMutableMetadataError::MalformedComponent`] without
    /// stringifying.
    pub fn component_id(&self) -> Option<ComponentId> {
        match self {
            Self::UnknownComponent(id)
            | Self::NotImplemented(id)
            | Self::ImmutableUpdate(id)
            | Self::MismatchedMutation(id)
            | Self::MalformedComponentValue {
                component_id: id, ..
            }
            | Self::FieldBoundExceeded {
                component_id: id, ..
            } => Some(*id),
            _ => None,
        }
    }
}

impl From<ComponentTypedError> for ComponentSourceError {
    /// Surface trait-layer errors at the dispatch boundary. The
    /// dispatch layer adds `UnknownComponent` / `NotImplemented` /
    /// `UnknownMetadataField` / `GroupMutableMetadata` for things the
    /// trait can't see; the variants below are the trait's domain
    /// and round-trip 1:1.
    fn from(err: ComponentTypedError) -> Self {
        match err {
            ComponentTypedError::ImmutableUpdate(id) => Self::ImmutableUpdate(id),
            ComponentTypedError::MismatchedMutation(id) => Self::MismatchedMutation(id),
            ComponentTypedError::MalformedValue {
                component_id,
                reason,
            } => Self::MalformedComponentValue {
                component_id,
                reason,
            },
            ComponentTypedError::InvalidInboxId(e) => Self::InvalidInboxId(e),
            ComponentTypedError::TlsCodec(e) => Self::TlsCodec(e),
            ComponentTypedError::TlsSetApply(e) => Self::TlsSetApply(e),
            ComponentTypedError::TlsMapApply(e) => Self::TlsMapApply(e),
            ComponentTypedError::UnspecifiedType(id) => Self::MalformedComponentValue {
                component_id: id,
                reason: "registered ComponentType is Unspecified".to_string(),
            },
            // A COMPONENT_REGISTRY delta that violates the registry's
            // write invariants (reserved/hardcoded/out-of-space id,
            // immutable overwrite, undecodable metadata). Structurally
            // a malformed value for the registry component; the reason
            // string carries the specific violation.
            ComponentTypedError::RegistryMutation(e) => Self::MalformedComponentValue {
                component_id: ComponentId::COMPONENT_REGISTRY,
                reason: e.to_string(),
            },
        }
    }
}

impl From<ComponentSourceError> for GroupMutableMetadataError {
    /// Preserve structure where possible. If the source already wraps a
    /// `GroupMutableMetadataError`, unwrap and return that inner variant
    /// verbatim so callers can
    /// match on `MissingExtension` / `MissingMetadataField` / etc.
    ///
    /// For every other variant, surface as `MalformedComponent` and
    /// preserve the offending `component_id` when it's available so
    /// downstream consumers (bindings, error-mapping) can match
    /// structurally on it. Variants without one surface as
    /// `component_id: None`; the display string stays the
    /// authoritative diagnostic.
    fn from(err: ComponentSourceError) -> Self {
        if let ComponentSourceError::GroupMutableMetadata(inner) = err {
            return inner;
        }
        let component_id = err.component_id();
        GroupMutableMetadataError::MalformedComponent {
            component_id,
            reason: err.to_string(),
        }
    }
}

/// Hardcoded logical type of a well-known component. Returns `None` for
/// app-range components (`0xC000-0xFEFF`) and for any well-known id that
/// is not yet wired into this match.
pub fn component_type(id: ComponentId) -> Option<ComponentType> {
    match id {
        // Hardcoded registry / list components. ComponentRegistry itself is a
        // TlsMap, but permissions are enforced in code — it never flows
        // through this module.
        ComponentId::COMPONENT_REGISTRY => Some(ComponentType::TlsMapBytesBytes),
        ComponentId::SUPER_ADMIN_LIST => Some(ComponentType::TlsSetInboxId),
        ComponentId::ADMIN_LIST => Some(ComponentType::TlsSetInboxId),

        // GroupMembership — TlsMap<InboxId, bytes>
        ComponentId::GROUP_MEMBERSHIP => Some(ComponentType::TlsMapInboxIdBytes),
        ComponentId::USER_DISPLAY_NAME => Some(ComponentType::TlsMapInboxIdString),

        // GroupMutableMetadata-backed string components.
        ComponentId::GROUP_NAME
        | ComponentId::GROUP_DESCRIPTION
        | ComponentId::GROUP_IMAGE_URL
        | ComponentId::APP_DATA
        | ComponentId::MIN_SUPPORTED_PROTOCOL_VERSION => Some(ComponentType::String),

        // GroupMutableMetadata-backed bytes components.
        ComponentId::MESSAGE_DISAPPEAR_FROM_NS
        | ComponentId::MESSAGE_DISAPPEAR_IN_NS
        | ComponentId::COMMIT_LOG_SIGNER
        | ComponentId::GROUP_IMAGE => Some(ComponentType::Bytes),

        // Immutable metadata (not flowable through AppDataUpdate writes,
        // but we still advertise the type for completeness).
        ComponentId::CONVERSATION_TYPE
        | ComponentId::CREATOR_INBOX_ID
        | ComponentId::ONESHOT_MESSAGE => Some(ComponentType::Bytes),
        ComponentId::DM_MEMBERS => Some(ComponentType::TlsSetInboxId),

        _ => None,
    }
}

use crate::group_mutable_metadata::METADATA_FIELD_COMPONENT_MAP;

/// Map a [`MetadataField`] string to its corresponding `ComponentId`.
///
/// Returns `None` for unknown field names so this can also be called with a
/// raw string coming from a legacy intent payload.
pub fn metadata_field_to_component_id(field_name: &str) -> Option<ComponentId> {
    METADATA_FIELD_COMPONENT_MAP
        .iter()
        .find(|(field, _)| field.as_str() == field_name)
        .map(|(_, id)| *id)
}

/// Map a `ComponentId` back to the `MetadataField` attribute name that stores
/// it in the legacy `GroupMutableMetadata` extension.
///
/// Returns `None` for component ids that are not backed by a
/// `GroupMutableMetadata` attribute (e.g. `ADMIN_LIST`, `GROUP_MEMBERSHIP`,
/// or anything outside the mutable metadata family).
pub fn component_id_to_metadata_field(id: ComponentId) -> Option<MetadataField> {
    METADATA_FIELD_COMPONENT_MAP
        .iter()
        .find(|(_, component_id)| *component_id == id)
        .map(|(field, _)| *field)
}

/// Read the component's current bytes from the OpenMLS AppData dictionary.
pub fn read_component_bytes(
    id: ComponentId,
    extensions: &Extensions<GroupContext>,
) -> Result<Option<Vec<u8>>, ComponentSourceError> {
    Ok(read_from_app_data_dict_from_extensions(id, extensions))
}

/// Compute the post-commit value of a single component
/// by overlaying the staged commit's `AppDataUpdate` proposals on top of
/// the pre-commit dict. Last-write-wins matches the lazy-batching apply
/// order in `xmtp_mls`'s `accumulate_app_data_updates`: every `Update(payload)`
/// is decoded against the running value (so collection deltas compose),
/// and `Remove` collapses to `None`.
///
/// Returns `Ok(None)` when the component is absent both before and after
/// the commit, or when it was explicitly removed. Returns `Err` only when
/// an `Update` payload fails to decode against the running value — the
/// same condition `validate_app_data_update_proposals_in_commit` would
/// also reject upstream, so callers can treat decode failure here as
/// "validator will surface the real error" and short-circuit.
///
/// Used by the commit validator to evaluate per-component invariants
/// (notably `MIN_SUPPORTED_PROTOCOL_VERSION`) that need the post-commit
/// view from the AppData dictionary.
///
/// # Registry semantics
///
/// `registry` is the **pre-commit** `COMPONENT_REGISTRY` (i.e. the state
/// of the dictionary entry before the staged commit applies). Callers
/// should load it once via `xmtp_mls`'s `load_component_registry` on the
/// live `mls_group` and reuse it across all validator helpers — same
/// registry feeds `validate_app_data_update_proposals_in_commit`
/// and any other per-component checks.
///
/// **Implication for commits that modify `COMPONENT_REGISTRY` in the
/// same commit as a write to a newly-registered component**: such a
/// write fails with `UnknownComponent` here because the new entry is
/// not yet visible in the pre-commit registry. This matches what the
/// receiver-side validator
/// (`validate_app_data_update_proposals_in_commit`) enforces
/// today and is the documented convention across the commit
/// path: registry mutations and writes that depend on those mutations
/// MUST land in separate commits.
///
pub fn read_post_commit_component_bytes(
    id: ComponentId,
    mls_group: &OpenMlsGroup,
    staged_commit: &StagedCommit,
    registry: &ComponentRegistry,
) -> Result<Option<Vec<u8>>, ComponentSourceError> {
    let openmls_id: openmls::component::ComponentId = id.as_u16();

    // Owned snapshot of operations targeting this specific component.
    // Iterating `app_data_update_proposals()` yields short-lived
    // `QueuedAppDataUpdateProposal` views that borrow into the staged
    // commit — we can't hold their byte slices across iterations, so
    // we materialize an owned form up front. `Update` payloads are
    // typically tiny (version strings, single-key deltas), so the
    // clone cost is negligible.
    enum Op {
        Update(Vec<u8>),
        Remove,
    }
    let ops: Vec<Op> = staged_commit
        .app_data_update_proposals()
        .filter_map(|queued| {
            let proposal = queued.app_data_update_proposal();
            if proposal.component_id() != openmls_id {
                return None;
            }
            Some(match proposal.operation() {
                AppDataUpdateOperation::Update(payload) => Op::Update(payload.as_slice().to_vec()),
                AppDataUpdateOperation::Remove => Op::Remove,
            })
        })
        .collect();
    if ops.is_empty() {
        return Ok(read_from_app_data_dict(id, mls_group));
    }

    let mut current = read_from_app_data_dict(id, mls_group);
    for op in &ops {
        match op {
            Op::Update(payload) => {
                current = Some(apply_app_data_update_payload(
                    id,
                    payload,
                    current.as_deref(),
                    registry,
                )?);
            }
            Op::Remove => current = None,
        }
    }
    Ok(current)
}

/// Look up the component's bytes in the OpenMLS AppData dictionary.
///
/// The commit validator in `xmtp_mls` pulls the pre-commit stored bytes
/// for a component and threads them into
/// [`expand_app_data_update_to_changes`] as `old_value` — the validator
/// uses that to resolve `RemoveByHash` mutations back to their underlying
/// inbox id. `xmtp_mls`'s `app_data` module also uses it from
/// `process_message_with_app_data`, `stage_app_data_propose_and_commit`,
/// and `pending_app_data_updates`.
pub fn read_from_app_data_dict(id: ComponentId, mls_group: &OpenMlsGroup) -> Option<Vec<u8>> {
    read_from_app_data_dict_from_extensions(id, mls_group.extensions())
}

/// Extensions-only counterpart of [`read_from_app_data_dict`] — reads the
/// component's bytes straight from a group's `GroupContext` extensions, with no
/// full `OpenMlsGroup`. openmls keys the dictionary by its own `ComponentId`,
/// which is just a `u16` alias, so `id.as_u16()` unwraps our newtype to the key.
pub fn read_from_app_data_dict_from_extensions(
    id: ComponentId,
    extensions: &Extensions<GroupContext>,
) -> Option<Vec<u8>> {
    extensions
        .app_data_dictionary()
        .and_then(|ext| ext.dictionary().get(&id.as_u16()))
        .map(|bytes| bytes.to_vec())
}

use crate::app_data::typed::ExpandedComponentChange;

/// Expand an `AppDataUpdate` proposal payload into the per-element changes
/// that should be checked against the component registry.
///
/// - `Bytes` components: returns a single `Update` change with the new
///   payload bytes.
/// - Collection components (`ADMIN_LIST` / `SUPER_ADMIN_LIST`): parses the
///   payload as a `TlsSetDelta<InboxId>` and emits one entry per
///   mutation, with `op = Insert` for `Insert`, `op = Delete` for
///   `Remove` / `RemoveByHash`.
/// - `AppDataUpdateOperation::Remove` (any component): a single
///   `Delete` entry with no value.
///
/// `old_value` is the component's pre-commit stored bytes (from the
/// AppData dictionary). It's only consulted for `RemoveByHash`
/// resolution on collection components: given the prior `TlsSet<InboxId>`,
/// we build a `hash → InboxId` index and resolve each `RemoveByHash` back
/// to the concrete key being removed so the validator sees the inbox id
/// the peer is targeting. If the hash doesn't match any prior key (or
/// `old_value` is `None`), the expansion surfaces `value: None` and the
/// subsequent CRDT apply step surfaces the real error.
///
/// Used on the receiver side to feed `validate_component_write` for each
/// distinct change inside a single `AppDataUpdate` proposal.
///
/// The steady-state validator dispatches through `lookup_component`
/// directly so it can also call `Component::validate_invariant`
/// without a second binary search. This wrapper is retained for
/// callers that don't need the invariant hook.
pub fn expand_app_data_update_to_changes(
    component_id: ComponentId,
    operation: &AppDataUpdateOperation,
    old_value: Option<&[u8]>,
    registry: &ComponentRegistry,
) -> Result<Vec<ExpandedComponentChange>, ComponentSourceError> {
    if let Some(component) = lookup_component(component_id) {
        return component
            .expand_to_changes(operation, old_value)
            .map_err(Into::into);
    }

    // No per-id [`Component`] impl on this client. The closed type
    // universe (7 variants) means every shape — including `TlsSet` /
    // `TlsMap` deltas — surfaces a proper per-element change list to the
    // validator. Old and new clients converge on the same dict state for
    // the same wire bytes.
    let ty = resolved_type(component_id, registry)?;
    expand_to_changes_for_type(component_id, ty, operation, old_value).map_err(Into::into)
}

/// Decode an incoming `AppDataUpdateOperation::Update(bytes)` payload
/// and produce the new full bytes of the component, given the prior
/// stored bytes (if any). `Update`-only — `Remove` carries no payload
/// and is handled directly by the caller.
///
/// Immutable components are rejected with
/// [`ComponentSourceError::ImmutableUpdate`] **only when a prior
/// value already exists** — the bootstrap commit is the canonical
/// first-insert path for immutable seeds, so this layer must allow
/// an `Update` whose `old_value` is `None`. The bootstrap validator
/// catches malicious initial values upstream via byte-compare.
///
/// A bounded component's payload is held to the element bound before
/// anything else, so the result does not depend on the state it applies
/// to; the new value is then held to the snapshot bound.
pub fn apply_app_data_update_payload(
    id: ComponentId,
    payload: &[u8],
    old_value: Option<&[u8]>,
    registry: &ComponentRegistry,
) -> Result<Vec<u8>, ComponentSourceError> {
    check_update_payload_bounds(id, payload, registry)?;
    // Immutability gate. Reject only on overwrite — a fresh insert
    // (no prior value) is the bootstrap commit's first write of an
    // immutable seed and must succeed for honest receivers to reach
    // the migrated state. Steady-state immutables always have a prior
    // (inserted at bootstrap), so a Byzantine peer trying to mutate
    // them post-bootstrap still hits this branch and gets rejected.
    if id.is_immutable() && old_value.is_some() {
        return Err(ComponentSourceError::ImmutableUpdate(id));
    }

    // A per-id `Component` impl handles the well-known mutable components
    // with a typed decoder; every other id dispatches on its resolved type.
    let new_value = match lookup_component(id) {
        Some(component) => component.apply_update_payload(payload, old_value)?,
        None => {
            apply_update_payload_for_type(id, resolved_type(id, registry)?, payload, old_value)?
        }
    };
    if is_field_bounded(id) {
        check_field_bounds(id, resolved_type(id, registry)?, &new_value)?;
    }
    Ok(new_value)
}

/// The type of component `id`, from two sources tried in order:
///
/// 1. In-code [`component_type`] mapping — covers well-known XMTP ids,
///    including those with no typed decoder (immutable seeds like
///    CREATOR_INBOX_ID — bootstrap-only first-write path).
/// 2. On-dict [`ComponentRegistry`] entry — covers components a *newer*
///    release ships that this client has never heard of; the registry's
///    `component_type` tag is the type oracle.
fn resolved_type(
    id: ComponentId,
    registry: &ComponentRegistry,
) -> Result<ComponentType, ComponentSourceError> {
    component_type(id).map_or_else(|| registered_component_type(id, registry), Ok)
}

/// Whether the field bounds cover component `id`: application components,
/// `USER_DISPLAY_NAME`, `GROUP_IMAGE`, and `COMPONENT_REGISTRY`, whose
/// application-range entries a DM participant may write.
fn is_field_bounded(id: ComponentId) -> bool {
    id.is_app_range()
        || [
            ComponentId::USER_DISPLAY_NAME,
            ComponentId::GROUP_IMAGE,
            ComponentId::COMPONENT_REGISTRY,
        ]
        .contains(&id)
}

/// Reject an `AppDataUpdate::Update` payload for component `id` that names
/// a scalar, collection key, or collection value longer than
/// [`MAX_FIELD_ELEMENT_BYTES`], including one that a later mutation of the
/// same payload removes and the key of a removal. It reads only the payload,
/// so it holds whatever state the update is applied to.
// implements: META-068
fn check_update_payload_bounds(
    id: ComponentId,
    payload: &[u8],
    registry: &ComponentRegistry,
) -> Result<(), ComponentSourceError> {
    if !is_field_bounded(id) {
        return Ok(());
    }
    // A removal by hash names no key bytes.
    let longest = match resolved_type(id, registry)? {
        ComponentType::Bytes | ComponentType::String => Some(payload.len()),
        ComponentType::TlsSetBytes => TlsSetDelta::<VLBytes>::tls_deserialize_exact(payload)?
            .mutations
            .iter()
            .map(|m| match m {
                TlsSetMutation::Insert(key) | TlsSetMutation::Remove(key) => key.as_slice().len(),
                TlsSetMutation::RemoveByHash(_) => 0,
            })
            .max(),
        ComponentType::TlsMapInboxIdBytes | ComponentType::TlsMapInboxIdString => {
            longest_delta_value::<InboxId>(payload)?
        }
        ComponentType::TlsMapBytesBytes if id == ComponentId::COMPONENT_REGISTRY => {
            longest_delta_value::<ComponentId>(payload)?
        }
        ComponentType::TlsMapBytesBytes => {
            TlsMapDelta::<VLBytes, VLBytes>::tls_deserialize_exact(payload)?
                .mutations
                .iter()
                .map(|m| match m {
                    TlsMapMutation::Insert { key, value }
                    | TlsMapMutation::Update { key, value } => {
                        key.as_slice().len().max(value.as_slice().len())
                    }
                    TlsMapMutation::Delete { key } => key.as_slice().len(),
                })
                .max()
        }
        ComponentType::TlsSetInboxId | ComponentType::Unspecified => None,
    };
    check_element_bound(id, longest.unwrap_or(0))
}

/// The longest value that a map delta with fixed-size `K` keys inserts or
/// updates.
fn longest_delta_value<K>(payload: &[u8]) -> Result<Option<usize>, tls_codec::Error>
where
    TlsMapDelta<K, VLBytes>: Deserialize,
{
    Ok(TlsMapDelta::<K, VLBytes>::tls_deserialize_exact(payload)?
        .mutations
        .iter()
        .map(|m| match m {
            TlsMapMutation::Insert { value, .. } | TlsMapMutation::Update { value, .. } => {
                value.as_slice().len()
            }
            TlsMapMutation::Delete { .. } => 0,
        })
        .max())
}

/// Reject `value`, the new bytes of component `id` of type `ty`, when a
/// scalar, collection key, or collection value is longer than
/// [`MAX_FIELD_ELEMENT_BYTES`] or a collection snapshot is longer than
/// [`MAX_FIELD_SNAPSHOT_BYTES`]. `InboxId` and `ComponentId` keys have a
/// fixed size, so only byte keys and values are measured element by element.
// implements: META-068
fn check_field_bounds(
    id: ComponentId,
    ty: ComponentType,
    value: &[u8],
) -> Result<(), ComponentSourceError> {
    let longest_element = match ty {
        ComponentType::Bytes | ComponentType::String => Some(value.len()),
        _ if value.len() > MAX_FIELD_SNAPSHOT_BYTES => {
            return Err(ComponentSourceError::FieldBoundExceeded {
                component_id: id,
                len: value.len(),
                max: MAX_FIELD_SNAPSHOT_BYTES,
            });
        }
        ComponentType::TlsSetBytes => TlsSet::<VLBytes>::tls_deserialize_exact(value)?
            .iter()
            .map(|v| v.as_slice().len())
            .max(),
        ComponentType::TlsMapInboxIdBytes | ComponentType::TlsMapInboxIdString => {
            longest_map_value::<InboxId>(value)?
        }
        ComponentType::TlsMapBytesBytes if id == ComponentId::COMPONENT_REGISTRY => {
            longest_map_value::<ComponentId>(value)?
        }
        ComponentType::TlsMapBytesBytes => {
            TlsMap::<VLBytes, VLBytes>::tls_deserialize_exact(value)?
                .iter()
                .flat_map(|(k, v)| [k.as_slice().len(), v.as_slice().len()])
                .max()
        }
        ComponentType::TlsSetInboxId | ComponentType::Unspecified => None,
    }
    .unwrap_or(0);
    check_element_bound(id, longest_element)
}

/// The longest value of a map with fixed-size `K` keys.
fn longest_map_value<K: Ord>(value: &[u8]) -> Result<Option<usize>, tls_codec::Error>
where
    TlsMap<K, VLBytes>: Deserialize,
{
    Ok(TlsMap::<K, VLBytes>::tls_deserialize_exact(value)?
        .values()
        .map(|v| v.as_slice().len())
        .max())
}

/// Reject `len`, the longest scalar, key, or value written to component
/// `id`, when it is over [`MAX_FIELD_ELEMENT_BYTES`].
fn check_element_bound(id: ComponentId, len: usize) -> Result<(), ComponentSourceError> {
    if len > MAX_FIELD_ELEMENT_BYTES {
        return Err(ComponentSourceError::FieldBoundExceeded {
            component_id: id,
            len,
            max: MAX_FIELD_ELEMENT_BYTES,
        });
    }
    Ok(())
}

/// Look up the [`ComponentType`] registered for a component id in the
/// on-dict [`ComponentRegistry`]. Returns
/// [`ComponentSourceError::UnknownComponent`] when no registry entry
/// exists — the deny-by-default rule that keeps unrecognized payloads
/// from being applied opaquely.
fn registered_component_type(
    id: ComponentId,
    registry: &ComponentRegistry,
) -> Result<ComponentType, ComponentSourceError> {
    let meta = registry
        .get(&id)
        .map_err(|e| ComponentSourceError::MalformedComponentValue {
            component_id: id,
            reason: format!("registry lookup: {e}"),
        })?
        .ok_or(ComponentSourceError::UnknownComponent(id))?;
    ComponentType::try_from(meta.component_type).map_err(|_| {
        ComponentSourceError::MalformedComponentValue {
            component_id: id,
            reason: format!(
                "registry entry has unknown component_type tag {}",
                meta.component_type
            ),
        }
    })
}

/// Overlay AppData-dict component values onto a base [`GroupMutableMetadata`].
///
/// Wire formats (must match what the sender emits via
/// [`apply_app_data_update_payload`]):
/// - Bytes components: raw UTF-8 string bytes.
/// - `ADMIN_LIST` / `SUPER_ADMIN_LIST`: TLS-serialized `TlsSet<InboxId>`,
///   each id hex-encoded back to string form.
///
/// ## Independence from `COMPONENT_REGISTRY` parseability
///
/// This function reads metadata field entries directly from the dict and
/// **never** loads or validates the `COMPONENT_REGISTRY` payload — it
/// does not use the `COMPONENT_REGISTRY` payload as a gate. So a malformed
/// `COMPONENT_REGISTRY` blob does NOT cause
/// metadata reads to drop authoritative data: as long as the individual
/// metadata field bytes (`GROUP_NAME`, `ADMIN_LIST`, …) decode
/// correctly, they round-trip into the returned GMM. Registry corruption
/// is surfaced loudly on the *write* paths instead — the sender gate in
/// `mls_sync.rs` and the commit validator in `validated_commit.rs` both
/// call `xmtp_mls`'s `load_component_registry` and propagate decode errors
/// — so a corrupt registry blocks state changes without making readable
/// data unreachable. See
/// `merge_with_malformed_registry_returns_valid_field` for the test
/// that pins this invariant.
pub fn merge_app_data_into_mutable_metadata(
    base: &mut GroupMutableMetadata,
    mls_group: &OpenMlsGroup,
) -> Result<(), ComponentSourceError> {
    merge_app_data_into_mutable_metadata_from_extensions(base, mls_group.extensions())
}

/// Extract [`GroupMutableMetadata`] from the AppData dictionary.
pub fn extract_group_mutable_metadata_capability_aware(
    mls_group: &OpenMlsGroup,
) -> Result<GroupMutableMetadata, ComponentSourceError> {
    let mut base =
        GroupMutableMetadata::new(std::collections::HashMap::new(), Vec::new(), Vec::new());
    merge_app_data_into_mutable_metadata(&mut base, mls_group)?;
    Ok(base)
}

/// Same as [`extract_group_mutable_metadata_capability_aware`], but driven from
/// the group's `GroupContext` extensions alone — no full `OpenMlsGroup::load`.
/// The mutable metadata lives entirely in the context extensions, so a single
/// `StorageProvider::group_context` read (one KV round-trip, no ratchet tree)
/// is all this needs.
pub fn extract_group_mutable_metadata_capability_aware_from_extensions(
    extensions: &Extensions<GroupContext>,
) -> Result<GroupMutableMetadata, ComponentSourceError> {
    let mut base =
        GroupMutableMetadata::new(std::collections::HashMap::new(), Vec::new(), Vec::new());
    merge_app_data_into_mutable_metadata_from_extensions(&mut base, extensions)?;
    Ok(base)
}

/// Extensions-only variant of [`merge_app_data_into_mutable_metadata`].
pub fn merge_app_data_into_mutable_metadata_from_extensions(
    base: &mut GroupMutableMetadata,
    extensions: &openmls::extensions::Extensions<openmls::group::GroupContext>,
) -> Result<(), ComponentSourceError> {
    // The merge body lives in `xmtp_mls_common` so crates below
    // `xmtp_mls` in the dependency graph (the archive exporter) can
    // reuse it. Map the per-component error back to
    // `MalformedComponentValue` so this function's error shape (which
    // callers and tests match on, and `component_id()` extracts from)
    // is unchanged by the move.
    crate::group_mutable_metadata::merge_dict_into_mutable_metadata(base, extensions).map_err(|e| {
        match e {
            GroupMutableMetadataError::MalformedComponent {
                component_id: Some(component_id),
                reason,
            } => ComponentSourceError::MalformedComponentValue {
                component_id,
                reason,
            },
            other => ComponentSourceError::GroupMutableMetadata(other),
        }
    })
}

// ============================================================================
// Inbox-id encoding helpers
// ============================================================================
//
// Inbox ids are SHA-256 hashes (see `xmtp_id::associations::member::inbox_id`).
// Their canonical string form is a 64-character hex string. Anything we put
// on the wire through the new `AppDataUpdate` path uses the
// versioned `InboxId` newtype instead — see the module-level docs for
// the rationale and `crate::inbox_id` for the full contract.

/// Decode a hex-string inbox id into an [`InboxId`].
///
/// Returns [`ComponentSourceError::InvalidInboxId`] wrapping either
/// [`InboxIdError::InvalidHex`] (input wasn't hex) or
/// [`InboxIdError::InvalidLength`] (wrong byte length after decoding).
/// Callers that need to distinguish the failure modes can match the
/// inner variant.
pub fn inbox_id_str_to_bytes(inbox_id: &str) -> Result<InboxId, ComponentSourceError> {
    InboxId::from_hex(inbox_id).map_err(Into::into)
}

/// Read the super-admin list from the AppData dictionary.
pub fn read_super_admin_list_from_dict(
    mls_group: &OpenMlsGroup,
) -> Result<Option<Vec<String>>, ComponentSourceError> {
    read_super_admin_list_from_extensions(mls_group.extensions())
}

/// Extensions-only variant of [`read_super_admin_list_from_dict`]. Use
/// the shim above when an `OpenMlsGroup` is at hand; this form is
/// available primarily for unit testing and for commit-validation
/// paths that only carry an `Extensions` reference.
pub fn read_super_admin_list_from_extensions(
    extensions: &Extensions<GroupContext>,
) -> Result<Option<Vec<String>>, ComponentSourceError> {
    let Some(ext) = extensions.app_data_dictionary() else {
        return Ok(None);
    };
    let Some(bytes) = ext
        .dictionary()
        .get(&ComponentId::SUPER_ADMIN_LIST.as_u16())
    else {
        return Ok(None);
    };
    let set = TlsSet::<InboxId>::tls_deserialize_exact(bytes).map_err(|e| {
        ComponentSourceError::MalformedComponentValue {
            component_id: ComponentId::SUPER_ADMIN_LIST,
            reason: format!("invalid TlsSet<InboxId>: {e}"),
        }
    })?;
    Ok(Some(set.iter().map(|id| id.to_hex()).collect()))
}

/// Synthesize a [`GroupMetadata`] from the AppData dictionary.
/// Returns `Ok(None)` if the critical immutable seeds are absent.
///
/// Encoding mirrors the sender-side synthesis in
/// [`crate::app_data::migration::synthesize_canonical_subset_for_validation`]:
/// - `CONVERSATION_TYPE`: 4 big-endian bytes of `ConversationType as i32`
///   (see `encode_conversation_type` there).
/// - `CREATOR_INBOX_ID`: the versioned `InboxId` TLS wire form
///   (`varint(version) || 32-byte payload`) — the same shape every
///   other inbox-id-bearing component on the new path uses. Reader
///   hex-encodes the decoded id back into the legacy
///   `GroupMetadata::creator_inbox_id: String` slot.
/// - `DM_MEMBERS`: `TlsSet<InboxId>` with exactly two elements —
///   matches the declared `ComponentType::TlsSetInboxId` and the
///   sender's `encode_dm_members`. The writer rejects self-DMs
///   (identical slots) up front; readers that see a 1-element set
///   surface `MalformedComponentValue`.
/// - `ONESHOT_MESSAGE`: prost-encoded `OneshotMessage`.
pub fn read_group_metadata_from_dict(
    mls_group: &OpenMlsGroup,
) -> Result<Option<GroupMetadataReturn>, ComponentSourceError> {
    read_group_metadata_from_extensions(mls_group.extensions())
}

/// Extensions-only variant of [`read_group_metadata_from_dict`]. Same
/// rationale for the split as [`read_super_admin_list_from_extensions`].
pub fn read_group_metadata_from_extensions(
    extensions: &Extensions<GroupContext>,
) -> Result<Option<GroupMetadataReturn>, ComponentSourceError> {
    use prost::Message;
    use xmtp_proto::xmtp::mls::message_contents::{
        ConversationType, DmMembers as DmMembersProto, Inbox as InboxProto, OneshotMessage,
    };

    let Some(ext) = extensions.app_data_dictionary() else {
        return Ok(None);
    };
    let dict = ext.dictionary();

    let Some(ct_bytes) = dict.get(&ComponentId::CONVERSATION_TYPE.as_u16()) else {
        return Ok(None);
    };
    let Some(creator_bytes) = dict.get(&ComponentId::CREATOR_INBOX_ID.as_u16()) else {
        return Ok(None);
    };

    let ct_arr: [u8; 4] =
        ct_bytes
            .try_into()
            .map_err(|_| ComponentSourceError::MalformedComponentValue {
                component_id: ComponentId::CONVERSATION_TYPE,
                reason: format!("expected 4 bytes, got {}", ct_bytes.len()),
            })?;
    let conversation_type = i32::from_be_bytes(ct_arr);

    let creator_inbox_id = InboxId::tls_deserialize_exact(creator_bytes)
        .map_err(|e| ComponentSourceError::MalformedComponentValue {
            component_id: ComponentId::CREATOR_INBOX_ID,
            reason: format!("invalid InboxId TLS encoding: {e}"),
        })?
        .to_hex();

    // `DM_MEMBERS` on the wire is `TlsSet<InboxId>`; re-shape to
    // `DmMembersProto` so downstream `GroupMetadata::try_from` is unchanged.
    // The pair grants DM-only authority, so it is malformed outside a DM.
    // implements: META-030
    let dm_members = match dict.get(&ComponentId::DM_MEMBERS.as_u16()) {
        Some(_) if conversation_type != ConversationType::Dm as i32 => {
            return Err(ComponentSourceError::MalformedComponentValue {
                component_id: ComponentId::DM_MEMBERS,
                reason: format!("present in conversation type {conversation_type}"),
            });
        }
        Some(b) => {
            let set = TlsSet::<InboxId>::tls_deserialize_exact(b).map_err(|e| {
                ComponentSourceError::MalformedComponentValue {
                    component_id: ComponentId::DM_MEMBERS,
                    reason: format!("invalid TlsSet<InboxId>: {e}"),
                }
            })?;
            let ids: Vec<InboxId> = set.iter().copied().collect();
            if ids.len() != 2 {
                return Err(ComponentSourceError::MalformedComponentValue {
                    component_id: ComponentId::DM_MEMBERS,
                    reason: format!("expected 2 inbox ids, got {}", ids.len()),
                });
            }
            Some(DmMembersProto {
                dm_member_one: Some(InboxProto {
                    inbox_id: ids[0].to_hex(),
                }),
                dm_member_two: Some(InboxProto {
                    inbox_id: ids[1].to_hex(),
                }),
            })
        }
        None => None,
    };

    let oneshot = match dict.get(&ComponentId::ONESHOT_MESSAGE.as_u16()) {
        Some(b) => Some(OneshotMessage::decode(b).map_err(|e| {
            ComponentSourceError::MalformedComponentValue {
                component_id: ComponentId::ONESHOT_MESSAGE,
                reason: format!("OneshotMessage prost decode: {e}"),
            }
        })?),
        None => None,
    };

    Ok(Some(GroupMetadataReturn {
        conversation_type,
        creator_inbox_id,
        dm_members,
        oneshot,
    }))
}

/// Intermediate proto-shaped result of [`read_group_metadata_from_extensions`].
/// Caller converts to the final [`crate::group_metadata::GroupMetadata`].
#[derive(Debug)]
pub struct GroupMetadataReturn {
    pub conversation_type: i32,
    pub creator_inbox_id: String,
    pub dm_members: Option<xmtp_proto::xmtp::mls::message_contents::DmMembers>,
    pub oneshot: Option<xmtp_proto::xmtp::mls::message_contents::OneshotMessage>,
}

/// Read the `GROUP_MEMBERSHIP` dict entry and decode it into the
/// legacy `GroupMembership` proto shape. Used by `extract_group_membership` on the
/// receive-side validator to bridge the dict-stored membership back
/// into the existing `GroupMembership` Rust type without rewriting
/// every caller.
pub fn read_group_membership_from_dict(
    extensions: &Extensions<GroupContext>,
) -> Result<Option<xmtp_proto::xmtp::mls::message_contents::GroupMembership>, ComponentSourceError>
{
    use crate::app_data::migration::decode_group_membership_dict;
    use xmtp_proto::xmtp::mls::message_contents::GroupMembership as GroupMembershipProto;

    let Some(ext) = extensions.app_data_dictionary() else {
        return Ok(None);
    };
    let Some(bytes) = ext
        .dictionary()
        .get(&ComponentId::GROUP_MEMBERSHIP.as_u16())
    else {
        return Ok(None);
    };

    let entries = decode_group_membership_dict(bytes).map_err(|e| {
        ComponentSourceError::MalformedComponentValue {
            component_id: ComponentId::GROUP_MEMBERSHIP,
            reason: format!("TlsMap decode: {e}"),
        }
    })?;

    // Flatten per-inbox GroupMembershipEntryV1 back into the legacy
    // proto shape: members (inbox_id → sequence_id), failed_installations
    // (flat Vec). The proto still has a flat failed_installations field
    // for backward compat — we concatenate per-inbox failed lists for
    // callers that still read the flat list.
    //
    // `decode_group_membership_dict` already rejects entries with
    // `version: None` (`MigrationError::GroupMembershipEntryUnknownVersion`),
    // so the only legal post-decode shape today is `Some(Version::V1(_))`.
    // Anything else (a future Version variant we can't interpret) is a
    // forward-compat hazard and surfaces as `MalformedComponentValue`.
    use xmtp_proto::xmtp::mls::message_contents::group_membership_entry::Version as GroupMembershipEntryVersion;
    let mut members: std::collections::HashMap<String, u64> = std::collections::HashMap::new();
    let mut failed: Vec<Vec<u8>> = Vec::new();
    for (inbox_id, entry) in entries {
        let v1 = match entry.version {
            Some(GroupMembershipEntryVersion::V1(v1)) => v1,
            None => {
                return Err(ComponentSourceError::MalformedComponentValue {
                    component_id: ComponentId::GROUP_MEMBERSHIP,
                    reason: format!(
                        "GroupMembershipEntry for {} has no version",
                        inbox_id.to_hex()
                    ),
                });
            }
        };
        members.insert(inbox_id.to_hex(), v1.sequence_id);
        failed.extend(v1.failed_installations);
    }

    Ok(Some(GroupMembershipProto {
        members,
        failed_installations: failed,
    }))
}

/// Encode a list of hex inbox ids as a TLS-serialized `TlsSet<InboxId>`.
// Only tests call it. `expect` (not `allow`) so the compiler flags this
// once a non-test caller in this module uses it.
#[cfg_attr(not(test), expect(dead_code))]
fn encode_inbox_id_set(inbox_ids: &[String]) -> Result<Vec<u8>, ComponentSourceError> {
    let ids: Vec<InboxId> = inbox_ids
        .iter()
        .map(|s| inbox_id_str_to_bytes(s))
        .collect::<Result<Vec<_>, _>>()?;
    let set: TlsSet<InboxId> = ids.into_iter().collect();
    Ok(set.tls_serialize_detached()?)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        app_data::{
            component_permissions::component_permissions,
            component_registry::{ComponentOp, new_component_metadata},
        },
        inbox_id::INBOX_ID_BYTE_LEN,
        tls_set::TlsKeyHash,
    };
    use prost::Message;
    use xmtp_proto::xmtp::mls::message_contents::{
        MetadataPolicy as MetadataPolicyProto,
        metadata_policy::{Kind as MetadataPolicyKind, MetadataBasePolicy},
    };

    /// Build a deterministic 64-character hex inbox id from a tag byte. The
    /// tag is repeated 32 times, giving a unique inbox id per call without
    /// needing real cryptographic generation.
    fn fake_inbox_id(tag: u8) -> String {
        hex::encode([tag; INBOX_ID_BYTE_LEN])
    }

    /// Build the [`InboxId`] form of [`fake_inbox_id`] directly.
    fn fake_inbox(tag: u8) -> InboxId {
        InboxId::from_bytes([tag; INBOX_ID_BYTE_LEN])
    }

    /// Empty registry constant for tests that exercise known-id paths —
    /// `lookup_component` resolves first, so the registry is never
    /// consulted and an empty one is sufficient.
    fn empty_registry() -> ComponentRegistry {
        ComponentRegistry::new()
    }

    /// A one-mutation `TlsSetDelta<InboxId>` payload.
    fn inbox_set_delta(mutation: TlsSetMutation<InboxId>) -> Vec<u8> {
        TlsSetDelta {
            mutations: vec![mutation],
        }
        .tls_serialize_detached()
        .unwrap()
    }

    /// Build a single-entry registry for tests that exercise the
    /// type-aware fallback on unknown ids. Permissions are `Allow` for
    /// every op so the policy layer does not interfere with the
    /// dispatch test under question.
    fn registry_with(id: ComponentId, ty: ComponentType) -> ComponentRegistry {
        fn allow() -> MetadataPolicyProto {
            MetadataPolicyProto {
                kind: Some(MetadataPolicyKind::Base(MetadataBasePolicy::Allow as i32)),
            }
        }
        let perms = component_permissions()
            .insert(allow())
            .update(allow())
            .delete(allow())
            .call();
        let meta = new_component_metadata(perms, ty);
        let mut reg = ComponentRegistry::new();
        reg.set(id, meta).unwrap();
        reg
    }

    // --- inbox-id helpers --------------------------------------------------

    #[xmtp_common::test]
    fn test_inbox_id_round_trip() {
        let original = fake_inbox_id(0xAB);
        let id = inbox_id_str_to_bytes(&original).unwrap();
        assert_eq!(id.as_bytes(), &[0xAB; 32]);
        assert_eq!(id.to_hex(), original);
    }

    #[xmtp_common::test]
    fn test_inbox_id_invalid_hex() {
        let err = inbox_id_str_to_bytes("not_hex").unwrap_err();
        assert!(
            matches!(
                err,
                ComponentSourceError::InvalidInboxId(InboxIdError::InvalidHex(_))
            ),
            "got {err:?}"
        );
    }

    #[xmtp_common::test]
    fn test_inbox_id_wrong_length() {
        // Valid hex but too short — only 16 bytes.
        let err = inbox_id_str_to_bytes(&"ab".repeat(16)).unwrap_err();
        assert!(
            matches!(
                err,
                ComponentSourceError::InvalidInboxId(InboxIdError::InvalidLength {
                    expected: INBOX_ID_BYTE_LEN,
                    actual: 16,
                })
            ),
            "got {err:?}"
        );
    }

    // --- component_type lookups --------------------------------------------

    #[xmtp_common::test]
    fn test_component_type_string_family() {
        // GMM-backed components whose wire format is UTF-8 text — names,
        // descriptions, URLs, and the app-data string blob.
        for id in [
            ComponentId::GROUP_NAME,
            ComponentId::GROUP_DESCRIPTION,
            ComponentId::GROUP_IMAGE_URL,
            ComponentId::APP_DATA,
            ComponentId::MIN_SUPPORTED_PROTOCOL_VERSION,
        ] {
            assert_eq!(component_type(id), Some(ComponentType::String));
        }
    }

    #[xmtp_common::test]
    fn test_component_type_bytes_family() {
        // Components whose payload is opaque bytes, not UTF-8 text:
        // timestamped disappearance windows and the commit-log signer key.
        for id in [
            ComponentId::MESSAGE_DISAPPEAR_FROM_NS,
            ComponentId::MESSAGE_DISAPPEAR_IN_NS,
            ComponentId::COMMIT_LOG_SIGNER,
        ] {
            assert_eq!(component_type(id), Some(ComponentType::Bytes));
        }
    }

    #[xmtp_common::test]
    fn test_component_type_set_inbox_id_family() {
        for id in [
            ComponentId::ADMIN_LIST,
            ComponentId::SUPER_ADMIN_LIST,
            ComponentId::DM_MEMBERS,
        ] {
            assert_eq!(component_type(id), Some(ComponentType::TlsSetInboxId));
        }
    }

    #[xmtp_common::test]
    fn test_component_type_group_membership() {
        assert_eq!(
            component_type(ComponentId::GROUP_MEMBERSHIP),
            Some(ComponentType::TlsMapInboxIdBytes)
        );
    }

    /// The two newest well-known ids carry the fixed types every client
    /// must agree on: a UTF-8 inbox map for display names, opaque bytes for
    /// the group image.
    // verifies: META-010
    #[xmtp_common::test]
    fn test_component_type_profile_and_image() {
        assert_eq!(
            component_type(ComponentId::USER_DISPLAY_NAME),
            Some(ComponentType::TlsMapInboxIdString)
        );
        assert_eq!(
            component_type(ComponentId::GROUP_IMAGE),
            Some(ComponentType::Bytes)
        );
    }

    #[xmtp_common::test]
    fn test_component_type_app_range_is_none() {
        assert_eq!(component_type(ComponentId::new(0xC000)), None);
        assert_eq!(component_type(ComponentId::new(0xFDAB)), None);
    }

    // --- MetadataField <-> ComponentId mapping -----------------------------

    #[xmtp_common::test]
    fn test_metadata_field_round_trip() {
        for field in [
            MetadataField::GroupName,
            MetadataField::Description,
            MetadataField::GroupImageUrlSquare,
            MetadataField::MessageDisappearFromNS,
            MetadataField::MessageDisappearInNS,
            MetadataField::MinimumSupportedProtocolVersion,
            MetadataField::CommitLogSigner,
            MetadataField::AppData,
        ] {
            let id = metadata_field_to_component_id(field.as_str())
                .expect("every MetadataField has a ComponentId");
            let back = component_id_to_metadata_field(id)
                .expect("every mapped ComponentId has a MetadataField");
            assert_eq!(back, field, "round-trip mismatch for {field:?}");
        }
    }

    #[xmtp_common::test]
    fn test_metadata_field_unknown_name_returns_none() {
        assert!(metadata_field_to_component_id("nonexistent_field").is_none());
    }

    #[xmtp_common::test]
    fn test_component_id_to_metadata_field_non_gmm_returns_none() {
        assert!(component_id_to_metadata_field(ComponentId::ADMIN_LIST).is_none());
        assert!(component_id_to_metadata_field(ComponentId::GROUP_MEMBERSHIP).is_none());
        assert!(component_id_to_metadata_field(ComponentId::new(0xC000)).is_none());
    }

    // --- apply_app_data_update_payload -------------------------------------

    #[xmtp_common::test]
    fn test_apply_bytes_payload_returns_payload_verbatim() {
        let payload = b"new_name";
        let new_value = apply_app_data_update_payload(
            ComponentId::GROUP_NAME,
            payload,
            None,
            &empty_registry(),
        )
        .unwrap();
        assert_eq!(new_value, payload);
    }

    #[xmtp_common::test]
    fn test_apply_bytes_payload_ignores_old_value() {
        // Full replacement — the old value is irrelevant.
        let new_value = apply_app_data_update_payload(
            ComponentId::GROUP_DESCRIPTION,
            b"replacement",
            Some(b"old_description"),
            &empty_registry(),
        )
        .unwrap();
        assert_eq!(new_value, b"replacement");
    }

    #[xmtp_common::test]
    fn test_apply_admin_list_insert_against_none() {
        // Apply an insert delta against a group that has no prior admin list.
        // The synthesized full value should be a TlsSet with the one inbox id.
        let inbox = fake_inbox_id(0x44);
        let insert_payload =
            inbox_set_delta(TlsSetMutation::Insert(InboxId::from_hex(&inbox).unwrap()));

        let new_bytes = apply_app_data_update_payload(
            ComponentId::ADMIN_LIST,
            &insert_payload,
            None,
            &empty_registry(),
        )
        .unwrap();

        let set = TlsSet::<InboxId>::tls_deserialize_exact(&new_bytes).unwrap();
        assert_eq!(set.len(), 1);
        assert!(set.contains(&fake_inbox(0x44)));
    }

    #[xmtp_common::test]
    fn test_apply_admin_list_insert_against_existing_set() {
        // Build a prior set with one entry, then apply a delta that inserts a
        // second entry. Confirm both entries are present in the new value.
        let alice = fake_inbox_id(0x01);
        let bob = fake_inbox_id(0x02);

        let prior = encode_inbox_id_set(std::slice::from_ref(&alice)).unwrap();
        let insert_payload =
            inbox_set_delta(TlsSetMutation::Insert(InboxId::from_hex(&bob).unwrap()));

        let new_bytes = apply_app_data_update_payload(
            ComponentId::ADMIN_LIST,
            &insert_payload,
            Some(&prior),
            &empty_registry(),
        )
        .unwrap();

        let set = TlsSet::<InboxId>::tls_deserialize_exact(&new_bytes).unwrap();
        assert_eq!(set.len(), 2);
        assert!(set.contains(&fake_inbox(0x01)));
        assert!(set.contains(&fake_inbox(0x02)));
    }

    #[xmtp_common::test]
    fn test_apply_admin_list_remove_against_existing_set() {
        let alice = fake_inbox_id(0x01);
        let bob = fake_inbox_id(0x02);

        let prior = encode_inbox_id_set(&[alice.clone(), bob.clone()]).unwrap();
        let remove_payload =
            inbox_set_delta(TlsSetMutation::Remove(InboxId::from_hex(&alice).unwrap()));

        let new_bytes = apply_app_data_update_payload(
            ComponentId::ADMIN_LIST,
            &remove_payload,
            Some(&prior),
            &empty_registry(),
        )
        .unwrap();

        let set = TlsSet::<InboxId>::tls_deserialize_exact(&new_bytes).unwrap();
        assert_eq!(set.len(), 1);
        assert!(set.contains(&fake_inbox(0x02)));
        assert!(!set.contains(&fake_inbox(0x01)));
    }

    #[xmtp_common::test]
    fn test_apply_super_admin_list_delta() {
        let owner = fake_inbox_id(0xAA);
        let new_sa = fake_inbox_id(0xBB);

        let prior = encode_inbox_id_set(std::slice::from_ref(&owner)).unwrap();
        let add_payload =
            inbox_set_delta(TlsSetMutation::Insert(InboxId::from_hex(&new_sa).unwrap()));

        let new_bytes = apply_app_data_update_payload(
            ComponentId::SUPER_ADMIN_LIST,
            &add_payload,
            Some(&prior),
            &empty_registry(),
        )
        .unwrap();

        let set = TlsSet::<InboxId>::tls_deserialize_exact(&new_bytes).unwrap();
        assert_eq!(set.len(), 2);
        assert!(set.contains(&fake_inbox(0xAA)));
        assert!(set.contains(&fake_inbox(0xBB)));
    }

    #[xmtp_common::test]
    fn test_apply_admin_list_malformed_payload_returns_tls_codec_error() {
        // Garbage bytes that aren't a valid TlsSetDelta — exercises the
        // wire-format decode failure path that ProcessMessageWithAppDataError
        // ::AppDataDecode wraps in production. The receiver-side
        // `process_message_with_app_data` propagates this through the new
        // GroupMessageProcessingError::OpenMlsProcessMessageWithAppData
        // variant rather than masking it as `FoundAppDataUpdateProposal`.
        let err = apply_app_data_update_payload(
            ComponentId::ADMIN_LIST,
            &[0xff, 0xff, 0xff, 0xff],
            None,
            &empty_registry(),
        )
        .unwrap_err();
        assert!(
            matches!(err, ComponentSourceError::TlsCodec(_)),
            "got {err:?}"
        );
    }

    #[xmtp_common::test]
    fn test_apply_admin_list_malformed_old_value_returns_tls_codec_error() {
        // The delta payload is well-formed, but the old_value isn't a
        // valid TlsSet — also a TlsCodec error, just from the other side.
        let inbox = fake_inbox_id(0x55);
        let payload = inbox_set_delta(TlsSetMutation::Insert(InboxId::from_hex(&inbox).unwrap()));
        let err = apply_app_data_update_payload(
            ComponentId::ADMIN_LIST,
            &payload,
            Some(&[0xde, 0xad, 0xbe, 0xef]),
            &empty_registry(),
        )
        .unwrap_err();
        assert!(
            matches!(err, ComponentSourceError::TlsCodec(_)),
            "got {err:?}"
        );
    }

    #[xmtp_common::test]
    fn test_apply_immutable_first_insert_allowed() {
        // Bootstrap-shape: an `Update(payload)` against an immutable
        // component with no prior value is the bootstrap commit's
        // first-write path. Apply must store the payload bytes
        // verbatim — the bootstrap validator's byte-compare catches a
        // peer that crafts a malicious initial value, so the apply
        // layer doesn't need its own decode/check step.
        let bytes = apply_app_data_update_payload(
            ComponentId::CONVERSATION_TYPE,
            b"seed",
            None,
            &empty_registry(),
        )
        .unwrap();
        assert_eq!(bytes, b"seed");
    }

    #[xmtp_common::test]
    fn test_apply_immutable_overwrite_rejected() {
        // Steady-state: an `Update(payload)` against an immutable
        // component that already has a prior value must fail. This is
        // the only path Byzantine peers have to mutate immutables
        // post-bootstrap, and the apply layer is the gatekeeper.
        let err = apply_app_data_update_payload(
            ComponentId::CONVERSATION_TYPE,
            b"junk",
            Some(b"prior"),
            &empty_registry(),
        )
        .unwrap_err();
        assert!(matches!(err, ComponentSourceError::ImmutableUpdate(_)));
    }

    #[xmtp_common::test]
    fn test_apply_component_registry_delta_against_empty() {
        // Bootstrap shape: a `TlsMapDelta<ComponentId, VLBytes>` of
        // all-`Insert` mutations applied against an empty map produces
        // a materialized snapshot containing those entries. Values must
        // be structurally valid `ComponentMetadata` — registry deltas
        // are entry-validated at apply.
        let id_a = ComponentId::GROUP_NAME;
        let id_b = ComponentId::GROUP_DESCRIPTION;
        let meta_a = registry_with(id_a, ComponentType::String)
            .get(&id_a)
            .unwrap()
            .unwrap()
            .encode_to_vec();
        let meta_b = registry_with(id_b, ComponentType::Bytes)
            .get(&id_b)
            .unwrap()
            .unwrap()
            .encode_to_vec();
        let delta = TlsMapDelta::<ComponentId, VLBytes>::new()
            .insert(id_a, VLBytes::new(meta_a.clone()))
            .insert(id_b, VLBytes::new(meta_b.clone()));
        let payload = delta.tls_serialize_detached().unwrap();

        let new_bytes = apply_app_data_update_payload(
            ComponentId::COMPONENT_REGISTRY,
            &payload,
            None,
            &empty_registry(),
        )
        .unwrap();
        let map = TlsMap::<ComponentId, VLBytes>::tls_deserialize_exact(&new_bytes).unwrap();
        assert_eq!(map.len(), 2);
        assert_eq!(
            map.get(&id_a).map(|v| v.as_slice()),
            Some(meta_a.as_slice())
        );
        assert_eq!(
            map.get(&id_b).map(|v| v.as_slice()),
            Some(meta_b.as_slice())
        );
    }

    #[xmtp_common::test]
    fn test_apply_group_membership_delta_against_existing_map() {
        // Post-bootstrap shape: an `Update` mutation applied on top of
        // a prior `TlsMap<InboxId, VLBytes>` snapshot produces a new
        // snapshot with the updated value.
        let alice = fake_inbox(0xAA);
        let bob = fake_inbox(0xBB);
        let mut prior: TlsMap<InboxId, VLBytes> = TlsMap::new();
        prior.set(alice, VLBytes::new(vec![0x01]));
        prior.set(bob, VLBytes::new(vec![0x02]));
        let prior_bytes = prior.tls_serialize_detached().unwrap();

        let delta = TlsMapDelta::<InboxId, VLBytes>::new()
            .update(alice, VLBytes::new(vec![0x99]))
            .delete(bob);
        let payload = delta.tls_serialize_detached().unwrap();

        let new_bytes = apply_app_data_update_payload(
            ComponentId::GROUP_MEMBERSHIP,
            &payload,
            Some(&prior_bytes),
            &empty_registry(),
        )
        .unwrap();
        let map = TlsMap::<InboxId, VLBytes>::tls_deserialize_exact(&new_bytes).unwrap();
        assert_eq!(map.len(), 1);
        assert_eq!(
            map.get(&alice).map(|v| v.as_slice()),
            Some([0x99].as_slice())
        );
        assert!(!map.contains_key(&bob));
    }

    #[xmtp_common::test]
    fn test_apply_map_component_malformed_delta_returns_codec_error() {
        // Garbage bytes that aren't a valid TlsMapDelta surface as a
        // TLS-codec error, same shape as the set-component path.
        let err = apply_app_data_update_payload(
            ComponentId::COMPONENT_REGISTRY,
            &[0xff, 0xff, 0xff, 0xff],
            None,
            &empty_registry(),
        )
        .unwrap_err();
        assert!(
            matches!(err, ComponentSourceError::TlsCodec(_)),
            "got {err:?}"
        );
    }

    #[xmtp_common::test]
    fn test_apply_map_component_apply_failure_surfaces_apply_error() {
        // A delta that updates a key not present in the prior snapshot
        // fails at apply time — surfaced as `TlsMapApply(KeyNotFound)`.
        let alice = fake_inbox(0x01);
        let delta = TlsMapDelta::<InboxId, VLBytes>::new().update(alice, VLBytes::new(vec![0x42]));
        let payload = delta.tls_serialize_detached().unwrap();
        let err = apply_app_data_update_payload(
            ComponentId::GROUP_MEMBERSHIP,
            &payload,
            None,
            &empty_registry(),
        )
        .unwrap_err();
        assert!(
            matches!(
                err,
                ComponentSourceError::TlsMapApply(TlsMapError::KeyNotFound)
            ),
            "got {err:?}"
        );
    }

    // --- Unknown-component tolerance via type-aware dispatch ---------------
    //
    // No per-id `Component` impl exists for these ids on this client; the
    // sender shipped a newer release. Old clients look up the
    // `ComponentType` registered for the id in the on-dict
    // [`ComponentRegistry`] and route the payload through the
    // type-level decoder. The seven `ComponentType` variants cover the
    // wire-format universe, so any future well-known or
    // application-range component lands convergently — including
    // `TlsSet` / `TlsMap` deltas, which previously needed a per-id
    // impl to apply correctly.

    #[xmtp_common::test]
    fn test_apply_unknown_id_with_bytes_registry_entry_stores_payload() {
        // Bytes shape: opaque passthrough, registry entry supplies the
        // type tag so the dispatch knows *not* to try a TLS-delta
        // decode (which would corrupt the bytes).
        let id = ComponentId::new(0xC123);
        let registry = registry_with(id, ComponentType::Bytes);
        let new_value = apply_app_data_update_payload(id, b"opaque", None, &registry).unwrap();
        assert_eq!(new_value, b"opaque");
    }

    #[xmtp_common::test]
    fn test_apply_unknown_id_with_string_registry_entry_validates_utf8() {
        // String shape: payload must be valid UTF-8. Bad bytes surface
        // as `MalformedComponentValue` rather than silently corrupting
        // the dict.
        let id = ComponentId::new(0xC222);
        let registry = registry_with(id, ComponentType::String);
        let ok = apply_app_data_update_payload(id, b"hello", None, &registry).unwrap();
        assert_eq!(ok, b"hello");
        let err = apply_app_data_update_payload(id, &[0xC3, 0x28], None, &registry).unwrap_err();
        assert!(matches!(
            err,
            ComponentSourceError::MalformedComponentValue { .. }
        ));
    }

    #[xmtp_common::test]
    fn test_apply_unknown_id_with_tls_set_inbox_id_applies_delta() {
        // The whole point of registry-typed dispatch: a new
        // `TlsSet<InboxId>` component lands as a typed delta apply, not
        // as an opaque blob replacement — old and new clients converge
        // on the same `TlsSet` snapshot byte-for-byte.
        let id = ComponentId::new(0xC333);
        let registry = registry_with(id, ComponentType::TlsSetInboxId);
        let bob = fake_inbox(0x02);
        let delta = TlsSetDelta::<InboxId>::new().insert(bob);
        let payload = delta.tls_serialize_detached().unwrap();
        let new_bytes = apply_app_data_update_payload(id, &payload, None, &registry).unwrap();
        let set = TlsSet::<InboxId>::tls_deserialize_exact(&new_bytes).unwrap();
        assert_eq!(set.len(), 1);
        assert!(set.contains(&bob));
    }

    #[xmtp_common::test]
    fn test_apply_unknown_id_with_no_registry_entry_rejected() {
        // Deny-by-default: no per-id impl AND no registry entry means
        // we have no type to decode against. Reject rather than store
        // bytes whose shape we can't reason about — the alternative
        // would fork the dict the moment a typed client did know the
        // shape.
        let err =
            apply_app_data_update_payload(ComponentId::new(0xC456), b"x", None, &empty_registry())
                .unwrap_err();
        assert!(matches!(err, ComponentSourceError::UnknownComponent(_)));
    }

    /// Immutability is enforced range-by-id, not per-`Component`-impl,
    /// so an unknown id sitting in the XMTP immutable range
    /// (`0xBE00-0xBFFF`) still gets the bootstrap-style "first insert
    /// allowed, subsequent overwrite rejected" contract — even when
    /// the type-aware dispatcher (not a per-id `Component` impl) is
    /// the path being exercised. Pins the reviewer's concern that the
    /// tolerance branch might bypass immutability for newly-defined
    /// immutable components that a NEWER release ships.
    #[xmtp_common::test]
    fn test_apply_unknown_id_in_xmtp_immutable_range_first_write_allowed() {
        let id = ComponentId::new(0xBE05); // XMTP immutable range
        assert!(id.is_immutable());
        let registry = registry_with(id, ComponentType::Bytes);
        let new_value = apply_app_data_update_payload(id, b"seed", None, &registry).unwrap();
        assert_eq!(new_value, b"seed");
    }

    #[xmtp_common::test]
    fn test_apply_unknown_id_in_xmtp_immutable_range_overwrite_rejected() {
        let id = ComponentId::new(0xBE05);
        assert!(id.is_immutable());
        let registry = registry_with(id, ComponentType::Bytes);
        let err = apply_app_data_update_payload(id, b"new", Some(b"old"), &registry).unwrap_err();
        assert!(matches!(err, ComponentSourceError::ImmutableUpdate(_)));
    }

    /// Same contract for the application immutable range
    /// (`0xFD00-0xFEFF`) — overwrites of an unknown immutable
    /// application component are rejected even though no per-id impl
    /// exists on this client.
    #[xmtp_common::test]
    fn test_apply_unknown_id_in_app_immutable_range_overwrite_rejected() {
        let id = ComponentId::new(0xFD42); // app immutable range
        assert!(id.is_immutable());
        let registry = registry_with(id, ComponentType::Bytes);
        let err = apply_app_data_update_payload(id, b"new", Some(b"old"), &registry).unwrap_err();
        assert!(matches!(err, ComponentSourceError::ImmutableUpdate(_)));
    }

    #[xmtp_common::test]
    fn test_apply_reserved_range_component_rejected_with_or_without_registry() {
        // 0xFF00+ is the reserved range. `ComponentRegistry::set`
        // refuses to insert reserved ids, so a registry entry can't
        // even be constructed — the apply path falls through to
        // `UnknownComponent`.
        let err =
            apply_app_data_update_payload(ComponentId::new(0xFF01), b"x", None, &empty_registry())
                .unwrap_err();
        assert!(matches!(err, ComponentSourceError::UnknownComponent(_)));
    }

    #[xmtp_common::test]
    fn test_apply_out_of_range_component_rejected() {
        // Ids below 0x8000 are outside the AppData address space.
        let err =
            apply_app_data_update_payload(ComponentId::new(0x0042), b"x", None, &empty_registry())
                .unwrap_err();
        assert!(matches!(err, ComponentSourceError::UnknownComponent(_)));
    }

    #[xmtp_common::test]
    fn test_apply_unknown_xmtp_range_id_with_registry_dispatches() {
        // Same path applies to the XMTP-defined range, not just the
        // app range: any 0x8000-0xBFFF id that lacks a per-id impl
        // routes through the registry.
        let id = ComponentId::new(0x8FFF);
        assert!(id.is_xmtp_range());
        let registry = registry_with(id, ComponentType::Bytes);
        let new_value = apply_app_data_update_payload(id, b"opaque", None, &registry).unwrap();
        assert_eq!(new_value, b"opaque");
    }

    // --- expand_app_data_update_to_changes ---------------------------------

    /// Helper: wrap a `TlsSetDelta<InboxId>` payload in an
    /// `AppDataUpdateOperation::Update(...)`.
    fn update_op(delta: TlsSetDelta<InboxId>) -> AppDataUpdateOperation {
        AppDataUpdateOperation::Update(delta.tls_serialize_detached().unwrap().into())
    }

    #[xmtp_common::test]
    fn test_expand_insert_surfaces_inbox_id_bytes() {
        let alice = fake_inbox(0x11);
        let delta: TlsSetDelta<InboxId> = TlsSetDelta {
            mutations: vec![TlsSetMutation::Insert(alice)],
        };
        let changes = expand_app_data_update_to_changes(
            ComponentId::ADMIN_LIST,
            &update_op(delta),
            None,
            &empty_registry(),
        )
        .unwrap();
        assert_eq!(changes.len(), 1);
        assert_eq!(changes[0].op, ComponentOp::Insert);
        assert_eq!(
            changes[0].value.as_deref(),
            Some(alice.as_bytes().as_slice())
        );
    }

    #[xmtp_common::test]
    fn test_expand_remove_surfaces_inbox_id_bytes() {
        let alice = fake_inbox(0x22);
        let delta: TlsSetDelta<InboxId> = TlsSetDelta {
            mutations: vec![TlsSetMutation::Remove(alice)],
        };
        let changes = expand_app_data_update_to_changes(
            ComponentId::ADMIN_LIST,
            &update_op(delta),
            None,
            &empty_registry(),
        )
        .unwrap();
        assert_eq!(changes.len(), 1);
        assert_eq!(changes[0].op, ComponentOp::Delete);
        assert_eq!(
            changes[0].value.as_deref(),
            Some(alice.as_bytes().as_slice())
        );
    }

    #[xmtp_common::test]
    fn test_expand_remove_by_hash_resolves_to_inbox_id_from_old_value() {
        // Prior set has alice + bob; RemoveByHash(hash(alice)) should
        // resolve back to alice's raw 32 bytes so the validator sees
        // *which* identity is being removed.
        let alice = fake_inbox(0xAA);
        let bob = fake_inbox(0xBB);
        let prior: TlsSet<InboxId> = [alice, bob].into_iter().collect();
        let old_bytes = prior.tls_serialize_detached().unwrap();

        let hash = TlsKeyHash::of(&alice).unwrap();
        let delta: TlsSetDelta<InboxId> = TlsSetDelta {
            mutations: vec![TlsSetMutation::RemoveByHash(hash)],
        };
        let changes = expand_app_data_update_to_changes(
            ComponentId::ADMIN_LIST,
            &update_op(delta),
            Some(&old_bytes),
            &empty_registry(),
        )
        .unwrap();
        assert_eq!(changes.len(), 1);
        assert_eq!(changes[0].op, ComponentOp::Delete);
        assert_eq!(
            changes[0].value.as_deref(),
            Some(alice.as_bytes().as_slice())
        );
    }

    #[xmtp_common::test]
    fn test_expand_remove_by_hash_miss_surfaces_none_value() {
        // Prior set has alice; RemoveByHash targets bob's hash (not in set).
        // Expansion surfaces value: None — the CRDT apply step will later
        // reject with KeyNotFound. Expansion's job is reshape, not auth.
        let alice = fake_inbox(0x01);
        let bob = fake_inbox(0x02);
        let prior: TlsSet<InboxId> = [alice].into_iter().collect();
        let old_bytes = prior.tls_serialize_detached().unwrap();

        let hash = TlsKeyHash::of(&bob).unwrap();
        let delta: TlsSetDelta<InboxId> = TlsSetDelta {
            mutations: vec![TlsSetMutation::RemoveByHash(hash)],
        };
        let changes = expand_app_data_update_to_changes(
            ComponentId::ADMIN_LIST,
            &update_op(delta),
            Some(&old_bytes),
            &empty_registry(),
        )
        .unwrap();
        assert_eq!(changes.len(), 1);
        assert_eq!(changes[0].op, ComponentOp::Delete);
        assert!(changes[0].value.is_none());
    }

    #[xmtp_common::test]
    fn test_expand_remove_by_hash_with_no_old_value_surfaces_none() {
        // No prior bytes → no set to search → every RemoveByHash misses.
        let hash = TlsKeyHash::of(&fake_inbox(0x33)).unwrap();
        let delta: TlsSetDelta<InboxId> = TlsSetDelta {
            mutations: vec![TlsSetMutation::RemoveByHash(hash)],
        };
        let changes = expand_app_data_update_to_changes(
            ComponentId::SUPER_ADMIN_LIST,
            &update_op(delta),
            None,
            &empty_registry(),
        )
        .unwrap();
        assert_eq!(changes.len(), 1);
        assert_eq!(changes[0].op, ComponentOp::Delete);
        assert!(changes[0].value.is_none());
    }

    #[xmtp_common::test]
    fn test_expand_remove_by_hash_malformed_old_value_surfaces_codec_error() {
        // Our own dict bytes are corrupt — surface the decode error
        // loudly rather than silently degrading to value: None, since a
        // corrupt prior set signals a local-state bug, not a peer bug.
        let hash = TlsKeyHash::of(&fake_inbox(0x44)).unwrap();
        let delta: TlsSetDelta<InboxId> = TlsSetDelta {
            mutations: vec![TlsSetMutation::RemoveByHash(hash)],
        };
        let err = expand_app_data_update_to_changes(
            ComponentId::ADMIN_LIST,
            &update_op(delta),
            Some(&[0xde, 0xad, 0xbe, 0xef]),
            &empty_registry(),
        )
        .unwrap_err();
        assert!(
            matches!(err, ComponentSourceError::TlsCodec(_)),
            "got {err:?}"
        );
    }

    /// XIP §2.2: an unknown component id with `Update(payload)` and a
    /// registered [`ComponentType::Bytes`] expands to a single Update
    /// change carrying the payload, so registry-policy validation runs
    /// against bytes a typed client would also accept opaquely.
    #[xmtp_common::test]
    fn test_expand_unknown_component_bytes_typed_emits_single_update() {
        let id = ComponentId::new(0x80FF);
        let registry = registry_with(id, ComponentType::Bytes);
        let changes = expand_app_data_update_to_changes(
            id,
            &AppDataUpdateOperation::Update(b"opaque".to_vec().into()),
            None,
            &registry,
        )
        .unwrap();
        assert_eq!(changes.len(), 1);
        assert_eq!(changes[0].op, ComponentOp::Update);
        assert_eq!(changes[0].value.as_deref(), Some(b"opaque".as_slice()));
    }

    /// Unknown id with `Remove` and any registered type expands to a
    /// single Delete change — the wipe semantics are type-agnostic.
    #[xmtp_common::test]
    fn test_expand_unknown_component_remove_emits_single_delete() {
        let id = ComponentId::new(0x80FF);
        let registry = registry_with(id, ComponentType::Bytes);
        let changes =
            expand_app_data_update_to_changes(id, &AppDataUpdateOperation::Remove, None, &registry)
                .unwrap();
        assert_eq!(changes.len(), 1);
        assert_eq!(changes[0].op, ComponentOp::Delete);
        assert!(changes[0].value.is_none());
    }

    /// Type-aware dispatch is the whole point: an unknown
    /// `TlsSet<InboxId>` component expands to a per-element change
    /// list, not a single opaque blob — so the validator policy loop
    /// inspects each `Insert` individually, the same as it would for a
    /// known set-shaped component.
    #[xmtp_common::test]
    fn test_expand_unknown_component_tls_set_inbox_id_emits_per_element_changes() {
        let id = ComponentId::new(0x80FE);
        let registry = registry_with(id, ComponentType::TlsSetInboxId);
        let alice = fake_inbox(0x11);
        let bob = fake_inbox(0x22);
        let delta = TlsSetDelta::<InboxId>::new().insert(alice).insert(bob);
        let payload = delta.tls_serialize_detached().unwrap();
        let changes = expand_app_data_update_to_changes(
            id,
            &AppDataUpdateOperation::Update(payload.into()),
            None,
            &registry,
        )
        .unwrap();
        assert_eq!(changes.len(), 2);
        assert!(changes.iter().all(|c| c.op == ComponentOp::Insert));
    }

    /// Reserved-range ids are not tolerated. `ComponentRegistry::set`
    /// rejects reserved ids at construction time, so a registry entry
    /// can't even be built — the dispatcher falls through to the
    /// `UnknownComponent` rejection.
    #[xmtp_common::test]
    fn test_expand_reserved_range_id_still_rejected() {
        let err = expand_app_data_update_to_changes(
            ComponentId::new(0xFF02),
            &AppDataUpdateOperation::Update(b"x".to_vec().into()),
            None,
            &empty_registry(),
        )
        .unwrap_err();
        assert!(matches!(err, ComponentSourceError::UnknownComponent(_)));
    }

    /// Deny-by-default: an unknown id with no registry entry is
    /// rejected. Without a type tag we have no idea how to decode the
    /// payload, and storing it opaquely would diverge from any future
    /// typed client that DID know the shape.
    #[xmtp_common::test]
    fn test_expand_unknown_component_no_registry_entry_rejected() {
        let err = expand_app_data_update_to_changes(
            ComponentId::new(0x80FF),
            &AppDataUpdateOperation::Update(b"opaque".to_vec().into()),
            None,
            &empty_registry(),
        )
        .unwrap_err();
        assert!(matches!(err, ComponentSourceError::UnknownComponent(_)));
    }

    #[xmtp_common::test]
    fn test_expand_skips_old_value_decode_when_no_remove_by_hash() {
        // Delta has only Insert/Remove — we should never touch old_value,
        // so even a garbage old_value must not fail the expansion.
        let alice = fake_inbox(0x55);
        let delta: TlsSetDelta<InboxId> = TlsSetDelta {
            mutations: vec![TlsSetMutation::Insert(alice)],
        };
        let changes = expand_app_data_update_to_changes(
            ComponentId::ADMIN_LIST,
            &update_op(delta),
            Some(&[0xff, 0xff, 0xff]), // intentionally malformed
            &empty_registry(),
        )
        .unwrap();
        assert_eq!(changes.len(), 1);
        assert_eq!(changes[0].op, ComponentOp::Insert);
    }

    // ========================================================================
    // Dict-reader helpers — happy path and malformed coverage
    // ========================================================================
    //
    // The three `read_*_from_dict` helpers execute before bootstrap is
    // wired end-to-end; the unit tests below pin the wire-format
    // contract so a decoder drift can't silently ship a broken read path.

    use openmls::extensions::{
        AppDataDictionary, AppDataDictionaryExtension, Extension as OpenMlsExtension, Extensions,
    };

    /// Build a synthetic `Extensions<GroupContext>` with an AppData dictionary.
    fn extensions_with_entries(
        entries: &[(u16, Vec<u8>)],
    ) -> Extensions<openmls::group::GroupContext> {
        let mut dict = AppDataDictionary::new();
        for (id, bytes) in entries {
            let _ = dict.insert(*id, bytes.clone());
        }
        Extensions::from_vec(vec![OpenMlsExtension::AppDataDictionary(
            AppDataDictionaryExtension::new(dict),
        )])
        .expect("valid group-context extension set")
    }

    // --- read_super_admin_list_from_extensions ------------------------------

    #[xmtp_common::test]
    fn read_super_admin_list_absent_returns_none() {
        // The dict has no SUPER_ADMIN_LIST entry —
        // `Ok(None)` rather than surfacing a malformed-value error.
        let exts = extensions_with_entries(&[]);
        assert!(
            read_super_admin_list_from_extensions(&exts)
                .unwrap()
                .is_none()
        );
    }

    #[xmtp_common::test]
    fn read_super_admin_list_happy_path() {
        let ids = vec![fake_inbox_id(0xAA), fake_inbox_id(0xBB)];
        let bytes = encode_inbox_id_set(&ids).unwrap();
        let exts = extensions_with_entries(&[(ComponentId::SUPER_ADMIN_LIST.as_u16(), bytes)]);
        let got = read_super_admin_list_from_extensions(&exts)
            .unwrap()
            .unwrap();
        assert_eq!(got.len(), 2);
        // TlsSet sorts by value so 0xAA sorts before 0xBB.
        assert_eq!(got[0], fake_inbox_id(0xAA));
        assert_eq!(got[1], fake_inbox_id(0xBB));
    }

    #[xmtp_common::test]
    fn read_super_admin_list_malformed_bytes_surface_error() {
        let exts = extensions_with_entries(&[(
            ComponentId::SUPER_ADMIN_LIST.as_u16(),
            vec![0x00, 0xDE, 0xAD],
        )]);
        let err = read_super_admin_list_from_extensions(&exts).unwrap_err();
        assert!(matches!(
            err,
            ComponentSourceError::MalformedComponentValue {
                component_id,
                ..
            } if component_id == ComponentId::SUPER_ADMIN_LIST
        ));
    }

    // --- read_group_metadata_from_extensions --------------------------------

    fn encode_conv_type_bytes(value: i32) -> Vec<u8> {
        value.to_be_bytes().to_vec()
    }

    fn encode_dm_pair(tag_a: u8, tag_b: u8) -> Vec<u8> {
        encode_inbox_id_set(&[fake_inbox_id(tag_a), fake_inbox_id(tag_b)]).unwrap()
    }

    /// Encode a single tagged inbox id in the `CREATOR_INBOX_ID` wire
    /// form (versioned `InboxId` TLS encoding).
    fn encode_creator_bytes(tag: u8) -> Vec<u8> {
        fake_inbox(tag).tls_serialize_detached().unwrap()
    }

    #[xmtp_common::test]
    fn read_group_metadata_missing_required_seeds_returns_none() {
        // A missing CONVERSATION_TYPE seed is not malformed.
        let exts = extensions_with_entries(&[]);
        assert!(
            read_group_metadata_from_extensions(&exts)
                .unwrap()
                .is_none()
        );
    }

    #[xmtp_common::test]
    fn read_group_metadata_happy_path_non_dm() {
        let exts = extensions_with_entries(&[
            (
                ComponentId::CONVERSATION_TYPE.as_u16(),
                encode_conv_type_bytes(1),
            ),
            (
                ComponentId::CREATOR_INBOX_ID.as_u16(),
                encode_creator_bytes(0x11),
            ),
        ]);
        let got = read_group_metadata_from_extensions(&exts).unwrap().unwrap();
        assert_eq!(got.conversation_type, 1);
        assert_eq!(got.creator_inbox_id, fake_inbox_id(0x11));
        assert!(got.dm_members.is_none());
        assert!(got.oneshot.is_none());
    }

    #[xmtp_common::test]
    fn read_group_metadata_dm_happy_path() {
        // DM group — DM_MEMBERS decodes as TlsSet<InboxId>, re-shaped
        // to the proto's two-slot form.
        let exts = extensions_with_entries(&[
            (
                ComponentId::CONVERSATION_TYPE.as_u16(),
                encode_conv_type_bytes(2),
            ),
            (
                ComponentId::CREATOR_INBOX_ID.as_u16(),
                encode_creator_bytes(0x22),
            ),
            (ComponentId::DM_MEMBERS.as_u16(), encode_dm_pair(0x22, 0x33)),
        ]);
        let got = read_group_metadata_from_extensions(&exts).unwrap().unwrap();
        let dm = got.dm_members.unwrap();
        assert_eq!(dm.dm_member_one.unwrap().inbox_id, fake_inbox_id(0x22));
        assert_eq!(dm.dm_member_two.unwrap().inbox_id, fake_inbox_id(0x33));
    }

    #[xmtp_common::test]
    fn read_group_metadata_dm_wrong_cardinality_errors() {
        // A 1-element TlsSet<InboxId> is invalid for DM_MEMBERS —
        // surfaces `MalformedComponentValue`.
        let one_element = encode_inbox_id_set(&[fake_inbox_id(0x44)]).unwrap();
        let exts = extensions_with_entries(&[
            (
                ComponentId::CONVERSATION_TYPE.as_u16(),
                encode_conv_type_bytes(2),
            ),
            (
                ComponentId::CREATOR_INBOX_ID.as_u16(),
                encode_creator_bytes(0x44),
            ),
            (ComponentId::DM_MEMBERS.as_u16(), one_element),
        ]);
        let err = read_group_metadata_from_extensions(&exts).unwrap_err();
        assert!(matches!(
            err,
            ComponentSourceError::MalformedComponentValue {
                component_id,
                ..
            } if component_id == ComponentId::DM_MEMBERS
        ));
    }

    /// A `DM_MEMBERS` pair outside a DM is malformed. Commit authorization
    /// reads metadata through this parser, and the pair grants DM-only
    /// authority, so a group persisted by an older client with the pair
    /// must not reach the policy checks.
    // verifies: META-030
    #[xmtp_common::test(unwrap_try = true)]
    fn read_group_metadata_rejects_dm_members_outside_a_dm() {
        let read = |conversation_type: i32| {
            read_group_metadata_from_extensions(&extensions_with_entries(&[
                (
                    ComponentId::CONVERSATION_TYPE.as_u16(),
                    encode_conv_type_bytes(conversation_type),
                ),
                (
                    ComponentId::CREATOR_INBOX_ID.as_u16(),
                    encode_creator_bytes(0x22),
                ),
                (ComponentId::DM_MEMBERS.as_u16(), encode_dm_pair(0x22, 0x33)),
            ]))
        };
        read(2)?;
        for non_dm in [0, 1, 3, 4] {
            assert!(matches!(
                read(non_dm),
                Err(ComponentSourceError::MalformedComponentValue { component_id, .. })
                    if component_id == ComponentId::DM_MEMBERS
            ));
        }
    }

    #[xmtp_common::test]
    fn read_group_metadata_malformed_creator_errors() {
        // CREATOR_INBOX_ID is the versioned `InboxId` TLS encoding —
        // a few stray bytes won't satisfy the varint length prefix
        // plus 32-byte payload, so deserialization must fail loud as
        // `MalformedComponentValue` rather than silently producing
        // a phantom inbox id.
        let exts = extensions_with_entries(&[
            (
                ComponentId::CONVERSATION_TYPE.as_u16(),
                encode_conv_type_bytes(1),
            ),
            (
                ComponentId::CREATOR_INBOX_ID.as_u16(),
                vec![0xFF, 0xFE, 0xFD],
            ),
        ]);
        let err = read_group_metadata_from_extensions(&exts).unwrap_err();
        assert!(matches!(
            err,
            ComponentSourceError::MalformedComponentValue {
                component_id,
                ..
            } if component_id == ComponentId::CREATOR_INBOX_ID
        ));
    }

    // --- read_group_membership_from_dict ------------------------------------

    #[xmtp_common::test]
    fn read_group_membership_happy_path_flattens_per_inbox() {
        use crate::app_data::migration::encode_group_membership_dict;
        use std::collections::BTreeMap;
        use xmtp_proto::xmtp::mls::message_contents::{
            GroupMembershipEntry,
            group_membership_entry::{V1 as GroupMembershipEntryV1, Version},
        };
        let mut entries: BTreeMap<InboxId, GroupMembershipEntry> = BTreeMap::new();
        entries.insert(
            InboxId::from_bytes([0x11; INBOX_ID_BYTE_LEN]),
            GroupMembershipEntry {
                version: Some(Version::V1(GroupMembershipEntryV1 {
                    sequence_id: 7,
                    failed_installations: vec![vec![0xA1; 16]],
                })),
            },
        );
        entries.insert(
            InboxId::from_bytes([0x22; INBOX_ID_BYTE_LEN]),
            GroupMembershipEntry {
                version: Some(Version::V1(GroupMembershipEntryV1 {
                    sequence_id: 42,
                    failed_installations: vec![vec![0xB1; 16]],
                })),
            },
        );
        let bytes = encode_group_membership_dict(&entries).unwrap();
        let exts = extensions_with_entries(&[(ComponentId::GROUP_MEMBERSHIP.as_u16(), bytes)]);

        let proto = read_group_membership_from_dict(&exts).unwrap().unwrap();
        // `members` is flat <hex_inbox_id, seq>
        assert_eq!(proto.members.len(), 2);
        assert_eq!(proto.members.get(&fake_inbox_id(0x11)), Some(&7));
        assert_eq!(proto.members.get(&fake_inbox_id(0x22)), Some(&42));
        // `failed_installations` concatenates the per-inbox lists.
        assert_eq!(proto.failed_installations.len(), 2);
    }

    #[xmtp_common::test]
    fn read_group_membership_malformed_bytes_surface_error() {
        let exts = extensions_with_entries(&[(
            ComponentId::GROUP_MEMBERSHIP.as_u16(),
            vec![0xDE, 0xAD, 0xBE, 0xEF],
        )]);
        let err = read_group_membership_from_dict(&exts).unwrap_err();
        assert!(matches!(
            err,
            ComponentSourceError::MalformedComponentValue {
                component_id,
                ..
            } if component_id == ComponentId::GROUP_MEMBERSHIP
        ));
    }

    // ========================================================================
    // merge_app_data_into_mutable_metadata_from_extensions —
    //   independence from COMPONENT_REGISTRY parseability
    // ========================================================================
    //
    // These pin the call-graph invariant that a malformed
    // `COMPONENT_REGISTRY` does **not** cause `mutable_metadata()` to
    // drop authoritative dict-backed fields. The merge function reads
    // each metadata field directly from the dict; nothing on this
    // read path calls `load_component_registry`. Registry corruption is
    // surfaced loudly on the *write* paths (sender gate in `mls_sync.rs`
    // and the commit validator in `validated_commit.rs`), where it
    // belongs.
    //
    use crate::group_mutable_metadata::{GroupMutableMetadata, MetadataField};

    fn empty_base_gmm() -> GroupMutableMetadata {
        GroupMutableMetadata::new(std::collections::HashMap::new(), Vec::new(), Vec::new())
    }

    #[xmtp_common::test]
    fn merge_with_malformed_registry_returns_valid_field() {
        // Reviewer's alleged "data loss" scenario: COMPONENT_REGISTRY
        // contains malformed bytes, but a metadata field (GROUP_NAME)
        // is present and valid. The merge function does NOT validate
        // the registry — it reads GROUP_NAME directly — so the result
        // must contain "My Group" with no error.
        let exts = extensions_with_entries(&[
            (ComponentId::COMPONENT_REGISTRY.as_u16(), vec![0xCA; 4]),
            (ComponentId::GROUP_NAME.as_u16(), b"My Group".to_vec()),
        ]);
        let mut base = empty_base_gmm();
        merge_app_data_into_mutable_metadata_from_extensions(&mut base, &exts)
            .expect("merge ignores registry parseability and reads field directly");
        assert_eq!(
            base.attributes
                .get(MetadataField::GroupName.as_str())
                .map(String::as_str),
            Some("My Group"),
        );
    }

    #[xmtp_common::test]
    fn merge_with_malformed_field_surfaces_error_not_silent_loss() {
        // The other half of the invariant: when a metadata field's bytes
        // ARE malformed, the merge fails loudly with
        // `MalformedComponentValue` carrying the offending component id —
        // it never silently swallows the value. Pairs with the test
        // above to disprove "all metadata may be lost" framings.
        let exts =
            extensions_with_entries(&[(ComponentId::ADMIN_LIST.as_u16(), vec![0xff, 0xff, 0xff])]);
        let mut base = empty_base_gmm();
        let err =
            merge_app_data_into_mutable_metadata_from_extensions(&mut base, &exts).unwrap_err();
        assert!(
            matches!(
                err,
                ComponentSourceError::MalformedComponentValue { component_id, .. }
                    if component_id == ComponentId::ADMIN_LIST
            ),
            "expected MalformedComponentValue for ADMIN_LIST, got: {err:?}"
        );
    }

    // --- field bounds ------------------------------------------------------

    fn bytes(len: usize) -> VLBytes {
        VLBytes::new(vec![b'a'; len])
    }

    /// A scalar of an application component or `GROUP_IMAGE` may hold 8192
    /// bytes and no more, so one writer cannot fill the commit envelope.
    // verifies: META-068
    #[xmtp_common::test(unwrap_try = true)]
    fn bounded_scalar_holds_at_most_the_element_limit() {
        let app = ComponentId::new(0xC068);
        let registry = registry_with(app, ComponentType::Bytes);
        for id in [app, ComponentId::GROUP_IMAGE] {
            let at_limit = vec![0; MAX_FIELD_ELEMENT_BYTES];
            assert_eq!(
                apply_app_data_update_payload(id, &at_limit, None, &registry)?,
                at_limit
            );
            let over = [0; MAX_FIELD_ELEMENT_BYTES + 1];
            assert!(matches!(
                apply_app_data_update_payload(id, &over, None, &registry),
                Err(ComponentSourceError::FieldBoundExceeded {
                    component_id,
                    len: 8193,
                    max: MAX_FIELD_ELEMENT_BYTES,
                }) if component_id == id
            ));
        }
    }

    /// Each key and value of an application byte map, and each
    /// `USER_DISPLAY_NAME` value, may hold 8192 bytes and no more.
    // verifies: META-068
    #[xmtp_common::test(unwrap_try = true)]
    fn bounded_map_elements_hold_at_most_the_element_limit() {
        let app = ComponentId::new(0xC069);
        let registry = registry_with(app, ComponentType::TlsMapBytesBytes);
        let app_entry = |key: usize, value: usize| {
            let delta = TlsMapDelta::<VLBytes, VLBytes>::new().insert(bytes(key), bytes(value));
            apply_app_data_update_payload(
                app,
                &delta.tls_serialize_detached().unwrap(),
                None,
                &registry,
            )
        };
        let name = |len: usize| {
            let delta = TlsMapDelta::<InboxId, VLBytes>::new().insert(fake_inbox(1), bytes(len));
            apply_app_data_update_payload(
                ComponentId::USER_DISPLAY_NAME,
                &delta.tls_serialize_detached().unwrap(),
                None,
                &registry,
            )
        };
        let over = |result| {
            matches!(
                result,
                Err(ComponentSourceError::FieldBoundExceeded {
                    len: 8193,
                    max: MAX_FIELD_ELEMENT_BYTES,
                    ..
                })
            )
        };

        app_entry(MAX_FIELD_ELEMENT_BYTES, MAX_FIELD_ELEMENT_BYTES)?;
        assert!(over(app_entry(MAX_FIELD_ELEMENT_BYTES + 1, 1)));
        assert!(over(app_entry(1, MAX_FIELD_ELEMENT_BYTES + 1)));
        name(MAX_FIELD_ELEMENT_BYTES)?;
        assert!(over(name(MAX_FIELD_ELEMENT_BYTES + 1)));
    }

    /// A serialized collection may hold 65536 bytes and no more, even when
    /// every element is within its own limit.
    // verifies: META-068
    #[xmtp_common::test(unwrap_try = true)]
    fn bounded_collection_holds_at_most_the_snapshot_limit() {
        let app = ComponentId::new(0xC06A);
        let registry = registry_with(app, ComponentType::TlsMapBytesBytes);
        // Eight 8000-byte entries plus one of `last` bytes.
        let write = |last: usize| {
            let delta = (0..8u8)
                .fold(TlsMapDelta::<VLBytes, VLBytes>::new(), |d, k| {
                    d.insert(VLBytes::new(vec![k]), bytes(8000))
                })
                .insert(VLBytes::new(vec![8]), bytes(last));
            apply_app_data_update_payload(
                app,
                &delta.tls_serialize_detached().unwrap(),
                None,
                &registry,
            )
        };
        // The value length prefix is two bytes from 64 up, so the snapshot
        // grows one byte per value byte from there.
        let last = 64 + MAX_FIELD_SNAPSHOT_BYTES - write(64)?.len();

        assert_eq!(write(last)?.len(), MAX_FIELD_SNAPSHOT_BYTES);
        assert!(matches!(
            write(last + 1),
            Err(ComponentSourceError::FieldBoundExceeded {
                len: 65537,
                max: MAX_FIELD_SNAPSHOT_BYTES,
                ..
            })
        ));
    }

    /// An element over the bound is refused even when a later mutation of
    /// the same payload removes it, so the final value cannot hide it.
    // verifies: META-068
    #[xmtp_common::test(unwrap_try = true)]
    fn apply_bounds_elements_that_the_payload_later_removes() {
        let set = ComponentId::new(0xC06D);
        let map = ComponentId::new(0xC06E);
        let over = || bytes(MAX_FIELD_ELEMENT_BYTES + 1);
        let apply = |id, payload: Vec<u8>, ty| {
            apply_app_data_update_payload(id, &payload, None, &registry_with(id, ty))
        };
        let exceeded = |result| {
            matches!(
                result,
                Err(ComponentSourceError::FieldBoundExceeded { len: 8193, .. })
            )
        };

        let set_payload = TlsSetDelta::<VLBytes>::new()
            .insert(over())
            .remove(over())
            .tls_serialize_detached()?;
        assert!(exceeded(apply(
            set,
            set_payload,
            ComponentType::TlsSetBytes
        )));
        let value_payload = TlsMapDelta::<VLBytes, VLBytes>::new()
            .insert(bytes(1), over())
            .delete(bytes(1))
            .tls_serialize_detached()?;
        assert!(exceeded(apply(
            map,
            value_payload,
            ComponentType::TlsMapBytesBytes
        )));
        let key_payload = TlsMapDelta::<VLBytes, VLBytes>::new()
            .insert(over(), bytes(1))
            .delete(over())
            .tls_serialize_detached()?;
        assert!(exceeded(apply(
            map,
            key_payload,
            ComponentType::TlsMapBytesBytes
        )));
    }

    /// A registry entry padded to `len` bytes by an unknown protobuf field,
    /// which the registry decodes past and stores whole.
    fn padded_registry_entry(len: usize) -> VLBytes {
        let id = ComponentId::new(0xC100);
        let mut entry = registry_with(id, ComponentType::Bytes)
            .get(&id)
            .unwrap()
            .unwrap()
            .encode_to_vec();
        // Field 15, length-delimited, with a two-byte length varint.
        let pad = len - entry.len() - 3;
        assert!((128..16384).contains(&pad));
        entry.extend([0x7A, pad as u8 | 0x80, (pad >> 7) as u8]);
        entry.resize(len, 0);
        VLBytes::new(entry)
    }

    /// A registry entry may hold 8192 bytes and no more, unknown protobuf
    /// fields included: a DM participant may write application-range
    /// entries, and the registry keeps their bytes whole.
    // verifies: META-068
    #[xmtp_common::test(unwrap_try = true)]
    fn registry_entries_hold_at_most_the_element_limit() {
        let registry = ComponentRegistry::new();
        let insert = |len| {
            TlsMapDelta::<ComponentId, VLBytes>::new()
                .insert(ComponentId::new(0xC100), padded_registry_entry(len))
                .tls_serialize_detached()
                .unwrap()
        };
        let apply = |payload: &[u8]| {
            apply_app_data_update_payload(ComponentId::COMPONENT_REGISTRY, payload, None, &registry)
                .map(drop)
        };
        let check = |payload: &[u8]| {
            check_update_payload_bounds(ComponentId::COMPONENT_REGISTRY, payload, &registry)
        };
        let exceeded = |result| {
            matches!(
                result,
                Err(ComponentSourceError::FieldBoundExceeded {
                    len: 8193,
                    max: MAX_FIELD_ELEMENT_BYTES,
                    ..
                })
            )
        };

        let at_limit = insert(MAX_FIELD_ELEMENT_BYTES);
        check(&at_limit)?;
        apply(&at_limit)?;
        let over = insert(MAX_FIELD_ELEMENT_BYTES + 1);
        assert!(exceeded(check(&over)));
        assert!(exceeded(apply(&over)));
    }

    /// A serialized registry may hold 65536 bytes and no more, so repeated
    /// in-bound inserts cannot grow every member's group context without
    /// limit.
    // verifies: META-068
    #[xmtp_common::test(unwrap_try = true)]
    fn registry_holds_at_most_the_snapshot_limit() {
        // Eight 8000-byte entries plus one of `last` bytes.
        let write = |last: usize| {
            let delta = (0..8)
                .fold(TlsMapDelta::<ComponentId, VLBytes>::new(), |d, k| {
                    d.insert(ComponentId::new(0xC100 + k), padded_registry_entry(8000))
                })
                .insert(ComponentId::new(0xC108), padded_registry_entry(last));
            apply_app_data_update_payload(
                ComponentId::COMPONENT_REGISTRY,
                &delta.tls_serialize_detached().unwrap(),
                None,
                &ComponentRegistry::new(),
            )
        };
        // The entry's length prefixes are two bytes from 256 up, so the
        // snapshot grows one byte per entry byte from there.
        let last = 256 + MAX_FIELD_SNAPSHOT_BYTES - write(256)?.len();

        assert_eq!(write(last)?.len(), MAX_FIELD_SNAPSHOT_BYTES);
        assert!(matches!(
            write(last + 1),
            Err(ComponentSourceError::FieldBoundExceeded {
                component_id: ComponentId::COMPONENT_REGISTRY,
                len: 65537,
                max: MAX_FIELD_SNAPSHOT_BYTES,
            })
        ));
    }

    /// The bounds cover application components, `USER_DISPLAY_NAME`,
    /// `GROUP_IMAGE`, and `COMPONENT_REGISTRY` only. Other XMTP components
    /// keep their own limits.
    // verifies: META-068
    #[xmtp_common::test(unwrap_try = true)]
    fn field_bounds_skip_other_xmtp_components() {
        let registered = ComponentId::new(0x80FF);
        let over = [b'a'; MAX_FIELD_ELEMENT_BYTES + 1];
        let registry = registry_with(registered, ComponentType::Bytes);
        apply_app_data_update_payload(registered, &over, None, &registry)?;
        apply_app_data_update_payload(ComponentId::GROUP_DESCRIPTION, &over, None, &registry)?;
    }

    /// A payload's element bound covers every byte key and value it names,
    /// removal keys included: an oversized key can never match a stored one,
    /// so admitting it only spends proposal storage.
    // verifies: META-068
    #[xmtp_common::test(unwrap_try = true)]
    fn payload_bounds_measure_every_named_element() {
        let set = ComponentId::new(0xC06B);
        let map = ComponentId::new(0xC06C);
        let registries = [
            registry_with(set, ComponentType::TlsSetBytes),
            registry_with(map, ComponentType::TlsMapBytesBytes),
        ];
        let over = MAX_FIELD_ELEMENT_BYTES + 1;
        let check = |id, payload: Vec<u8>| {
            check_update_payload_bounds(id, &payload, &registries[usize::from(id == map)])
        };
        let exceeded = |result| matches!(result, Err(ComponentSourceError::FieldBoundExceeded { len, .. }) if len == over);

        let set_delta = |delta: TlsSetDelta<VLBytes>| delta.tls_serialize_detached().unwrap();
        check(
            set,
            set_delta(TlsSetDelta::new().insert(bytes(MAX_FIELD_ELEMENT_BYTES))),
        )?;
        assert!(exceeded(check(
            set,
            set_delta(TlsSetDelta::new().insert(bytes(over)))
        )));
        assert!(exceeded(check(
            set,
            set_delta(TlsSetDelta::new().remove(bytes(over)))
        )));
        check(
            set,
            set_delta(TlsSetDelta::new().remove(bytes(MAX_FIELD_ELEMENT_BYTES))),
        )?;

        let map_delta =
            |delta: TlsMapDelta<VLBytes, VLBytes>| delta.tls_serialize_detached().unwrap();
        assert!(exceeded(check(
            map,
            map_delta(TlsMapDelta::new().update(bytes(over), bytes(1)))
        )));
        assert!(exceeded(check(
            map,
            map_delta(TlsMapDelta::new().update(bytes(1), bytes(over)))
        )));
        assert!(exceeded(check(
            map,
            map_delta(TlsMapDelta::new().delete(bytes(over)))
        )));
        check(
            map,
            map_delta(TlsMapDelta::new().delete(bytes(MAX_FIELD_ELEMENT_BYTES))),
        )?;

        check(ComponentId::GROUP_DESCRIPTION, vec![b'a'; over])?;
    }
}