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