coven_replication/sync/store/
authorization.rs

1use super::*;
2use coven_database::BlockedWriteDiscard;
3use coven_protocol::store_commit::StoreRootRef;
4use std::sync::Arc;
5
6mod authorized_store;
7mod candidate_cleanup;
8pub(crate) mod history;
9mod history_construction;
10pub(crate) mod keyring;
11pub(crate) use keyring::load_wrapped_store_key;
12#[cfg(test)]
13mod recovery_blob_tests;
14#[cfg(test)]
15mod recovery_publication_tests;
16mod registration;
17pub(crate) mod registration_outbox;
18#[cfg(test)]
19mod registration_recovery_tests;
20
21mod store_test_support;
22
23use crate::sync::store::device_join::transport;
24pub(crate) use authorized_store::AuthorizedStore;
25pub(crate) use candidate_cleanup::{
26    delete_candidate_cleanup_targets, retire_store_write_candidates,
27};
28use history::AuthorizedStoreHistory;
29pub use history_construction::HistoryConstructionAuthority;
30pub use keyring::StoreKeyrings;
31pub use registration::StoreRegistrationError;
32use registration_outbox::RegistrationOutbox;
33
34#[doc(hidden)]
35pub struct Store {
36    database: StoreDatabase,
37    storage: Arc<dyn CloudSyncObjectStorage>,
38    store_dir: StoreDir,
39    blob_cache: crate::sync::store::blob::StoreBlobCache,
40    identity: UserKeypair,
41    device_id: Option<String>,
42    routing_encryption: Option<coven_keys::encryption::EncryptionService>,
43    root: crate::sync::store::protocol_root::VerifiedStoreRoot,
44}
45
46impl Store {
47    /// The provider-operation counter of the home this Store works through, so
48    /// a run over it can report each stage's count beside its wall time.
49    pub fn provider_requests(
50        &self,
51    ) -> Option<Arc<dyn coven_foundation::stage_timing::ProviderRequests>> {
52        self.storage.provider_requests()
53    }
54}
55
56#[doc(hidden)]
57pub struct StoreRestoreMembership {
58    pub store_root: StoreRootRef,
59    pub founder_pubkey: String,
60    pub membership_floor: coven_protocol::membership::MembershipFloor,
61}
62
63pub(crate) struct InitializedStore {
64    store: Store,
65    device_id: String,
66}
67
68impl InitializedStore {
69    pub(crate) fn new(store: Store, device_id: String) -> Self {
70        Self { store, device_id }
71    }
72
73    pub(crate) fn into_parts(self) -> (Store, String) {
74        (self.store, self.device_id)
75    }
76}
77
78#[derive(Debug, thiserror::Error)]
79pub enum StoreInitializationError {
80    #[error("Store protocol root failed: {0}")]
81    ProtocolRoot(#[from] crate::sync::store::protocol_root::StoreProtocolRootError),
82    #[error("Store history verification failed: {0}")]
83    History(#[from] crate::sync::store::pull::StorePullError),
84    #[error("Store initialization database state failed: {0}")]
85    Database(#[from] coven_database::DbError),
86    #[error("membership chain bootstrap/anchor failed: {0}")]
87    MembershipAnchor(#[from] crate::sync::store::membership::AnchoredChainError),
88    #[error("Store founder device installation failed: {0}")]
89    Registration(#[from] crate::sync::store::authorization::registration::StoreRegistrationError),
90    #[error("opening a Store for a non-founder requires an installed local device")]
91    NonFounderDeviceMissing,
92    #[error("initialized Store has no local device registration id")]
93    LocalDeviceMissing,
94    #[error("Store founder state is invalid: {0}")]
95    FounderState(String),
96}
97
98impl Store {
99    pub(crate) fn device_join_transport(&self) -> transport::StoreDeviceJoinTransport<'_> {
100        transport::StoreDeviceJoinTransport::new(self)
101    }
102
103    pub(crate) async fn allocate_device_join_transport_bundle(
104        &self,
105        offer: coven_protocol::store_commit::device_join_exchange::DeviceJoinOffer,
106    ) -> Result<transport::DeviceJoinOfferBundle, transport::DeviceJoinTransportError> {
107        let attempt_namespace = transport::attempt_namespace(offer.attempt_id);
108        let context = transport::slot_context(offer.store_root.store_root_hash);
109        let mut slots = std::collections::BTreeMap::new();
110        let allocations = futures_util::future::join_all(
111            transport::DeviceJoinTransportKind::ALL
112                .into_iter()
113                .map(|kind| {
114                    let context = &context;
115                    let attempt_namespace = &attempt_namespace;
116                    async move {
117                        self.storage
118                            .allocate_protocol_slot(
119                                context,
120                                &transport::semantic_prefix(attempt_namespace, kind),
121                                ".json",
122                            )
123                            .await
124                            .map(|slot| (kind, slot))
125                    }
126                }),
127        )
128        .await;
129        for allocation in allocations {
130            let (kind, slot) = allocation?;
131            slots.insert(kind, slot);
132        }
133        Ok(transport::DeviceJoinOfferBundle {
134            version: coven_protocol::store_commit::STORE_PROTOCOL_VERSION,
135            offer,
136            transport: transport::DeviceJoinTransportParams::new(
137                attempt_namespace,
138                slots,
139                coven_keys::encryption::MasterKeyring::generate(),
140            ),
141        })
142    }
143
144    pub(crate) async fn publish_device_join_transport_artifact(
145        &self,
146        bundle: &transport::DeviceJoinOfferBundle,
147        action: &crate::sync::store::DeviceJoinAction,
148    ) -> Result<(), transport::DeviceJoinTransportError> {
149        transport::DeviceJoinTransport::open(
150            self.storage.as_ref(),
151            bundle,
152            crate::sync::store::DeviceJoinRole::Owner,
153        )?
154        .publish(action)
155        .await
156    }
157
158    pub(crate) async fn await_device_join_transport_artifact<T: transport::DeviceJoinArtifact>(
159        &self,
160        bundle: &transport::DeviceJoinOfferBundle,
161        timing: transport::DeviceJoinTransportTiming,
162    ) -> Result<T, transport::DeviceJoinTransportError> {
163        transport::DeviceJoinTransport::open(
164            self.storage.as_ref(),
165            bundle,
166            crate::sync::store::DeviceJoinRole::Owner,
167        )?
168        .await_artifact::<T>(timing)
169        .await
170    }
171
172    pub(crate) async fn device_join_transport_status(
173        &self,
174        attempt_id: coven_protocol::store_commit::DeviceJoinAttemptId,
175        role: crate::sync::store::DeviceJoinRole,
176    ) -> Result<Option<crate::sync::store::DeviceJoinStatus>, transport::DeviceJoinTransportError>
177    {
178        Ok(self.database.device_join_status(attempt_id, role).await?)
179    }
180
181    /// Drop the admitting side's row for an attempt that has finished.
182    ///
183    /// The row is the resume anchor for the terminal step and nothing more, so
184    /// it goes once that step's artifact is at its slot. After that its absence
185    /// is what says the attempt is over: an abandonment asked for again has
186    /// nothing to abandon, and the driver has nothing left to deliver.
187    pub(crate) async fn retire_device_join_row(
188        &self,
189        attempt_id: coven_protocol::store_commit::DeviceJoinAttemptId,
190        role: crate::sync::store::DeviceJoinRole,
191    ) -> Result<(), transport::DeviceJoinTransportError> {
192        Ok(self.database.retire_device_join(attempt_id, role).await?)
193    }
194
195    /// Refuse to drive an attempt this device is not the admitting side of.
196    ///
197    /// One party admits, and the offer says which: the device whose activated
198    /// registration the offer names as its owner. That same registration is the
199    /// offer's provider administrator, so there is nothing else to weigh.
200    pub(crate) async fn require_device_join_admitter(
201        &self,
202        offer: &coven_protocol::store_commit::device_join_exchange::DeviceJoinOffer,
203    ) -> Result<(), transport::DeviceJoinTransportError> {
204        let local = self
205            .database
206            .local_activated_registration_ref()
207            .await
208            .map_err(crate::sync::store::DeviceJoinError::from)?
209            .ok_or(crate::sync::store::DeviceJoinError::ActiveDeviceRequired)?;
210        if local != offer.owner_registration {
211            return Err(crate::sync::store::DeviceJoinError::ActiveDeviceRequired.into());
212        }
213        Ok(())
214    }
215
216    pub(crate) fn circles(&self) -> StoreCircleCommands<'_> {
217        StoreCircleCommands::new(self)
218    }
219
220    fn local_author_pubkey(&self) -> String {
221        coven_keys::keys::public_key_hex(&self.identity)
222    }
223
224    #[doc(hidden)]
225    pub(crate) fn host_write_blob_staging(
226        &self,
227        runtime: tokio::runtime::Handle,
228    ) -> HostWriteBlobStaging {
229        HostWriteBlobStaging::new(
230            runtime,
231            Arc::clone(&self.storage),
232            self.root.reference().clone(),
233            self.store_dir.clone(),
234        )
235    }
236
237    pub(crate) async fn create(
238        database: StoreDatabase,
239        storage: Arc<dyn CloudSyncObjectStorage>,
240        store_dir: StoreDir,
241        founder_timestamp: &str,
242        identity: &UserKeypair,
243        routing_encryption: Option<coven_keys::encryption::EncryptionService>,
244    ) -> Result<InitializedStore, StoreInitializationError> {
245        let blob_cache =
246            crate::sync::store::blob::StoreBlobCache::new(database.clone(), store_dir.clone());
247        crate::sync::store::founder_creation::FounderStoreCreation::begin(
248            database,
249            storage,
250            &store_dir,
251            blob_cache,
252            founder_timestamp,
253            identity,
254            routing_encryption,
255        )
256        .await
257        .execute()
258        .await
259    }
260
261    pub(crate) async fn open(
262        database: StoreDatabase,
263        storage: Arc<dyn CloudSyncObjectStorage>,
264        store_dir: StoreDir,
265        expected_root: &StoreRootRef,
266        identity: &UserKeypair,
267        routing_encryption: Option<coven_keys::encryption::EncryptionService>,
268    ) -> Result<InitializedStore, StoreInitializationError> {
269        let root = crate::sync::store::protocol_root::VerifiedStoreRoot::open(
270            &database,
271            &*storage,
272            expected_root,
273        )
274        .await?;
275        let authority = HistoryConstructionAuthority::store();
276        let history_verifier = authority
277            .bind_verified(storage.as_ref(), root.clone())
278            .await?;
279        let blob_source = crate::sync::store::blob::RemoteBlobSource::authorized(
280            database.clone(),
281            storage.as_ref(),
282            root.reference().clone(),
283        );
284        let keyrings = keyring::StoreKeyrings::new(storage.as_ref(), root.reference().clone());
285        let blob_cache =
286            crate::sync::store::blob::StoreBlobCache::new(database.clone(), store_dir.clone());
287        AuthorizedStoreHistory::new(
288            database,
289            routing_encryption,
290            &storage,
291            &store_dir,
292            blob_cache,
293            history_verifier,
294            blob_source,
295            keyrings,
296        )
297        .finish_initialization(identity)
298        .await
299    }
300
301    #[doc(hidden)]
302    pub async fn load(
303        database: StoreDatabase,
304        storage: Arc<dyn CloudSyncObjectStorage>,
305        store_dir: StoreDir,
306        identity: UserKeypair,
307        routing_encryption: Option<coven_keys::encryption::EncryptionService>,
308    ) -> Result<Self, StoreError> {
309        let store_root =
310            database
311                .local_store_root_ref()
312                .await?
313                .ok_or(StoreError::MissingState {
314                    key: commit_plan::STORE_ROOT_AUTHORITY,
315                })?;
316        let root = crate::sync::store::protocol_root::VerifiedStoreRoot::open(
317            &database,
318            &*storage,
319            &store_root,
320        )
321        .await
322        .map_err(StoreError::from)?;
323        let device_id = database
324            .get_protocol_state(coven_database::LOCAL_DEVICE_ID_STATE_KEY)
325            .await?;
326        Ok(Self::new(
327            database,
328            storage,
329            store_dir,
330            identity,
331            device_id,
332            root,
333            routing_encryption,
334        ))
335    }
336
337    fn new(
338        database: StoreDatabase,
339        storage: Arc<dyn CloudSyncObjectStorage>,
340        store_dir: StoreDir,
341        identity: UserKeypair,
342        device_id: Option<String>,
343        root: crate::sync::store::protocol_root::VerifiedStoreRoot,
344        routing_encryption: Option<coven_keys::encryption::EncryptionService>,
345    ) -> Self {
346        let blob_cache =
347            crate::sync::store::blob::StoreBlobCache::new(database.clone(), store_dir.clone());
348        Self {
349            database,
350            storage,
351            store_dir,
352            blob_cache,
353            identity,
354            device_id,
355            root,
356            routing_encryption,
357        }
358    }
359    pub(crate) fn store_root(&self) -> &StoreRootRef {
360        self.root.reference()
361    }
362
363    pub(crate) fn blob_path_scheme(&self) -> BlobPathScheme {
364        self.storage.blob_path_scheme()
365    }
366
367    pub(crate) async fn circle_close_status(
368        &self,
369        circle_id: coven_protocol::circle::CircleId,
370    ) -> Result<coven_protocol::circle::CircleCloseStatus, CircleOperationError> {
371        let (current, _) = self
372            .database
373            .circle_closing_context(circle_id, &self.local_author_pubkey())
374            .await?;
375        let coven_protocol::circle::CircleControlState::EpochClose(close) =
376            current.control.value.state()
377        else {
378            return Err(CircleOperationError::InvalidState(
379                "Circle close-status inspection received an active control".to_string(),
380            ));
381        };
382        let context = coven_protocol::objects::ProtocolObjectContext::store_encrypted(
383            current.control.value.store_root_hash,
384            coven_protocol::objects::ProtocolObjectDomain::CircleEpochCloseResponse,
385        );
386        let mut participants = Vec::with_capacity(close.participants.len());
387        for participant in &close.participants {
388            let prefix = coven_protocol::circle::circle_epoch_close_response_semantic_prefix(
389                current.control.value.circle_id,
390                close.close_id,
391                participant.registration.device_id,
392            );
393            let settlement = match self
394                .storage
395                .read_protocol_slot(&context, &participant.response_slot, &prefix)
396                .await
397            {
398                Ok((bytes, _)) => {
399                    match coven_protocol::circle::CircleEpochCloseResponseSlotValue::parse(&bytes)?
400                    {
401                        coven_protocol::circle::CircleEpochCloseResponseSlotValue::Response(_) => {
402                            coven_protocol::circle::CircleCloseSettlement::Responded
403                        }
404                        coven_protocol::circle::CircleEpochCloseResponseSlotValue::Exclusion(_) => {
405                            coven_protocol::circle::CircleCloseSettlement::Excluded
406                        }
407                    }
408                }
409                Err(coven_protocol::objects::StorageError::NotFound(_)) => {
410                    coven_protocol::circle::CircleCloseSettlement::Pending
411                }
412                Err(error) => {
413                    return Err(coven_protocol::objects::StoreObjectError::from(error).into())
414                }
415            };
416            participants.push(coven_protocol::circle::CircleCloseParticipant {
417                device_id: participant.registration.device_id,
418                settlement,
419            });
420        }
421        Ok(coven_protocol::circle::CircleCloseStatus {
422            circle_id,
423            close_id: close.close_id,
424            participants,
425        })
426    }
427
428    #[doc(hidden)]
429    pub(crate) async fn discard_blocked_write(
430        &self,
431        write_id: coven_protocol::write::WriteId,
432        _routing_encryption: Option<&coven_keys::encryption::EncryptionService>,
433    ) -> Result<Vec<coven_protocol::write::WriteId>, crate::sync::store::StoreError> {
434        let _authorship = self.database.author_own_stream().await;
435        if let Some(active) = self.database.active_store_publication().await? {
436            if active.owner()
437                == &coven_database::ActiveStorePublicationOwner::StoreWrite(write_id.clone())
438                && (active.is_awaiting_preparation() || active.is_discarding())
439            {
440                let active = self
441                    .database
442                    .begin_retired_store_write_discard(active)
443                    .await?;
444                candidate_cleanup::retire_store_write_candidates(
445                    &self.database,
446                    self.storage.as_ref(),
447                    active,
448                )
449                .await?;
450            }
451        }
452        match self.database.discard_blocked_write(&write_id).await? {
453            BlockedWriteDiscard::Discarded(discarded) => Ok(discarded),
454            BlockedWriteDiscard::RemoteResolutionRequired => Err(StoreError::InvalidOutbound(
455                "Store publication outcome must be settled before the blocked write can be discarded"
456                    .to_string(),
457            )),
458        }
459    }
460
461    pub(crate) async fn propose_device_exclusion_for_device(
462        &self,
463        device_id: coven_protocol::store_commit::StoreDeviceId,
464    ) -> Result<
465        coven_protocol::store_commit::StoreDeviceExclusionProposalRef,
466        device_exclusion::StoreDeviceExclusionError,
467    > {
468        let mut writer = self.authorize_exclusion_writer().await?;
469        device_exclusion::propose_for_device(&self.database, &mut writer, device_id).await
470    }
471
472    pub(crate) async fn cancel_device_exclusion_proposal(
473        &self,
474        proposal: &coven_protocol::store_commit::StoreDeviceExclusionProposalRef,
475    ) -> Result<(), device_exclusion::StoreDeviceExclusionError> {
476        let mut writer = self.authorize_exclusion_writer().await?;
477        device_exclusion::cancel_proposal(&mut writer, proposal).await
478    }
479
480    pub(crate) async fn finalize_device_exclusion_proposal(
481        &self,
482        proposal: &coven_protocol::store_commit::StoreDeviceExclusionProposalRef,
483    ) -> Result<(), device_exclusion::StoreDeviceExclusionError> {
484        let mut writer = self.authorize_exclusion_writer().await?;
485        device_exclusion::finalize_proposal(&mut writer, proposal).await
486    }
487
488    #[cfg(any(test, feature = "test-utils"))]
489    pub(crate) async fn propose_device_exclusion(
490        &self,
491        target: &coven_protocol::store_commit::StoreDeviceRegistrationRef,
492    ) -> Result<
493        device_exclusion::StoreDeviceExclusionResult,
494        device_exclusion::StoreDeviceExclusionError,
495    > {
496        let mut writer = self.authorize_exclusion_writer().await?;
497        writer.device_exclusion().propose(target).await
498    }
499
500    #[cfg(any(test, feature = "test-utils"))]
501    pub(crate) async fn cancel_device_exclusion(
502        &self,
503        proposal: &coven_protocol::store_commit::StoreDeviceExclusionProposalRef,
504    ) -> Result<
505        device_exclusion::StoreDeviceExclusionResult,
506        device_exclusion::StoreDeviceExclusionError,
507    > {
508        let mut writer = self.authorize_exclusion_writer().await?;
509        writer.device_exclusion().cancel(proposal).await
510    }
511
512    #[cfg(any(test, feature = "test-utils"))]
513    pub(crate) async fn finalize_device_exclusion(
514        &self,
515        proposal: &coven_protocol::store_commit::StoreDeviceExclusionProposalRef,
516    ) -> Result<
517        device_exclusion::StoreDeviceExclusionResult,
518        device_exclusion::StoreDeviceExclusionError,
519    > {
520        let mut writer = self.authorize_exclusion_writer().await?;
521        writer.device_exclusion().exclude(proposal).await
522    }
523
524    #[cfg(any(test, feature = "test-utils"))]
525    pub(crate) async fn device_exclusion_operations_for_test(
526        &self,
527    ) -> Result<
528        Vec<device_exclusion::StoreDeviceExclusionOperationInfo>,
529        device_exclusion::StoreDeviceExclusionError,
530    > {
531        device_exclusion::operations_for_test(&self.database).await
532    }
533
534    #[cfg(any(test, feature = "test-utils"))]
535    pub(crate) async fn stage_uploaded_device_exclusion_proposal_for_test(
536        &self,
537    ) -> Result<
538        coven_protocol::store_commit::StoreDeviceExclusionProposalRef,
539        device_exclusion::StoreDeviceExclusionError,
540    > {
541        let mut writer = self.authorize_exclusion_writer().await?;
542        device_exclusion::stage_uploaded_proposal_for_test(&self.database, &mut writer).await
543    }
544
545    async fn authorize_exclusion_writer(
546        &self,
547    ) -> Result<AuthorizedWriterOperation<'_>, device_exclusion::StoreDeviceExclusionError> {
548        self.authorize_writer()
549            .await
550            .map_err(StoreError::from)
551            .map_err(device_exclusion::StoreDeviceExclusionError::from)
552    }
553
554    #[doc(hidden)]
555    pub async fn members(
556        &self,
557    ) -> Result<Vec<coven_protocol::membership::MemberInfo>, membership::MembershipOpsError> {
558        let authorization = self
559            .authorize()
560            .await
561            .map_err(StoreError::from)
562            .map_err(membership::MembershipOpsError::from)?;
563        Ok(authorization.members(Some(&self.identity.public_key())))
564    }
565
566    #[doc(hidden)]
567    pub async fn restore_membership(
568        &self,
569    ) -> Result<StoreRestoreMembership, membership::MembershipOpsError> {
570        let authorization = self
571            .authorize()
572            .await
573            .map_err(StoreError::from)
574            .map_err(membership::MembershipOpsError::from)?;
575        authorization.restore_membership()
576    }
577
578    async fn authorize_history(&self) -> Result<AuthorizedStoreHistory<'_>, SyncCycleFailure> {
579        let authority = HistoryConstructionAuthority::store();
580        let history_verifier = authority
581            .bind_verified(self.storage.as_ref(), self.root.clone())
582            .await
583            .map_err(|error| SyncCycleFailure::operation("open Store history authority", error))?;
584        let blob_source = crate::sync::store::blob::RemoteBlobSource::authorized(
585            self.database.clone(),
586            self.storage.as_ref(),
587            self.root.reference().clone(),
588        );
589        let keyrings =
590            keyring::StoreKeyrings::new(self.storage.as_ref(), self.root.reference().clone());
591        let mut history = AuthorizedStoreHistory::new(
592            self.database.clone(),
593            self.routing_encryption.clone(),
594            &self.storage,
595            &self.store_dir,
596            self.blob_cache.clone(),
597            history_verifier,
598            blob_source,
599            keyrings,
600        );
601        history.seed_retained_history().await.map_err(|error| {
602            SyncCycleFailure::operation("load installed Store history authority", error)
603        })?;
604        Ok(history)
605    }
606
607    pub(crate) async fn authorize(&self) -> Result<AuthorizedStore<'_>, SyncCycleFailure> {
608        self.authorize_history()
609            .await?
610            .authorize_store(&self.identity, self.device_id.as_deref())
611            .await
612    }
613
614    pub(crate) async fn authorize_writer(
615        &self,
616    ) -> Result<
617        AuthorizedWriterOperation<'_>,
618        crate::sync::store::commit_publication::StoreWriterAuthorizationError,
619    > {
620        RegistrationOutbox::new(self.database.clone(), &*self.storage)
621            .drain()
622            .await
623            .map_err(
624                crate::sync::store::commit_publication::StoreWriterAuthorizationError::Registration,
625            )?;
626        self.authorize()
627            .await
628            .map_err(crate::sync::store::commit_publication::StoreWriterAuthorizationError::StoreAuthority)?
629            .into_writer()
630            .await
631            .map_err(crate::sync::store::commit_publication::StoreWriterAuthorizationError::Registration)
632    }
633
634    #[doc(hidden)]
635    pub(crate) async fn begin_device_join(
636        &self,
637        member_pubkey: &str,
638    ) -> Result<
639        coven_protocol::store_commit::device_join_exchange::DeviceJoinOffer,
640        crate::sync::store::DeviceJoinError,
641    > {
642        let mut writer = self
643            .authorize_writer()
644            .await
645            .map_err(crate::sync::store::DeviceJoinError::from)?;
646        writer.join_operation().begin(member_pubkey).await
647    }
648
649    pub(crate) async fn begin_device_join_bundle(
650        &self,
651        member_pubkey: &str,
652    ) -> Result<
653        crate::sync::store::DeviceJoinOfferBundle,
654        crate::sync::store::DeviceJoinTransportError,
655    > {
656        let mut writer = self
657            .authorize_writer()
658            .await
659            .map_err(crate::sync::store::DeviceJoinError::from)?;
660        let offer = writer.join_operation().begin(member_pubkey).await?;
661        self.device_join_transport().allocate_bundle(offer).await
662    }
663
664    pub(crate) async fn begin_owner_promotion_for_device(
665        &self,
666        device_id: coven_protocol::store_commit::StoreDeviceId,
667    ) -> Result<
668        coven_protocol::store_commit::OwnerPromotionRequest,
669        owner_role_promotion::OwnerPromotionError,
670    > {
671        let registration = self
672            .database
673            .activated_store_device_registration_for_device(device_id)
674            .await?
675            .ok_or_else(|| {
676                owner_role_promotion::OwnerPromotionError::Protocol(
677                    "the target Store device is not active".to_string(),
678                )
679            })?;
680        self.begin_owner_promotion(registration.reference().clone())
681            .await
682    }
683
684    pub(crate) async fn begin_owner_promotion(
685        &self,
686        member_registration: coven_protocol::store_commit::StoreDeviceRegistrationRef,
687    ) -> Result<
688        coven_protocol::store_commit::OwnerPromotionRequest,
689        owner_role_promotion::OwnerPromotionError,
690    > {
691        let mut writer = self
692            .authorize_writer()
693            .await
694            .map_err(owner_role_promotion::OwnerPromotionError::from)?;
695        writer.owner_promotion().begin(member_registration).await
696    }
697
698    pub(crate) async fn accept_owner_promotion(
699        &self,
700        request: coven_protocol::store_commit::OwnerPromotionRequest,
701    ) -> Result<
702        coven_protocol::store_commit::OwnerPromotionAcceptance,
703        owner_role_promotion::OwnerPromotionError,
704    > {
705        let mut writer = self
706            .authorize_writer()
707            .await
708            .map_err(owner_role_promotion::OwnerPromotionError::from)?;
709        writer.owner_promotion().accept(request).await
710    }
711
712    pub(crate) async fn finalize_owner_promotion(
713        &self,
714        encryption: &coven_keys::encryption::EncryptionService,
715        acceptance: coven_protocol::store_commit::OwnerPromotionAcceptance,
716    ) -> Result<
717        coven_protocol::circle_control::StoreMembershipStateRef,
718        owner_role_promotion::OwnerPromotionError,
719    > {
720        let mut writer = self
721            .authorize_writer()
722            .await
723            .map_err(owner_role_promotion::OwnerPromotionError::from)?;
724        writer
725            .owner_promotion()
726            .finalize(encryption, acceptance)
727            .await
728    }
729
730    #[allow(clippy::too_many_arguments)]
731    pub(crate) async fn admit_member(
732        &self,
733        public_key_hex: &str,
734        member_email: Option<&str>,
735        role: coven_protocol::membership::MemberRole,
736        encryption: &coven_keys::encryption::EncryptionService,
737        store_id: &str,
738        store_name: &str,
739    ) -> Result<
740        crate::sync::store::membership::MemberAdmission,
741        crate::sync::store::membership::MembershipOpsError,
742    > {
743        let mut authorization = self
744            .authorize_writer()
745            .await
746            .map_err(StoreError::from)
747            .map_err(membership::MembershipOpsError::from)?;
748        authorization
749            .admit_member(
750                public_key_hex,
751                member_email,
752                role,
753                encryption,
754                store_id,
755                store_name,
756            )
757            .await
758    }
759
760    #[allow(clippy::too_many_arguments)]
761    pub(crate) async fn remove_member(
762        &self,
763        public_key_hex: &str,
764        encryption: &coven_keys::encryption::EncryptionService,
765        master_keys: &dyn coven_keys::keys::MasterKeyCustody,
766        cipher: &dyn coven_storage::CloudSyncCipherStateAccess,
767        pending_rotation: &dyn coven_storage::CloudSyncRotationStateAccess,
768    ) -> Result<String, crate::sync::store::membership::MembershipOpsError> {
769        let mut authorization = self
770            .authorize_writer()
771            .await
772            .map_err(StoreError::from)
773            .map_err(membership::MembershipOpsError::from)?;
774        authorization
775            .remove_member(
776                public_key_hex,
777                encryption,
778                master_keys,
779                cipher,
780                pending_rotation,
781            )
782            .await
783    }
784
785    #[cfg(any(test, feature = "test-utils"))]
786    pub(crate) async fn circle_epoch_access(
787        &self,
788        circle_id: coven_protocol::circle::CircleId,
789        expected_control: coven_protocol::circle::CircleControlCoord,
790    ) -> Result<Option<coven_protocol::circle_activation::CircleEpochAccess>, coven_database::DbError>
791    {
792        self.database
793            .circle_epoch_access(self.root.reference().clone(), circle_id, expected_control)
794            .await
795    }
796
797    #[cfg(any(test, feature = "test-utils"))]
798    pub(crate) async fn latest_local_store_position(
799        &self,
800    ) -> Result<Option<coven_protocol::store_commit::StoreBatchCommitRef>, StoreError> {
801        let writer = self.authorize_writer().await.map_err(StoreError::from)?;
802        writer
803            .latest_local_store_position()
804            .await
805            .map_err(Into::into)
806    }
807
808    #[cfg(any(test, feature = "test-utils"))]
809    pub(crate) async fn load_exact_materialized_commit(
810        &self,
811        stream_id: &str,
812        sequence: u64,
813    ) -> Result<
814        Option<(
815            coven_protocol::store_commit::StoreBatchCommitRef,
816            coven_protocol::store_commit::VerifiedStoreBatchCommit,
817        )>,
818        StoreError,
819    > {
820        let Some(reference) = self
821            .database
822            .exact_materialized_ref(stream_id, sequence)
823            .await?
824        else {
825            return Ok(None);
826        };
827        let mut history = self.authorize_history().await.map_err(StoreError::from)?;
828        let commit = history
829            .load_commit(&reference)
830            .await
831            .map_err(StoreError::from)?;
832        Ok(Some((reference, commit)))
833    }
834}
835
836#[cfg(test)]
837mod tests;
coven_replication/sync/store/authorization.rs

coven_replication/sync/store/
authorization.rs
