coven_domain/joining/
client.rs

1//! Join an existing shared store using a recipient-sealed device invitation.
2//!
3//! Shared across all platforms (macOS, iOS, CLI).
4
5use std::sync::Arc;
6
7use tokio::sync::watch;
8use tracing::{info, warn};
9
10use coven_database::supported_version;
11use coven_database::Migration;
12use coven_database::{CovenMigrationPolicy, Database};
13#[cfg(feature = "oauth-providers")]
14use coven_foundation::config::CloudProvider;
15use coven_foundation::config::{Config, ConfigError, HomeStorage};
16use coven_foundation::stage_timing::StageTimings;
17use coven_foundation::store_dir::{StoreDir, StoreLayout};
18use coven_keys::encryption::{EncryptionError, EncryptionService, MasterKeyring};
19use coven_keys::identity_custody::IdentityCustody;
20use coven_keys::keys::{
21    CloudHomeCredentials, DeviceIdentityCustody, KeyError, MasterKeyCustody, StoreKeys, UserKeypair,
22};
23use coven_protocol::synced_schema::SyncedTable;
24use coven_replication::sync::store::{
25    MembershipMutationError, PreparedDeviceJoinSnapshot, PreparedSnapshotBootstrap, PullError,
26    SnapshotError,
27};
28use coven_replication::sync::MemberAdmission;
29use coven_storage::cloud::{CloudHomeError, CloudHomeJoinInfo, ExactCloudHome};
30use coven_storage::{BlobPathScheme, CloudCipher, CloudSyncConnection};
31
32/// Why joining or restoring a store failed. Both are the same operation —
33/// bootstrap a store from the cloud — differing only in their entry data (an
34/// admission that wraps the store key vs a restore code that carries the bucket
35/// credentials), so they share one error shape rather than two that duplicate
36/// most of their variants and then have to map between each other.
37#[derive(Debug, thiserror::Error)]
38pub enum BootstrapError {
39    #[error("cloud home: {0}")]
40    CloudHome(#[from] CloudHomeError),
41    #[error("encryption: {0}")]
42    Encryption(#[from] EncryptionError),
43    #[error("membership mutation: {0}")]
44    MembershipMutation(#[source] Box<MembershipMutationError>),
45    #[error("snapshot: {0}")]
46    Snapshot(SnapshotError),
47    #[error("pull: {0}")]
48    Pull(#[from] PullError),
49    #[error("Store pull: {0}")]
50    StorePull(#[from] coven_replication::sync::store::StorePullError),
51    #[error("Store device registration: {0}")]
52    StoreRegistration(#[from] coven_replication::sync::store::StoreRegistrationError),
53    #[error("Store device join: {0}")]
54    DeviceJoin(#[from] coven_replication::sync::DeviceJoinError),
55    #[error("Store device join transport: {0}")]
56    DeviceJoinTransport(#[from] coven_replication::sync::store::DeviceJoinTransportError),
57    #[error("storage: {0}")]
58    Storage(#[from] coven_protocol::objects::StorageError),
59    #[error("config: {0}")]
60    Config(#[from] ConfigError),
61    #[error("keyring: {0}")]
62    Key(#[from] KeyError),
63    #[error("I/O: {0}")]
64    Io(#[from] std::io::Error),
65    #[error("device invitation: {0}")]
66    DeviceInvite(#[from] crate::joining::DeviceInviteError),
67    #[error("device pairing: {0}")]
68    Pairing(#[from] crate::joining::DevicePairingTransportError),
69    #[error("device pairing state: {0}")]
70    PairingState(#[from] crate::joining::DevicePairingError),
71    #[error("device join invite version {0} is not supported")]
72    UnsupportedDeviceInviteVersion(u32),
73    #[error("invalid store id: {0}")]
74    InvalidStoreId(#[from] coven_foundation::store_dir::PathTokenError),
75    #[error("invalid restore code: {0}")]
76    RestoreCode(#[from] crate::restoration::RestoreCodeError),
77    #[error("store already exists locally: {0}")]
78    StoreExists(String),
79    /// A hard crash left a store directory with no saved config, and clearing
80    /// that torn-bootstrap residue before retrying failed — so the retry can't
81    /// proceed over the leftover directory or keyring entries.
82    #[error("could not clear a torn bootstrap for {store_id}: {failures}")]
83    TornBootstrapCleanup {
84        store_id: String,
85        failures: BootstrapCleanupFailures,
86    },
87    #[error("could not remove cancelled join state for {store_id}: {failures}")]
88    CancelledJoinCleanup {
89        store_id: String,
90        failures: BootstrapCleanupFailures,
91    },
92    #[error("provider: {0}")]
93    Provider(String),
94    #[cfg(feature = "oauth-providers")]
95    #[error("OAuth client configuration: {0}")]
96    OAuthClient(#[from] coven_storage::oauth::OAuthClientCredsError),
97    #[error("{provider:?} cannot provide exact protocol and blob slots with this configuration")]
98    ExactSlotsUnavailable {
99        provider: coven_foundation::config::CloudProvider,
100    },
101    #[error("database open: {0}")]
102    DatabaseOpen(#[from] coven_database::OpenError),
103    #[error("invalid signing key: {0}")]
104    InvalidSigningKey(#[from] SigningKeyError),
105    /// The caller's cancel signal fired at a phase boundary, so the join or
106    /// restore stopped before saving the store. This returns through the same
107    /// failure-cleanup path a real error takes — removing the partly-created
108    /// store directory and any per-store keyring entries written so far — so a
109    /// cancelled bootstrap leaves no residue in either place.
110    #[error("the operation was cancelled")]
111    Cancelled,
112    /// Bootstrap failed AND cleaning up what it had durably written also failed.
113    /// Both are carried: `cause` is the original bootstrap failure that
114    /// triggered the cleanup, `cleanup` is why the cleanup itself failed — the
115    /// cause is preserved as a value, not flattened into a string.
116    #[error(
117        "could not clean up the partial store after bootstrap failed: {cleanup} (bootstrap error: {cause})"
118    )]
119    Cleanup {
120        cleanup: BootstrapCleanupFailures,
121        cause: Box<BootstrapError>,
122    },
123}
124
125impl From<SnapshotError> for BootstrapError {
126    fn from(error: SnapshotError) -> Self {
127        match error {
128            SnapshotError::Cancelled => Self::Cancelled,
129            error => Self::Snapshot(error),
130        }
131    }
132}
133
134#[derive(Debug, thiserror::Error)]
135pub enum SigningKeyError {
136    #[error("{0}")]
137    Material(#[from] coven_foundation::code_envelope::FixedHexError),
138    #[error("activated continuation has no device signing key")]
139    MissingContinuationSigner,
140    #[error("Owner recovery cannot carry an activated device signer")]
141    UnexpectedOwnerRecoverySigner,
142}
143
144#[derive(Debug, thiserror::Error)]
145pub enum BootstrapCleanupFailure {
146    #[error("store directory: {0}")]
147    StoreDirectory(#[source] std::io::Error),
148    #[error("master key: {0}")]
149    MasterKey(#[source] KeyError),
150    #[error("identity: {0}")]
151    Identity(#[source] KeyError),
152    #[error("cloud home credentials: {0}")]
153    CloudHomeCredentials(#[source] KeyError),
154}
155
156#[derive(Debug)]
157pub struct BootstrapCleanupFailures(Vec<BootstrapCleanupFailure>);
158
159impl BootstrapCleanupFailures {
160    fn is_empty(&self) -> bool {
161        self.0.is_empty()
162    }
163}
164
165impl std::fmt::Display for BootstrapCleanupFailures {
166    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
167        for (index, failure) in self.0.iter().enumerate() {
168            if index > 0 {
169                formatter.write_str("; ")?;
170            }
171            write!(formatter, "{failure}")?;
172        }
173        Ok(())
174    }
175}
176
177impl std::error::Error for BootstrapCleanupFailures {
178    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
179        self.0.first().map(|failure| failure as _)
180    }
181}
182
183impl From<MembershipMutationError> for BootstrapError {
184    fn from(error: MembershipMutationError) -> Self {
185        Self::MembershipMutation(Box::new(error))
186    }
187}
188
189/// The complete local cleanup capability for one bootstrap attempt.
190pub(crate) struct BootstrapCleanup<'a> {
191    store_dir: &'a StoreDir,
192    store_keys: &'a StoreKeys,
193    custody: &'a dyn MasterKeyCustody,
194    identity_custody: &'a dyn DeviceIdentityCustody,
195}
196
197impl<'a> BootstrapCleanup<'a> {
198    pub(crate) fn new(
199        store_dir: &'a StoreDir,
200        store_keys: &'a StoreKeys,
201        custody: &'a dyn MasterKeyCustody,
202        identity_custody: &'a dyn DeviceIdentityCustody,
203    ) -> Self {
204        Self {
205            store_dir,
206            store_keys,
207            custody,
208            identity_custody,
209        }
210    }
211
212    /// Refuse a completed store and remove all local state from a torn attempt.
213    pub(crate) fn refuse_completed_or_clear(&self, store_id: &str) -> Result<(), BootstrapError> {
214        if self.store_dir.config_path().exists() {
215            return Err(BootstrapError::StoreExists(store_id.to_string()));
216        }
217
218        if self.store_dir.exists() {
219            warn!(
220                store_dir = %self.store_dir.display(),
221                "clearing a torn bootstrap: a store directory with no saved config, left by a restore that a crash interrupted before completion"
222            );
223            let failures = self.remove();
224            if !failures.is_empty() {
225                return Err(BootstrapError::TornBootstrapCleanup {
226                    store_id: store_id.to_string(),
227                    failures,
228                });
229            }
230        }
231
232        Ok(())
233    }
234
235    /// Remove partial local state and preserve the initiating bootstrap error.
236    pub(crate) fn after_failure(&self, cause: BootstrapError) -> BootstrapError {
237        let failures = self.remove();
238        if failures.is_empty() {
239            cause
240        } else {
241            BootstrapError::Cleanup {
242                cleanup: failures,
243                cause: Box::new(cause),
244            }
245        }
246    }
247
248    /// Remove every local artifact the bound bootstrap attempt may have written.
249    pub(crate) fn remove(&self) -> BootstrapCleanupFailures {
250        let mut failures = Vec::new();
251
252        if let Err(error) = self.store_dir.remove_tree() {
253            failures.push(BootstrapCleanupFailure::StoreDirectory(error));
254        }
255        if let Err(error) = self.custody.forget() {
256            failures.push(BootstrapCleanupFailure::MasterKey(error));
257        }
258        if let Err(error) = self.identity_custody.forget() {
259            failures.push(BootstrapCleanupFailure::Identity(error));
260        }
261        if let Err(error) = self.store_keys.delete_cloud_home_credentials() {
262            failures.push(BootstrapCleanupFailure::CloudHomeCredentials(error));
263        }
264
265        BootstrapCleanupFailures(failures)
266    }
267}
268
269async fn build_cloud_home_for_join(
270    join_info: &CloudHomeJoinInfo,
271    lib_ks: &StoreKeys,
272    cloud_homes: &coven_storage::cloud::CloudHomeFactory,
273    oauth_tokens: Option<coven_storage::oauth::OAuthTokens>,
274    cloudkit_ops: Option<Arc<dyn coven_storage::cloud::cloudkit::CloudKitOps>>,
275    clock: coven_foundation::clock::ClockRef,
276    exact_upload_verification: coven_foundation::config::ExactUploadVerification,
277) -> Result<Arc<dyn ExactCloudHome>, BootstrapError> {
278    use coven_storage::cloud::*;
279
280    #[cfg(not(feature = "oauth-providers"))]
281    let _ = (&lib_ks, cloud_homes, &oauth_tokens, &clock);
282    #[cfg(feature = "oauth-providers")]
283    let credential_custody =
284        coven_keys::keys::CloudHomeCredentialsOwner::new(lib_ks.clone()).current();
285
286    match join_info {
287        CloudHomeJoinInfo::S3 {
288            bucket,
289            region,
290            endpoint,
291            access_key,
292            secret_key,
293            key_prefix,
294        } => {
295            let s3 = cloud_homes
296                .open_s3(
297                    bucket.clone(),
298                    region.clone(),
299                    endpoint.clone(),
300                    access_key.clone(),
301                    secret_key.clone(),
302                    key_prefix.clone(),
303                    exact_upload_verification,
304                    clock.clone(),
305                )
306                .await?;
307            Ok(Arc::new(s3))
308        }
309        #[cfg(feature = "oauth-providers")]
310        CloudHomeJoinInfo::GoogleDrive { folder_id } => {
311            let tokens = oauth_tokens.ok_or_else(|| {
312                BootstrapError::Provider("Google Drive join requires an OAuth token".to_string())
313            })?;
314            let oauth_config = cloud_homes.oauth_config_for(CloudProvider::GoogleDrive)?;
315            let session = oauth_session::OAuthSession::new(
316                tokens,
317                credential_custody.clone(),
318                clock,
319                oauth_config,
320                "Google Drive",
321            );
322            Ok(Arc::new(google_drive::GoogleDriveCloudHome::new(
323                folder_id.clone(),
324                session,
325                exact_upload_verification,
326            )))
327        }
328        #[cfg(feature = "oauth-providers")]
329        CloudHomeJoinInfo::Dropbox { folder_path } => {
330            let tokens = oauth_tokens.ok_or_else(|| {
331                BootstrapError::Provider("Dropbox join requires an OAuth token".to_string())
332            })?;
333            let oauth_config = cloud_homes.oauth_config_for(CloudProvider::Dropbox)?;
334            let session = oauth_session::OAuthSession::new(
335                tokens,
336                credential_custody.clone(),
337                clock,
338                oauth_config,
339                "Dropbox",
340            );
341            Ok(Arc::new(dropbox::DropboxCloudHome::new(
342                folder_path.clone(),
343                session,
344                exact_upload_verification,
345            )))
346        }
347        #[cfg(feature = "oauth-providers")]
348        CloudHomeJoinInfo::OneDrive {
349            drive_id,
350            folder_id,
351        } => {
352            let tokens = oauth_tokens.ok_or_else(|| {
353                BootstrapError::Provider("OneDrive join requires an OAuth token".to_string())
354            })?;
355            let oauth_config = cloud_homes.oauth_config_for(CloudProvider::OneDrive)?;
356            let session = oauth_session::OAuthSession::new(
357                tokens,
358                credential_custody,
359                clock,
360                oauth_config,
361                "OneDrive",
362            );
363            Ok(Arc::new(onedrive::OneDriveCloudHome::new(
364                drive_id.clone(),
365                folder_id.clone(),
366                session,
367                exact_upload_verification,
368            )))
369        }
370        #[cfg(not(feature = "oauth-providers"))]
371        CloudHomeJoinInfo::GoogleDrive { .. }
372        | CloudHomeJoinInfo::Dropbox { .. }
373        | CloudHomeJoinInfo::OneDrive { .. } => Err(BootstrapError::Provider(
374            "OAuth cloud providers are not supported in this build".to_string(),
375        )),
376        CloudHomeJoinInfo::CloudKit => {
377            let ops = cloudkit_ops.ok_or_else(|| {
378                BootstrapError::Provider("CloudKit driver not provided".to_string())
379            })?;
380            Ok(Arc::new(cloudkit::CloudKitCloudHome::new_private(
381                ops,
382                exact_upload_verification,
383            )))
384        }
385        CloudHomeJoinInfo::CloudKitShare {
386            share_url,
387            owner_name,
388            zone_name,
389        } => {
390            let ops = cloudkit_ops.ok_or_else(|| {
391                BootstrapError::Provider("CloudKit driver not provided".to_string())
392            })?;
393            let accepted = cloudkit::accept_share(ops.clone(), share_url.clone()).await?;
394            if accepted.owner_name != *owner_name || accepted.zone_name != *zone_name {
395                return Err(BootstrapError::Provider(format!(
396                    "CloudKit accepted share zone mismatch: invite owner/zone {owner_name}/{zone_name}, accepted {}/{}",
397                    accepted.owner_name, accepted.zone_name
398                )));
399            }
400            let home = Arc::new(cloudkit::CloudKitCloudHome::new_shared(
401                ops.clone(),
402                owner_name.clone(),
403                zone_name.clone(),
404                exact_upload_verification,
405            ));
406            Ok(home)
407        }
408    }
409}
410
411pub(crate) enum EnrollmentProviderAccess {
412    Supplied(Option<coven_storage::oauth::OAuthTokens>),
413    Stored,
414    #[cfg(any(test, feature = "test-utils"))]
415    InjectedHome,
416}
417
418#[cfg(feature = "oauth-providers")]
419pub(crate) fn enrollment_oauth_tokens(
420    join_info: &CloudHomeJoinInfo,
421    store_keys: &StoreKeys,
422    access: EnrollmentProviderAccess,
423) -> Result<Option<coven_storage::oauth::OAuthTokens>, BootstrapError> {
424    let provider = match join_info {
425        CloudHomeJoinInfo::GoogleDrive { .. } => "Google Drive",
426        CloudHomeJoinInfo::Dropbox { .. } => "Dropbox",
427        CloudHomeJoinInfo::OneDrive { .. } => "OneDrive",
428        CloudHomeJoinInfo::S3 { .. }
429        | CloudHomeJoinInfo::CloudKit
430        | CloudHomeJoinInfo::CloudKitShare { .. } => return Ok(None),
431    };
432    match access {
433        EnrollmentProviderAccess::Supplied(Some(tokens)) => {
434            store_keys.set_cloud_home_oauth_tokens(&tokens)?;
435            Ok(Some(tokens))
436        }
437        EnrollmentProviderAccess::Supplied(None) | EnrollmentProviderAccess::Stored => store_keys
438            .get_cloud_home_oauth_tokens()?
439            .map(Some)
440            .ok_or_else(|| {
441                BootstrapError::Provider(format!(
442                    "{provider} device enrollment requires OAuth authorization"
443                ))
444            }),
445        #[cfg(any(test, feature = "test-utils"))]
446        EnrollmentProviderAccess::InjectedHome => Ok(None),
447    }
448}
449
450#[cfg(not(feature = "oauth-providers"))]
451pub(crate) fn enrollment_oauth_tokens(
452    join_info: &CloudHomeJoinInfo,
453    _store_keys: &StoreKeys,
454    access: EnrollmentProviderAccess,
455) -> Result<Option<coven_storage::oauth::OAuthTokens>, BootstrapError> {
456    match access {
457        EnrollmentProviderAccess::Supplied(tokens) => drop(tokens),
458        EnrollmentProviderAccess::Stored => {}
459        #[cfg(any(test, feature = "test-utils"))]
460        EnrollmentProviderAccess::InjectedHome => {}
461    }
462    match join_info {
463        CloudHomeJoinInfo::GoogleDrive { .. }
464        | CloudHomeJoinInfo::Dropbox { .. }
465        | CloudHomeJoinInfo::OneDrive { .. } => Err(BootstrapError::Provider(
466            "OAuth cloud providers are not supported in this build".to_string(),
467        )),
468        CloudHomeJoinInfo::S3 { .. }
469        | CloudHomeJoinInfo::CloudKit
470        | CloudHomeJoinInfo::CloudKitShare { .. } => Ok(None),
471    }
472}
473
474/// A joining device's local half of the four-transfer admission exchange.
475/// The journal lives outside the incomplete store directory, so every method
476/// can be retried after process termination without losing its exact predecessor.
477pub(crate) struct DeviceJoinClient {
478    admission: MemberAdmission,
479    member_pubkey: String,
480    layout: StoreLayout,
481    synced_tables: Vec<SyncedTable>,
482    migrations: Vec<Migration>,
483    coven_migration_policy: CovenMigrationPolicy,
484    exact_upload_verification: coven_foundation::config::ExactUploadVerification,
485    transfer_limits: coven_protocol::blob::TransferLimits,
486    store_keys: StoreKeys,
487    custody: Arc<dyn MasterKeyCustody>,
488    identity_custody: Arc<dyn DeviceIdentityCustody>,
489    cloud_homes: coven_storage::cloud::CloudHomeFactory,
490    oauth_tokens: Option<coven_storage::oauth::OAuthTokens>,
491    cloudkit_ops: Option<Arc<dyn coven_storage::cloud::cloudkit::CloudKitOps>>,
492    clock: coven_foundation::clock::ClockRef,
493    #[cfg(any(test, feature = "test-utils"))]
494    test_home: Option<Arc<dyn ExactCloudHome>>,
495}
496
497struct DeviceJoinStorage {
498    storage: Arc<dyn coven_storage::CloudSyncObjectStorage>,
499    keyring: MasterKeyring,
500    /// The owner-anchored membership chain the keyring open already walked and
501    /// verified. Installing this device's owner anchor needs the same chain, so
502    /// it is kept rather than walked from the cloud a second time.
503    membership: coven_replication::sync::store::AcceptedMembershipAuthority,
504}
505
506impl DeviceJoinClient {
507    #[allow(clippy::too_many_arguments)]
508    pub(crate) fn new(
509        admission: MemberAdmission,
510        member_pubkey: String,
511        layout: StoreLayout,
512        synced_tables: Vec<SyncedTable>,
513        migrations: Vec<Migration>,
514        coven_migration_policy: CovenMigrationPolicy,
515        exact_upload_verification: coven_foundation::config::ExactUploadVerification,
516        transfer_limits: coven_protocol::blob::TransferLimits,
517        key_custody: coven_keys::custody::KeyCustody,
518        identity_custody: IdentityCustody,
519        oauth_clients: coven_storage::oauth::OAuthClients,
520        oauth_tokens: Option<coven_storage::oauth::OAuthTokens>,
521        cloudkit_ops: Option<Arc<dyn coven_storage::cloud::cloudkit::CloudKitOps>>,
522        clock: coven_foundation::clock::ClockRef,
523    ) -> Result<Self, BootstrapError> {
524        if admission.wrapped_key.recipient_pubkey != member_pubkey {
525            return Err(crate::joining::DeviceInviteError::RecipientMismatch.into());
526        }
527        coven_storage::cloud::setup::require_exact_slot_capabilities_join_info(
528            &admission.join_info,
529            exact_upload_verification,
530        )
531        .map_err(|provider| BootstrapError::ExactSlotsUnavailable { provider })?;
532        coven_foundation::store_dir::validate_path_token(&admission.store_id)?;
533        let store_dir = layout.store_dir(&admission.store_id);
534        let store_keys = StoreKeys::bind(admission.store_id.clone());
535        let custody = key_custody.resolve(&store_keys, &store_dir);
536        let identity_custody = identity_custody.resolve(&store_keys, &store_dir);
537        Ok(Self {
538            admission,
539            member_pubkey,
540            layout,
541            synced_tables,
542            migrations,
543            coven_migration_policy,
544            exact_upload_verification,
545            transfer_limits,
546            store_keys,
547            custody,
548            identity_custody,
549            cloud_homes: coven_storage::cloud::CloudHomeFactory::new(oauth_clients),
550            oauth_tokens,
551            cloudkit_ops,
552            clock,
553            #[cfg(any(test, feature = "test-utils"))]
554            test_home: None,
555        })
556    }
557
558    #[cfg(any(test, feature = "test-utils"))]
559    pub(crate) fn with_test_bootstrap_home(mut self, home: Arc<dyn ExactCloudHome>) -> Self {
560        self.test_home = Some(home);
561        self
562    }
563
564    pub(crate) async fn prepare_provider_access_request(
565        &self,
566        offer: coven_replication::sync::DeviceJoinOffer,
567    ) -> Result<coven_replication::sync::DeviceProviderAccessRequest, BootstrapError> {
568        self.require_offer(&offer)?;
569        let signer = coven_keys::keys::peek_pending_identity(&offer.member_pubkey)?;
570        let storage: Arc<dyn coven_storage::CloudSyncObjectStorage> =
571            Arc::new(self.transport_storage().await?);
572        let pending = self.open_pending_journal()?;
573        let observation = coven_replication::sync::store::PendingDeviceJoinObservation::open(
574            &pending,
575            &storage,
576            &offer.store_root,
577            offer.attempt_id,
578        )
579        .await?;
580        let authority = coven_replication::sync::store::PendingDeviceJoinAuthority::open(
581            observation,
582            &signer,
583            offer,
584        )
585        .await?;
586        Ok(authority.prepare_provider_access_request().await?)
587    }
588
589    pub(crate) async fn accept_device_join_abandonment(
590        &self,
591        abandonment: coven_replication::sync::DeviceJoinAbandonment,
592    ) -> Result<coven_replication::sync::DeviceJoinAbandonment, BootstrapError> {
593        let storage: Arc<dyn coven_storage::CloudSyncObjectStorage> =
594            Arc::new(self.transport_storage().await?);
595        let pending = self.open_pending_journal()?;
596        let mut observation = coven_replication::sync::store::PendingDeviceJoinObservation::open(
597            &pending,
598            &storage,
599            &self.admission.store_root,
600            abandonment.abandonment.attempt_id,
601        )
602        .await?;
603        Ok(observation.observe_abandonment(abandonment).await?)
604    }
605
606    /// The library this join produced, when it already exists.
607    ///
608    /// This is the joining device's finished marker. It used to be a journal
609    /// row that outlived the join it described; the config file is written by
610    /// the same completion, says the same thing, and is the file the library is
611    /// opened from — so the row was a second copy of it that nothing else read.
612    pub(crate) fn completed_library(&self) -> Result<Option<Config>, BootstrapError> {
613        let store_dir = self.layout.store_dir(&self.admission.store_id);
614        if !store_dir.config_path().exists() {
615            return Ok(None);
616        }
617        let config = Config::load_from_config_yaml(&store_dir)?;
618        if config.store_id != self.admission.store_id {
619            return Err(coven_replication::sync::DeviceJoinError::JournalConflict.into());
620        }
621        Ok(Some(config))
622    }
623
624    pub(crate) fn device_join_status(
625        &self,
626        attempt_id: coven_protocol::DeviceJoinAttemptId,
627    ) -> Result<Option<coven_replication::sync::DeviceJoinStatus>, BootstrapError> {
628        let pending = self.open_pending_journal()?;
629        Ok(pending.status(attempt_id)?)
630    }
631
632    #[cfg(test)]
633    pub(crate) fn pending_journal_records_for_test(
634        &self,
635    ) -> Result<
636        Vec<coven_protocol::store_commit::device_join_journal::DeviceJoinJournalRecord>,
637        BootstrapError,
638    > {
639        Ok(self.open_pending_journal()?.records()?)
640    }
641
642    #[cfg(test)]
643    pub(crate) fn resume_device_joins(
644        &self,
645    ) -> Result<Vec<coven_replication::sync::DeviceJoinAction>, BootstrapError> {
646        let pending = self.open_pending_journal()?;
647        Ok(pending.actions()?)
648    }
649
650    pub(crate) async fn prepare_registration_request(
651        &self,
652        approval: coven_replication::sync::DeviceProviderAdmissionApproval,
653    ) -> Result<coven_replication::sync::DeviceRegistrationRequest, BootstrapError> {
654        let offer = &approval.request.offer;
655        self.require_offer(offer)?;
656        let signer = coven_keys::keys::peek_pending_identity(&offer.member_pubkey)?;
657        let storage: Arc<dyn coven_storage::CloudSyncObjectStorage> =
658            Arc::new(self.transport_storage().await?);
659        let pending = self.open_pending_journal()?;
660        let observation = coven_replication::sync::store::PendingDeviceJoinObservation::open(
661            &pending,
662            &storage,
663            &offer.store_root,
664            offer.attempt_id,
665        )
666        .await?;
667        let mut authority = coven_replication::sync::store::PendingDeviceJoinAuthority::open(
668            observation,
669            &signer,
670            offer.as_ref().clone(),
671        )
672        .await?;
673        Ok(authority.prepare_registration_request(approval).await?)
674    }
675
676    pub(crate) fn record_same_principal_registration_request(
677        &self,
678        approval: coven_replication::sync::DeviceProviderAdmissionApproval,
679    ) -> Result<coven_replication::sync::DeviceRegistrationRequest, BootstrapError> {
680        let offer = approval.request.offer.as_ref().clone();
681        self.require_offer(&offer)?;
682        let pending = self.open_pending_journal()?;
683        Ok(coven_replication::sync::store::PendingDeviceJoinAuthority::record_same_principal_registration_request(
684            &pending,
685            &offer,
686            approval,
687        )?)
688    }
689
690    /// Report the breakdown whichever way the bootstrap ends. A join that took
691    /// four minutes is exactly the one whose stage timings are wanted, and a
692    /// join that failed partway still reached the stages it reached.
693    pub(crate) async fn bootstrap_pending_device(
694        &self,
695        bootstrap: coven_replication::sync::ProviderReadyDeviceBootstrap,
696        on_progress: &coven_replication::sync::JoiningDeviceJoinProgressObserver,
697        cancel: &watch::Receiver<bool>,
698    ) -> Result<
699        coven_protocol::store_commit::device_join_exchange::DeviceJoinReadiness,
700        BootstrapError,
701    > {
702        // Opened before the run so the run counts through it. Opening one is
703        // local, so nothing measurable happens ahead of the first stage.
704        let cloud = self.build_cloud_home().await?;
705        let mut timings =
706            StageTimings::counting("Device join bootstrap", cloud.provider_requests());
707        let outcome = Box::pin(self.bootstrap_pending_device_staged(
708            cloud,
709            bootstrap,
710            on_progress,
711            cancel,
712            &mut timings,
713        ))
714        .await;
715        timings.report();
716        outcome
717    }
718
719    async fn bootstrap_pending_device_staged(
720        &self,
721        cloud: Arc<dyn ExactCloudHome>,
722        bootstrap: coven_replication::sync::ProviderReadyDeviceBootstrap,
723        on_progress: &coven_replication::sync::JoiningDeviceJoinProgressObserver,
724        cancel: &watch::Receiver<bool>,
725        timings: &mut StageTimings,
726    ) -> Result<
727        coven_protocol::store_commit::device_join_exchange::DeviceJoinReadiness,
728        BootstrapError,
729    > {
730        let offer = &bootstrap.bootstrap.request.approval().request.offer;
731        self.require_offer(offer)?;
732        let attempt_id = bootstrap.bootstrap.publication_authorization.attempt_id;
733        let pending = self.open_pending_journal()?;
734        if *cancel.borrow() {
735            return Err(BootstrapError::Cancelled);
736        }
737        let signer = coven_keys::keys::peek_pending_identity(&offer.member_pubkey)?;
738        let join = timings
739            .stage("open Store storage", self.build_storage(cloud, &signer))
740            .await?;
741        let store_dir = self.layout.store_dir(&self.admission.store_id);
742        if let Some(readiness) = pending.completed_joiner_readiness(attempt_id)? {
743            if store_dir.db_path().exists() {
744                return Ok(readiness);
745            }
746        }
747        if store_dir.config_path().exists() {
748            return Err(BootstrapError::StoreExists(self.admission.store_id.clone()));
749        }
750        store_dir.ensure_created()?;
751        let db_path = store_dir.db_path();
752        let history_verifier = timings
753            .stage(
754                "read Store root",
755                coven_replication::sync::store::HistoryConstructionAuthority::for_snapshot()
756                    .open_pinned(join.storage.as_ref(), &offer.store_root),
757            )
758            .await
759            .map_err(SnapshotError::from)?;
760        // Selecting the snapshot resolves membership at the admission floor on
761        // this verifier, which is a second one and holds nothing the first
762        // walked. The rollup is what keeps that from being a second walk.
763        timings
764            .stage(
765                "read the membership rollup",
766                history_verifier.adopt_published_membership_rollup(),
767            )
768            .await;
769        let snapshot = timings
770            .stage(
771                "download snapshot",
772                PreparedSnapshotBootstrap::prepare_device_join(
773                    &join.storage,
774                    history_verifier,
775                    &self.admission.membership_floor,
776                    supported_version(&self.migrations),
777                    &db_path,
778                    &signer,
779                    std::sync::Arc::clone(on_progress),
780                    cancel,
781                    &bootstrap
782                        .bootstrap
783                        .publication_authorization
784                        .attempt_activation,
785                ),
786            )
787            .await?;
788        on_progress(coven_replication::sync::JoiningDeviceJoinProgress::InstallingSnapshot);
789        let routing_encryption = EncryptionService::from(join.keyring.clone());
790        let device_id = bootstrap
791            .bootstrap
792            .request
793            .expected_registration()
794            .device_id
795            .to_string();
796        let opened = timings
797            .stage(
798                "install snapshot",
799                snapshot.install(
800                    &store_dir,
801                    self.synced_tables.clone(),
802                    coven_protocol::blob::BLOB_TOMBSTONE_GRACE,
803                    self.transfer_limits,
804                    device_id,
805                    self.clock.clone(),
806                    &self.migrations,
807                    self.coven_migration_policy,
808                    Some(&routing_encryption),
809                ),
810            )
811            .await?;
812        let published_at = self.clock.now().to_rfc3339();
813        let mut joining = timings
814            .stage(
815                "load membership",
816                opened.begin_device_join(&pending, offer.as_ref().clone()),
817            )
818            .await?;
819        Ok(timings
820            .stage(
821                "install history",
822                joining.bootstrap(bootstrap, &published_at, Some(&routing_encryption)),
823            )
824            .await?)
825    }
826
827    pub(crate) async fn complete_device_join(
828        &self,
829        activation: coven_replication::sync::DeviceJoinActivation,
830        on_progress: &coven_replication::sync::JoiningDeviceJoinProgressObserver,
831    ) -> Result<Config, BootstrapError> {
832        // Opened before the run so the run counts through it. Opening one is
833        // local, so nothing measurable happens ahead of the first stage.
834        let cloud = self.build_cloud_home().await?;
835        let mut timings =
836            StageTimings::counting("Device join completion", cloud.provider_requests());
837        let outcome = Box::pin(self.complete_device_join_staged(
838            cloud,
839            activation,
840            on_progress,
841            &mut timings,
842        ))
843        .await;
844        timings.report();
845        outcome
846    }
847
848    async fn complete_device_join_staged(
849        &self,
850        cloud: Arc<dyn ExactCloudHome>,
851        activation: coven_replication::sync::DeviceJoinActivation,
852        on_progress: &coven_replication::sync::JoiningDeviceJoinProgressObserver,
853        timings: &mut StageTimings,
854    ) -> Result<Config, BootstrapError> {
855        let attempt_id = activation.attempt_id;
856        let pending = self.open_pending_journal()?;
857        let store_dir = self.layout.store_dir(&self.admission.store_id);
858        let completed_config = if store_dir.config_path().exists() {
859            Some(Config::load_from_config_yaml(&store_dir)?)
860        } else {
861            None
862        };
863        if completed_config
864            .as_ref()
865            .is_some_and(|config| config.store_id != self.admission.store_id)
866        {
867            return Err(coven_replication::sync::DeviceJoinError::JournalConflict.into());
868        }
869        let signer = match completed_config.as_ref() {
870            Some(_) => coven_keys::keys::require_identity(self.identity_custody.as_ref())?,
871            None => coven_keys::keys::peek_pending_identity(&self.member_pubkey)?,
872        };
873        let join = timings
874            .stage("open Store storage", self.build_storage(cloud, &signer))
875            .await?;
876        let pending_readiness = pending.observe_joiner_activation_if_pending(&activation)?;
877        let device_id = match (pending_readiness.as_ref(), completed_config.as_ref()) {
878            (Some(readiness), _) => readiness.proof.registration.device_id.to_string(),
879            (None, Some(config)) => config.device_id.clone(),
880            (None, None) => {
881                return Err(coven_replication::sync::DeviceJoinError::JournalConflict.into());
882            }
883        };
884        let db_path = store_dir.db_path();
885        let db = Database::open(
886            &db_path,
887            self.synced_tables.clone(),
888            coven_protocol::blob::BLOB_TOMBSTONE_GRACE,
889            self.transfer_limits,
890            device_id.clone(),
891            self.clock.clone(),
892            self.coven_migration_policy,
893            &self.migrations,
894        )?;
895        let database = coven_database::StoreDatabase::from_database(db.clone());
896        let routing_encryption = EncryptionService::from(join.keyring.clone());
897        let observation = timings
898            .stage(
899                "read Store root",
900                coven_replication::sync::store::PendingDeviceJoinObservation::open(
901                    &pending,
902                    &join.storage,
903                    &self.admission.store_root,
904                    attempt_id,
905                ),
906            )
907            .await?;
908        let mut joining = timings
909            .stage(
910                "load membership",
911                observation.into_joining_store(
912                    database,
913                    &store_dir,
914                    signer.clone(),
915                    Some(join.membership.chain().clone()),
916                    Some(routing_encryption.clone()),
917                ),
918            )
919            .await?;
920        on_progress(coven_replication::sync::JoiningDeviceJoinProgress::CatchingUp);
921        timings
922            .stage(
923                "pull history",
924                joining.pull_store_history(Some(&routing_encryption)),
925            )
926            .await?;
927        let joined = timings
928            .stage(
929                "materialize activation",
930                joining.materialize(activation.clone()),
931            )
932            .await?;
933        if pending_readiness
934            .as_ref()
935            .is_some_and(|readiness| joined.registration != readiness.proof.registration)
936            || joined.registration.device_id.to_string() != device_id
937        {
938            return Err(coven_replication::sync::DeviceJoinError::JournalConflict.into());
939        }
940        on_progress(coven_replication::sync::JoiningDeviceJoinProgress::SavingLibrary);
941        self.custody.persist(&join.keyring)?;
942        self.identity_custody.establish(&signer)?;
943        if let Some(credentials) = derive_credentials(&self.admission.join_info) {
944            self.store_keys.set_cloud_home_credentials(&credentials)?;
945        }
946        let cipher = CloudCipher::Encrypted(join.keyring.clone().into());
947        let mut config = super::build_config(
948            &self.admission.store_id,
949            &device_id,
950            &self.admission.store_name,
951            &self.admission.join_info,
952            &cipher,
953        );
954        config.cloud_home.exact_upload_verification = self.exact_upload_verification;
955        config.save_to_config_yaml(&store_dir)?;
956        timings
957            .stage("close join journal", joining.complete(activation))
958            .await?;
959        coven_keys::keys::discard_pending_identity(&self.member_pubkey)?;
960        info!(store_id = %self.admission.store_id, "joined Store device");
961        Ok(config)
962    }
963
964    pub(crate) async fn install_same_principal_device_join(
965        &self,
966        join: coven_replication::sync::SamePrincipalDeviceJoin,
967        on_progress: &coven_replication::sync::JoiningDeviceJoinProgressObserver,
968        cancel: &watch::Receiver<bool>,
969    ) -> Result<Config, BootstrapError> {
970        // Opened before the run so the run counts through it. Opening one is
971        // local, so nothing measurable happens ahead of the first stage.
972        let cloud = self.build_cloud_home().await?;
973        let mut timings =
974            StageTimings::counting("Same-provider device join", cloud.provider_requests());
975        let outcome = Box::pin(self.install_same_principal_device_join_staged(
976            cloud,
977            join,
978            on_progress,
979            cancel,
980            &mut timings,
981        ))
982        .await;
983        timings.report();
984        outcome
985    }
986
987    async fn install_same_principal_device_join_staged(
988        &self,
989        cloud: Arc<dyn ExactCloudHome>,
990        join: coven_replication::sync::SamePrincipalDeviceJoin,
991        on_progress: &coven_replication::sync::JoiningDeviceJoinProgressObserver,
992        cancel: &watch::Receiver<bool>,
993        timings: &mut StageTimings,
994    ) -> Result<Config, BootstrapError> {
995        join.verify_shape()
996            .map_err(coven_replication::sync::DeviceJoinError::from)?;
997        let offer = &join.bootstrap.bootstrap.request.approval().request.offer;
998        self.require_offer(offer)?;
999        if *cancel.borrow() {
1000            return Err(BootstrapError::Cancelled);
1001        }
1002        let pending = self.open_pending_journal()?;
1003        let store_dir = self.layout.store_dir(&self.admission.store_id);
1004        if store_dir.config_path().exists() {
1005            return Err(BootstrapError::StoreExists(self.admission.store_id.clone()));
1006        }
1007        let signer = coven_keys::keys::peek_pending_identity(&offer.member_pubkey)?;
1008        let storage = timings
1009            .stage("open Store storage", self.build_storage(cloud, &signer))
1010            .await?;
1011        store_dir.ensure_created()?;
1012        let prepared = timings
1013            .stage(
1014                "download snapshot",
1015                PreparedDeviceJoinSnapshot::prepare(
1016                    &storage.storage,
1017                    (*join.installation).clone(),
1018                    &storage.membership,
1019                    supported_version(&self.migrations),
1020                    &store_dir.db_path(),
1021                    on_progress,
1022                    cancel,
1023                ),
1024            )
1025            .await?;
1026        if *cancel.borrow() {
1027            return Err(BootstrapError::Cancelled);
1028        }
1029        on_progress(coven_replication::sync::JoiningDeviceJoinProgress::InstallingSnapshot);
1030        let routing_encryption = EncryptionService::from(storage.keyring.clone());
1031        let device_id = join
1032            .bootstrap
1033            .bootstrap
1034            .request
1035            .expected_registration()
1036            .device_id
1037            .to_string();
1038        let installed = timings
1039            .stage("install snapshot", async {
1040                prepared.install(
1041                    self.synced_tables.clone(),
1042                    coven_protocol::blob::BLOB_TOMBSTONE_GRACE,
1043                    self.transfer_limits,
1044                    device_id.clone(),
1045                    self.clock.clone(),
1046                    &self.migrations,
1047                    self.coven_migration_policy,
1048                    &routing_encryption,
1049                )
1050            })
1051            .await?;
1052        let completion = timings.stage("install history", coven_replication::sync::store::PendingDeviceJoinAuthority::prepare_same_principal_completion(
1053            &pending,
1054            &storage.storage,
1055            &store_dir,
1056            &signer,
1057            join,
1058            installed,
1059            &self.clock.now().to_rfc3339(),
1060            Some(&routing_encryption),
1061            Some(storage.membership.chain().clone()),
1062        ))
1063        .await?;
1064        if completion.joined().registration.device_id.to_string() != device_id {
1065            return Err(coven_replication::sync::DeviceJoinError::JournalConflict.into());
1066        }
1067        on_progress(coven_replication::sync::JoiningDeviceJoinProgress::SavingLibrary);
1068        self.custody.persist(&storage.keyring)?;
1069        self.identity_custody.establish(&signer)?;
1070        if let Some(credentials) = derive_credentials(&self.admission.join_info) {
1071            self.store_keys.set_cloud_home_credentials(&credentials)?;
1072        }
1073        let cipher = CloudCipher::Encrypted(storage.keyring.clone().into());
1074        let mut config = super::build_config(
1075            &self.admission.store_id,
1076            &device_id,
1077            &self.admission.store_name,
1078            &self.admission.join_info,
1079            &cipher,
1080        );
1081        config.cloud_home.exact_upload_verification = self.exact_upload_verification;
1082        config.save_to_config_yaml(&store_dir)?;
1083        timings
1084            .stage("close join journal", completion.complete())
1085            .await?;
1086        coven_keys::keys::discard_pending_identity(&self.member_pubkey)?;
1087        info!(store_id = %self.admission.store_id, "joined Store device");
1088        Ok(config)
1089    }
1090
1091    fn require_offer(
1092        &self,
1093        offer: &coven_replication::sync::DeviceJoinOffer,
1094    ) -> Result<(), BootstrapError> {
1095        if offer.store_root != self.admission.store_root
1096            || offer.member_pubkey != self.member_pubkey
1097        {
1098            return Err(coven_replication::sync::DeviceJoinError::OfferMismatch.into());
1099        }
1100        Ok(())
1101    }
1102
1103    fn open_pending_journal(
1104        &self,
1105    ) -> Result<coven_replication::sync::DeviceJoinJournalDatabase, BootstrapError> {
1106        let directory = self.layout.stores_root().join(".pending-device-joins");
1107        Ok(coven_replication::sync::DeviceJoinJournalDatabase::open(
1108            directory.join(format!("{}.sqlite", self.admission.store_id)),
1109        )?)
1110    }
1111
1112    /// Open the home this join reads through, counting every operation asked
1113    /// of it.
1114    ///
1115    /// A join opens two storages over this one home — a plaintext one to pin
1116    /// the Store root and walk the membership chain, then an encrypted one for
1117    /// everything after — so counting here rather than per storage is what puts
1118    /// the whole join's operations in one running total. The home a test
1119    /// supplies is left alone and counts nothing.
1120    async fn build_cloud_home(&self) -> Result<Arc<dyn ExactCloudHome>, BootstrapError> {
1121        #[cfg(any(test, feature = "test-utils"))]
1122        if let Some(home) = &self.test_home {
1123            return Ok(home.clone());
1124        }
1125        let home = build_cloud_home_for_join(
1126            &self.admission.join_info,
1127            &self.store_keys,
1128            &self.cloud_homes,
1129            self.oauth_tokens.clone(),
1130            self.cloudkit_ops.clone(),
1131            self.clock.clone(),
1132            self.exact_upload_verification,
1133        )
1134        .await?;
1135        Ok(Arc::new(coven_storage::cloud::CountingCloudHome::new(home)))
1136    }
1137
1138    /// Plaintext storage over the joining device's cloud home.
1139    ///
1140    /// Both the pre-key bootstrap reads and the device-join transport go
1141    /// through this: the transport's objects carry their own per-attempt seal,
1142    /// so they need no store key — which is what lets a joiner publish its
1143    /// access request before it has unwrapped the store keyring at all.
1144    pub(super) async fn transport_storage(&self) -> Result<CloudSyncConnection, BootstrapError> {
1145        let signer = coven_keys::keys::peek_pending_identity(&self.member_pubkey)?;
1146        let cloud = self.build_cloud_home().await?;
1147        self.plaintext_storage(cloud, &signer)
1148    }
1149
1150    fn plaintext_storage(
1151        &self,
1152        home: Arc<dyn ExactCloudHome>,
1153        signer: &UserKeypair,
1154    ) -> Result<CloudSyncConnection, BootstrapError> {
1155        Ok(CloudSyncConnection::new(
1156            home,
1157            CloudCipher::Plaintext,
1158            BlobPathScheme::for_storage(HomeStorage::Opaque),
1159            self.admission.store_id.clone(),
1160            signer.clone(),
1161        ))
1162    }
1163
1164    /// Unwrap the store keyring over the home this join already opened.
1165    ///
1166    /// Timed as one step by its callers, where it has been the second-largest
1167    /// on a live join. Constructing the home is local — no bucket check, no
1168    /// auth probe — and its caller does it, so all of this time is the two
1169    /// reads that pin the Store root and its founder, the membership rollup,
1170    /// the membership chain walk, and the wrapped-key reads.
1171    ///
1172    /// The walk used to grow with the Store's whole membership history: a
1173    /// listing and a read per head, then a read per entry, back to the founding
1174    /// entry. It now takes everything up to the newest published snapshot's
1175    /// membership frontier from that snapshot's rollup, in one read, and walks
1176    /// the provider only for what was published after it — so what is left is
1177    /// the probe that finds each stream's end, and the tail itself.
1178    async fn build_storage(
1179        &self,
1180        cloud: Arc<dyn ExactCloudHome>,
1181        signer: &UserKeypair,
1182    ) -> Result<DeviceJoinStorage, BootstrapError> {
1183        let mut timings =
1184            StageTimings::counting("Device join Store storage", cloud.provider_requests());
1185        let outcome = Box::pin(self.build_storage_staged(cloud, signer, &mut timings)).await;
1186        timings.report();
1187        outcome
1188    }
1189
1190    async fn build_storage_staged(
1191        &self,
1192        cloud: Arc<dyn ExactCloudHome>,
1193        signer: &UserKeypair,
1194        timings: &mut StageTimings,
1195    ) -> Result<DeviceJoinStorage, BootstrapError> {
1196        let bootstrap_storage = self.plaintext_storage(cloud.clone(), signer)?;
1197        let recipient = hex::encode(signer.public_key());
1198        if self.admission.wrapped_key.recipient_pubkey != recipient {
1199            return Err(
1200                coven_replication::sync::store::MembershipMutationError::Crypto(
1201                    "admission wrapped-key ref names another recipient".to_string(),
1202                )
1203                .into(),
1204            );
1205        }
1206        self.admission
1207            .membership_floor
1208            .validate()
1209            .map_err(coven_replication::sync::store::MembershipMutationError::MembershipFloor)?;
1210        let history = timings
1211            .stage(
1212                "pin the Store root",
1213                coven_replication::sync::store::HistoryConstructionAuthority::admission()
1214                    .open_pinned(&bootstrap_storage, &self.admission.store_root),
1215            )
1216            .await
1217            .map_err(coven_replication::sync::store::MembershipMutationError::from)?;
1218        // What the walk below would otherwise fetch two round trips at a time,
1219        // in one read. Advisory: a Store with no published rollup, or one that
1220        // does not authenticate, leaves the walk exactly as it was.
1221        timings
1222            .stage(
1223                "read the membership rollup",
1224                history.adopt_published_membership_rollup(),
1225            )
1226            .await;
1227        let chain = timings
1228            .stage(
1229                "walk the membership chain",
1230                history.load_accepted_membership_authority(
1231                    &self.admission.membership_floor.0,
1232                    Some(&self.admission.owner_pubkey),
1233                ),
1234            )
1235            .await
1236            .map_err(coven_replication::sync::store::MembershipMutationError::from)?;
1237        let encryption = timings
1238            .stage(
1239                "open the keyring",
1240                coven_replication::sync::store::StoreKeyrings::new(
1241                    &bootstrap_storage,
1242                    self.admission.store_root.clone(),
1243                )
1244                .open_containing(
1245                    signer,
1246                    chain.chain(),
1247                    &self.admission.wrapped_key,
1248                ),
1249            )
1250            .await?;
1251        let keyring = MasterKeyring::from(encryption.clone());
1252        let storage = CloudSyncConnection::new(
1253            cloud,
1254            CloudCipher::Encrypted(encryption),
1255            BlobPathScheme::for_storage(HomeStorage::Opaque),
1256            self.admission.store_id.clone(),
1257            signer.clone(),
1258        );
1259        Ok(DeviceJoinStorage {
1260            storage: Arc::new(storage),
1261            keyring,
1262            membership: chain,
1263        })
1264    }
1265}
1266
1267/// The credentials to persist for this join, or `None` when the provider needs
1268/// no stored secret (OAuth tokens are already saved; CloudKit uses the container).
1269pub(crate) fn derive_credentials(join_info: &CloudHomeJoinInfo) -> Option<CloudHomeCredentials> {
1270    match join_info {
1271        CloudHomeJoinInfo::S3 {
1272            access_key,
1273            secret_key,
1274            ..
1275        } => Some(CloudHomeCredentials::S3 {
1276            access_key: access_key.clone(),
1277            secret_key: secret_key.clone(),
1278        }),
1279        _ => None,
1280    }
1281}
1282
1283#[cfg(test)]
1284#[path = "client_tests.rs"]
1285mod tests;
coven_domain/joining/client.rs

coven_domain/joining/
client.rs
