xmtp_mls/
builder.rs

1#[cfg(test)]
2use crate::GroupCommitLock;
3use crate::{
4    StorageError, XmtpApi,
5    attachments::AttachmentRuntime,
6    client::{Client, ClientError, DeviceSync},
7    context::{XmtpMlsLocalContext, XmtpSharedContext},
8    groups::change_callbacks::UnstableChangeCallbacks,
9    identity::{Identity, IdentityStrategy},
10    identity_updates::load_identity_updates,
11    mutex_registry::MutexRegistry,
12    server_configuration::ServerConfigurationHandle,
13    utils::{VersionInfo, cleanup_duplicate_updates},
14    worker::{WorkerRunner, tasks::TaskWorker},
15    worker::{device_sync::worker::SyncWorker, disappearing_messages::DisappearingMessagesWorker},
16};
17use futures::FutureExt;
18use prost::Message as _;
19use std::sync::Arc;
20use std::sync::atomic::AtomicBool;
21use thiserror::Error;
22use tokio_util::sync::CancellationToken;
23use tracing::debug;
24use xmtp_api::ApiClientWrapper;
25use xmtp_api_backend::TrackedStatsClient;
26use xmtp_common::snippet::Snippet;
27use xmtp_common::{ErrorCode, Event, Retry};
28use xmtp_cryptography::signature::IdentifierValidationError;
29use xmtp_db::{DbConnection, XmtpMlsStorageProvider, prelude::*};
30use xmtp_db::{XmtpDb, sql_key_store::SqlKeyStore};
31use xmtp_id::scw_verifier::SmartContractSignatureVerifier;
32use xmtp_macro::log_event;
33use xmtp_proto::api_client::XmtpBackendClient;
34use xmtp_proto::xmtp::mls::database::{
35    ProcessPendingSelfRemove, Task as TaskProto, task::Task as TaskKind,
36};
37
38type ContextParts<Api, S, Db> = Arc<XmtpMlsLocalContext<Api, Db, S>>;
39/// Opens the store at the paths `build` resolves for a storage location.
40#[cfg(not(target_arch = "wasm32"))]
41type LocationStoreOpener<Db> = Box<
42    dyn FnOnce(
43            crate::storage_location::ResolvedPaths,
44        ) -> xmtp_common::BoxDynFuture<'static, Result<Db, ClientBuilderError>>
45        + Send
46        + Sync,
47>;
48#[cfg(target_arch = "wasm32")]
49type LocationStoreOpener<Db> = Box<
50    dyn FnOnce(
51        crate::storage_location::ResolvedPaths,
52    ) -> xmtp_common::BoxDynFuture<'static, Result<Db, ClientBuilderError>>,
53>;
54
55fn open_location_store(
56    paths: crate::storage_location::ResolvedPaths,
57    key: xmtp_db::EncryptionKey,
58) -> xmtp_common::BoxDynFuture<'static, Result<xmtp_db::DefaultStore, ClientBuilderError>> {
59    Box::pin(async move {
60        #[cfg(not(target_arch = "wasm32"))]
61        let db = {
62            let parent = paths
63                .db_path
64                .parent()
65                .ok_or(crate::storage_location::StorageLocationError::InboxId)?;
66            xmtp_attachments::create_private_directory(parent)
67                .await
68                .map_err(crate::storage_location::StorageLocationError::from)?;
69            xmtp_db::NativeDb::builder()
70                .persistent(paths.db_path.to_string_lossy().into_owned())
71                .key(key)
72                .build()?
73        };
74        #[cfg(target_arch = "wasm32")]
75        let db = xmtp_db::WasmDb::new(&xmtp_db::StorageOption::Persistent(
76            paths.db_path.to_string_lossy().into_owned(),
77        ))
78        .await
79        .map_err(xmtp_db::StorageError::from)?;
80        #[cfg(target_arch = "wasm32")]
81        let _ = key;
82        Ok(xmtp_db::EncryptedMessageStore::new(db)?)
83    })
84}
85
86// Every identity request uses an admitted snapshot, including the lookup
87// needed to find a data directory's database.
88fn admit_configuration<ApiClient: XmtpBackendClient>(
89    api_client: &mut ApiClientWrapper<ApiClient>,
90    configuration: &xmtp_configuration::ServerConfiguration,
91    client_version: &semver::Version,
92) -> Result<(), ClientError> {
93    crate::server_configuration::check_minimum_version(configuration, client_version)?;
94    // implements: CONF-051
95    if configuration.auth.enabled && !api_client.has_credential_source() {
96        return Err(ClientError::AuthRequired {
97            required_scopes: configuration.auth.required_scopes.clone(),
98        });
99    }
100    let snapshot = Arc::new(configuration.clone());
101    api_client
102        .api_client
103        .set_limits(Arc::new(snapshot.limits.clone()));
104    api_client.set_configuration(snapshot);
105    Ok(())
106}
107
108#[derive(Error, Debug, ErrorCode)]
109pub enum ClientBuilderError {
110    /// The deployment storage path could not be resolved or opened.
111    /// May be retryable if local storage becomes available.
112    #[error(transparent)]
113    #[error_code("StorageLocation")]
114    StorageLocation(#[from] crate::storage_location::StorageLocationError),
115    /// Attachment storage could not be prepared or cleaned.
116    /// May be retryable if local storage becomes available.
117    #[error(transparent)]
118    #[error_code("Attachment")]
119    Attachment(#[from] crate::attachments::AttachmentClientError),
120    #[error(transparent)]
121    #[error_code(inherit)]
122    AddressValidation(#[from] IdentifierValidationError),
123    /// Missing parameter.
124    ///
125    /// Required builder parameter not provided. Not retryable.
126    #[error("Missing parameter: {parameter}")]
127    MissingParameter { parameter: &'static str },
128    /// Client error.
129    ///
130    /// Client operation failed during build. May be retryable.
131    #[error(transparent)]
132    ClientError(#[from] crate::client::ClientError),
133    /// Storage error.
134    ///
135    /// Storage initialization failed. Not retryable.
136    #[error("Storage Error")]
137    StorageError(#[from] StorageError),
138    /// Identity error.
139    ///
140    /// Identity creation/loading failed. Not retryable.
141    #[error(transparent)]
142    Identity(#[from] crate::identity::IdentityError),
143    /// API error.
144    ///
145    /// API client initialization failed. Retryable.
146    #[error(transparent)]
147    WrappedApiError(#[from] xmtp_api::ApiError),
148    /// Group error.
149    ///
150    /// Group operation failed during build. Not retryable.
151    #[error(transparent)]
152    GroupError(#[from] Box<crate::groups::GroupError>),
153    /// Device sync error.
154    ///
155    /// Device sync setup failed. Not retryable.
156    #[error(transparent)]
157    DeviceSync(#[from] Box<crate::worker::device_sync::DeviceSyncError>),
158    /// Offline build failed.
159    ///
160    /// Builder tried to access the network in offline mode. Not retryable.
161    #[error("Offline build failed, builder tried to access the network")]
162    OfflineBuildFailed,
163}
164
165impl From<crate::worker::device_sync::DeviceSyncError> for ClientBuilderError {
166    fn from(value: crate::worker::device_sync::DeviceSyncError) -> Self {
167        ClientBuilderError::DeviceSync(Box::new(value))
168    }
169}
170
171impl From<crate::groups::GroupError> for ClientBuilderError {
172    fn from(value: crate::groups::GroupError) -> Self {
173        ClientBuilderError::GroupError(Box::new(value))
174    }
175}
176
177pub struct ClientBuilder<ApiClient, S, Db = xmtp_db::DefaultStore> {
178    pub(crate) deployment_recorder: Option<crate::storage_location::DeploymentRecorder>,
179    pub(crate) data_location: Option<crate::storage_location::StorageLocation>,
180    pub(crate) location_store_opener: Option<LocationStoreOpener<Db>>,
181    pub(crate) mls_storage_factory: Option<fn(&Db) -> S>,
182    pub(crate) storage_location_selected: bool,
183    pub(crate) storage_location_conflict: bool,
184    pub(crate) attachments_dir: Option<std::path::PathBuf>,
185    pub(crate) attachment_options: xmtp_attachments::AttachmentOptions,
186    pub(crate) api_client: Option<ApiClient>,
187    pub(crate) identity: Option<Identity>,
188    pub(crate) store: Option<Db>,
189    pub(crate) identity_strategy: IdentityStrategy,
190    pub(crate) scw_verifier: Option<Box<dyn SmartContractSignatureVerifier>>,
191    /// Whether the app supplied its own verifier, exempt from the chain check.
192    pub(crate) custom_scw_verifier: bool,
193    pub(crate) device_sync_worker_mode: DeviceSyncMode,
194    pub(crate) fork_recovery_opts: Option<ForkRecoveryOpts>,
195    /// Unstable: group-change callbacks the host registered at construction.
196    pub(crate) change_callbacks: UnstableChangeCallbacks,
197    pub(crate) stream_policy: crate::subscriptions::policy::StreamPolicy,
198    pub(crate) incoming_factory:
199        Option<Arc<dyn crate::subscriptions::incoming::SubscriptionFactory>>,
200    pub(crate) version_info: VersionInfo,
201    pub(crate) allow_offline: bool,
202    /// Whether the strategy's identifier must belong to the inbox of the
203    /// identity the build loads.
204    pub(crate) require_identifier_in_inbox: bool,
205    pub(crate) disable_commit_log_worker: bool,
206    pub(crate) mls_storage: Option<S>,
207    pub(crate) disable_workers: bool,
208    pub(crate) worker_config: crate::worker::WorkerConfig,
209    /// A snapshot supplied by the caller. When present the client
210    /// never fetches, stores, refreshes, or checks the identifier. Rust tests
211    /// only; not exposed through the bindings.
212    pub(crate) config_provider: Option<Arc<dyn xmtp_configuration::ConfigProvider>>,
213    #[cfg(any(test, feature = "test-utils"))]
214    application_components_for_test:
215        Option<Vec<xmtp_configuration::ApplicationComponentDefinition>>,
216}
217
218#[derive(Clone, Copy, Debug, PartialEq, Eq)]
219pub enum DeviceSyncMode {
220    Disabled,
221    Enabled,
222}
223
224#[derive(Clone, Debug, PartialEq)]
225pub enum ForkRecoveryPolicy {
226    None,
227    AllowlistedGroups,
228    All,
229}
230
231#[derive(Clone, Debug)]
232pub struct ForkRecoveryOpts {
233    pub enable_recovery_requests: ForkRecoveryPolicy,
234    pub groups_to_request_recovery: Vec<String>,
235    pub disable_recovery_responses: bool,
236    pub worker_interval_ns: Option<u64>,
237}
238
239impl Default for ForkRecoveryOpts {
240    fn default() -> Self {
241        Self {
242            enable_recovery_requests: ForkRecoveryPolicy::None,
243            groups_to_request_recovery: Vec::new(),
244            disable_recovery_responses: false,
245            worker_interval_ns: None,
246        }
247    }
248}
249
250impl Client<()> {
251    /// Get the builder for this [`Client`]
252    pub fn builder(strategy: IdentityStrategy) -> ClientBuilder<(), ()> {
253        ClientBuilder::<(), ()>::new(strategy)
254    }
255}
256
257impl<ApiClient, S, Db> ClientBuilder<ApiClient, S, Db> {
258    /// Override internal limits for controlled tests.
259    #[cfg(test)]
260    pub(crate) fn stream_policy(
261        mut self,
262        settings: crate::subscriptions::policy::StreamPolicy,
263    ) -> Self {
264        self.stream_policy = settings;
265        self
266    }
267
268    #[tracing::instrument(level = "trace", skip_all)]
269    pub fn new(identity_strategy: IdentityStrategy) -> Self {
270        Self {
271            deployment_recorder: None,
272            data_location: None,
273            location_store_opener: None,
274            mls_storage_factory: None,
275            storage_location_selected: false,
276            storage_location_conflict: false,
277            attachments_dir: None,
278            attachment_options: xmtp_attachments::AttachmentOptions::default(),
279            identity_strategy,
280            api_client: None,
281            identity: None,
282            store: None,
283            scw_verifier: None,
284            custom_scw_verifier: false,
285            device_sync_worker_mode: DeviceSyncMode::Enabled,
286            fork_recovery_opts: None,
287            change_callbacks: UnstableChangeCallbacks::default(),
288            stream_policy: crate::subscriptions::policy::StreamPolicy::default(),
289            incoming_factory: None,
290            version_info: VersionInfo::default(),
291            allow_offline: false,
292            require_identifier_in_inbox: false,
293            disable_commit_log_worker: false,
294            mls_storage: None,
295            disable_workers: false,
296            worker_config: crate::worker::WorkerConfig::default(),
297            config_provider: None,
298            #[cfg(any(test, feature = "test-utils"))]
299            application_components_for_test: None,
300        }
301    }
302}
303
304#[cfg(any(test, feature = "test-utils"))]
305impl<ApiClient, S, Db> ClientBuilder<ApiClient, S, Db>
306where
307    ApiClient: Clone,
308    Db: Clone,
309    S: Clone,
310{
311    pub fn from_client(
312        client: Client<ContextParts<ApiClient, S, Db>>,
313    ) -> ClientBuilder<ApiClient, S, Db> {
314        let cloned_api: ApiClient = client.context.api_client.api_client.raw_for_test().clone();
315        ClientBuilder {
316            deployment_recorder: None,
317            data_location: None,
318            location_store_opener: None,
319            mls_storage_factory: None,
320            storage_location_selected: false,
321            storage_location_conflict: false,
322            attachments_dir: client.context.attachments.dir.clone(),
323            attachment_options: client.context.attachments.options.clone(),
324            api_client: Some(cloned_api),
325            identity: Some(client.context.identity.clone()),
326            store: Some(client.context.store.clone()),
327            identity_strategy: IdentityStrategy::CachedOnly,
328            scw_verifier: Some(Box::new(client.context.scw_verifier.clone())),
329            custom_scw_verifier: client.context.server_configuration.custom_verifier(),
330            device_sync_worker_mode: client.context.device_sync.mode,
331            fork_recovery_opts: Some(client.context.fork_recovery_opts.clone()),
332            change_callbacks: client.context.change_callbacks.clone(),
333            stream_policy: client.context.incoming_runtime.policy().clone(),
334            incoming_factory: client.context.incoming_runtime.original_factory.clone(),
335            version_info: client.context.version_info.clone(),
336            allow_offline: false,
337            require_identifier_in_inbox: false,
338            disable_commit_log_worker: false,
339            mls_storage: Some(client.context.mls_storage.clone()),
340            disable_workers: false,
341            worker_config: client.context.worker_config.clone(),
342            config_provider: None,
343            #[cfg(any(test, feature = "test-utils"))]
344            application_components_for_test: None,
345        }
346    }
347}
348
349/// One-time backfill of `ProcessPendingSelfRemove` tasks for groups that already
350/// had pending leave requests before self-removal became event-driven. Such rows
351/// have no incoming LeaveRequest to re-fire, so without this they'd only be
352/// processed if a new one arrived. Best-effort and idempotent (deduped per group).
353///
354/// TODO(#3748): removable once every client has shipped a release that enqueues
355/// these tasks inline — safe to delete by the next-next stable release.
356fn backfill_pending_self_remove_tasks<C>(context: &C) -> Result<(), StorageError>
357where
358    C: XmtpSharedContext,
359{
360    let db = context.db();
361    for raw_id in db.get_groups_have_pending_leave_request()? {
362        let Ok(group_id) = xmtp_proto::types::GroupId::try_from(raw_id.as_slice()) else {
363            continue;
364        };
365        let now = xmtp_common::time::now_ns();
366        let proto = TaskProto {
367            task: Some(TaskKind::ProcessPendingSelfRemove(
368                ProcessPendingSelfRemove {
369                    group_id: group_id.to_vec(),
370                },
371            )),
372        };
373        let task = xmtp_db::tasks::NewTask::builder()
374            .originating_message_sequence_id(0)
375            .created_at_ns(now)
376            .next_attempt_at_ns(now)
377            .build(proto)?;
378        // Insert-if-absent per group: leaves any live retrying task (and its
379        // backoff) untouched, only replacing dead rows. Safe to call on every
380        // startup without resurrecting exhausted tasks.
381        db.upsert_pending_self_remove_task(&group_id, task)?;
382    }
383    Ok(())
384}
385
386// TODO: the return type is temp and
387// will be modified in subsequent PRs
388impl<ApiClient, S, Db> ClientBuilder<ApiClient, S, Db> {
389    /// build a client in offline mode.
390    /// returns an error if the client failed to build as offline
391    pub fn build_offline(self) -> Result<Client<ContextParts<ApiClient, S, Db>>, ClientBuilderError>
392    where
393        ApiClient: XmtpApi + 'static,
394        Db: xmtp_db::XmtpDb + 'static,
395        S: XmtpMlsStorageProvider + 'static,
396    {
397        self.build()
398            .now_or_never()
399            .ok_or(ClientBuilderError::OfflineBuildFailed)
400            .flatten()
401    }
402
403    #[tracing::instrument(err, skip_all, fields(operation = "mls.build_client"))]
404    pub async fn build(self) -> Result<Client<ContextParts<ApiClient, S, Db>>, ClientBuilderError>
405    where
406        ApiClient: XmtpApi + 'static,
407        Db: xmtp_db::XmtpDb + 'static,
408        S: XmtpMlsStorageProvider + 'static,
409    {
410        let ClientBuilder {
411            mut deployment_recorder,
412            data_location,
413            location_store_opener,
414            mls_storage_factory,
415            storage_location_conflict,
416            mut attachments_dir,
417            attachment_options,
418            mut api_client,
419            identity,
420            mut store,
421            mut identity_strategy,
422            mut scw_verifier,
423            custom_scw_verifier,
424
425            device_sync_worker_mode,
426            fork_recovery_opts,
427            change_callbacks,
428            stream_policy,
429            incoming_factory,
430            version_info,
431            allow_offline,
432            require_identifier_in_inbox,
433            disable_commit_log_worker,
434            mut mls_storage,
435            disable_workers,
436            worker_config,
437            config_provider,
438            #[cfg(any(test, feature = "test-utils"))]
439            application_components_for_test,
440            ..
441        } = self;
442
443        if storage_location_conflict {
444            return Err(crate::storage_location::StorageLocationError::ConflictingStore.into());
445        }
446
447        let api_client = api_client
448            .take()
449            .ok_or(ClientBuilderError::MissingParameter {
450                parameter: "api_client",
451            })?;
452
453        let mut scw_verifier = scw_verifier
454            .take()
455            .ok_or(ClientBuilderError::MissingParameter {
456                parameter: "scw_verifier",
457            })?;
458
459        let mut api_client = ApiClientWrapper::new(api_client, Retry::default());
460        let config_provider = config_provider.map(ServerConfigurationHandle::new);
461        let mut data_dir_opened_identifier = None;
462        if let Some(location) = data_location {
463            use crate::storage_location::StorageLocationError;
464            let data_dir = matches!(
465                location,
466                crate::storage_location::StorageLocation::DataDir(_)
467            );
468            // A data directory's path holds the inbox. Only an online build
469            // can look up an unknown one.
470            let look_up_inbox = data_dir && identity_strategy.inbox_id().is_none();
471            if look_up_inbox
472                && (allow_offline
473                    || !matches!(
474                        identity_strategy,
475                        IdentityStrategy::CreateForIdentifier { .. }
476                    ))
477            {
478                return Err(StorageLocationError::InboxId.into());
479            }
480            let backend_url = api_client.backend_url().unwrap_or_default().to_owned();
481            if data_dir && backend_url.trim_end_matches('/').is_empty() {
482                return Err(StorageLocationError::BackendUrl.into());
483            }
484            let recorder = location.recorder(&backend_url);
485            let recorded = match &recorder {
486                Some(recorder) => recorder.lookup().await?,
487                None => None,
488            };
489            let mut fetched = None;
490            // The request that looks up an inbox carries the identifier, and
491            // the database it finds may hold no configuration. So that build
492            // asks the deployment for its configuration first, even for a
493            // recorded URL.
494            let opened_identifier = match recorded {
495                Some(identifier) if !look_up_inbox => Some(identifier),
496                _ if !data_dir => None,
497                recorded => {
498                    if allow_offline {
499                        return Err(StorageLocationError::OfflineMissingDeployment.into());
500                    }
501                    let response = api_client.get_configuration().await.map_err(|error| {
502                        ClientBuilderError::ClientError(ClientError::ConfigurationUnavailable(
503                            Box::new(crate::server_configuration::ConfigurationFetchError::Api(
504                                error,
505                            )),
506                        ))
507                    })?;
508                    let configuration = crate::server_configuration::validated(&response)
509                        .map_err(ClientError::from)?;
510                    let identifier = configuration.identifier.clone();
511                    if let Some(recorded) = recorded
512                        && recorded != identifier
513                    {
514                        return Err(ClientError::BackendMismatch {
515                            stored: recorded,
516                            received: identifier,
517                        }
518                        .into());
519                    }
520                    if look_up_inbox {
521                        let admission = config_provider
522                            .as_ref()
523                            .map(ServerConfigurationHandle::configuration)
524                            .unwrap_or(&configuration);
525                        if admission.identifier != identifier {
526                            return Err(StorageLocationError::DeploymentMismatch.into());
527                        }
528                        admit_configuration(
529                            &mut api_client,
530                            admission,
531                            version_info.pkg_semver().semver(),
532                        )?;
533                    }
534                    fetched = Some((identifier.clone(), response));
535                    Some(identifier)
536                }
537            };
538            // An explicit database is opened first, and its build looks up
539            // an unknown inbox after the configuration checks below.
540            if look_up_inbox {
541                identity_strategy = identity_strategy.with_inbox(&api_client).await?;
542            }
543            let paths = location.resolve_identifier(
544                identity_strategy.inbox_id().unwrap_or_default(),
545                opened_identifier.as_deref().unwrap_or_default(),
546            )?;
547            let opener = location_store_opener.ok_or(StorageLocationError::ConflictingStore)?;
548            let opened = opener(paths.clone()).await?;
549            if let Some((identifier, response)) = fetched {
550                // A stored configuration stays, and the checks below compare
551                // it with this backend.
552                if opened.db().server_configuration()?.is_none() {
553                    opened.db().store_server_configuration(
554                        &identifier,
555                        backend_url.trim_end_matches('/'),
556                        &response.encode_to_vec(),
557                        xmtp_common::time::now_ns(),
558                    )?;
559                }
560                if let Some(recorder) = &recorder {
561                    recorder.record(&identifier).await?;
562                }
563            }
564            store = Some(opened);
565            attachments_dir = Some(paths.attachments_dir);
566            if data_dir {
567                data_dir_opened_identifier = opened_identifier.clone();
568            }
569            deployment_recorder = recorder.map(|recorder| match opened_identifier {
570                Some(identifier) => recorder.for_opened_identifier(identifier),
571                None => recorder,
572            });
573        }
574        let store = store
575            .take()
576            .ok_or(ClientBuilderError::MissingParameter { parameter: "store" })?;
577        let mls_storage = mls_storage
578            .take()
579            .or_else(|| mls_storage_factory.map(|factory| factory(&store)))
580            .ok_or(ClientBuilderError::MissingParameter {
581                parameter: "mls_storage",
582            })?;
583
584        let conn = store.db();
585
586        // The configuration is resolved before any identity
587        // work, so a deployment that refuses this client refuses it before the
588        // database gains an identity. A caller-supplied provider
589        // short-circuits every network and database path here, which is what
590        // keeps `build_offline` free of a pending future.
591        let has_config_provider = config_provider.is_some();
592        let server_configuration = match config_provider {
593            Some(provider) => provider,
594            None => crate::server_configuration::resolve(&api_client, &conn, allow_offline).await?,
595        };
596
597        #[cfg(any(test, feature = "test-utils"))]
598        let server_configuration = match application_components_for_test {
599            Some(components) => server_configuration.with_application_components(components),
600            None => server_configuration,
601        };
602
603        if has_config_provider && let Some(opened_identifier) = data_dir_opened_identifier {
604            let provider_identifier = &server_configuration.configuration().identifier;
605            let stored_identifier = conn.server_configuration()?.map(|row| row.identifier);
606            if provider_identifier != &opened_identifier
607                || stored_identifier.as_deref().is_some_and(|identifier| {
608                    !identifier.is_empty() && identifier != provider_identifier
609                })
610            {
611                return Err(
612                    crate::storage_location::StorageLocationError::DeploymentMismatch.into(),
613                );
614            }
615        }
616
617        let server_configuration = server_configuration.with_chain_restriction(custom_scw_verifier);
618        if server_configuration.requires_preflight() {
619            api_client.api_client.require_preflight();
620        }
621        if !custom_scw_verifier {
622            scw_verifier = Box::new(api_client.clone());
623        }
624        if let Some(recorder) = deployment_recorder {
625            if !has_config_provider && !server_configuration.configuration().identifier.is_empty() {
626                recorder
627                    .record(&server_configuration.configuration().identifier)
628                    .await?;
629            }
630            server_configuration.set_deployment_recorder(recorder);
631        }
632
633        admit_configuration(
634            &mut api_client,
635            server_configuration.configuration(),
636            version_info.pkg_semver().semver(),
637        )?;
638
639        let identifier_check = match &identity_strategy {
640            IdentityStrategy::CreateIfNotFound {
641                identifier, nonce, ..
642            }
643            | IdentityStrategy::CreateForIdentifier { identifier, nonce }
644                if require_identifier_in_inbox =>
645            {
646                Some((identifier.clone(), *nonce))
647            }
648            _ => None,
649        };
650        let mut identity = if let Some(identity) = identity {
651            identity
652        } else {
653            identity_strategy
654                .initialize_identity(&api_client, &mls_storage, &scw_verifier)
655                .await?
656        };
657
658        // The registration request is handed to the app to
659        // sign, so bind it to the chains the deployment accepts before it
660        // leaves the client.
661        if let Some(request) = identity.signature_request.as_mut() {
662            server_configuration.restrict(request);
663        }
664
665        debug!(
666            inbox_id = identity.inbox_id(),
667            installation_id = identity.installation_keys.public_bytes().snippet(),
668            "Initialized identity"
669        );
670        if !allow_offline {
671            // get sequence_id from identity updates and loaded into the DB
672            load_identity_updates(
673                &api_client,
674                &conn,
675                vec![identity.inbox_id.as_str()].as_slice(),
676            )
677            .await?;
678        }
679        // The identity store trusts a stored identity of the same inbox. The
680        // check runs on the identity updates loaded above and before any
681        // worker, so a rejected identifier makes no request as the stored
682        // installation.
683        if let Some((identifier, nonce)) = identifier_check {
684            let membership = crate::identity_updates::identifier_membership(
685                &conn,
686                &identity.inbox_id,
687                &identifier,
688                &scw_verifier,
689            )
690            .await?;
691            if !crate::identity_updates::identifier_opens_inbox(
692                membership,
693                &identifier,
694                nonce,
695                &identity.inbox_id,
696            ) {
697                return Err(crate::identity::IdentityError::IdentifierNotInInbox {
698                    inbox_id: identity.inbox_id.clone(),
699                }
700                .into());
701            }
702        }
703
704        // Fold the legacy single-worker toggles into the unified enable map so
705        // there is one source of truth for "is worker X enabled" that code paths
706        // when it registers worker subscriptions. The old fields keep working;
707        // they just write here.
708        let mut worker_config = worker_config;
709        if disable_workers {
710            // Disable optional workers. The HMAC epoch observer still runs for
711            // every open client.
712            for kind in [
713                crate::worker::WorkerKind::DeviceSync,
714                crate::worker::WorkerKind::DisappearingMessages,
715                crate::worker::WorkerKind::CommitLog,
716                crate::worker::WorkerKind::TaskRunner,
717                crate::worker::WorkerKind::ConfigurationRefresh,
718                crate::worker::WorkerKind::AttachmentCleanup,
719            ] {
720                worker_config.enabled.insert(kind, false);
721            }
722        }
723        if matches!(device_sync_worker_mode, DeviceSyncMode::Disabled) {
724            worker_config
725                .enabled
726                .entry(crate::worker::WorkerKind::DeviceSync)
727                .or_insert(false);
728        }
729        if disable_commit_log_worker {
730            worker_config
731                .enabled
732                .entry(crate::worker::WorkerKind::CommitLog)
733                .or_insert(false);
734        }
735
736        let events = xmtp_events::EventBus::new();
737        let events_for_lifetime = events.clone();
738        let public_event_writer: Arc<dyn xmtp_events::EventWriter<()>> =
739            Arc::new(xmtp_events::PublicBusWriter::new(&events));
740        server_configuration.set_event_writer(public_event_writer.clone());
741        api_client
742            .api_client
743            .register_client_event_writer(&public_event_writer);
744        let mut workers = WorkerRunner::new();
745        let incoming_runtime =
746            crate::subscriptions::incoming::IncomingRuntime::new(stream_policy, incoming_factory)
747                .with_preflight(api_client.api_client.clone());
748        let attachments =
749            Arc::new(AttachmentRuntime::new(attachments_dir, attachment_options).await?);
750        let context = Arc::new(XmtpMlsLocalContext {
751            attachments,
752            identity,
753            mls_storage,
754            store,
755            api_client,
756            version_info,
757            server_configuration,
758            scw_verifier: Arc::new(scw_verifier),
759            mutexes: MutexRegistry::new(),
760            #[cfg(test)]
761            mls_commit_lock: Arc::new(GroupCommitLock::new()),
762            events,
763            public_event_writer,
764            registration_event_pending: Arc::new(AtomicBool::new(false)),
765            device_sync: DeviceSync {
766                mode: device_sync_worker_mode,
767            },
768            fork_recovery_opts: fork_recovery_opts.unwrap_or_default(),
769            change_callbacks,
770            incoming_runtime: Arc::new(incoming_runtime),
771            worker_config,
772
773            worker_metrics: workers.metrics().clone(),
774            task_channels: workers.task_channels().clone(),
775            cancellation_token: CancellationToken::new(),
776            shutdown_complete: Arc::new(AtomicBool::new(false)),
777            foreground_calls: Default::default(),
778            #[cfg(any(test, target_arch = "wasm32"))]
779            message_clock: Default::default(),
780            delivery_owner: Default::default(),
781            identity_resolutions: Default::default(),
782        });
783
784        crate::server_configuration::bind_preflight(&context);
785
786        // register workers
787        if let Err(error) = context.attachments.sweep(&context).await {
788            tracing::warn!(%error, "attachment cleanup failed during client build");
789        }
790        if let Err(error) = context.attachments.ensure_reconciled(&context).await {
791            tracing::warn!(%error, "attachment reconciliation failed during client build");
792        }
793        if !disable_workers {
794            use crate::worker::WorkerKind;
795            // One source of truth for enablement: the folded WorkerConfig map.
796            let enabled = |k| context.worker_config().worker_enabled(k);
797
798            // Keep the original `device_sync_worker_enabled()` AND-check to
799            // preserve the exact legacy semantics alongside the map gate.
800            if enabled(WorkerKind::DeviceSync) && context.device_sync_worker_enabled() {
801                workers.register_new_worker::<SyncWorker<ContextParts<ApiClient, S, Db>>, _>(
802                    context.clone(),
803                );
804            }
805            if enabled(WorkerKind::DisappearingMessages) {
806                workers
807                    .register_new_worker::<DisappearingMessagesWorker<ContextParts<ApiClient, S, Db>>, _>(
808                        context.clone(),
809                    );
810            }
811            // Enable CommitLogWorker based on configuration
812            // The deployment decides whether the commit log runs,
813            // falling back to the compiled default when it says nothing.
814            if enabled(WorkerKind::CommitLog)
815                && context
816                    .server_configuration()
817                    .configuration()
818                    .mls
819                    .commit_log_enabled()
820                && !disable_commit_log_worker
821            {
822                workers.register_new_worker::<
823                crate::groups::commit_log::CommitLogWorker<ContextParts<ApiClient, S, Db>>,
824                _,
825                >(context.clone());
826            }
827            // Only a client that reads its configuration refreshes it.
828            // A caller-supplied provider owns its own values.
829            if enabled(WorkerKind::ConfigurationRefresh) && !has_config_provider {
830                workers
831                    .register_new_worker::<crate::server_configuration::worker::ConfigurationWorker<
832                        ContextParts<ApiClient, S, Db>,
833                    >, _>(context.clone());
834            }
835            if enabled(WorkerKind::TaskRunner) {
836                workers.register_new_worker::<TaskWorker<ContextParts<ApiClient, S, Db>>, _>(
837                    context.clone(),
838                );
839                // KP maintenance is deliberately coupled to the TaskRunner (no
840                // standalone fallback): disabling the TaskRunner — per-kind via
841                // WorkerConfig or globally via disable_workers — disables KP
842                // rotation/deletion with it. Seeding failure is fatal to the
843                // build: the DB was already opened/migrated above, so an error
844                // here means it is broken; building a client whose critical
845                // maintenance silently never got seeded would be worse.
846                crate::worker::key_package_maintenance::seed_and_reconcile_kp_tasks(&context)?;
847                // One-time backfill: pending self-removes recorded before the
848                // worker became event-driven have no LeaveRequest to re-fire, so
849                // seed a ProcessPendingSelfRemove task for each already-flagged
850                // group. Best-effort (logged, never fails the build).
851                //
852                // TODO: remove this migration once all clients have shipped a
853                // release that enqueues these tasks inline — safe to delete by the
854                // next-next stable release.
855                if let Err(e) = backfill_pending_self_remove_tasks(&context) {
856                    tracing::warn!(
857                        "pending-self-remove backfill failed (will rely on next sync): {e}"
858                    );
859                }
860            }
861            if enabled(WorkerKind::AttachmentCleanup) && context.attachments.store.is_some() {
862                workers.register_new_worker::<crate::attachments::cleanup::AttachmentCleanup<
863                    ContextParts<ApiClient, S, Db>,
864                >, _>(context.clone());
865            }
866        }
867
868        // Every open client observes HMAC epoch changes, including clients
869        // that disable the other background workers.
870        workers.register_new_worker::<
871            crate::worker::hmac_epoch::HmacEpochWorker<ContextParts<ApiClient, S, Db>>,
872            _,
873        >(context.clone());
874
875        let workers = Arc::new(workers);
876
877        workers.spawn(context.clone());
878
879        log_event!(
880            Event::ClientCreated,
881            context.installation_id(),
882            inbox_id = context.inbox_id(),
883            device_sync_enabled = context.device_sync_worker_enabled(),
884            disabled_workers = disable_workers,
885        );
886
887        let installation_id = context.installation_id();
888        let client = Client {
889            context,
890            installation_id,
891            workers,
892            app_lifetime: Arc::new(crate::client::AppLifetime {
893                events: events_for_lifetime,
894            }),
895        };
896
897        // Cleanup old unstitched group updated messages.
898        let conn = DbConnection::new(client.db());
899        let cancel = client.context.cancellation_token().clone();
900        xmtp_common::spawn(None, async move {
901            tokio::select! {
902                _ = cancel.cancelled() => {}
903                _ = cleanup_duplicate_updates::perform(conn) => {}
904            }
905        });
906
907        Ok(client)
908    }
909
910    pub fn identity(self, identity: Identity) -> Self {
911        Self {
912            identity: Some(identity),
913            ..self
914        }
915    }
916
917    pub fn attachments_dir(mut self, path: impl Into<std::path::PathBuf>) -> Self {
918        self.storage_location_conflict |= self.storage_location_selected;
919        self.attachments_dir = Some(path.into());
920        self
921    }
922
923    pub fn attachment_options(mut self, options: xmtp_attachments::AttachmentOptions) -> Self {
924        self.attachment_options = options;
925        self
926    }
927
928    /// Save a deployment-scoped location. `build` resolves it after all options are set.
929    pub async fn data_location(
930        self,
931        location: crate::storage_location::StorageLocation,
932        key: xmtp_db::EncryptionKey,
933    ) -> Result<ClientBuilder<ApiClient, S, xmtp_db::DefaultStore>, ClientBuilderError> {
934        self.data_location_with(location, move |paths| open_location_store(paths, key))
935    }
936
937    /// Save a deployment-scoped location that `opener` opens. `build` resolves
938    /// the paths after all options are set and passes them to `opener`, so a
939    /// host can open the store with its own connection options.
940    pub fn data_location_with<NewDb>(
941        self,
942        location: crate::storage_location::StorageLocation,
943        opener: impl FnOnce(
944            crate::storage_location::ResolvedPaths,
945        )
946            -> xmtp_common::BoxDynFuture<'static, Result<NewDb, ClientBuilderError>>
947        + xmtp_common::MaybeSend
948        + xmtp_common::MaybeSync
949        + 'static,
950    ) -> Result<ClientBuilder<ApiClient, S, NewDb>, ClientBuilderError> {
951        location.validate()?;
952        let conflict = self.storage_location_conflict
953            || self.store.is_some()
954            || self.mls_storage.is_some()
955            || self.mls_storage_factory.is_some()
956            || self.storage_location_selected
957            || self.attachments_dir.is_some();
958        Ok(ClientBuilder {
959            deployment_recorder: None,
960            data_location: Some(location),
961            location_store_opener: Some(Box::new(opener)),
962            mls_storage_factory: None,
963            storage_location_selected: true,
964            storage_location_conflict: conflict,
965            attachments_dir: self.attachments_dir,
966            attachment_options: self.attachment_options,
967            api_client: self.api_client,
968            identity: self.identity,
969            store: None,
970            identity_strategy: self.identity_strategy,
971            scw_verifier: self.scw_verifier,
972            custom_scw_verifier: self.custom_scw_verifier,
973            device_sync_worker_mode: self.device_sync_worker_mode,
974            fork_recovery_opts: self.fork_recovery_opts,
975            change_callbacks: self.change_callbacks,
976            stream_policy: self.stream_policy,
977            incoming_factory: self.incoming_factory,
978            version_info: self.version_info,
979            allow_offline: self.allow_offline,
980            disable_commit_log_worker: self.disable_commit_log_worker,
981            require_identifier_in_inbox: self.require_identifier_in_inbox,
982            mls_storage: self.mls_storage,
983            disable_workers: self.disable_workers,
984            worker_config: self.worker_config,
985            config_provider: self.config_provider,
986            #[cfg(any(test, feature = "test-utils"))]
987            application_components_for_test: self.application_components_for_test,
988        })
989    }
990
991    /// Unstable: register callbacks notified when group state changes.
992    ///
993    /// Registration is construction-time by necessity — the changes these
994    /// report arrive from the stream and sync paths, where no SDK call is on
995    /// the stack. Passing [`UnstableChangeCallbacks::default`] (nothing set)
996    /// is equivalent to not calling this at all.
997    ///
998    /// See [`crate::groups::change_callbacks`] for the delivery contract.
999    pub fn unstable_change_callbacks(self, change_callbacks: UnstableChangeCallbacks) -> Self {
1000        Self {
1001            change_callbacks,
1002            ..self
1003        }
1004    }
1005
1006    pub fn store<NewDb>(self, db: NewDb) -> ClientBuilder<ApiClient, S, NewDb> {
1007        ClientBuilder {
1008            deployment_recorder: self.deployment_recorder,
1009            data_location: self.data_location,
1010            location_store_opener: None,
1011            mls_storage_factory: None,
1012            storage_location_selected: self.storage_location_selected,
1013            storage_location_conflict: self.storage_location_conflict
1014                || self.storage_location_selected,
1015            store: Some(db),
1016            api_client: self.api_client,
1017            identity: self.identity,
1018            identity_strategy: self.identity_strategy,
1019            scw_verifier: self.scw_verifier,
1020            custom_scw_verifier: self.custom_scw_verifier,
1021            device_sync_worker_mode: self.device_sync_worker_mode,
1022            fork_recovery_opts: self.fork_recovery_opts,
1023            change_callbacks: self.change_callbacks,
1024            stream_policy: self.stream_policy,
1025            incoming_factory: self.incoming_factory,
1026            version_info: self.version_info,
1027            allow_offline: self.allow_offline,
1028            disable_commit_log_worker: self.disable_commit_log_worker,
1029            require_identifier_in_inbox: self.require_identifier_in_inbox,
1030            mls_storage: self.mls_storage,
1031            disable_workers: self.disable_workers,
1032            worker_config: self.worker_config,
1033            config_provider: self.config_provider,
1034            #[cfg(any(test, feature = "test-utils"))]
1035            application_components_for_test: self.application_components_for_test,
1036            attachments_dir: self.attachments_dir,
1037            attachment_options: self.attachment_options,
1038        }
1039    }
1040
1041    /// Use the default SQlite MLS Key-Value Store
1042    pub fn default_mls_store(
1043        self,
1044    ) -> Result<
1045        ClientBuilder<ApiClient, SqlKeyStore<<Db as XmtpDb>::DbQuery>, Db>,
1046        ClientBuilderError,
1047    >
1048    where
1049        Db: XmtpDb,
1050    {
1051        Ok(ClientBuilder {
1052            deployment_recorder: self.deployment_recorder,
1053            data_location: self.data_location,
1054            location_store_opener: self.location_store_opener,
1055            mls_storage_factory: Some(|store: &Db| SqlKeyStore::new(store.db())),
1056            storage_location_selected: self.storage_location_selected,
1057            storage_location_conflict: self.storage_location_conflict,
1058            api_client: self.api_client,
1059            identity: self.identity,
1060            identity_strategy: self.identity_strategy,
1061            scw_verifier: self.scw_verifier,
1062            custom_scw_verifier: self.custom_scw_verifier,
1063            device_sync_worker_mode: self.device_sync_worker_mode,
1064            fork_recovery_opts: self.fork_recovery_opts,
1065            change_callbacks: self.change_callbacks,
1066            stream_policy: self.stream_policy,
1067            incoming_factory: self.incoming_factory,
1068            version_info: self.version_info,
1069            allow_offline: self.allow_offline,
1070            disable_commit_log_worker: self.disable_commit_log_worker,
1071            require_identifier_in_inbox: self.require_identifier_in_inbox,
1072            mls_storage: self
1073                .store
1074                .as_ref()
1075                .map(|store| SqlKeyStore::new(store.db())),
1076            store: self.store,
1077            disable_workers: self.disable_workers,
1078            worker_config: self.worker_config,
1079            config_provider: self.config_provider,
1080            #[cfg(any(test, feature = "test-utils"))]
1081            application_components_for_test: self.application_components_for_test,
1082            attachments_dir: self.attachments_dir,
1083            attachment_options: self.attachment_options,
1084        })
1085    }
1086
1087    pub fn mls_storage<NewS>(self, mls_storage: NewS) -> ClientBuilder<ApiClient, NewS, Db> {
1088        let storage_location_conflict =
1089            self.storage_location_conflict || self.storage_location_selected;
1090        ClientBuilder {
1091            deployment_recorder: self.deployment_recorder,
1092            data_location: self.data_location,
1093            location_store_opener: self.location_store_opener,
1094            mls_storage_factory: None,
1095            storage_location_selected: self.storage_location_selected,
1096            storage_location_conflict,
1097            store: self.store,
1098            api_client: self.api_client,
1099            identity: self.identity,
1100            identity_strategy: self.identity_strategy,
1101            scw_verifier: self.scw_verifier,
1102            custom_scw_verifier: self.custom_scw_verifier,
1103            device_sync_worker_mode: self.device_sync_worker_mode,
1104            fork_recovery_opts: self.fork_recovery_opts,
1105            change_callbacks: self.change_callbacks,
1106            stream_policy: self.stream_policy,
1107            incoming_factory: self.incoming_factory,
1108            version_info: self.version_info,
1109            allow_offline: self.allow_offline,
1110            disable_commit_log_worker: self.disable_commit_log_worker,
1111            require_identifier_in_inbox: self.require_identifier_in_inbox,
1112            mls_storage: Some(mls_storage),
1113            disable_workers: self.disable_workers,
1114            worker_config: self.worker_config,
1115            config_provider: self.config_provider,
1116            #[cfg(any(test, feature = "test-utils"))]
1117            application_components_for_test: self.application_components_for_test,
1118            attachments_dir: self.attachments_dir,
1119            attachment_options: self.attachment_options,
1120        }
1121    }
1122
1123    pub fn with_disable_workers(mut self, disable_workers: bool) -> Self {
1124        self.disable_workers = disable_workers;
1125        self
1126    }
1127
1128    pub fn with_device_sync_worker_mode(self, mode: Option<DeviceSyncMode>) -> Self {
1129        Self {
1130            device_sync_worker_mode: mode.unwrap_or(DeviceSyncMode::Enabled),
1131            ..self
1132        }
1133    }
1134
1135    pub fn device_sync_worker_mode(self, mode: DeviceSyncMode) -> Self {
1136        Self {
1137            device_sync_worker_mode: mode,
1138            ..self
1139        }
1140    }
1141
1142    pub fn fork_recovery_opts(self, opts: ForkRecoveryOpts) -> Self {
1143        Self {
1144            fork_recovery_opts: Some(opts),
1145            ..self
1146        }
1147    }
1148
1149    /// Configure background-worker intervals, jitter, and per-worker
1150    /// enablement. See [`crate::worker::WorkerConfig`].
1151    pub fn worker_config(mut self, cfg: crate::worker::WorkerConfig) -> Self {
1152        self.worker_config = cfg;
1153        self
1154    }
1155
1156    /// Replace only the application catalogue after normal configuration admission.
1157    /// Used by SDK metadata test fixtures.
1158    #[cfg(any(test, feature = "test-utils"))]
1159    pub fn application_components_for_test(
1160        mut self,
1161        components: Vec<xmtp_configuration::ApplicationComponentDefinition>,
1162    ) -> Self {
1163        self.application_components_for_test = Some(components);
1164        self
1165    }
1166
1167    /// Supply the server configuration instead of reading it.
1168    ///
1169    /// With a provider in place the client does not fetch, store, or refresh
1170    /// the configuration. A data directory still checks its deployment identifier.
1171    /// Rust callers only; the bindings do not expose this.
1172    pub fn config_provider(
1173        mut self,
1174        provider: Arc<dyn xmtp_configuration::ConfigProvider>,
1175    ) -> Self {
1176        self.config_provider = Some(provider);
1177        self
1178    }
1179
1180    /// Attach a query-only API client. Receipt uses ordered Query pages.
1181    /// Standard streaming clients use `api_client_with_streams` at construction.
1182    pub fn api_client<A>(self, api_client: A) -> ClientBuilder<A, S, Db> {
1183        ClientBuilder {
1184            deployment_recorder: self.deployment_recorder,
1185            data_location: self.data_location,
1186            location_store_opener: self.location_store_opener,
1187            mls_storage_factory: self.mls_storage_factory,
1188            storage_location_selected: self.storage_location_selected,
1189            storage_location_conflict: self.storage_location_conflict,
1190            api_client: Some(api_client),
1191            identity: self.identity,
1192            identity_strategy: self.identity_strategy,
1193            scw_verifier: self.scw_verifier,
1194            custom_scw_verifier: self.custom_scw_verifier,
1195            store: self.store,
1196            device_sync_worker_mode: self.device_sync_worker_mode,
1197            fork_recovery_opts: self.fork_recovery_opts,
1198            change_callbacks: self.change_callbacks,
1199            stream_policy: self.stream_policy,
1200            incoming_factory: None,
1201            version_info: self.version_info,
1202            allow_offline: self.allow_offline,
1203            disable_commit_log_worker: self.disable_commit_log_worker,
1204            require_identifier_in_inbox: self.require_identifier_in_inbox,
1205            mls_storage: self.mls_storage,
1206            disable_workers: self.disable_workers,
1207            worker_config: self.worker_config,
1208            config_provider: self.config_provider,
1209            #[cfg(any(test, feature = "test-utils"))]
1210            application_components_for_test: self.application_components_for_test,
1211            attachments_dir: self.attachments_dir,
1212            attachment_options: self.attachment_options,
1213        }
1214    }
1215
1216    xmtp_common::if_native! {
1217        /// Attach a native API client with a lazy shared Bidi receiver.
1218        /// No transport task or connection starts until the first receiving interest.
1219        pub fn api_client_with_streams<A>(self, api_client: A) -> ClientBuilder<A, S, Db>
1220        where
1221            A: xmtp_proto::api_client::XmtpMlsBidiStreams
1222                + crate::subscriptions::router_callbacks::ApiClientIdentity
1223                + Clone
1224                + Send
1225                + Sync
1226                + 'static,
1227            A::SubscribeStream: 'static,
1228        {
1229            let factory = crate::subscriptions::incoming::BidiSubscriptionFactory::new(api_client.clone());
1230            let mut builder = self.api_client(api_client);
1231            builder.incoming_factory = Some(Arc::new(factory));
1232            builder
1233        }
1234    }
1235
1236    xmtp_common::if_wasm! {
1237        /// Attach a browser API client with a lazy static-stream receiver.
1238        /// The factory owns the API client, not the client context.
1239        pub fn api_client_with_streams<A>(self, api_client: A) -> ClientBuilder<A, S, Db>
1240        where
1241            A: xmtp_proto::api_client::XmtpMlsStreams + Clone + 'static,
1242        {
1243            let api = api_client.clone();
1244            let mut builder = self.api_client(api_client);
1245            builder.incoming_factory = Some(Arc::new(move |cursors: xmtp_proto::types::TopicCursor, limits| -> crate::subscriptions::incoming::SubscriptionFuture {
1246                let api = api.clone();
1247                Box::pin(async move {
1248                    api.subscribe_envelopes_with_cursors(&cursors, limits)
1249                        .await
1250                        .map(|subscription| subscription.map_error(xmtp_proto::api::NetworkError::new))
1251                        .map_err(xmtp_proto::api::NetworkError::new)
1252                })
1253            }));
1254            builder
1255        }
1256    }
1257
1258    pub fn maybe_version(
1259        mut self,
1260        version: Option<VersionInfo>,
1261    ) -> ClientBuilder<ApiClient, S, Db> {
1262        if let Some(v) = version {
1263            self.version_info = v;
1264        }
1265        self
1266    }
1267
1268    pub fn version(self, version_info: VersionInfo) -> ClientBuilder<ApiClient, S, Db> {
1269        Self {
1270            version_info,
1271            ..self
1272        }
1273    }
1274
1275    /// Skip network calls when building a client
1276    pub fn with_allow_offline(
1277        self,
1278        allow_offline: Option<bool>,
1279    ) -> ClientBuilder<ApiClient, S, Db> {
1280        Self {
1281            allow_offline: allow_offline.unwrap_or(false),
1282            ..self
1283        }
1284    }
1285
1286    /// Require the identity strategy's identifier to belong to the inbox of
1287    /// the identity the build loads, including a stored one. The build checks
1288    /// the association state after it loads the inbox's identity updates and
1289    /// before it starts any worker. Before the inbox's first identity update,
1290    /// only the identifier whose inbox at the strategy's nonce is that inbox
1291    /// belongs to it. A failure is
1292    /// [`IdentityError::IdentifierNotInInbox`](crate::identity::IdentityError::IdentifierNotInInbox).
1293    pub fn require_identifier_in_inbox(self) -> Self {
1294        Self {
1295            require_identifier_in_inbox: true,
1296            ..self
1297        }
1298    }
1299
1300    /// Control whether the CommitLogWorker background task is enabled.
1301    /// Useful for tests that need deterministic commit log operations.
1302    #[cfg(any(test, feature = "test-utils"))]
1303    pub fn with_commit_log_worker(self, enabled: bool) -> Self {
1304        Self {
1305            disable_commit_log_worker: !enabled,
1306            ..self
1307        }
1308    }
1309
1310    #[cfg(any(test, feature = "test-utils"))]
1311    pub fn enable_sqlite_triggers(self) -> Self
1312    where
1313        Db: XmtpDb,
1314    {
1315        let db = self.store.as_ref().expect("unwrapping in test env").conn();
1316        let db = xmtp_db::DbConnection::new(db);
1317        db.register_triggers();
1318        db.disable_memory_security();
1319        self
1320    }
1321
1322    pub fn enable_api_stats(
1323        self,
1324    ) -> Result<ClientBuilder<TrackedStatsClient<ApiClient>, S, Db>, ClientBuilderError> {
1325        if self.api_client.is_none() {
1326            return Err(ClientBuilderError::MissingParameter {
1327                parameter: "api_client",
1328            });
1329        }
1330
1331        Ok(ClientBuilder {
1332            deployment_recorder: self.deployment_recorder,
1333            data_location: self.data_location,
1334            location_store_opener: self.location_store_opener,
1335            mls_storage_factory: self.mls_storage_factory,
1336            storage_location_selected: self.storage_location_selected,
1337            storage_location_conflict: self.storage_location_conflict,
1338            api_client: Some(TrackedStatsClient::new(
1339                self.api_client.expect("checked for none"),
1340            )),
1341            identity: self.identity,
1342            identity_strategy: self.identity_strategy,
1343            scw_verifier: self.scw_verifier,
1344            custom_scw_verifier: self.custom_scw_verifier,
1345            store: self.store,
1346
1347            device_sync_worker_mode: self.device_sync_worker_mode,
1348            fork_recovery_opts: self.fork_recovery_opts,
1349            change_callbacks: self.change_callbacks,
1350            stream_policy: self.stream_policy,
1351            incoming_factory: self.incoming_factory,
1352            version_info: self.version_info,
1353            allow_offline: self.allow_offline,
1354            disable_commit_log_worker: self.disable_commit_log_worker,
1355            require_identifier_in_inbox: self.require_identifier_in_inbox,
1356            mls_storage: self.mls_storage,
1357            disable_workers: self.disable_workers,
1358            worker_config: self.worker_config,
1359            config_provider: self.config_provider,
1360            #[cfg(any(test, feature = "test-utils"))]
1361            application_components_for_test: self.application_components_for_test,
1362            attachments_dir: self.attachments_dir,
1363            attachment_options: self.attachment_options,
1364        })
1365    }
1366
1367    pub fn with_scw_verifier(
1368        self,
1369        verifier: impl SmartContractSignatureVerifier + 'static,
1370    ) -> ClientBuilder<ApiClient, S, Db> {
1371        ClientBuilder {
1372            deployment_recorder: self.deployment_recorder,
1373            data_location: self.data_location,
1374            location_store_opener: self.location_store_opener,
1375            mls_storage_factory: self.mls_storage_factory,
1376            storage_location_selected: self.storage_location_selected,
1377            storage_location_conflict: self.storage_location_conflict,
1378            api_client: self.api_client,
1379            identity: self.identity,
1380            identity_strategy: self.identity_strategy,
1381            scw_verifier: Some(Box::new(verifier)),
1382            custom_scw_verifier: true,
1383            store: self.store,
1384
1385            device_sync_worker_mode: self.device_sync_worker_mode,
1386            fork_recovery_opts: self.fork_recovery_opts,
1387            change_callbacks: self.change_callbacks,
1388            stream_policy: self.stream_policy,
1389            incoming_factory: self.incoming_factory,
1390            version_info: self.version_info,
1391            allow_offline: self.allow_offline,
1392            disable_commit_log_worker: self.disable_commit_log_worker,
1393            require_identifier_in_inbox: self.require_identifier_in_inbox,
1394            mls_storage: self.mls_storage,
1395            disable_workers: self.disable_workers,
1396            worker_config: self.worker_config,
1397            config_provider: self.config_provider,
1398            #[cfg(any(test, feature = "test-utils"))]
1399            application_components_for_test: self.application_components_for_test,
1400            attachments_dir: self.attachments_dir,
1401            attachment_options: self.attachment_options,
1402        }
1403    }
1404
1405    /// Build the client with a default remote verifier
1406    /// requires the 'api' to be set.
1407    pub fn with_remote_verifier(self) -> Result<ClientBuilder<ApiClient, S, Db>, ClientBuilderError>
1408    where
1409        ApiClient: Clone + XmtpApi + 'static,
1410    {
1411        let api = self
1412            .api_client
1413            .clone()
1414            .ok_or(ClientBuilderError::MissingParameter {
1415                parameter: "api_client",
1416            })?;
1417
1418        Ok(ClientBuilder {
1419            deployment_recorder: self.deployment_recorder,
1420            data_location: self.data_location,
1421            location_store_opener: self.location_store_opener,
1422            mls_storage_factory: self.mls_storage_factory,
1423            storage_location_selected: self.storage_location_selected,
1424            storage_location_conflict: self.storage_location_conflict,
1425            api_client: self.api_client,
1426            identity: self.identity,
1427            identity_strategy: self.identity_strategy,
1428            scw_verifier: Some(Box::new(ApiClientWrapper::new(api, Retry::default()))
1429                as Box<dyn SmartContractSignatureVerifier>),
1430            // The default verifier replaces the caller's verifier, so the
1431            // exemption from the chain restriction ends here.
1432            custom_scw_verifier: false,
1433            store: self.store,
1434            device_sync_worker_mode: self.device_sync_worker_mode,
1435            fork_recovery_opts: self.fork_recovery_opts,
1436            change_callbacks: self.change_callbacks,
1437            stream_policy: self.stream_policy,
1438            incoming_factory: self.incoming_factory,
1439            version_info: self.version_info,
1440            allow_offline: self.allow_offline,
1441            disable_commit_log_worker: self.disable_commit_log_worker,
1442            require_identifier_in_inbox: self.require_identifier_in_inbox,
1443            mls_storage: self.mls_storage,
1444            disable_workers: self.disable_workers,
1445            worker_config: self.worker_config,
1446            config_provider: self.config_provider,
1447            #[cfg(any(test, feature = "test-utils"))]
1448            application_components_for_test: self.application_components_for_test,
1449            attachments_dir: self.attachments_dir,
1450            attachment_options: self.attachment_options,
1451        })
1452    }
1453}
1454
1455#[cfg(test)]
1456mod worker_registration_tests {
1457    use crate::tester;
1458    use crate::worker::{WorkerConfig, WorkerKind};
1459
1460    #[xmtp_common::test(unwrap_try = true)]
1461    #[cfg_attr(target_arch = "wasm32", ignore)]
1462    async fn disabled_worker_is_not_registered() {
1463        let mut cfg = WorkerConfig::default();
1464        cfg.enabled.insert(WorkerKind::DisappearingMessages, false);
1465        tester!(alix, worker_config: cfg);
1466
1467        let kinds = alix.client.workers.registered_kinds();
1468        assert!(
1469            !kinds.contains(&WorkerKind::DisappearingMessages),
1470            "disabled worker must not be registered, got {kinds:?}"
1471        );
1472        assert!(
1473            kinds.contains(&WorkerKind::TaskRunner),
1474            "un-disabled worker must still be registered, got {kinds:?}"
1475        );
1476        let subscriptions = alix.client.workers.subscribed_kinds();
1477        assert!(!subscriptions.contains(&WorkerKind::DisappearingMessages));
1478        assert!(subscriptions.contains(&WorkerKind::TaskRunner));
1479    }
1480
1481    #[xmtp_common::test(unwrap_try = true)]
1482    #[cfg_attr(target_arch = "wasm32", ignore)]
1483    async fn disabled_optional_workers_keep_the_epoch_observer() {
1484        tester!(alix, disable_workers);
1485        assert_eq!(
1486            alix.client.workers.registered_kinds(),
1487            vec![WorkerKind::HmacEpoch]
1488        );
1489        assert!(alix.client.workers.is_running());
1490    }
1491}
xmtp_mls/builder.rs

xmtp_mls/
builder.rs
