xmtp_mls/

client.rs

use crate::{
    attachments::Attachments,
    context::XmtpSharedContext,
    groups::{
        ConversationListItem, GroupError, MlsGroup, group_permissions::PolicySet,
        welcome_sync::WelcomeService,
    },
    identity::{Identity, IdentityError, parse_credential},
    identity_updates::{
        IdentityUpdateError, IdentityUpdates, load_identity_updates,
        load_identity_updates_for_client,
    },
    inbox_handles::{self, HandleError, IdentityTarget, account_targets, resolve_verified_in},
    mls_store::{MlsStore, MlsStoreError},
    server_configuration::{BlockedConnection, ServerConfigError},
    subscriptions::{
        LocalEventError,
        internal::{emit_consent_changes, emit_deleted_messages},
    },
    utils::VersionInfo,
    worker::device_sync::{DeviceSyncClient, worker::SyncMetric},
    worker::{WorkerRunner, metrics::WorkerMetrics},
};
pub mod notifications;
pub use crate::core::config::DeviceSync;
use crate::subscriptions::catch_up::{CatchUpError, CatchUpSummary};
#[cfg(not(target_arch = "wasm32"))]
use crate::subscriptions::router_callbacks::{self, ApiClientIdentity};
#[cfg(any(test, feature = "test-utils"))]
use crate::subscriptions::stream_messages::stream_stats::StreamWithStats;
use crate::{
    core::events::PreferenceUpdate,
    subscriptions::{self, SubscribeError},
};
use crate::{
    groups::welcome_sync::GroupSyncSummary,
    identity_updates::{batch_get_association_state_with_verifier, get_creation_signature_kind},
    messages::{
        decoded_message::DecodedMessage,
        enrichment::{EnrichMessageError, enrich_messages},
    },
};
use futures::Stream;
use notifications::{NotificationConfig, NotificationError, NotificationState};
use openmls::prelude::tls_codec::Error as TlsCodecError;
use std::{
    collections::HashMap,
    sync::{Arc, atomic::Ordering},
};
use thiserror::Error;
use xmtp_api::{ApiClientWrapper, XmtpApi};
use xmtp_common::{ErrorCode, Event, Retry, retry_async, retryable};
use xmtp_common::{MaybeSend, StreamHandle};
use xmtp_configuration::{
    CONVERSATION_LIST_DEFAULT_LIMIT, CREATE_PQ_KEY_PACKAGE_EXTENSION,
    KEY_PACKAGE_ROTATION_INTERVAL_NS, LOCAL_PAGE_MAX_LIMIT,
};
use xmtp_cryptography::signature::IdentifierValidationError;
use xmtp_db::TransactionOutcome::Continue;
use xmtp_db::consent_record::StoredConsentRecord;
use xmtp_db::{
    ConnectionExt, NotFound, StorageError, TransactionOutcome, XmtpDb,
    consent_record::{
        ConsentActError, ConsentKey, ConsentState, ConsentType, ConsentWrite, Decision, Owed,
        conversation_consent, write_consent,
    },
    db_connection::DbConnection,
    encrypted_store::conversation_list::ConversationListItem as DbConversationListItem,
    group::{ConversationType, GroupMembershipState, GroupQueryArgs},
    group_message::StoredGroupMessage,
    identity::StoredIdentity,
};
use xmtp_db::{group::GroupQueryOrderBy, prelude::*};
use xmtp_events::DeletionCause;
use xmtp_id::associations::handle::Handle;
use xmtp_id::key_package::{KeyPackageVerificationError, VerifiedKeyPackageV2};
use xmtp_id::{
    AsIdRef, InboxId, InboxIdRef, InboxOwner,
    associations::{
        AssociationError, AssociationState, Identifier, MemberIdentifier, SignatureError,
        builder::{SignatureRequest, SignatureRequestError},
    },
    scw_verifier::SmartContractSignatureVerifier,
};
use xmtp_macro::log_event;
use xmtp_mls_group_state::{
    group::{DMMetadataOptions, GroupMetadataOptions},
    group_metadata::DmMembers,
    group_mutable_metadata::MessageDisappearingSettings,
};
#[cfg(not(target_arch = "wasm32"))]
use xmtp_proto::api_client::XmtpMlsBidiStreams;
use xmtp_proto::api_client::XmtpMlsStreams;
use xmtp_proto::types::InstallationId;
use xmtp_proto::{
    api::HasStats,
    api_client::{ApiStats, IdentityStats, XmtpBackendClient},
};

use xmtp_proto::types::GroupId;
/// Enum representing the network the Client is connected to
#[derive(Clone, Copy, Default, Debug)]
pub enum Network {
    Local(&'static str),
    #[default]
    Dev,
    Prod,
}

/// Timeout for waiting until a registration publish can be read.
#[derive(Debug, Clone)]
pub struct VisibilityConfirmationOptions {
    pub timeout_ms: u64,
}

const REGISTRATION_INITIAL_BACKOFF: std::time::Duration = std::time::Duration::from_millis(50);
const REGISTRATION_MAX_BACKOFF: std::time::Duration = std::time::Duration::from_secs(1);

impl Default for VisibilityConfirmationOptions {
    fn default() -> Self {
        Self { timeout_ms: 30_000 }
    }
}

/// An app's consent decision about one entity. It carries no
/// time: the client times it.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ConsentAct {
    pub entity_type: ConsentType,
    pub entity: String,
    pub state: ConsentState,
}

impl ConsentAct {
    pub fn new(entity_type: ConsentType, state: ConsentState, entity: String) -> Self {
        Self {
            entity_type,
            entity,
            state,
        }
    }
}

#[derive(Debug, Error, ErrorCode)]
pub enum ClientError {
    #[error(transparent)]
    #[error_code(inherit)]
    AddressValidation(#[from] IdentifierValidationError),
    /// Could not publish.
    ///
    /// Failed to publish messages to the network. May be retryable.
    #[error("could not publish: {0}")]
    PublishError(String),
    /// Storage error.
    ///
    /// Database operation failed. May be retryable.
    #[error("storage error: {0}")]
    Storage(#[from] StorageError),
    /// API error.
    ///
    /// Network request to XMTP backend failed. Retryable.
    #[error("API error: {0}")]
    Api(#[from] xmtp_api::ApiError),
    /// Identity error.
    ///
    /// Problem with identity operations. Not retryable.
    #[error("identity error: {0}")]
    Identity(#[from] crate::identity::IdentityError),
    /// TLS Codec error.
    ///
    /// Encoding/decoding MLS TLS structures failed. Not retryable.
    #[error("TLS Codec error: {0}")]
    TlsError(#[from] TlsCodecError),
    /// Key package verification failed.
    ///
    /// Invalid key package received from network. Not retryable.
    #[error("key package verification: {0}")]
    KeyPackageVerification(#[from] KeyPackageVerificationError),
    /// Stream inconsistency.
    ///
    /// Message stream state became inconsistent. Not retryable.
    #[error("Stream inconsistency error: {0}")]
    StreamInconsistency(String),
    /// Association error.
    ///
    /// Identity association operation failed. Not retryable.
    #[error("Association error: {0}")]
    Association(#[from] AssociationError),
    /// Signature validation error.
    ///
    /// A signature failed verification. Not retryable.
    #[error("signature validation error: {0}")]
    SignatureValidation(#[from] SignatureError),
    /// Identity update error.
    ///
    /// Failed to process identity update. Not retryable.
    #[error(transparent)]
    IdentityUpdate(#[from] IdentityUpdateError),
    /// Signature request error.
    ///
    /// Failed to create/process signature request. Not retryable.
    #[error(transparent)]
    SignatureRequest(#[from] SignatureRequestError),
    /// Group error.
    ///
    /// Group operation failed. May be retryable.
    // the box is to prevent infinite cycle between client and group errors
    #[error(transparent)]
    Group(Box<GroupError>),
    /// Local event error.
    ///
    /// Failed to process local event. Not retryable.
    #[error(transparent)]
    LocalEvent(#[from] LocalEventError),
    /// Database connection error.
    ///
    /// Connection to database failed. Retryable.
    #[error(transparent)]
    Db(#[from] xmtp_db::ConnectionError),
    /// Message decode failed.
    ///
    /// Enrichment returned no decoded message for a stored message. Not
    /// retryable.
    #[error("Failed to decode message")]
    MessageDecodeFailed,
    /// MLS store error.
    ///
    /// OpenMLS key store operation failed. Not retryable.
    #[error(transparent)]
    MlsStore(#[from] MlsStoreError),
    /// Message enrichment error.
    ///
    /// Failed to enrich message content. Not retryable.
    #[error(transparent)]
    EnrichMessage(#[from] EnrichMessageError),
    /// Conversion Error
    ///
    /// Data type failed to convert. Not retryable.
    #[error(transparent)]
    Conversion(#[from] xmtp_proto::ConversionError),
    #[error(transparent)]
    #[error_code(inherit)]
    Handle(#[from] crate::inbox_handles::HandleError),
    /// Registration not visible.
    ///
    /// The registration receipt was invalid or was not visible before the timeout.
    /// Call `register_identity` again to retry a pending registration.
    #[error("Registration is not visible")]
    RegistrationNotVisible,
    /// Client is closed.
    ///
    /// Operation was attempted on a client that has been shut down via
    /// `Client::close`. Not retryable — build a new client instead.
    #[error("client is closed")]
    AlreadyClosed,
    /// Server configuration unavailable.
    ///
    /// The backend did not serve its configuration, or the answer could not be
    /// stored. A backend older than spec 006 answers `UNIMPLEMENTED`; there is
    /// no compatibility shim. Retryable exactly when the wrapped failure is:
    /// an unreachable backend is worth another attempt, `UNIMPLEMENTED` is not.
    #[error("server configuration unavailable: {0}")]
    ConfigurationUnavailable(#[source] Box<crate::server_configuration::ConfigurationFetchError>),
    /// Server configuration invalid.
    ///
    /// The backend published a configuration this client cannot use: a missing
    /// or malformed identifier, a minimum version that does not parse, or a
    /// chain that is not a CAIP-2 identifier. Not retryable.
    #[error("server configuration invalid: {0}")]
    ConfigurationInvalid(#[from] xmtp_configuration::ServerConfigurationError),
    /// Backend mismatch.
    ///
    /// This database is bound to one backend deployment and a different one
    /// answered. Not retryable — use a database created for the backend this
    /// app now points at.
    #[error("this database is bound to backend {stored}, but {received} answered")]
    BackendMismatch { stored: String, received: String },
    /// Client version too old.
    ///
    /// The backend requires a newer libxmtp than this build. Not retryable —
    /// ship an updated client.
    #[error("client version {client} is below the {minimum} this backend requires")]
    ClientVersionTooOld { client: String, minimum: String },
    /// Authentication required.
    ///
    /// The backend requires a credential and none was configured. Not
    /// retryable — supply an auth callback or handle before building.
    #[error(
        "this backend requires authentication; required scopes: [{}]",
        required_scopes.join(", ")
    )]
    AuthRequired { required_scopes: Vec<String> },
    /// Chain not accepted.
    ///
    /// The backend does not verify smart contract wallet signatures on this
    /// chain. Not retryable — use a chain the deployment accepts.
    #[error(
        "this backend does not accept smart contract wallet signatures on {chain}; accepted: [{}]",
        accepted.join(", ")
    )]
    ChainNotAccepted {
        chain: String,
        accepted: Vec<String>,
    },
    /// Invalid list options.
    ///
    /// A conversation listing named a limit below 1 or above the maximum page
    /// size, a before conversation id without the before time of its list
    /// order, or sync or one-shot groups. Not retryable — fix the options.
    #[error("invalid list options: {0}")]
    InvalidListOptions(&'static str),
    /// Consent act refused.
    ///
    /// The consent entity cannot be stored, or names a conversation the
    /// client does not hold. Not retryable.
    #[error("consent act: {0}")]
    #[error_code(inherit)]
    ConsentAct(#[from] ConsentActError),
}

impl ClientError {
    #[deny(clippy::wildcard_enum_match_arm)]
    pub fn db_needs_connection(&self) -> bool {
        use crate::worker::NeedsDbReconnect;
        match self {
            Self::Storage(s) | Self::ConsentAct(ConsentActError::Storage(s)) => {
                s.db_needs_connection()
            }
            Self::Db(c) => c.db_needs_connection(),
            Self::Group(error) => error.needs_db_reconnect(),
            Self::Identity(error) => error.needs_db_reconnect(),
            Self::IdentityUpdate(error) => error.needs_db_reconnect(),
            Self::MlsStore(error) => error.needs_db_reconnect(),
            Self::ConfigurationUnavailable(error) => error.db_needs_connection(),
            Self::ConsentAct(_)
            | Self::AddressValidation(..)
            | Self::PublishError(..)
            | Self::Api(..)
            | Self::TlsError(..)
            | Self::KeyPackageVerification(..)
            | Self::StreamInconsistency(..)
            | Self::Association(..)
            | Self::SignatureValidation(..)
            | Self::SignatureRequest(..)
            | Self::LocalEvent(..)
            | Self::MessageDecodeFailed
            | Self::EnrichMessage(..)
            | Self::Conversion(..)
            | Self::Handle(..)
            | Self::RegistrationNotVisible
            | Self::AlreadyClosed
            | Self::ConfigurationInvalid(..)
            | Self::BackendMismatch { .. }
            | Self::ClientVersionTooOld { .. }
            | Self::AuthRequired { .. }
            | Self::ChainNotAccepted { .. }
            | Self::InvalidListOptions(..) => false,
        }
    }
}

impl From<NotFound> for ClientError {
    fn from(value: NotFound) -> Self {
        ClientError::Storage(StorageError::NotFound(value))
    }
}

impl From<GroupError> for ClientError {
    fn from(err: GroupError) -> ClientError {
        ClientError::Group(Box::new(err))
    }
}

impl From<BlockedConnection> for ClientError {
    fn from(blocked_connection: BlockedConnection) -> Self {
        match blocked_connection {
            BlockedConnection::BackendMismatch { stored, received } => {
                ClientError::BackendMismatch { stored, received }
            }
            BlockedConnection::ClientVersionTooOld { client, minimum } => {
                ClientError::ClientVersionTooOld { client, minimum }
            }
        }
    }
}

impl From<ServerConfigError> for ClientError {
    fn from(error: ServerConfigError) -> Self {
        match error {
            ServerConfigError::Unavailable(error) => ClientError::ConfigurationUnavailable(error),
            ServerConfigError::Invalid(error) => ClientError::ConfigurationInvalid(error),
            ServerConfigError::Blocked(blocked_connection) => blocked_connection.into(),
            ServerConfigError::Closed => ClientError::AlreadyClosed,
        }
    }
}

impl xmtp_common::RetryableError for ClientError {
    fn is_retryable(&self) -> bool {
        match self {
            ClientError::Group(group_error) => retryable!(group_error),
            ClientError::IdentityUpdate(error) => retryable!(error),
            ClientError::Api(api_error) => retryable!(api_error),
            ClientError::Storage(storage_error) => retryable!(storage_error),
            ClientError::Db(db) => retryable!(db),
            // SCW verification errors carry retryability through SignatureError;
            // transient RPC provider failures must not advance the welcome cursor.
            // See xmtp/libxmtp#3394.
            ClientError::SignatureValidation(e) => retryable!(e),
            // A backend that was briefly unreachable, or a database that was
            // briefly locked, is worth another attempt: the refresh worker's
            // backoff depends on this answer.
            ClientError::ConfigurationUnavailable(e) => retryable!(e),
            _ => false,
        }
    }

    fn rate_limit_delay(&self) -> Option<xmtp_common::time::Duration> {
        match self {
            ClientError::Group(group_error) => group_error.rate_limit_delay(),
            ClientError::IdentityUpdate(error) => error.rate_limit_delay(),
            ClientError::Api(api_error) => api_error.rate_limit_delay(),
            ClientError::SignatureValidation(e) => e.rate_limit_delay(),
            ClientError::ConfigurationUnavailable(e) => e.rate_limit_delay(),
            _ => None,
        }
    }
}

/// The structured stream failure in `error`, read as
/// [`stream_failure_details`](crate::subscriptions::stream_failure::stream_failure_details)
/// does, and also through [`ClientError::Group`], whose `source()` skips the
/// group error because the variant is transparent.
pub fn stream_failure_details(
    error: &(dyn std::error::Error + 'static),
) -> Option<crate::subscriptions::stream_failure::StreamFailureDetails> {
    crate::subscriptions::stream_failure::stream_failure_details_opening(error, |error| {
        match error.downcast_ref::<ClientError>() {
            Some(ClientError::Group(error)) => Some(error.as_ref()),
            _ => None,
        }
    })
}

/// Clients manage access to the network, identity, and data store
pub struct Client<Context> {
    pub context: Context,
    pub installation_id: InstallationId,
    pub(crate) workers: Arc<WorkerRunner>,
    pub(crate) app_lifetime: Arc<AppLifetime>,
}

pub(crate) struct AppLifetime {
    pub(crate) events: xmtp_events::EventBus<crate::subscriptions::internal::InternalEvent>,
}

impl Drop for AppLifetime {
    fn drop(&mut self) {
        self.events.close_app_subscriptions();
    }
}

impl<Context> Drop for Client<Context> {
    fn drop(&mut self) {
        log_event!(Event::ClientDropped, self.installation_id);
    }
}

// most of these things are `Arc`'s
impl<Context: Clone> Clone for Client<Context> {
    fn clone(&self) -> Self {
        Self {
            context: self.context.clone(),
            installation_id: self.installation_id,
            workers: self.workers.clone(),
            app_lifetime: self.app_lifetime.clone(),
        }
    }
}

impl<Context> Client<Context>
where
    Context: XmtpSharedContext,
{
    pub fn identity_updates(&self) -> IdentityUpdates<&Context> {
        IdentityUpdates::new(&self.context)
    }

    pub fn mls_store(&self) -> MlsStore<Context> {
        MlsStore::new(self.context.clone())
    }

    pub fn scw_verifier(&self) -> Arc<Box<dyn SmartContractSignatureVerifier>> {
        self.context.scw_verifier()
    }

    pub fn version_info(&self) -> &VersionInfo {
        self.context.version_info()
    }
}

impl<Context> Client<Context>
where
    Context: XmtpSharedContext,
    Context::ApiClient: HasStats,
{
    pub fn api_stats(&self) -> ApiStats {
        self.context.api().api_client.mls_stats()
    }

    pub fn identity_api_stats(&self) -> IdentityStats {
        self.context.api().api_client.identity_stats()
    }

    pub fn clear_stats(&self) {
        self.context.api().api_client.mls_stats().clear();
        self.context.api().api_client.identity_stats().clear();
    }
}

/// Get the [`AssociationState`] for each `inbox_id`
pub async fn inbox_addresses_with_verifier<ApiClient: XmtpApi>(
    api_client: &ApiClientWrapper<ApiClient>,
    conn: &impl DbQuery,
    inbox_ids: Vec<InboxIdRef<'_>>,
    scw_verifier: &impl SmartContractSignatureVerifier,
) -> Result<Vec<AssociationState>, ClientError> {
    load_identity_updates(api_client, conn, &inbox_ids).await?;
    let state = batch_get_association_state_with_verifier(
        conn,
        &inbox_ids.into_iter().map(|i| (i, None)).collect::<Vec<_>>(),
        scw_verifier,
    )
    .await?;
    Ok(state)
}

impl<Context> Client<Context>
where
    Context: XmtpSharedContext + 'static,
{
    /// Reconnect to the client's database if it has previously been released
    pub fn reconnect_db(&self) -> Result<(), ClientError> {
        // Hold the call gate until the workers start. `close` waits for this
        // call before it stops the workers, so it also stops the ones started
        // here.
        let _call = self
            .context
            .foreground_calls()
            .enter()
            .ok_or(ClientError::AlreadyClosed)?;
        if self.context.is_closed() {
            return Err(ClientError::AlreadyClosed);
        }
        #[cfg(all(test, not(target_arch = "wasm32")))]
        test_hooks::pause_reconnect(&self.context.installation_id().to_vec());
        self.context.db().reconnect().map_err(StorageError::from)?;
        self.workers.spawn(self.context.clone());
        Ok(())
    }

    /// Cleanly shut down this client: cancel in-flight workers and streams,
    /// then release the DB connection. Idempotent — a second call is `Ok(())`.
    ///
    /// Callers (notably the Node binding consumers) should `await` this before
    /// deleting the SQLite file or dropping the client wrapper, to avoid late
    /// log spew from detached workers/streams firing against a dead DB.
    pub async fn close(&self) -> Result<(), ClientError> {
        // `shutdown_complete` is distinct from `is_closed()` (which reflects
        // cancellation): only set after the DB actually disconnects. If
        // `disconnect()` errors below, callers can retry `close()` and we'll
        // attempt disconnect again rather than silently short-circuiting.
        if self.context.shutdown_complete() {
            return Ok(());
        }
        self.context
            .api()
            .api_client
            .unregister_client_event_writer(self.context.public_event_writer());
        self.context.events().close_app_subscriptions();
        self.context.cancellation_token().cancel();
        // A running host call can hold a pooled connection, and a running
        // reconnect can start workers. Wait for both before the workers stop
        // and the database disconnects.
        self.context.foreground_calls().close_and_wait().await;
        let delivery_result = self.context.close_message_delivery();
        self.workers.shutdown().await;
        self.context.events().close_internal_subscriptions();
        // Keep the database connected when lease release fails so a later
        // close call can retry it.
        delivery_result?;
        self.context
            .db()
            .disconnect()
            .map_err(xmtp_db::StorageError::from)?;
        self.context.mark_shutdown_complete();
        log_event!(Event::ClientClosed, self.installation_id);
        Ok(())
    }

    /// yields until the sync worker notifies that it is initialized and running.
    pub async fn wait_for_sync_worker_init(&self) {
        self.workers.wait_for_sync_worker_init().await;
    }

    pub fn sync_metrics(&self) -> Option<Arc<WorkerMetrics<SyncMetric>>> {
        self.workers.sync_metrics()
    }
}

impl<Context> Client<Context>
where
    Context: XmtpSharedContext,
{
    /// What this deployment published about itself, as resolved at build.
    /// A refresh rewrites the stored copy; it never changes
    /// this value.
    pub fn server_configuration(&self) -> &xmtp_configuration::ServerConfiguration {
        self.context.server_configuration().configuration()
    }

    /// Fetch the deployment configuration now and rewrite the stored copy.
    ///
    /// Applies the same validation, storage, and identifier
    /// binding the refresh worker applies. The snapshot this client is
    /// holding is unchanged: a new value takes effect at the next build.
    pub async fn refresh_server_configuration(
        &self,
    ) -> Result<xmtp_configuration::ServerConfiguration, ClientError> {
        Ok(crate::core::preflight::refresh(&self.context).await?)
    }

    /// Retrieves the client's installation public key, sometimes also called `installation_id`
    pub fn installation_public_key(&self) -> InstallationId {
        self.context.installation_id()
    }
    /// Retrieves the client's inbox ID
    pub fn inbox_id(&self) -> InboxIdRef<'_> {
        self.context.identity().inbox_id()
    }

    /// get a reference to the monolithic Database object where
    /// higher-level queries are defined
    pub fn db(&self) -> <Context::Db as XmtpDb>::DbQuery {
        self.context.db()
    }

    /// This associates an installation_id with a human-readable
    /// name and makes the logs a little easier to read.
    #[cfg(any(test, feature = "test-utils"))]
    pub fn set_name(&self, name: &str) {
        log_event!(Event::AssociateName, self.context.installation_id(), name);
    }

    pub fn device_sync_client(&self) -> DeviceSyncClient<Context> {
        let metrics = self.context.sync_metrics();
        DeviceSyncClient::new(
            self.context.clone(),
            metrics.unwrap_or(Arc::new(WorkerMetrics::new(self.context.installation_id()))),
        )
    }

    /// The inbox that holds `identifier`, or `None`. The backend's answer
    /// counts only when the newest state the client derives for the answered
    /// inbox holds the identifier.
    pub async fn find_inbox_id_from_identifier(
        &self,
        identifier: Identifier,
    ) -> Result<Option<String>, ClientError> {
        Ok(
            resolve_verified_in(&self.context, &[IdentityTarget::Account(identifier)])
                .await?
                .pop()
                .flatten(),
        )
    }

    /// Get the highest `sequence_id` from the local database for the client's `inbox_id`.
    /// This may not be consistent with the latest state on the backend.
    pub fn inbox_sequence_id(
        &self,
        conn: &DbConnection<<Context::Db as XmtpDb>::Connection>,
    ) -> Result<i64, StorageError> {
        self.context
            .identity()
            .sequence_id(conn)
            .map_err(Into::into)
    }

    /// Get the [`AssociationState`] for the client's `inbox_id`
    pub async fn inbox_state(
        &self,
        refresh_from_network: bool,
    ) -> Result<AssociationState, ClientError> {
        let conn = self.context.db();
        let inbox_id = self.inbox_id();
        if refresh_from_network {
            load_identity_updates_for_client(&self.context, &conn, &[inbox_id]).await?;
        }
        let identity_service = IdentityUpdates::new(&self.context);
        let state = identity_service
            .get_association_state(&conn, inbox_id, None)
            .await?;
        Ok(state)
    }

    /// Get the [`AssociationState`] for each `inbox_id`
    pub async fn inbox_addresses(
        &self,
        refresh_from_network: bool,
        inbox_ids: Vec<InboxIdRef<'_>>,
    ) -> Result<Vec<AssociationState>, ClientError> {
        let conn = self.context.db();
        if refresh_from_network {
            load_identity_updates_for_client(&self.context, &conn, &inbox_ids).await?;
        }
        let identity_service = IdentityUpdates::new(&self.context);
        let state = identity_service
            .batch_get_association_state(
                &conn,
                &inbox_ids.into_iter().map(|i| (i, None)).collect::<Vec<_>>(),
            )
            .await?;
        Ok(state)
    }

    /// Get the total number of inbox updates for `inbox_ids`. `refresh_from_network` will force
    /// a network refresh. May still access network if an inbox_id does not yet exist in the local
    /// cache.
    pub async fn fetch_inbox_updates_count(
        &self,
        refresh_from_network: bool,
        inbox_ids: Vec<InboxIdRef<'_>>,
    ) -> Result<HashMap<InboxId, u32>, ClientError> {
        let conn = self.context.db();
        if refresh_from_network {
            load_identity_updates_for_client(&self.context, &conn, &inbox_ids).await?;
        }
        let inbox_id_strs = inbox_ids.to_vec();
        let counts = conn.count_inbox_updates(&inbox_id_strs)?;
        Ok(counts.into_iter().map(|(k, v)| (k, v as u32)).collect())
    }

    /// Get the total number of inbox updates for the client's inbox_id.
    /// Setting `refresh_from_network` forces a network refresh, otherwise
    /// this operation is offline.
    pub async fn fetch_own_inbox_updates_count(
        &self,
        refresh_from_network: bool,
    ) -> Result<u32, ClientError> {
        let inbox_id = self.inbox_id();
        Ok(self
            .fetch_inbox_updates_count(refresh_from_network, vec![inbox_id])
            .await?
            .get(inbox_id)
            .copied()
            .unwrap_or(0))
    }

    /// Get the signature kind used to create an inbox.
    ///
    /// # Arguments
    /// * `inbox_id` - The inbox ID to check
    /// * `refresh_from_network` - Whether to fetch updates from the network first
    ///
    /// # Returns
    /// * `Some(SignatureKind)` - The signature kind used to create the inbox
    /// * `None` - Inbox doesn't exist or creation info is unavailable
    pub async fn inbox_creation_signature_kind(
        &self,
        inbox_id: InboxIdRef<'_>,
        refresh_from_network: bool,
    ) -> Result<Option<xmtp_id::associations::SignatureKind>, ClientError> {
        let conn = self.context.db();

        // Load the first identity update (creation update) for this inbox if requested
        if refresh_from_network {
            load_identity_updates_for_client(&self.context, &conn, &[inbox_id]).await?;
        }

        let verifier = self.context.scw_verifier();

        let signature_kind = get_creation_signature_kind(&conn, verifier, inbox_id).await?;

        Ok(signature_kind)
    }

    /// Store the app's consent decisions in one transaction. An
    /// act on a group id the client does not hold fails the whole call, and
    /// nothing is stored.
    pub async fn set_consent_states(&self, acts: &[ConsentAct]) -> Result<(), ClientError> {
        let own = self.context.inbox_id();
        let keys = acts
            .iter()
            .map(|act| ConsentKey::for_app(&self.context.db(), own, act.entity_type, &act.entity))
            .collect::<Result<Vec<_>, _>>()?;
        crate::state_tx::state_write_with_events(
            self.context.mls_storage(),
            self.context.events(),
            |tx, events| {
                let storage = tx.storage();
                let db = storage.db();
                let mut changes = vec![];
                for (key, act) in keys.iter().zip(acts) {
                    let decision = ConsentWrite::Decide(Decision::Set(act.state));
                    changes.extend(write_consent(&db, key, decision, Owed::Mark)?);
                }
                emit_consent_changes(&self.context, events, &db, &changes)?;
                Ok::<_, StorageError>(Continue(()))
            },
        )?
        .into_continued();

        Ok(())
    }

    /// The consent an app reads for an entity: a conversation's under
    /// its key, held or not, or an inbox's record.
    // implements: CONS-001
    pub async fn get_consent_state(
        &self,
        entity_type: ConsentType,
        entity: String,
    ) -> Result<ConsentState, ClientError> {
        let db = self.context.db();
        let Some(key) = ConsentKey::parse(entity_type, &entity) else {
            return Ok(ConsentState::Unknown);
        };
        Ok(match key.group_id() {
            Some(id) => conversation_consent(&db, self.context.inbox_id(), &id)?,
            None => db
                .consent_row(&key)?
                .map_or(ConsentState::Unknown, |r| r.state),
        })
    }

    /// Release the client's database connection
    pub fn release_db_connection(&self) -> Result<(), ClientError> {
        self.context
            .db()
            .disconnect()
            .map_err(xmtp_db::StorageError::from)?;
        Ok(())
    }

    /// Get a reference to the client's identity struct
    pub fn identity(&self) -> &Identity {
        self.context.identity()
    }

    /// Ensures identity is ready before performing operations.
    /// Call `register_identity()` first if this fails.
    pub(crate) fn ensure_identity_ready<E: From<BlockedConnection> + From<IdentityError>>(
        &self,
    ) -> Result<(), E> {
        // Once the connection is blocked, every later call fails with the
        // reason. This is the gate every client-level operation already passes
        // through, so the check costs nothing extra.
        self.context.server_configuration().check()?;
        if !self.identity().is_ready() {
            tracing::warn!(
                inbox_id = %self.inbox_id(),
                "Operation attempted before register_identity() was called"
            );
            return Err(IdentityError::UninitializedIdentity.into());
        }
        Ok(())
    }

    /// Create a new group with the default settings
    /// Applies a custom [`PolicySet`] to the group if one is specified
    pub fn create_group(
        &self,
        permissions_policy_set: Option<PolicySet>,
        opts: Option<GroupMetadataOptions>,
    ) -> Result<MlsGroup<Context>, ClientError> {
        self.ensure_identity_ready::<ClientError>()?;

        let group: MlsGroup<Context> = MlsGroup::create_and_insert(
            self.context.clone(),
            ConversationType::Group,
            permissions_policy_set.unwrap_or_default(),
            opts.unwrap_or_default(),
            None,
        )?;

        log_event!(
            Event::CreatedGroup,
            self.context.installation_id(),
            group_id = group.group_id
        );

        Ok(group)
    }

    /// Create a group with an initial set of members added. Every account
    /// resolves before the group is stored, as do member identities and key
    /// packages. An account with no inbox fails with [`GroupError::AddressNotFound`].
    pub async fn create_group_with_identifiers(
        &self,
        account_identifiers: &[Identifier],
        permissions_policy_set: Option<PolicySet>,
        opts: Option<GroupMetadataOptions>,
    ) -> Result<MlsGroup<Context>, ClientError> {
        self.create_group_with_targets(
            &account_targets(account_identifiers),
            permissions_policy_set,
            opts,
        )
        .await
    }

    /// Returns the stored group if its first add is still pending when the wait
    /// ends. A later sync completes the add.
    #[tracing::instrument(level = "debug", skip_all, fields(size = inbox_ids.len()))]
    pub async fn create_group_with_members(
        &self,
        inbox_ids: &[impl AsIdRef],
        permissions_policy_set: Option<PolicySet>,
        opts: Option<GroupMetadataOptions>,
    ) -> Result<MlsGroup<Context>, ClientError> {
        tracing::info!("creating group");
        self.ensure_identity_ready::<ClientError>()?;
        let opts = opts.unwrap_or_default();
        let draft = MlsGroup::<Context>::prepare_new(
            &self.context,
            None,
            ConversationType::Group,
            permissions_policy_set.unwrap_or_default(),
            &opts,
            None,
        )?;
        let ids = inbox_ids.iter().map(AsIdRef::as_ref).collect::<Vec<_>>();
        // implements: GMOD-056
        let add = MlsGroup::<Context>::initial_add(&self.context, &draft, &ids).await?;
        let (stored, intent) = MlsGroup::<Context>::store_new(
            &self.context,
            draft,
            GroupMembershipState::Allowed,
            ConversationType::Group,
            opts,
            true,
            add,
        )?;
        let group = MlsGroup::new(
            self.context.clone(),
            stored.id,
            stored.dm_id,
            stored.conversation_type,
            stored.created_at_ns,
        );
        log_event!(
            Event::CreatedGroup,
            self.context.installation_id(),
            group_id = group.group_id
        );
        let Some(intent) = intent else {
            return Ok(group);
        };
        match group.sync_until_intent_resolved(intent.id).await {
            Err(error) if !group.intent_pending(intent.id)? => return Err(error.into()),
            Err(error) => tracing::warn!(?error, "first add pending; a sync finishes it"),
            Ok(_) => {}
        }

        Ok(group)
    }

    /// Create a new Direct Message with the default settings
    #[tracing::instrument(level = "debug", skip_all)]
    async fn create_dm_by_inbox_id(
        &self,
        target_inbox_id: InboxId,
        opts: Option<DMMetadataOptions>,
    ) -> Result<MlsGroup<Context>, ClientError> {
        let group: MlsGroup<Context> = MlsGroup::create_dm_and_insert(
            &self.context,
            GroupMembershipState::Allowed,
            target_inbox_id.clone(),
            GroupMetadataOptions {
                message_disappearing_settings: opts
                    .unwrap_or_default()
                    .message_disappearing_settings,
                ..Default::default()
            },
            None,
        )?;

        log_event!(
            Event::CreatedDM,
            self.context.installation_id(),
            group_id = group.group_id,
            target_inbox = target_inbox_id
        );
        group.add_members(&[target_inbox_id]).await?;

        Ok(group)
    }

    /// Find or create a Direct Message with the default settings
    pub async fn find_or_create_dm_by_identity(
        &self,
        target_identity: Identifier,
        opts: Option<DMMetadataOptions>,
    ) -> Result<MlsGroup<Context>, ClientError> {
        self.ensure_identity_ready::<ClientError>()?;
        tracing::info!("finding or creating dm with address: {target_identity}");
        let inbox_id = match self
            .find_inbox_id_from_identifier(target_identity.clone())
            .await?
        {
            Some(id) => id,
            None => {
                return Err(NotFound::InboxIdForAddress(target_identity.to_string()).into());
            }
        };

        self.find_or_create_dm(inbox_id, opts).await
    }

    /// Find or create a Direct Message by inbox_id with the default settings.
    /// A stored DM whose creation stopped before the peer was added gets the
    /// peer first, and any failure of that add is returned, as a new DM's
    /// creation returns its add's failure.
    pub async fn find_or_create_dm(
        &self,
        inbox_id: impl AsIdRef,
        opts: Option<DMMetadataOptions>,
    ) -> Result<MlsGroup<Context>, ClientError> {
        self.ensure_identity_ready::<ClientError>()?;
        let inbox_id = inbox_id.as_ref();
        tracing::info!("finding or creating dm with inbox_id: {}", inbox_id);
        let db = self.context.db();
        let group = db.find_active_dm_group(&DmMembers {
            member_one_inbox_id: self.inbox_id(),
            member_two_inbox_id: inbox_id,
        })?;

        if let Some(group) = group {
            let group = MlsGroup::new(
                self.context.clone(),
                group.id,
                group.dm_id,
                group.conversation_type,
                group.created_at_ns,
            );
            // implements: DMS-020
            group.add_missing_dm_peer().await?;
            return Ok(group);
        }
        self.create_dm_by_inbox_id(inbox_id.to_string(), opts).await
    }

    /// Look up a group by its ID
    ///
    /// Returns a [`MlsGroup`] if the group exists, or an error if it does not
    ///
    pub fn group(&self, group_id: &GroupId) -> Result<MlsGroup<Context>, ClientError> {
        MlsGroup::load_by_id(&self.context, group_id).map_err(Into::into)
    }

    /// Look up a group by its ID while stitching DMs
    ///
    /// Returns a [`MlsGroup`] if the group exists, or an error if it does not
    ///
    pub fn stitched_group(&self, group_id: &GroupId) -> Result<MlsGroup<Context>, ClientError> {
        let conn = self.context.db();
        let stored_group = conn.fetch_stitched(group_id)?;
        stored_group
            .map(|g| {
                MlsGroup::new(
                    self.context.clone(),
                    g.id,
                    g.dm_id,
                    g.conversation_type,
                    g.created_at_ns,
                )
            })
            .ok_or(NotFound::GroupById(*group_id))
            .map_err(Into::into)
    }

    /// Find all the duplicate dms for this group
    pub fn find_duplicate_dms_for_group(
        &self,
        group_id: &GroupId,
    ) -> Result<Vec<MlsGroup<Context>>, ClientError> {
        let (group, _) = MlsGroup::new_cached(self.context.clone(), group_id)?;
        Ok(group.find_duplicate_dms()?)
    }

    /// Fetches the message disappearing settings for a given group ID.
    ///
    /// Returns `Some(MessageDisappearingSettings)` if the group exists and has valid settings,
    /// `None` if the group or settings are missing, or `Err(ClientError)` on a database error.
    pub fn group_disappearing_settings(
        &self,
        group_id: &GroupId,
    ) -> Result<Option<MessageDisappearingSettings>, ClientError> {
        let (group, _) = MlsGroup::new_cached(self.context.clone(), group_id)?;
        Ok(group.disappearing_settings()?)
    }

    /**
     * Look up a DM group by the target's inbox_id.
     *
     * Returns a [`MlsGroup`] if the group exists, or an error if it does not
     */
    pub fn dm_group_from_target_inbox(
        &self,
        target_inbox_id: String,
    ) -> Result<MlsGroup<Context>, ClientError> {
        let conn = self.context.db();

        let group = conn
            .find_active_dm_group(&DmMembers {
                member_one_inbox_id: self.inbox_id(),
                member_two_inbox_id: &target_inbox_id,
            })?
            .ok_or(NotFound::DmByInbox(target_inbox_id))?;
        Ok(MlsGroup::new(
            self.context.clone(),
            group.id,
            group.dm_id,
            group.conversation_type,
            group.created_at_ns,
        ))
    }

    /// Look up a message by its ID
    /// Returns a [`StoredGroupMessage`] if the message exists, or an error if it does not
    pub fn message(&self, message_id: Vec<u8>) -> Result<StoredGroupMessage, ClientError> {
        let conn = &mut self.context.db();
        // implements: META-051
        let message =
            conn.get_app_visible_group_message(&message_id, xmtp_common::time::now_ns())?;
        Ok(message.ok_or(NotFound::MessageById(message_id))?)
    }

    /// Look up a stored message and its current conversation.
    /// A message in a duplicate DM resolves to the stitched conversation.
    pub async fn message_with_group(
        &self,
        id: &[u8],
    ) -> Result<Option<(StoredGroupMessage, MlsGroup<Context>)>, ClientError> {
        let message = match self.message(id.to_vec()) {
            Ok(message) => message,
            Err(ClientError::Storage(StorageError::NotFound(NotFound::MessageById(_)))) => {
                return Ok(None);
            }
            Err(error) => return Err(error),
        };
        let group = self.stitched_group(&message.group_id)?;
        Ok(Some((message, group)))
    }

    /// Look up and enrich a message by its ID, returning a [`DecodedMessage`]
    /// Returns an error if the message is not found or if it cannot be decoded/enriched
    #[xmtp_common::mls_span]
    pub fn message_v2(&self, message_id: Vec<u8>) -> Result<DecodedMessage, ClientError> {
        let conn = self.context.db();
        let message = conn
            .get_app_visible_group_message(&message_id, xmtp_common::time::now_ns())?
            .ok_or_else(|| NotFound::MessageById(message_id.clone()))?;

        let group_id = message.group_id;

        let enriched = enrich_messages(conn, &group_id, vec![message])?;

        // Since enrich_messages returns a Vec<DecodedMessage>, we can use .into_iter().next().ok_or(...) to take ownership without cloning.
        enriched
            .into_iter()
            .next()
            // In practice `enrich_messages` should always return an array of the same length as the input
            .ok_or(ClientError::MessageDecodeFailed)
    }

    /// Delete a message by its ID
    /// This method is idempotent and will not error if the message is not found
    /// Returns the number of messages deleted (0 or 1)
    pub fn delete_message(&self, message_id: Vec<u8>) -> Result<usize, ClientError> {
        Ok(crate::state_tx::state_write_with_events(
            self.context.mls_storage(),
            self.context.events(),
            |tx, events| {
                let storage = tx.storage();
                let db = storage.db();
                let message = db.get_group_message(&message_id)?;
                let deleted = db.delete_message_by_id(&message_id)?;
                if deleted > 0
                    && let Some(message) = message
                {
                    emit_deleted_messages(
                        events,
                        vec![message],
                        DeletionCause::DeletedLocally,
                        // A local removal of the row: no inbox sent a delete.
                        None,
                        &db,
                    )?;
                }
                Ok::<_, StorageError>(Continue(deleted))
            },
        )?
        .into_continued())
    }

    /// Query for groups with optional filters
    ///
    /// Filters:
    /// - allowed_states: only return groups with the given membership states
    /// - created_after_ns: only return groups created after the given timestamp (in nanoseconds)
    /// - created_before_ns: only return groups created before the given timestamp (in nanoseconds)
    /// - limit: only return the first `limit` groups
    pub fn find_groups(&self, args: GroupQueryArgs) -> Result<Vec<MlsGroup<Context>>, ClientError> {
        MlsGroup::find(&self.context, args).map_err(Into::into)
    }

    /// List one page of the app's conversations, in activity order unless
    /// `args` names another order. A page holds at most `args.limit`
    /// conversations, or [`CONVERSATION_LIST_DEFAULT_LIMIT`] when it names
    /// none. To read the next page, pass the last row's `listing_time_ns` as
    /// the order's before time and its group id as `before_id`.
    ///
    /// Internal callers that need every group use `find_groups` or
    /// `fetch_conversation_list`, which apply no default limit.
    // implements: CONS-046, CONS-049
    pub fn list_conversations(
        &self,
        mut args: GroupQueryArgs,
    ) -> Result<Vec<ConversationListItem<Context>>, ClientError> {
        if args.order_by.is_none() {
            args.order_by = Some(GroupQueryOrderBy::LastActivity);
        }
        let limit = args
            .limit
            .unwrap_or(i64::from(CONVERSATION_LIST_DEFAULT_LIMIT));
        if !(1..=i64::from(LOCAL_PAGE_MAX_LIMIT)).contains(&limit) {
            return Err(ClientError::InvalidListOptions(
                "limit must be at least 1 and at most the maximum page size",
            ));
        }
        args.limit = Some(limit);
        if args.before_id.is_some() && args.order_before_ns().is_none() {
            return Err(ClientError::InvalidListOptions(
                "before_id needs the before time of the list order",
            ));
        }
        // Requested sync groups bypass the page's limit, order and cursor, and
        // a sync group is not a conversation.
        if args.include_sync_groups || args.conversation_type.is_some_and(|kind| kind.is_virtual())
        {
            return Err(ClientError::InvalidListOptions(
                "a conversation listing cannot include sync or one-shot groups",
            ));
        }
        Ok(self
            .context
            .db()
            .fetch_conversation_list(args)?
            .into_iter()
            .map(
                |conversation_item: DbConversationListItem| ConversationListItem {
                    membership_state: conversation_item.membership_state,
                    added_by_inbox_id: conversation_item.added_by_inbox_id,
                    group: MlsGroup::new(
                        self.context.clone(),
                        conversation_item.id,
                        conversation_item.dm_id,
                        conversation_item.conversation_type,
                        conversation_item.created_at_ns,
                    ),
                    is_commit_log_forked: conversation_item.is_commit_log_forked,
                    listing_time_ns: conversation_item.listing_time_ns,
                },
            )
            .collect())
    }

    /// Register this installation with an inbox owner and wait until it is visible.
    pub async fn register_with_owner(&self, owner: &impl InboxOwner) -> Result<(), ClientError> {
        if self.identity().is_ready() {
            return self.ensure_registration_visible().await;
        }

        let mut signature_request = self
            .context
            .signature_request()
            .ok_or(SignatureRequestError::MissingSigner)?;
        let signature = owner
            .sign(&signature_request.signature_text())
            .map_err(SignatureError::from)?;
        signature_request
            .add_signature(signature, self.scw_verifier())
            .await?;
        self.register_identity(signature_request).await
    }

    /// Upload the key package before the identity update exposes this installation.
    /// Record its receipt for key retirement and retain the registration cursor.
    /// Return success only after the backend can read the registration receipt.
    #[xmtp_common::mls_span]
    pub async fn register_identity(
        &self,
        signature_request: SignatureRequest,
    ) -> Result<(), ClientError> {
        tracing::info!("registering identity");
        // Registration is a network call like any other.
        self.context.server_configuration().check()?;

        // Resume visibility confirmation after a stored registration.
        let stored_identity: Option<StoredIdentity> = self.context.db().fetch(&())?;
        if stored_identity.is_some() {
            tracing::info!("Resuming stored identity registration");
            self.identity().set_ready();
            return self.ensure_registration_visible().await;
        }

        // Step 1: Generate key package and store locally (not uploaded yet)
        let (kp_bytes, history_id) = crate::identity_lifecycle::generate_and_store_key_package(
            self.identity(),
            self.context.mls_storage(),
            CREATE_PQ_KEY_PACKAGE_EXTENSION,
        )?;

        // Step 2: Validate signatures (fails here if invalid - no network pollution)
        let identity_update = signature_request
            .build_identity_update()
            .map_err(IdentityUpdateError::from)?;
        identity_update
            .to_verified(&self.context.scw_verifier())
            .await?;

        // Step 3: Upload key package first (prevents race condition)
        let key_package_meta = self.context.api().upload_key_package(kp_bytes).await?;
        let key_package_cursor = key_package_meta
            .cursor
            .filter(|cursor| cursor.sequence_id > 0)
            .ok_or(xmtp_api::ApiError::InvalidResponse(
                "key package publish cursor",
            ))?;

        // Own-installation changes are compared from before publication. A
        // conflict retry can load another installation's update into this
        // database before the final reload, and those changes are reported too.
        let inbox_id = self.inbox_id().to_string();
        let registration_cursor = self
            .identity_updates()
            .with_own_installation_events(
                &self.context.db(),
                Box::pin(async {
                    // Step 4: Publish identity update (makes installation visible)
                    let registration_cursor = crate::identity_updates::publish_with_conflict_retry(
                        self.context.api(),
                        &self.context.db(),
                        identity_update,
                        &self.context.scw_verifier(),
                    )
                    .await
                    .map_err(|error| IdentityUpdateError::Publish(Box::new(error)))?;
                    // Step 5: Fetch and store in local DB (needed for group operations)
                    retry_async!(
                        Retry::default(),
                        (async {
                            load_identity_updates(
                                self.context.api(),
                                &self.context.db(),
                                &[inbox_id.as_str()],
                            )
                            .await
                        })
                    )?;
                    Ok(registration_cursor)
                }),
            )
            .await?;

        // Backend publication order can differ from local generation order.
        crate::state_tx::state_write(self.context.mls_storage(), |tx| {
            let storage = tx.storage();
            storage.db().record_key_package_publication(
                history_id,
                xmtp_proto::types::Cursor(key_package_cursor.sequence_id),
            )?;
            storage
                .db()
                .reset_key_package_rotation_queue(KEY_PACKAGE_ROTATION_INTERVAL_NS)?;
            Ok::<_, StorageError>(Continue(()))
        })
        .map(TransactionOutcome::into_continued)?;

        // Mark identity as ready
        let mut stored_identity = StoredIdentity::try_from(self.identity())?;
        stored_identity.registration_cursor_sequence_id = Some(registration_cursor.0 as i64);
        stored_identity.store(&self.context.db())?;
        self.context
            .registration_event_pending()
            .store(true, Ordering::Release);
        self.identity().set_ready();
        self.ensure_registration_visible().await
    }

    /// Whether registration has completed its visibility confirmation.
    /// A stored identity with a pending receipt is not yet registered.
    pub fn is_registration_visible(&self) -> Result<bool, ClientError> {
        let stored: Option<StoredIdentity> = self.context.db().fetch(&())?;
        Ok(self.identity().is_ready()
            && stored.is_some_and(|identity| identity.registration_cursor_sequence_id.is_none()))
    }

    /// Confirm a stored registration before reporting success.
    pub async fn ensure_registration_visible(&self) -> Result<(), ClientError> {
        let stored: Option<StoredIdentity> = self.context.db().fetch(&())?;
        if stored.is_some_and(|identity| identity.registration_cursor_sequence_id.is_some()) {
            self.wait_for_registration_visible(VisibilityConfirmationOptions::default())
                .await?;
        }
        Ok(())
    }

    /// Wait until the serving database exposes the registration identity-topic head.
    /// A missing or older head is polled. The timeout also bounds each request.
    pub async fn wait_for_registration_visible(
        &self,
        options: VisibilityConfirmationOptions,
    ) -> Result<(), ClientError> {
        use xmtp_common::time::{Duration, sleep, timeout};
        if !self.identity().is_ready() {
            return Err(ClientError::RegistrationNotVisible);
        }
        let stored: Option<StoredIdentity> = self.context.db().fetch(&())?;
        let stored = stored.ok_or(ClientError::RegistrationNotVisible)?;
        let Some(receipt) = stored.registration_cursor_sequence_id else {
            return Ok(());
        };
        self.context.server_configuration().check()?;
        let sequence_id = u64::try_from(receipt)
            .ok()
            .filter(|sequence_id| *sequence_id != 0)
            .ok_or(ClientError::RegistrationNotVisible)?;
        timeout(Duration::from_millis(options.timeout_ms), async {
            let mut delay = REGISTRATION_INITIAL_BACKOFF;
            let inbox = hex::decode(self.inbox_id())
                .map_err(|_| xmtp_api::ApiError::InvalidRequest("registration inbox id"))?;
            let topic = xmtp_proto::types::Topic::new_identity_update(inbox);
            xmtp_proto::types::Topic::parse(&topic)?;
            loop {
                let heads = self
                    .context
                    .api()
                    .newest_topic_cursors(vec![topic.clone()])
                    .await?;
                let head = heads
                    .get(&topic)
                    .ok_or(xmtp_api::ApiError::InvalidResponse(
                        "registration identity head",
                    ))?;
                // implements: IDENT-072
                if head.0 >= sequence_id {
                    return Ok::<(), ClientError>(());
                }
                sleep(delay).await;
                delay = (delay * 2).min(REGISTRATION_MAX_BACKOFF);
            }
        })
        .await
        .map_err(|_| ClientError::RegistrationNotVisible)??;
        if self
            .context
            .registration_event_pending()
            .swap(false, Ordering::AcqRel)
        {
            xmtp_events::EventWriter::emit(
                self.context.events(),
                Some(xmtp_events::ClientEvent::IdentityRegistered(
                    xmtp_events::IdentityRegistered {
                        inbox_id: self.inbox_id().to_string(),
                        installation_key: self.installation_id.to_vec(),
                    },
                )),
                None,
            );
        }
        self.context.db().clear_registration_cursor(receipt)?;
        Ok(())
    }

    /// If no key rotation is scheduled, queue it to occur in the next 5 seconds.
    pub fn queue_key_rotation(&self) -> Result<(), ClientError> {
        crate::worker::key_package_maintenance::queue_key_rotation(&self.context)?;
        Ok(())
    }

    /// Upload a new key package to the network replacing an existing key package.
    pub async fn rotate_and_upload_key_package(&self) -> Result<(), ClientError> {
        // A client with a blocked connection publishes nothing. This one
        // reaches `Identity` directly instead of going through a group or the
        // publish path, so it carries its own gate.
        self.ensure_identity_ready::<ClientError>()?;
        crate::identity_lifecycle::rotate_and_upload_key_package(
            self.identity(),
            self.context.api(),
            self.context.mls_storage(),
            CREATE_PQ_KEY_PACKAGE_EXTENSION,
        )
        .await?;
        // The rotation marked superseded KPs delete_at=now+grace; without this
        // the parked KpDeletion task would sweep them up to ~30d late.
        crate::worker::key_package_maintenance::nudge_deletion(&self.context)?;

        Ok(())
    }

    /// Fetches the current key package from the network for each of the `installation_id`s specified
    #[tracing::instrument(skip_all)]
    pub async fn get_key_packages_for_installation_ids(
        &self,
        installation_ids: Vec<Vec<u8>>,
    ) -> Result<
        HashMap<Vec<u8>, Result<VerifiedKeyPackageV2, KeyPackageVerificationError>>,
        ClientError,
    > {
        MlsStore::new(self.context.clone())
            .get_key_packages_for_installation_ids(installation_ids)
            .await
            .map_err(Into::into)
    }

    /// Download all unread welcome messages and converts to a group struct, ignoring malformed messages.
    /// Returns any new groups created in the operation
    #[tracing::instrument(skip_all)]
    pub async fn sync_welcomes(&self) -> Result<Vec<MlsGroup<Context>>, GroupError> {
        self.ensure_identity_ready::<GroupError>()?;
        WelcomeService::new(self.context.clone())
            .sync_welcomes()
            .await
            .map_err(|error| self.context.server_configuration().blocked_or(error))
    }

    /// Sync all groups for the current installation and return the number of groups that were synced.
    /// Only active groups will be synced.
    #[tracing::instrument(err, skip_all, fields(operation = "sync_all_groups"))]
    pub async fn sync_all_groups(
        &self,
        groups: Vec<MlsGroup<Context>>,
    ) -> Result<GroupSyncSummary, GroupError> {
        self.ensure_identity_ready::<GroupError>()?;
        WelcomeService::new(self.context.clone())
            .sync_all_groups(groups)
            .await
            .map_err(|error| self.context.server_configuration().blocked_or(error))
    }

    /// Sync all unread welcome messages and then sync all groups.
    /// Returns the total number of active groups synced.
    #[xmtp_common::mls_span]
    pub async fn sync_all_welcomes_and_groups(
        &self,
        consent_states: Option<Vec<ConsentState>>,
    ) -> Result<GroupSyncSummary, GroupError> {
        self.ensure_identity_ready::<GroupError>()?;
        WelcomeService::new(self.context.clone())
            .sync_all_welcomes_and_groups(consent_states)
            .await
            .map_err(|error| self.context.server_configuration().blocked_or(error))
    }

    /// Sweep every group flagged `paused_for_version` and clear the
    /// pause flag for any whose floor is now satisfied by this
    /// client's `pkg_version`. Pure local-state operation — no
    /// network calls. Returns the count of groups unstuck.
    ///
    /// `sync_all_welcomes_and_groups` already runs this sweep as a
    /// preamble; the standalone entry point is for SDKs that want a
    /// cheap "post-upgrade recovery" hook independent of the normal
    /// sync flow.
    pub async fn unstick_paused_groups(&self) -> Result<usize, GroupError> {
        self.ensure_identity_ready::<GroupError>()?;
        WelcomeService::new(self.context.clone())
            .unstick_paused_groups()
            .await
    }

    pub async fn sync_all_welcomes_and_device_sync_groups(
        &self,
    ) -> Result<GroupSyncSummary, ClientError> {
        self.sync_welcomes().await?;
        self.sync_all_device_sync_groups().await
    }

    pub async fn sync_all_device_sync_groups(&self) -> Result<GroupSyncSummary, ClientError> {
        let groups = self
            .context
            .db()
            .all_sync_groups()?
            .into_iter()
            .map(|g| {
                MlsGroup::new(
                    self.context.clone(),
                    g.id,
                    g.dm_id,
                    g.conversation_type,
                    g.created_at_ns,
                )
            })
            .collect();

        Ok(self.sync_all_groups(groups).await?)
    }

    /**
     * Validates a credential against the given installation public key
     *
     * This will go to the network and get the latest association state for the inbox.
     * It ensures that the installation_pub_key is in that association state
     */
    pub async fn validate_credential_against_network(
        &self,
        conn: &DbConnection<<Context::Db as XmtpDb>::Connection>,
        credential: &[u8],
        installation_pub_key: Vec<u8>,
    ) -> Result<InboxId, ClientError> {
        let inbox_id = parse_credential(credential)?;
        let association_state = IdentityUpdates::new(&self.context)
            .get_latest_association_state(conn, &inbox_id)
            .await?;
        let ident = MemberIdentifier::installation(installation_pub_key);

        match association_state.get(&ident) {
            Some(_) => Ok(inbox_id),
            None => Err(IdentityError::InstallationIdNotFound(inbox_id).into()),
        }
    }

    /// Reports whether the backend names an inbox for each identifier.
    /// This is an unverified availability hint. The client does not independently
    /// verify the lookup against current association state or check
    /// key packages. This is not a delivery guarantee.
    ///
    /// Arguments:
    /// - account_identifier: a list of account identifiers to check
    ///
    /// Returns:
    /// A map from each identifier to its availability hint.
    pub async fn can_message(
        &self,
        account_identifiers: &[Identifier],
    ) -> Result<HashMap<Identifier, bool>, ClientError> {
        // A client with a blocked connection issues no request. The blocked connection
        // alone, not `ensure_identity_ready`, because this answers before the
        // caller has registered an identity.
        self.context.server_configuration().check()?;
        let requests = account_identifiers.iter().map(Into::into).collect();

        let results = self.context.api().get_inbox_ids(requests).await?;
        Ok(account_identifiers
            .iter()
            .cloned()
            .zip(results.into_iter().map(|inbox_id| inbox_id.is_some()))
            .collect())
    }
}

impl<Context: XmtpSharedContext> Client<Context> {
    pub fn attachments(&self) -> Attachments<Context> {
        Attachments {
            context: self.context.clone(),
        }
    }
}

impl<Context: XmtpSharedContext> Client<Context> {
    /// Store configuration, then register inline. The task retries transient errors.
    pub async fn enable_notifications(
        &self,
        config: NotificationConfig,
    ) -> Result<NotificationState, NotificationError> {
        notifications::enable_notifications(&self.context, config).await
    }

    /// Disable locally before unregistering. The recipient identity and overrides stay.
    /// A failed unregister leaves the client disabled with its recipient reset owed.
    pub async fn disable_notifications(&self) -> Result<(), NotificationError> {
        notifications::disable_notifications(&self.context).await
    }

    /// Read the locally stored state without a backend call.
    pub fn notification_state(&self) -> Result<NotificationState, StorageError> {
        notifications::notification_state(&self.context)
    }
}

impl<Context: XmtpSharedContext> Client<Context> {
    /// The inbox that holds each handle, in order, or `None`. An answer from
    /// the backend counts only when the newest state the client derives for
    /// the answered inbox holds the handle. An old handle of an inbox still
    /// resolves to it.
    pub async fn resolve_handles(
        &self,
        handles: &[Handle],
    ) -> Result<Vec<Option<InboxId>>, ClientError> {
        Ok(inbox_handles::resolve_verified_in(
            &self.context,
            &inbox_handles::handle_targets(handles),
        )
        .await?)
    }

    /// Whether the backend names an inbox for each handle, in order. This
    /// checks no association state: use it only where a wrong answer costs a
    /// candidate, never to reach an inbox.
    pub async fn handles_have_owner(&self, handles: &[Handle]) -> Result<Vec<bool>, ClientError> {
        Ok(inbox_handles::handles_have_owner(&self.context, handles).await?)
    }

    /// The handles for `input` that no inbox holds. `input` is a handle as a
    /// user typed it; its ASCII capitals become lowercase. Input with a code
    /// checks only that handle. A name without a code checks the name and
    /// [`HANDLE_SEARCH_CODES`](inbox_handles::HANDLE_SEARCH_CODES) distinct random codes in one lookup, and the
    /// name comes first in the result. Text outside the handle grammar fails
    /// before any request. No policy applies here, so a returned handle can
    /// still be refused when it is claimed.
    pub async fn available_handles(&self, input: &str) -> Result<Vec<Handle>, ClientError> {
        Ok(inbox_handles::available_handles(&self.context, input).await?)
    }

    /// Claim `handle` for this client's inbox, signed by this installation's
    /// key. A claim of a handle the inbox already holds makes it the display
    /// handle again. The backend refuses a taken, reserved, short, or too
    /// frequent claim with `ApiError::HandleRejected`. After the publish, the
    /// client has stored the inbox's new identity update.
    pub async fn claim_handle(&self, handle: &Handle) -> Result<(), ClientError> {
        self.ensure_identity_ready::<ClientError>()?;
        Ok(inbox_handles::claim_handle(&self.context, handle).await?)
    }

    /// The inbox of each target, in order, from one call. Fails when any
    /// target has no inbox: an account as `find_or_create_dm_by_identity`
    /// does, and a handle with [`HandleError::NotFound`].
    pub async fn resolve_targets(
        &self,
        targets: &[IdentityTarget],
    ) -> Result<Vec<InboxId>, ClientError> {
        Ok(inbox_handles::resolve_targets(&self.context, targets).await?)
    }

    /// The active DM with `target`, or `None`. A target that no inbox holds
    /// has no DM.
    pub async fn find_dm_by_target(
        &self,
        target: &IdentityTarget,
    ) -> Result<Option<MlsGroup<Context>>, ClientError> {
        let Some(peer) =
            inbox_handles::resolve_verified_in(&self.context, std::slice::from_ref(target))
                .await?
                .pop()
                .flatten()
        else {
            return Ok(None);
        };
        let stored = self.context.db().find_active_dm_group(&DmMembers {
            member_one_inbox_id: self.inbox_id(),
            member_two_inbox_id: peer.as_str(),
        })?;
        stored.map(|stored| self.group(&stored.id)).transpose()
    }

    /// Find the DM with `target`, or create it. A handle resolves before the
    /// existing DM is looked up.
    pub async fn find_or_create_dm_by_target(
        &self,
        target: &IdentityTarget,
        opts: DMMetadataOptions,
    ) -> Result<MlsGroup<Context>, ClientError> {
        match target {
            IdentityTarget::Account(identifier) => {
                self.find_or_create_dm_by_identity(identifier.clone(), Some(opts))
                    .await
            }
            IdentityTarget::Handle(handle) => {
                let peer = self
                    .resolve_handles(std::slice::from_ref(handle))
                    .await?
                    .pop()
                    .flatten()
                    .ok_or(HandleError::NotFound)?;
                self.find_or_create_dm(peer, Some(opts)).await
            }
        }
    }

    /// Create a group with `targets` as its members, in one commit. Targets,
    /// member identities, and key packages resolve before the group is stored.
    /// A target with no inbox leaves no group: a handle fails with
    /// [`HandleError::NotFound`], an account with [`GroupError::AddressNotFound`].
    pub async fn create_group_with_targets(
        &self,
        targets: &[IdentityTarget],
        permissions_policy_set: Option<PolicySet>,
        opts: Option<GroupMetadataOptions>,
    ) -> Result<MlsGroup<Context>, ClientError> {
        let inboxes = inbox_handles::resolve_targets_in(&self.context, targets).await?;
        let inbox_ids = crate::groups::targets::require_account_inboxes(targets, inboxes)?;
        self.create_group_with_members(&inbox_ids, permissions_policy_set, opts)
            .await
    }
}

impl<Context> Client<Context>
where
    Context: XmtpSharedContext + 'static,
{
    /// Use push metadata as a target. Fetch its complete prefix before processing.
    pub async fn process_streamed_welcome_message(
        &self,
        envelope_bytes: Vec<u8>,
    ) -> Result<Vec<MlsGroup<Context>>, SubscribeError> {
        subscriptions::process_streamed_welcome_message(&self.context, envelope_bytes).await
    }

    pub async fn stream_conversations(
        &self,
        conversation_type: Option<ConversationType>,
        include_duplicate_dms: bool,
    ) -> Result<
        impl Stream<Item = Result<MlsGroup<Context>, SubscribeError>> + use<'_, Context>,
        SubscribeError,
    >
    where
        Context::ApiClient: XmtpMlsStreams,
    {
        subscriptions::stream_conversations(&self.context, conversation_type, include_duplicate_dms)
            .await
    }

    /// Stream conversations but decouple the lifetime of 'self' from the stream.
    pub async fn stream_conversations_owned(
        &self,
        conversation_type: Option<ConversationType>,
        include_duplicate_dms: bool,
    ) -> Result<
        impl Stream<Item = Result<MlsGroup<Context>, SubscribeError>> + 'static + use<Context>,
        SubscribeError,
    >
    where
        Context::ApiClient: XmtpMlsStreams,
    {
        subscriptions::stream_conversations_owned(
            self.context.clone(),
            conversation_type,
            include_duplicate_dms,
        )
        .await
    }
}

impl<Context> Client<Context>
where
    Context: XmtpSharedContext + 'static,
    Context::ApiClient: XmtpMlsStreams + 'static,
    Context::MlsStorage: 'static,
{
    pub fn stream_conversations_with_callback(
        client: Arc<Client<Context>>,
        conversation_type: Option<ConversationType>,
        convo_callback: impl FnMut(Result<MlsGroup<Context>, SubscribeError>) + MaybeSend + 'static,
        on_close: impl FnOnce() + MaybeSend + 'static,
        include_duplicate_dms: bool,
    ) -> impl StreamHandle<StreamOutput = Result<(), SubscribeError>> {
        subscriptions::stream_conversations_with_callback(
            client.context.clone(),
            client,
            conversation_type,
            convo_callback,
            on_close,
            include_duplicate_dms,
        )
    }

    pub async fn stream_all_messages(
        &self,
        conversation_type: Option<ConversationType>,
        consent_state: Option<Vec<ConsentState>>,
    ) -> Result<impl Stream<Item = Result<StoredGroupMessage, SubscribeError>> + '_, SubscribeError>
    {
        subscriptions::stream_all_messages(&self.context, conversation_type, consent_state).await
    }

    pub async fn stream_all_messages_owned(
        &self,
        conversation_type: Option<ConversationType>,
        consent_state: Option<Vec<ConsentState>>,
    ) -> Result<
        impl Stream<Item = Result<StoredGroupMessage, SubscribeError>> + 'static + use<Context>,
        SubscribeError,
    > {
        subscriptions::stream_all_messages_owned(
            self.context.clone(),
            conversation_type,
            consent_state,
        )
        .await
    }

    pub fn stream_consent_with_callback(
        client: Arc<Client<Context>>,
        callback: impl FnMut(Result<Vec<StoredConsentRecord>, SubscribeError>) + MaybeSend + 'static,
        on_close: impl FnOnce() + MaybeSend + 'static,
    ) -> impl StreamHandle<StreamOutput = Result<(), SubscribeError>> {
        subscriptions::stream_consent_with_callback(
            client.context.clone(),
            client,
            callback,
            on_close,
        )
    }

    pub fn stream_preferences_with_callback(
        client: Arc<Client<Context>>,
        callback: impl FnMut(Result<Vec<PreferenceUpdate>, SubscribeError>) + MaybeSend + 'static,
        on_close: impl FnOnce() + MaybeSend + 'static,
    ) -> impl StreamHandle<StreamOutput = Result<(), SubscribeError>> {
        subscriptions::stream_preferences_with_callback(
            client.context.clone(),
            client,
            callback,
            on_close,
        )
    }

    pub fn stream_message_deletions_with_callback(
        client: Arc<Client<Context>>,
        callback: impl FnMut(Result<DecodedMessage, SubscribeError>) + MaybeSend + 'static,
        on_close: impl FnOnce() + MaybeSend + 'static,
    ) -> impl StreamHandle<StreamOutput = Result<(), SubscribeError>> {
        subscriptions::stream_message_deletions_with_callback(
            client.context.clone(),
            client,
            callback,
            on_close,
        )
    }
}

impl<Context> Client<Context>
where
    Context: XmtpSharedContext + 'static,
    Context::ApiClient: XmtpMlsStreams + 'static,
    Context::MlsStorage: 'static,
    <Context::ApiClient as XmtpMlsStreams>::GroupMessageStream: Unpin,
    <Context::ApiClient as XmtpMlsStreams>::WelcomeMessageStream: Unpin,
{
    #[cfg(any(test, feature = "test-utils"))]
    pub async fn stream_all_messages_owned_with_stats(
        &self,
        conversation_type: Option<ConversationType>,
        consent_state: Option<Vec<ConsentState>>,
    ) -> Result<
        impl StreamWithStats<Item = Result<StoredGroupMessage, SubscribeError>> + 'static,
        SubscribeError,
    > {
        subscriptions::stream_all_messages_owned_with_stats(
            self.context.clone(),
            conversation_type,
            consent_state,
        )
        .await
    }
}

impl<C: XmtpSharedContext + 'static> Client<C> {
    /// Process fixed starting targets and only the discoveries from the enrolled Welcomes.
    pub async fn catch_up_to_live(
        &self,
        timeout: Option<xmtp_common::time::Duration>,
    ) -> Result<CatchUpSummary, CatchUpError> {
        subscriptions::catch_up::catch_up_to_live(&self.context, timeout).await
    }
}

#[cfg(not(target_arch = "wasm32"))]
impl<C> Client<C>
where
    C: XmtpSharedContext + 'static,
    C::ApiClient:
        XmtpMlsBidiStreams + XmtpMlsStreams + ApiClientIdentity + Clone + Send + Sync + 'static,
    <C::ApiClient as XmtpMlsBidiStreams>::SubscribeStream: 'static,
{
    pub fn stream_all_messages_with_callback_dispatch(
        client: Arc<Client<C>>,
        conversation_type: Option<ConversationType>,
        consent_states: Option<Vec<ConsentState>>,
        callback: impl FnMut(Result<StoredGroupMessage, SubscribeError>) + MaybeSend + 'static,
        on_close: impl FnOnce() + MaybeSend + 'static,
    ) -> impl StreamHandle<StreamOutput = Result<(), SubscribeError>> {
        router_callbacks::stream_all_messages_with_callback_dispatch(
            client.context.clone(),
            client,
            conversation_type,
            consent_states,
            callback,
            on_close,
        )
    }

    pub fn stream_conversations_with_callback_dispatch(
        client: Arc<Client<C>>,
        conversation_type: Option<ConversationType>,
        include_duplicate_dms: bool,
        callback: impl FnMut(Result<MlsGroup<C>, SubscribeError>) + MaybeSend + 'static,
        on_close: impl FnOnce() + MaybeSend + 'static,
    ) -> impl StreamHandle<StreamOutput = Result<(), SubscribeError>> {
        router_callbacks::stream_conversations_with_callback_dispatch(
            client.context.clone(),
            client,
            conversation_type,
            include_duplicate_dms,
            callback,
            on_close,
        )
    }
}

#[cfg(test)]
pub(crate) mod tests;

#[cfg(all(test, not(target_arch = "wasm32")))]
pub(crate) mod test_hooks {
    use std::sync::Arc;
    use tokio::sync::Barrier;

    type PauseHook = (Vec<u8>, Arc<Barrier>, Arc<Barrier>);
    /// Stops the next `reconnect_db` of one installation after its closed
    /// check. The test runs this synchronous call on a blocking thread.
    pub(crate) static PAUSE_RECONNECT: parking_lot::Mutex<Option<PauseHook>> =
        parking_lot::Mutex::new(None);

    pub(super) fn pause_reconnect(installation_id: &[u8]) {
        let paused = {
            let mut hook = PAUSE_RECONNECT.lock();
            if hook
                .as_ref()
                .is_some_and(|(installation, _, _)| installation.as_slice() == installation_id)
            {
                hook.take()
            } else {
                None
            }
        };
        if let Some((_, entered, release)) = paused {
            tokio::runtime::Handle::current().block_on(async {
                entered.wait().await;
                release.wait().await;
            });
        }
    }
}