coven_storage/providers/
connector.rs

1//! Provider construction from an explicit location and device credentials.
2
3use super::{
4    http_client::HttpClient, oauth_folders::OAuthFolders, DropboxStorage, OAuthSession,
5    OneDriveStorage, S3Storage, StorageFolder,
6};
7use crate::{
8    ProviderAuth, ProviderOps, S3Credentials, S3DetectionRun, S3DetectionRuns, StorageConfig,
9    StorageConnection, StorageCredentials, StorageError, StorageFailure,
10};
11use coven_foundation::{
12    clock::ClockRef,
13    id_source::{DeviceId, IdSourceRef, StoreId},
14};
15use std::sync::Arc;
16
17/// Constructs provider connections without committing settings or credentials.
18/// Tests replace the connection boundary with in-memory storage.
19#[async_trait::async_trait]
20pub trait StorageConnector: Send + Sync {
21    /// Read the signed-in account's display label before recording a member's location.
22    async fn account_label(
23        &self,
24        _auth: ProviderAuth,
25        _cooldown: Arc<dyn crate::Cooldown>,
26    ) -> Result<String, StorageError> {
27        Err(StorageFailure::ProviderUnavailable.into())
28    }
29
30    /// Read the signed-in OneDrive root.
31    async fn onedrive_root(
32        &self,
33        _auth: ProviderAuth,
34        _cooldown: Arc<dyn crate::Cooldown>,
35    ) -> Result<StorageFolder, StorageError> {
36        Err(StorageFailure::ProviderUnavailable.into())
37    }
38
39    /// Read all immediate OneDrive child folders.
40    async fn onedrive_folders(
41        &self,
42        _parent: &StorageConfig,
43        _auth: ProviderAuth,
44        _cooldown: Arc<dyn crate::Cooldown>,
45    ) -> Result<Vec<StorageFolder>, StorageError> {
46        Err(StorageFailure::ProviderUnavailable.into())
47    }
48
49    /// Create a OneDrive folder without renaming or replacing an occupied name.
50    async fn create_onedrive_folder(
51        &self,
52        _parent: &StorageConfig,
53        _name: &str,
54        _auth: ProviderAuth,
55        _cooldown: Arc<dyn crate::Cooldown>,
56    ) -> Result<StorageFolder, StorageError> {
57        Err(StorageFailure::ProviderUnavailable.into())
58    }
59
60    /// Read all shared Dropbox namespaces.
61    async fn dropbox_shared_folders(
62        &self,
63        _auth: ProviderAuth,
64        _cooldown: Arc<dyn crate::Cooldown>,
65    ) -> Result<Vec<StorageFolder>, StorageError> {
66        Err(StorageFailure::ProviderUnavailable.into())
67    }
68
69    /// Make a folder named `name` at the top of the signed-in Dropbox and
70    /// share it, refusing a name already taken.
71    async fn create_dropbox_shared_folder(
72        &self,
73        _name: &str,
74        _auth: ProviderAuth,
75        _cooldown: Arc<dyn crate::Cooldown>,
76    ) -> Result<StorageFolder, StorageError> {
77        Err(StorageFailure::ProviderUnavailable.into())
78    }
79
80    /// Create the owner's CloudKit location from the entitled container.
81    /// A connector without native CloudKit support refuses this capability.
82    async fn prepare_cloudkit(
83        &self,
84        _container: &str,
85        _store: StoreId,
86    ) -> Result<StorageConfig, StorageError> {
87        Err(StorageFailure::ProviderUnavailable.into())
88    }
89    /// Validate settings and wrap the selected native provider once. The caller
90    /// verifies the connection before publication. Every request waits for
91    /// the store's shared `cooldown`; a rejected OAuth token is refreshed in
92    /// the sign-in's custody (§20.2). A setup call makes its own
93    /// connection beside it (`StorageConnection::during_setup`) for its
94    /// requests to draw on its waits (§20.8).
95    /// The connection writes for `device`: what it stages sits in that
96    /// device's reserved folder (D10).
97    async fn connect(
98        &self,
99        config: StorageConfig,
100        auth: ProviderAuth,
101        device: DeviceId,
102        publications: Arc<dyn crate::PublicationGate>,
103        cooldown: Arc<dyn crate::Cooldown>,
104        transfers: Arc<crate::TransferBudget>,
105    ) -> Result<Arc<StorageConnection>, StorageError>;
106    /// Probe an S3 location's service (§20.7), writing only fresh objects under
107    /// the store's `.coven-check/<run>/` prefix for `runs.run` and deleting
108    /// them, also on failure, then removing what each of `runs.earlier`
109    /// left there. The caller recorded `runs.run` before calling; the answer
110    /// names the runs whose probes and uploads are confirmed gone.
111    /// A connector without native S3 refuses this capability.
112    /// A rate-limited probe is sent again while setup's `waits` allow (§20.8).
113    async fn detect_s3(
114        &self,
115        _config: StorageConfig,
116        _credentials: S3Credentials,
117        _store: StoreId,
118        runs: &S3DetectionRuns,
119        _waits: &crate::SetupWaits,
120    ) -> S3DetectionRun {
121        S3DetectionRun::unwritten(runs, Err(StorageFailure::ProviderUnavailable.into()))
122    }
123    /// Before a store's first setup at a location it needs to itself (a
124    /// Dropbox shared folder, a OneDrive folder, or an S3 bucket), refuse a
125    /// location whose root holds anything outside the store's folder or the
126    /// `coven-location` marker (§20.3).
127    /// Reads only. A connector without that provider refuses this capability.
128    /// A rate-limited listing is sent again while setup's `waits` allow.
129    async fn check_new_store(
130        &self,
131        _config: StorageConfig,
132        _auth: ProviderAuth,
133        _store: StoreId,
134        _waits: &crate::SetupWaits,
135    ) -> Result<(), StorageError> {
136        Err(StorageFailure::ProviderUnavailable.into())
137    }
138}
139
140/// The provider composition root retains the supplied clock and id source,
141/// and the one HTTP client every connection it makes sends through.
142pub struct ProviderConnector {
143    clock: ClockRef,
144    ids: IdSourceRef,
145    /// Building it fails only when TLS cannot be set up; each call needing it
146    /// then returns that failure.
147    http: Result<HttpClient, Arc<super::http_client::TlsSetupError>>,
148    #[cfg(any(test, feature = "test-utils"))]
149    folder_api: Option<String>,
150}
151impl ProviderConnector {
152    /// Retain the capabilities used to construct subsequent connections.
153    pub fn new(clock: ClockRef, ids: IdSourceRef) -> Self {
154        Self {
155            http: HttpClient::new(clock.clone()).map_err(Arc::new),
156            clock,
157            ids,
158            #[cfg(any(test, feature = "test-utils"))]
159            folder_api: None,
160        }
161    }
162    /// Send account and folder calls to an in-process provider API in application tests.
163    #[cfg(any(test, feature = "test-utils"))]
164    pub fn with_folder_api(mut self, api: String) -> Self {
165        self.folder_api = Some(api);
166        self
167    }
168    /// Delete every object and abort every unfinished upload in an S3
169    /// location's bucket, once its root shows it holds only stores, with
170    /// this connector's clock, ids and HTTP client (api.md E14). A
171    /// development tool outside the sync protocol.
172    #[cfg(feature = "dev-tools")]
173    pub async fn empty_s3_bucket(
174        &self,
175        config: StorageConfig,
176        credentials: S3Credentials,
177    ) -> Result<super::EmptiedS3Bucket, super::EmptyS3BucketError> {
178        let http = self
179            .http
180            .clone()
181            .map_err(|source| super::http_client::tls_failure(crate::CloudProvider::S3, source))
182            .map_err(super::EmptyS3BucketError::Storage)?;
183        super::empty_s3_bucket(config, credentials, self.clock.clone(), self.ids.clone(), http)
184            .await
185    }
186    async fn with_folders<T, F, Fut>(
187        &self,
188        auth: ProviderAuth,
189        cooldown: Arc<dyn crate::Cooldown>,
190        request: F,
191    ) -> Result<T, StorageError>
192    where
193        F: FnOnce(OAuthFolders) -> Fut,
194        Fut: std::future::Future<Output = Result<T, StorageError>>,
195    {
196        let ProviderAuth::OAuth { tokens, custody } = auth else {
197            return Err(StorageFailure::InvalidConfiguration
198                .with_source("account and folder requests require OAuth sign-in"));
199        };
200        let access = tokens.access;
201        let api = super::oauth_api(access.provider()).expect("an OAuth provider").to_owned();
202        #[cfg(any(test, feature = "test-utils"))]
203        let api = self.folder_api.clone().unwrap_or(api);
204        let session = OAuthSession::new(
205            access.provider(),
206            tokens,
207            custody,
208            self.clock.clone(),
209            self.http
210                .clone()
211                .map_err(|source| super::http_client::tls_failure(access.provider(), source))?,
212        )?;
213        request(OAuthFolders::new(session, access, api, cooldown)).await
214    }
215}
216
217#[async_trait::async_trait]
218impl StorageConnector for ProviderConnector {
219    async fn account_label(
220        &self,
221        auth: ProviderAuth,
222        cooldown: Arc<dyn crate::Cooldown>,
223    ) -> Result<String, StorageError> {
224        self.with_folders(auth, cooldown, |folders| async move { folders.account_label().await })
225            .await
226    }
227
228    /// Read the signed-in OneDrive root.
229    async fn onedrive_root(
230        &self,
231        auth: ProviderAuth,
232        cooldown: Arc<dyn crate::Cooldown>,
233    ) -> Result<StorageFolder, StorageError> {
234        self.with_folders(auth, cooldown, |folders| async move { folders.onedrive_root().await })
235            .await
236    }
237
238    /// Read all immediate OneDrive child folders.
239    async fn onedrive_folders(
240        &self,
241        parent: &StorageConfig,
242        auth: ProviderAuth,
243        cooldown: Arc<dyn crate::Cooldown>,
244    ) -> Result<Vec<StorageFolder>, StorageError> {
245        self.with_folders(auth, cooldown, |folders| async move {
246            folders.onedrive_folders(parent).await
247        })
248        .await
249    }
250
251    /// Create a OneDrive folder without renaming or replacing an occupied name.
252    async fn create_onedrive_folder(
253        &self,
254        parent: &StorageConfig,
255        name: &str,
256        auth: ProviderAuth,
257        cooldown: Arc<dyn crate::Cooldown>,
258    ) -> Result<StorageFolder, StorageError> {
259        self.with_folders(auth, cooldown, |folders| async move {
260            folders.create_onedrive_folder(parent, name).await
261        })
262        .await
263    }
264
265    /// Read all shared Dropbox namespaces.
266    async fn dropbox_shared_folders(
267        &self,
268        auth: ProviderAuth,
269        cooldown: Arc<dyn crate::Cooldown>,
270    ) -> Result<Vec<StorageFolder>, StorageError> {
271        self.with_folders(auth, cooldown, |folders| async move {
272            folders.dropbox_shared_folders().await
273        })
274        .await
275    }
276
277    /// Make a folder named `name` at the top of the signed-in Dropbox and
278    /// share it, refusing a name already taken.
279    async fn create_dropbox_shared_folder(
280        &self,
281        name: &str,
282        auth: ProviderAuth,
283        cooldown: Arc<dyn crate::Cooldown>,
284    ) -> Result<StorageFolder, StorageError> {
285        self.with_folders(auth, cooldown, |folders| async move {
286            folders.create_dropbox_shared_folder(name).await
287        })
288        .await
289    }
290
291    async fn prepare_cloudkit(
292        &self,
293        container: &str,
294        store: StoreId,
295    ) -> Result<StorageConfig, StorageError> {
296        #[cfg(target_vendor = "apple")]
297        {
298            super::cloudkit::prepare(container, store).await
299        }
300        #[cfg(not(target_vendor = "apple"))]
301        {
302            let _ = (container, store);
303            Err(StorageFailure::ProviderUnavailable.into())
304        }
305    }
306    async fn connect(
307        &self,
308        config: StorageConfig,
309        auth: ProviderAuth,
310        device: DeviceId,
311        publications: Arc<dyn crate::PublicationGate>,
312        cooldown: Arc<dyn crate::Cooldown>,
313        transfers: Arc<crate::TransferBudget>,
314    ) -> Result<Arc<StorageConnection>, StorageError> {
315        crate::ConnectionCredentials { location: config.clone(), credentials: auth.credentials() }
316            .validate()?;
317        let provider = config.provider();
318        let http = self
319            .http
320            .clone()
321            .map_err(|source| super::http_client::tls_failure(provider, source))?;
322        let session = |tokens, custody| {
323            OAuthSession::new(provider, tokens, custody, self.clock.clone(), http.clone())
324        };
325        let provider: Arc<dyn ProviderOps> = match (&config, auth) {
326            (StorageConfig::S3 { .. }, ProviderAuth::S3(credentials)) => Arc::new(S3Storage::new(
327                config,
328                credentials,
329                self.clock.clone(),
330                self.ids.clone(),
331                cooldown.clone(),
332                http.clone(),
333            )?),
334            (StorageConfig::Dropbox { .. }, ProviderAuth::OAuth { tokens, custody }) => {
335                let session = session(tokens, custody)?;
336                Arc::new(DropboxStorage::new(config, session)?)
337            }
338            (StorageConfig::OneDrive { .. }, ProviderAuth::OAuth { tokens, custody }) => {
339                let session = session(tokens, custody)?;
340                Arc::new(OneDriveStorage::new(config, session, self.ids.clone())?)
341            }
342            (StorageConfig::CloudKit { .. }, ProviderAuth::CloudKit) => {
343                #[cfg(target_vendor = "apple")]
344                {
345                    Arc::new(super::CloudKitStorage::new(config, self.ids.clone()).await?)
346                }
347                #[cfg(not(target_vendor = "apple"))]
348                {
349                    return Err(StorageFailure::ProviderUnavailable.into());
350                }
351            }
352            _ => {
353                return Err(StorageFailure::InvalidConfiguration
354                    .with_source("credentials do not match the provider"))
355            }
356        };
357        let connection = StorageConnection::with_gate(
358            provider,
359            device,
360            publications,
361            cooldown,
362            transfers,
363            self.clock.clone(),
364        );
365        Ok(Arc::new(connection))
366    }
367    async fn detect_s3(
368        &self,
369        config: StorageConfig,
370        credentials: S3Credentials,
371        store: StoreId,
372        runs: &S3DetectionRuns,
373        waits: &crate::SetupWaits,
374    ) -> S3DetectionRun {
375        let valid = crate::ConnectionCredentials {
376            location: config.clone(),
377            credentials: StorageCredentials::S3(credentials.clone()),
378        }
379        .validate();
380        let http = self
381            .http
382            .clone()
383            .map_err(|source| super::http_client::tls_failure(crate::CloudProvider::S3, source));
384        let http = match valid.and(http) {
385            Ok(http) => http,
386            Err(error) => return S3DetectionRun::unwritten(runs, Err(error)),
387        };
388        super::detect_s3(
389            config,
390            credentials,
391            store,
392            runs,
393            self.clock.clone(),
394            self.ids.clone(),
395            http,
396            waits,
397        )
398        .await
399    }
400    async fn check_new_store(
401        &self,
402        config: StorageConfig,
403        auth: ProviderAuth,
404        store: StoreId,
405        waits: &crate::SetupWaits,
406    ) -> Result<(), StorageError> {
407        crate::ConnectionCredentials { location: config.clone(), credentials: auth.credentials() }
408            .validate()?;
409        if !config.holds_one_store() {
410            return Err(StorageFailure::InvalidConfiguration
411                .with_source("the location holds several stores, so nothing checks it"));
412        }
413        let provider = config.provider();
414        let http = self
415            .http
416            .clone()
417            .map_err(|source| super::http_client::tls_failure(provider, source))?;
418        let session = |tokens, custody| {
419            OAuthSession::new(provider, tokens, custody, self.clock.clone(), http.clone())
420        };
421        // Each page is one listing request: a rate-limited one stored nothing.
422        let clock = self.clock.as_ref();
423        let holds_only = match (&config, auth) {
424            (StorageConfig::S3 { .. }, ProviderAuth::S3(credentials)) => {
425                return super::check_new_s3_store(
426                    config,
427                    credentials,
428                    store,
429                    self.clock.clone(),
430                    self.ids.clone(),
431                    http,
432                    waits,
433                )
434                .await
435            }
436            (StorageConfig::Dropbox { .. }, ProviderAuth::OAuth { tokens, custody }) => {
437                let dropbox = DropboxStorage::new(config, session(tokens, custody)?)?;
438                waits.resend(clock, || dropbox.holds_only(store)).await?
439            }
440            (StorageConfig::OneDrive { .. }, ProviderAuth::OAuth { tokens, custody }) => {
441                let onedrive =
442                    OneDriveStorage::new(config, session(tokens, custody)?, self.ids.clone())?;
443                waits.resend(clock, || onedrive.holds_only(store)).await?
444            }
445            _ => {
446                return Err(StorageFailure::InvalidConfiguration
447                    .with_source("credentials do not match the provider"))
448            }
449        };
450        if !holds_only {
451            return Err(StorageFailure::OwnLocationRequired.into());
452        }
453        Ok(())
454    }
455}
456
457#[cfg(all(test, not(target_vendor = "apple")))]
458#[path = "connector_tests.rs"]
459mod tests;
coven_storage/providers/connector.rs

coven_storage/providers/
connector.rs
