coven/

handle.rs

//! The data handle: one object a host constructs once that owns coven's
//! pieces and exposes the whole data interface as methods.
//!
//! coven owns the store's data — SQL rows and blobs, on disk first, cloud
//! optional. A host (a desktop/mobile app) talks to coven through this one
//! handle and never assembles coven's internals by hand or hands them back to
//! coven on every call. The handle delegates to retained owners for rows,
//! blobs, sync, security, membership, joining, recovery, and Circles; the
//! caller passes only descriptors (a [`RowBlobRef`], SQL, or a config).
//!
//! The stack runs on Tokio and is `Send + Sync` throughout.
//!
//! ## What it owns
//!
//! - **Rows** — SQL execution and row-and-blob writes.
//! - **Blobs** — exact row-bound reads, cache policy, locality transitions, and
//!   upload visibility.
//! - **Sync** — connection lifecycle, status, and explicit synchronization.
//! - **Security** — key custody, device identity, host secrets, and app-data
//!   sealing.
//! - **Membership, joining, recovery, and Circles** — their complete host
//!   workflows, each behind its retained domain owner.

use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;

use crate::device_pairing::StoreDevicePairing;
use crate::store_blobs::StoreBlobAccess;
use crate::store_blobs::StoreBlobs;
use crate::store_circles::StoreCircles;
use crate::store_cloud_storage::StoreCloudStorage;
use crate::store_joining::StoreJoining;
use crate::store_membership::StoreMembership;
use crate::store_recovery::StoreRecovery;
use crate::store_rows::StoreRows;
use crate::store_security::StoreSecurity;
use crate::store_sync::{ConfigProvider, StoreSync, SyncError};
use coven_database::store::StoreReads;
use coven_database::{Database, DbError, StoreDatabase};
use coven_foundation::clock::ClockRef;
use coven_foundation::store_dir::StoreDir;
use coven_foundation::store_dir::StoreOpenGuard;
use coven_keys::encryption::SealError;
use coven_keys::keys::{
    DeviceIdentityCustody, IdentityError, KeyError, MasterKeyCustody, MasterKeyError, StoreKeys,
};
use coven_protocol::blob::{BlobTransitionObserver, RowBlobRef};
use coven_protocol::membership::MemberInfo;
use coven_protocol::membership::MemberRole;
use coven_replication::blob::transition::{MakeLocalError, MakeRemoteError};
use coven_replication::blob::DrainOutcome;
use coven_replication::sync::store::blob::{LocalStoreBlobAccess, StoreBlobCache};
use coven_replication::sync::sync_loop::SyncLoopStatus;
use coven_replication::sync::EagerCacheFillStatus;
use coven_replication::sync::{BlobCacheError, BlobStream};
#[cfg(any(test, feature = "test-utils"))]
use coven_storage::cloud::ExactCloudHome;
#[cfg(any(test, feature = "test-utils"))]
use coven_storage::CloudCipher;
use tokio::sync::watch;

/// Why one blocked operation could not be handed back to the sync loop.
///
/// The three kinds keep their own refusal vocabularies — a write's, a Circle
/// operation's, a reclaim operation's — because a host that shows the operation
/// shows its error too, and flattening them would lose what it says.
#[derive(Debug, thiserror::Error)]
pub enum RetryBlockedOperationError {
    #[error("retry blocked write: {0}")]
    Write(#[from] crate::CovenError),
    #[error("retry circle operation: {0}")]
    Circle(#[from] crate::CircleError),
    #[error("retry stuck reclaim operation: {0}")]
    Reclaim(#[from] crate::SyncError),
}

/// The cipher a store's app-data sealing runs under, resolved from `custody`.
///
/// A store whose custody unlocks `None` has no key to seal under or open with,
/// which is [`SealError::Locked`] — the same discipline the sync engine's cipher
/// resolution keeps, where an opaque home with no established key refuses to
/// start rather than inventing one.
///
/// Shared by [`CovenHandle`] and [`CovenReadHandle`](crate::CovenReadHandle) so
/// both resolve the identical keyring the identical way; a payload one seals, the
/// other opens.
/// The handle over one coven store.
///
/// Open it once with [`Coven::builder`](crate::Coven::builder), then call methods. Cheap to
/// [`clone`](Clone) — every field is shared (an `Arc`, a `Clone` handle, or a
/// reference-counted lock), so a clone drives the same retained owners as the
/// original.
///
/// # Using the handle
///
/// The host builds the handle once at startup and then only calls methods on it
/// — it never assembles coven's internals by hand or hands them back to coven on
/// every call. Rows go through the connection coven owns; blobs go through the
/// handle's read/store methods; sync is optional.
///
/// ```no_run
/// # use coven::{CovenHandle, RowBlobRef};
/// # async fn use_store(handle: &CovenHandle, cover: &RowBlobRef)
/// #     -> Result<(), Box<dyn std::error::Error>> {
/// // Rows: run app SQL on the connection coven owns.
/// let note_count: i64 = handle
///     .read(|sql| {
///         sql.query_row("SELECT count(*) FROM notes", [], |row| row.get(0))
///             .map_err(coven::CovenError::from)
///     })
///     .await?;
///
/// // Blobs: read an exact row version. coven resolves locality — the user's own
/// // file, its local store, the cache, or a cloud fetch — and returns plaintext.
/// let bytes: Vec<u8> = handle.read_blob(cover).await?;
///
/// // Sync is optional. Connect a provider, then drive it; a store with no
/// // cloud home never calls these and stays fully usable on-device.
/// handle.connect_sync().await?;
/// handle.sync_now();
/// # let _ = note_count;
/// # Ok(())
/// # }
/// ```
#[derive(Clone)]
pub struct CovenHandle {
    rows: StoreRows,
    blobs: StoreBlobs,
    security: StoreSecurity,
    sync: StoreSync,
    membership: StoreMembership,
    joining: StoreJoining,
    pairing: StoreDevicePairing,
    recovery: StoreRecovery,
    circles: StoreCircles,
}

impl CovenHandle {
    /// Build the handle over an already-open [`Database`] and the store's
    /// directory. Does no I/O and opens no sync connection — a home-less store
    /// is fully usable (rows + Local blobs). Call
    /// [`connect_sync`](Self::connect_sync) when a cloud provider is connected.
    ///
    /// `config_provider` is read fresh on every call that needs the current
    /// config (the cloud-home selection, the blob-path scheme), so the host can
    /// reconnect a provider without rebuilding the handle. `observer` carries the
    /// host's transition bookkeeping; pass `None` if it surfaces none.
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn new(
        db: Database,
        read_database: StoreReads,
        store_dir: StoreDir,
        config_provider: ConfigProvider,
        key_service: StoreKeys,
        key_custody: Arc<dyn MasterKeyCustody>,
        identity_custody: Arc<dyn DeviceIdentityCustody>,
        oauth_clients: coven_storage::oauth::OAuthClients,
        clock: ClockRef,
        cloudkit_ops: Option<Arc<dyn coven_storage::cloud::cloudkit::CloudKitOps>>,
        observer: Option<Arc<dyn BlobTransitionObserver>>,
        open_guard: Arc<StoreOpenGuard>,
        blob_chunking: coven_storage::BlobChunking,
    ) -> Self {
        let database = StoreDatabase::from_database(db);
        let cloud_homes = coven_storage::cloud::CloudHomeFactory::new(oauth_clients);
        let credentials = coven_keys::keys::CloudHomeCredentialsOwner::new(key_service.clone());
        let security = StoreSecurity::new(
            key_service,
            key_custody.clone(),
            identity_custody,
            store_dir.clone(),
        );
        let cloud_storage = StoreCloudStorage::new(
            security.clone(),
            cloud_homes,
            credentials,
            clock.clone(),
            cloudkit_ops,
            blob_chunking,
        );
        let blob_cache = StoreBlobCache::new(database.clone(), store_dir.clone());
        let local_blob_access =
            LocalStoreBlobAccess::new(database.clone(), store_dir.clone(), blob_cache);
        let blob_access = StoreBlobAccess::new(
            database.clone(),
            config_provider.clone(),
            cloud_storage.clone(),
            local_blob_access.clone(),
        );
        let sync = StoreSync::new(
            config_provider.clone(),
            security.clone(),
            database.clone(),
            #[cfg(test)]
            store_dir.clone(),
            clock.clone(),
            observer,
            open_guard,
            cloud_storage,
            blob_access.clone(),
            Arc::new(coven_replication::sync::sync_loop::SystemSyncLoopRuntimeFactory),
        );
        let rows = StoreRows::new(
            coven_database::StoreRowWrites::new(database.clone()),
            read_database,
            key_custody,
            sync.clone(),
        );
        let blobs = StoreBlobs::new(database.clone(), blob_access, local_blob_access);
        let membership = StoreMembership::new(sync.clone());
        let joining = StoreJoining::new(database.clone(), membership.clone(), sync.clone());
        let pairing = StoreDevicePairing::new(
            config_provider,
            store_dir.device_pairing_journal_path(),
            clock,
            joining.clone(),
            sync.clone(),
        );
        let recovery = StoreRecovery::new(database.clone(), security.clone(), sync.clone());
        let circles = StoreCircles::new(
            database.clone(),
            membership.clone(),
            security.clone(),
            sync.clone(),
        );
        Self {
            rows,
            blobs,
            security,
            sync,
            membership,
            joining,
            pairing,
            recovery,
            circles,
        }
    }

    pub async fn write<F, R>(&self, sql: F) -> crate::CovenResult<crate::WriteReceipt<R>>
    where
        F: for<'context, 'connection> FnOnce(
                crate::SqlContext<'context, 'connection>,
            ) -> crate::CovenResult<R>
            + Send
            + 'static,
        R: Send + 'static,
    {
        self.rows.write(sql).await
    }

    /// Read one consistent snapshot when awaited. Attach `process` to compute
    /// a result on separate workers after releasing the connection.
    pub fn read<F, R>(&self, read: F) -> crate::Read<'_, F>
    where
        F: for<'connection> FnOnce(crate::SqlReadContext<'connection>) -> crate::CovenResult<R>
            + Send
            + 'static,
        R: Send + 'static,
    {
        self.rows.read(read)
    }

    /// Create a query that returns its initial value and runs again when a
    /// committed database change can affect it.
    ///
    /// The query uses the same [`crate::SqlReadContext`] as [`read`](Self::read).
    /// Coven records the tables and columns SQLite reads, and narrows supported
    /// single-table primary-key predicates to their bound values. Other
    /// predicates retain safe table-and-column invalidation.
    /// Attach [`process`](crate::LiveQuery::process) to move result processing
    /// off the read connection. Only the final delivered value needs `Clone`
    /// and `PartialEq`.
    pub fn subscribe<F, R>(&self, query: F) -> crate::LiveQuery<R>
    where
        F: for<'connection> Fn(crate::SqlReadContext<'connection>) -> crate::CovenResult<R>
            + Send
            + Sync
            + 'static,
        R: Send + 'static,
    {
        self.rows.subscribe(query)
    }

    /// Create a tracked query whose absolute request can be replaced while the
    /// subscription remains active.
    /// Attach [`process`](crate::ReconfigurableLiveQuery::process) to process
    /// each result together with the request that produced it.
    pub fn subscribe_reconfigurable<Request, F, R>(
        &self,
        initial_request: Request,
        query: F,
    ) -> crate::ReconfigurableLiveQuery<Request, R>
    where
        Request: Clone + PartialEq + Send + Sync + 'static,
        F: for<'connection> Fn(
                &Request,
                crate::SqlReadContext<'connection>,
            ) -> crate::CovenResult<R>
            + Send
            + Sync
            + 'static,
        R: Send + 'static,
    {
        self.rows.subscribe_reconfigurable(initial_request, query)
    }

    pub async fn write_with_blobs<F, S, R>(
        &self,
        build: F,
        sql: S,
    ) -> crate::CovenResult<crate::WriteReceipt<R>>
    where
        F: FnOnce(&mut crate::WriteBatch) -> crate::CovenResult<()> + Send + 'static,
        S: for<'context, 'connection> FnOnce(
                crate::SqlContext<'context, 'connection>,
            ) -> crate::CovenResult<R>
            + Send
            + 'static,
        R: Send + 'static,
    {
        self.rows.write_with_blobs(build, sql).await
    }

    // =========================================================================
    // Sync lifecycle
    // =========================================================================

    /// Subscribe to the sync loop's [`SyncLoopStatus`] stream. The channel is
    /// owned by this handle, not the loop, so the receiver keeps working across a
    /// reconnect and may be created before any provider is connected (it starts
    /// receiving once a loop runs). Infallible for that reason — there is no loop
    /// state to check.
    ///
    /// The receiver immediately contains the current value. Intermediate values
    /// may be coalesced; `Synchronized.row_changes` is a refresh hint rather than a
    /// complete change stream.
    pub fn subscribe_sync_status(&self) -> tokio::sync::watch::Receiver<SyncLoopStatus> {
        self.sync.subscribe_status()
    }

    /// Subscribe to the post-open CacheEager fill. Enrollment installs rows and
    /// returns without artwork; the connected library then reports discovery,
    /// bounded-cadence download progress, completion, cancellation, or failure.
    pub fn subscribe_eager_cache_fill_status(
        &self,
    ) -> tokio::sync::watch::Receiver<EagerCacheFillStatus> {
        self.sync.subscribe_eager_cache_status()
    }

    /// Stop post-open CacheEager downloads without stopping cloud sync.
    pub fn cancel_eager_cache_fill(&self) {
        self.sync.cancel_eager_cache_fill();
    }

    /// Writes that have shared rows and have not reached a published position.
    pub async fn pending_writes(&self) -> Result<Vec<crate::PendingWrite>, crate::CovenError> {
        self.rows
            .pending_writes()
            .await
            .map_err(crate::CovenError::from)
    }

    /// Writes stopped by a semantic publication fault and awaiting an explicit
    /// retry or discard decision.
    pub async fn blocked_writes(&self) -> Result<Vec<crate::PendingWrite>, crate::CovenError> {
        self.rows
            .blocked_writes()
            .await
            .map_err(crate::CovenError::from)
    }

    /// Requeue one blocked write for full production validation. A connected
    /// sync loop is woken after the durable transition.
    pub async fn retry_blocked_write(
        &self,
        write_id: &crate::WriteId,
    ) -> Result<Vec<crate::WriteId>, crate::CovenError> {
        self.rows.retry_blocked_write(write_id).await
    }

    /// Hand one blocked operation back to the sync loop, whichever kind it is.
    ///
    /// The host renders [`SyncLoopStatus::Blocked`]'s operations as one list
    /// with one button, so it retries them through one call; the id says which
    /// path the retry takes. Each kind revalidates from scratch, so an
    /// operation whose cause still stands simply blocks again.
    pub async fn retry_blocked_operation(
        &self,
        operation: crate::BlockedOperationId,
    ) -> Result<(), crate::RetryBlockedOperationError> {
        match operation {
            crate::BlockedOperationId::Write(write_id) => {
                self.rows.retry_blocked_write(&write_id).await?;
                Ok(())
            }
            crate::BlockedOperationId::CircleOperation(operation_id) => {
                Ok(self.circles.retry(operation_id).await?)
            }
            crate::BlockedOperationId::Reclaim(operation_id) => {
                Ok(self.sync.retry_stuck_reclaim(operation_id).await?)
            }
        }
    }

    /// Atomically discard a blocked write and reverse every later unpublished
    /// shared write whose working-row state depends on it.
    pub async fn discard_blocked_write(
        &self,
        write_id: &crate::WriteId,
    ) -> Result<Vec<crate::WriteId>, crate::CovenError> {
        self.rows.discard_blocked_write(write_id).await
    }

    /// Read the current durable status of one write.
    pub async fn write_status(
        &self,
        write_id: &crate::WriteId,
    ) -> Result<crate::WriteStatus, crate::CovenError> {
        self.rows
            .write_status(write_id)
            .await
            .map_err(crate::CovenError::from)
    }

    /// Subscribe to one write's current durable status. The initial value is
    /// reconstructed from SQLite before the receiver is returned.
    pub async fn subscribe_write_status(
        &self,
        write_id: &crate::WriteId,
    ) -> Result<tokio::sync::watch::Receiver<crate::WriteStatus>, crate::CovenError> {
        self.rows
            .subscribe_write_status(write_id)
            .await
            .map_err(crate::CovenError::from)
    }

    /// Build the connected cloud storage, start its sync loop, and install the
    /// connection. If the cloud home fails to build, no connection is installed.
    ///
    /// The at-rest cipher is resolved from the handle's custody per start: an
    /// opaque home unlocks the master keyring (failing with
    /// [`SyncError::MasterKeyNotEstablished`] if none is established), a
    /// browsable one never consults custody. Reconnecting a provider replaces
    /// the cloud home and loop while retaining the Store database and clock.
    pub async fn connect_sync(&self) -> Result<(), SyncError> {
        self.sync.connect().await
    }

    /// Build and probe the cloud home described by `config` without installing
    /// it as this handle's sync connection. Hosts use this to validate proposed
    /// provider settings before committing them to their config source.
    pub async fn probe_cloud_home(&self, config: &crate::Config) -> Result<(), SyncError> {
        self.sync.probe_cloud_home(config).await
    }

    /// Connect a new S3 cloud home and commit its credentials and any generated
    /// opaque-home master key only after the replacement connection is ready.
    pub async fn setup_s3_cloud_home(
        &self,
        cloud_home: crate::CloudHomeConfig,
        access_key: String,
        secret_key: String,
    ) -> Result<crate::ConnectedCloudHome, crate::CloudHomeSetupError> {
        self.sync.setup_s3(cloud_home, access_key, secret_key).await
    }

    /// Connect a new CloudKit cloud home and commit any generated opaque-home
    /// master key only after the replacement connection is ready.
    pub async fn setup_cloudkit_cloud_home(
        &self,
        cloud_home: crate::CloudHomeConfig,
        cloudkit_ops: Arc<dyn coven_storage::cloud::cloudkit::CloudKitOps>,
    ) -> Result<crate::ConnectedCloudHome, crate::CloudHomeSetupError> {
        self.sync.setup_cloudkit(cloud_home, cloudkit_ops).await
    }

    /// Authorize and connect a new Google Drive, Dropbox, or OneDrive home.
    /// Tokens remain proposed until the replacement connection is ready.
    #[cfg(feature = "oauth-providers")]
    pub async fn setup_oauth_cloud_home(
        &self,
        cloud_home: crate::CloudHomeConfig,
        cancel: tokio::sync::watch::Receiver<bool>,
    ) -> Result<crate::ConnectedCloudHome, crate::CloudHomeSetupError> {
        self.sync.setup_oauth(cloud_home, cancel).await
    }

    /// Whether a home with this storage policy needs and can unlock its key.
    pub fn cloud_home_key_state(
        &self,
        storage: crate::HomeStorage,
    ) -> Result<crate::CloudHomeKeyState, KeyError> {
        self.security.cloud_home_key_state(storage)
    }

    /// Import the master key for this returning opaque cloud home, verify it
    /// against the signed Store root, and connect without retaining a rejected key.
    pub async fn unlock_cloud_home(
        &self,
        serialized_master_key: &str,
    ) -> Result<crate::ConnectedCloudHome, crate::CloudHomeUnlockError> {
        self.sync.unlock(serialized_master_key).await
    }

    pub async fn connect_sync_with_cloudkit(
        &self,
        cloudkit_ops: Arc<dyn coven_storage::cloud::cloudkit::CloudKitOps>,
    ) -> Result<(), SyncError> {
        self.sync.connect_with_cloudkit(cloudkit_ops).await
    }

    /// Test-only: connect a started sync loop over an injected [`ExactCloudHome`]
    /// instead of one built from [`crate::Config`], so a host's integration tests drive
    /// the real make-Remote / make-Local / upload-drain and read paths over a mock
    /// cloud with no live provider.
    ///
    /// The test counterpart of [`connect_sync`](Self::connect_sync): it builds
    /// storage over `home`/`cipher`, prepares the configured home's master key,
    /// starts the loop, and commits a newly generated key and the connection
    /// together only after startup succeeds. The explicit cipher protects the
    /// injected storage; the master key separately protects Store routing data.
    ///
    /// The read path needs no separate hook: `blob_storage`
    /// serves reads from the connected loop's own `CloudSyncConnection`, which here
    /// wraps the injected `home`, so [`read_blob`](Self::read_blob) /
    /// [`pin`](Self::pin) resolve a Remote miss against the same test home the
    /// drain writes to.
    #[cfg(any(test, feature = "test-utils"))]
    pub fn connect_sync_with_test_home(
        &self,
        home: Arc<dyn ExactCloudHome>,
        cipher: CloudCipher,
    ) -> impl std::future::Future<Output = Result<(), crate::CloudHomeSetupError>> + Send + '_ {
        Box::pin(async move { self.sync.connect_with_test_home(home, cipher).await })
    }

    /// Test-only: atomically set up a proposed cloud home over an injected
    /// provider while exercising the production key and connection transaction.
    #[cfg(any(test, feature = "test-utils"))]
    pub async fn setup_cloud_home_with_test_home(
        &self,
        cloud_home: crate::CloudHomeConfig,
        home: Arc<dyn ExactCloudHome>,
        credentials: Option<crate::CloudHomeCredentials>,
    ) -> Result<crate::ConnectedCloudHome, crate::CloudHomeSetupError> {
        self.sync
            .setup_with_test_home(cloud_home, home, credentials)
            .await
    }

    /// Test-only: unlock a returning opaque home over an injected provider
    /// while exercising the production key-and-connection transaction.
    #[cfg(any(test, feature = "test-utils"))]
    pub async fn unlock_cloud_home_with_test_home(
        &self,
        serialized_master_key: &str,
        home: Arc<dyn ExactCloudHome>,
    ) -> Result<crate::ConnectedCloudHome, crate::CloudHomeUnlockError> {
        self.sync
            .unlock_with_test_home(serialized_master_key, home)
            .await
    }

    /// Test-only: connect over an injected [`ExactCloudHome`] exactly as
    /// [`connect_sync_with_test_home`](Self::connect_sync_with_test_home) does,
    /// but start no background loop — the caller drives sync itself.
    ///
    /// The loop-started connect and an explicit
    /// [`drain_uploads`](Self::drain_uploads) are two drainers of one queue. They
    /// take turns rather than overlap, so whichever runs second drains only what
    /// the first left — a host asserting on its own drain's count reads that as
    /// "nothing was queued" and fails intermittently. Here no cycle exists to
    /// share the queue with: the host's `drain_uploads` is the only drain, its
    /// count is the whole truth, and [`is_syncing`](Self::is_syncing) stays
    /// `false` for the connection's whole life.
    ///
    /// Everything a connected store can do is available — `make_remote`,
    /// `make_local`, the drain, membership — because none of it needs the loop
    /// thread. Circle *writes* are the exception: they are dispatched to that
    /// thread, so they refuse with
    /// [`CircleError::LoopNotRunning`](crate::CircleError::LoopNotRunning) here.
    #[cfg(any(test, feature = "test-utils"))]
    pub fn connect_sync_with_test_home_caller_driven(
        &self,
        home: Arc<dyn ExactCloudHome>,
        cipher: CloudCipher,
    ) -> impl std::future::Future<Output = Result<(), crate::CloudHomeSetupError>> + Send + '_ {
        Box::pin(async move {
            self.sync
                .connect_with_test_home_caller_driven(home, cipher)
                .await
        })
    }

    /// Test-only: connect over an injected [`ExactCloudHome`] while resolving the
    /// at-rest cipher from custody the way production
    /// [`connect_sync`](Self::connect_sync) does, instead of taking an explicit
    /// cipher like [`connect_sync_with_test_home`](Self::connect_sync_with_test_home).
    ///
    /// Where that method prepares a missing master key as part of its connection
    /// transaction, this requires an existing key and drives the connection path
    /// used by production, which unlocks the master keyring through the store's
    /// custody exactly as `start_sync` would — so a
    /// test can establish a key, connect over a mock home, and prove the traffic
    /// is sealed under that key. An opaque home with no key established fails
    /// [`SyncError::MasterKeyNotEstablished`] before the loop starts.
    #[cfg(any(test, feature = "test-utils"))]
    pub async fn connect_sync_with_test_home_custody(
        &self,
        home: Arc<dyn ExactCloudHome>,
    ) -> Result<(), SyncError> {
        self.sync.connect_with_test_home_custody(home).await
    }

    /// Start (or restart) the sync loop of the installed connection. A no-op
    /// when no provider is connected — a home-less store has nothing to start.
    /// Errors if the connected cloud home fails to build.
    pub async fn start_sync(&self) -> Result<(), SyncError> {
        self.sync.start().await
    }

    /// Stop the sync loop after the in-flight cycle while keeping the provider
    /// connected so [`start_sync`](Self::start_sync) can resume it. A no-op when
    /// no provider is connected.
    ///
    /// The material a running loop resolved from custody (the master keyring,
    /// the device signing identity) is cached only inside that loop for as
    /// long as it runs — nowhere else in the handle — and this is where it is
    /// purged. A subsequent [`start_sync`](Self::start_sync)/
    /// [`connect_sync`](Self::connect_sync) re-resolves fresh from whatever
    /// custody now serves, so a host's lock flow that stops sync as part of
    /// locking, then later reconnects, never resumes on stale material.
    pub fn stop_sync(&self) {
        self.sync.stop()
    }

    /// Disconnect the provider entirely: stop the loop and drop the connection.
    /// The store becomes home-less until the next
    /// [`connect_sync`](Self::connect_sync).
    ///
    /// Carries the same purge as [`stop_sync`](Self::stop_sync), so nothing about
    /// the previous connection — including which custody it resolved material
    /// from — survives into the next connect.
    pub fn disconnect_sync(&self) {
        self.sync.disconnect()
    }

    /// Disconnect the configured cloud home and remove its provider credentials.
    /// If credential removal fails, the installed connection is preserved.
    pub async fn disconnect_cloud_home(&self) -> Result<(), SyncError> {
        self.sync.disconnect_cloud_home().await
    }

    /// Wake the sync loop to run a cycle now rather than at the next idle tick. A
    /// no-op when no provider is connected.
    pub fn sync_now(&self) {
        self.sync.trigger()
    }

    /// Whether the sync loop is running. `false` for a home-less store.
    pub fn is_syncing(&self) -> bool {
        self.sync.is_syncing()
    }

    /// Whether a provider connection is installed. Distinct
    /// from [`is_syncing`](Self::is_syncing), which additionally requires the loop
    /// to be running: this is the predicate a host uses for "has a cloud home"
    /// without the loop-ready condition.
    pub fn is_connected(&self) -> bool {
        self.sync.is_connected()
    }

    // =========================================================================
    // Master-key lifecycle
    // =========================================================================

    /// Import a serialized master keyring a host already holds and establish it
    /// under the handle's custody, replacing whatever custody already holds.
    pub async fn import_master_key(&self, serialized: &str) -> Result<(), MasterKeyError> {
        self.sync.import_master_key(serialized).await
    }

    /// Remove the master key from custody and disconnect any operation retaining
    /// its unlocked value. If custody cannot remove the key, the connection is
    /// preserved and the error is returned.
    pub async fn forget_master_key(&self) -> Result<(), SyncError> {
        self.sync.forget_master_key().await
    }

    // =========================================================================
    // Identity lifecycle
    // =========================================================================

    /// Generate this store's signing identity and establish it under the
    /// handle's identity custody. Errors with
    /// [`IdentityError::AlreadyEstablished`] if custody already unlocks one —
    /// coven never generates over an existing identity. This is the identity
    /// counterpart of cloud-home setup's master-key transaction for a store a
    /// host is creating fresh (not joining or restoring, which each establish
    /// their own identity as part of what they do). Returns the established
    /// public key, hex-encoded.
    pub fn initialize_identity(&self) -> Result<String, IdentityError> {
        self.security.initialize_identity()
    }

    // =========================================================================
    // Host secrets
    // =========================================================================

    /// Set a host's own store-scoped secret — an API token, a service
    /// credential — under the same platform keyring, and the same access
    /// policy, as coven's own key material. `name` identifies the secret
    /// within the store; coven owns the account rendering and the entry's
    /// protection class. [`KeyError::InvalidSecretName`] if `name` collides
    /// with one of coven's own reserved slot names, is empty, or contains
    /// `:`.
    /// The concurrent blob-transfer limits in force: how many uploads an
    /// upload-drain pass runs at once and how many downloads a pin fetches at
    /// once.
    pub fn transfer_limits(&self) -> coven_protocol::blob::TransferLimits {
        self.blobs.transfer_limits()
    }

    /// Replace the transfer limits while the store is open. Every later
    /// upload-drain pass and pin call runs under the new limits; a pass
    /// already running keeps the limit it admitted under. The builder's
    /// `max_concurrent_uploads` / `max_concurrent_downloads` set the initial
    /// values.
    pub fn set_transfer_limits(&self, limits: coven_protocol::blob::TransferLimits) {
        self.blobs.set_transfer_limits(limits)
    }

    pub fn set_host_secret(&self, name: &str, value: &str) -> Result<(), KeyError> {
        self.security.set_host_secret(name, value)
    }

    /// Read a host secret set by [`set_host_secret`](Self::set_host_secret),
    /// `None` if never set. A present-but-empty entry is corrupt, not
    /// absent — the same discipline coven's own key reads apply.
    pub fn host_secret(&self, name: &str) -> Result<Option<String>, KeyError> {
        self.security.host_secret(name)
    }

    /// Remove a host secret. `Ok` whether or not one was set.
    pub fn delete_host_secret(&self, name: &str) -> Result<(), KeyError> {
        self.security.delete_host_secret(name)
    }

    // =========================================================================
    // App-data sealing
    // =========================================================================

    /// Seal `plaintext` under the store's current master-key generation, for a
    /// host to store in its own rows — a password entry's payload, an API token.
    /// coven's at-rest encryption is cloud-side; the local database is plaintext
    /// SQLite, so a host with a secret to keep in a row seals it here first.
    ///
    /// The output records the key fingerprint, so it stays openable after key
    /// rotation while that key remains in custody. `aad` binds the ciphertext to
    /// its context — the owning row's primary key, say — and
    /// [`open_app_data`](Self::open_app_data) with a different `aad` fails, so a
    /// payload moved to another row does not silently open there.
    ///
    /// [`SealError::Locked`] if the store has no established master key, the same
    /// gate [`connect_sync`](Self::connect_sync) applies before it seals cloud
    /// traffic.
    pub fn seal_app_data(&self, plaintext: &[u8], aad: &[u8]) -> Result<Vec<u8>, SealError> {
        self.security.seal_app_data(plaintext, aad)
    }

    /// Open a payload [`seal_app_data`](Self::seal_app_data) produced, under
    /// the key fingerprint it names. Rotation preserves access while custody
    /// retains the named key.
    ///
    /// [`SealError::Locked`] if the store is locked; a wrong `aad`, a tampered
    /// payload, an unreadable version, or a fingerprint this store's keyring lacks
    /// each surface their own typed error.
    pub fn open_app_data(&self, sealed: &[u8], aad: &[u8]) -> Result<Vec<u8>, SealError> {
        self.security.open_app_data(sealed, aad)
    }

    // =========================================================================
    // Blobs
    // =========================================================================

    /// Capture the exact current blob-bearing row version. Blob operations use
    /// this row-bound value so a later row replacement cannot redirect a read.
    pub async fn row_blob_ref(&self, table: &str, row_id: &str) -> Result<RowBlobRef, DbError> {
        self.blobs.row_blob_ref(table, row_id).await
    }

    /// Read a blob's whole plaintext through coven's locality-aware read: served
    /// from the user's file (Local user-provided), coven's local store (Local
    /// host-provided), the pinned/evictable cache on a Remote hit, or fetched
    /// from the cloud (into the cache) on a Remote miss. The host passes the
    /// [`RowBlobRef`] captured from [`row_blob_ref`](Self::row_blob_ref); coven
    /// holds the database, directory, and storage.
    pub async fn read_blob(&self, blob: &RowBlobRef) -> Result<Vec<u8>, BlobCacheError> {
        self.blobs.read(blob).await
    }

    /// Ensure the exact current row blob plaintext is durable on this device.
    /// Remote blobs materialize into their locator-keyed cache path; Local and
    /// pending-remote blobs exact-verify their authoritative local source.
    pub async fn materialize_row_blob(&self, blob: &RowBlobRef) -> Result<(), BlobCacheError> {
        self.blobs.materialize(blob).await
    }

    /// Open an exact row blob's plaintext for ranged reading, for streaming or
    /// seeking without loading the whole file. The ranged sibling of
    /// [`read_blob`](Self::read_blob), which stays the one-shot whole read.
    ///
    /// Opening resolves the blob's locality, proves the plaintext's size and
    /// content hash against the row, and holds the open file; every
    /// [`BlobStream::read_at`] then costs only the bytes it returns. Hold the
    /// stream for as long as the host is reading that blob — a stream per opened
    /// file, not per range — since re-opening re-proves the whole blob.
    pub async fn open_blob_stream(&self, blob: &RowBlobRef) -> Result<BlobStream, BlobCacheError> {
        self.blobs.open_stream(blob).await
    }

    /// Pin a Remote blob set for offline: coven fetches each into the protected
    /// cache (`storage/pinned/`) — from the evictable cache if already there, else
    /// the cloud — exempt from the size budget. Idempotent.
    pub async fn pin(&self, blobs: &[RowBlobRef]) -> Result<(), BlobCacheError> {
        self.blobs.pin(blobs).await
    }

    /// Unpin a Remote blob set: coven moves each from `storage/pinned/` to the
    /// evictable `storage/cache/` (still readable, now droppable). No cloud read.
    pub async fn unpin(&self, blobs: &[RowBlobRef]) -> Result<(), BlobCacheError> {
        self.blobs.unpin(blobs).await
    }

    /// Whether every blob in `blobs` is pinned for offline — present in coven's
    /// kept cache folder (`storage/pinned/`). The host answers "is this release
    /// kept offline" through this instead of stat-ing coven's cache layout itself.
    /// An empty set is vacuously pinned. A blob not pinned (in the evictable cache
    /// or absent) makes the whole set unpinned; an existence-check failure is
    /// surfaced, never read as "not pinned".
    pub async fn is_pinned(&self, blobs: &[RowBlobRef]) -> Result<bool, BlobCacheError> {
        self.blobs.all_pinned(blobs).await
    }

    /// Whether each of `table`'s `row_ids` is pinned for offline, one answer per
    /// id in the order given. `None` where an id names no live blob-bearing row.
    ///
    /// [`is_pinned`](Self::is_pinned) answers over blobs that together make up
    /// one thing — every blob of a release, pinned and unpinned together. This
    /// answers for many independent rows at once: a host drawing a "kept
    /// offline" marker per row of a page resolves and answers the whole page in
    /// one call, instead of a [`row_blob_ref`](Self::row_blob_ref) and an
    /// `is_pinned` per row.
    ///
    /// A row whose blob has no committed cloud object — one still Local, or one
    /// whose upload has not landed — has no kept copy to hold and reads as not
    /// pinned. An existence-check failure is still surfaced, never read as "not
    /// pinned".
    pub async fn rows_pinned(
        &self,
        table: &str,
        row_ids: Vec<String>,
    ) -> Result<Vec<Option<bool>>, BlobCacheError> {
        self.blobs.rows_pinned(table, row_ids).await
    }

    /// Remove one Remote blob's re-fetchable on-device cache copies from both
    /// `storage/pinned/` and `storage/cache/`. This never touches the local store,
    /// whose bytes may be the only usable copy owned by an unpublished write.
    /// It does not delete the cloud blob or its carrying row; a later read can
    /// fetch the bytes again.
    pub async fn evict_blob(&self, blob: &RowBlobRef) -> Result<(), BlobCacheError> {
        self.blobs.evict(blob).await
    }

    /// Make `(root_table, root_id)` Remote (Local → Remote): atomically enqueue
    /// every current blob-bearing row and record the transition intent, then
    /// return. Both provenances use this queue: user-provided blobs read their
    /// registered external file, and host-provided blobs read coven's local file.
    /// Once every upload is Created, coven records the gate change and its Store
    /// write together. Publication consumes the upload records and intent.
    /// `pin` retains uploaded plaintext as pinned offline copies. Errors with
    /// [`MakeRemoteError::SyncNotReady`] when no provider is connected.
    ///
    /// `refs` is the root's complete current blob set in the order the host wants
    /// uploads admitted. coven validates the set atomically before enqueueing it.
    /// `root_label` is what the host calls this root, snapshotted onto the queue
    /// rows and the intent. The queue outlives the root row on purpose — a
    /// cancelled or deleted root still has cloud objects to unwind — so an entry
    /// that had to read the row to name itself could not be rendered at exactly
    /// the moment a person most needs to see it.
    pub async fn make_remote(
        &self,
        root_table: &str,
        root_id: &str,
        root_label: &str,
        pin: bool,
        refs: Vec<coven_protocol::blob::RowBlobRef>,
    ) -> Result<(), MakeRemoteError> {
        self.sync
            .make_remote(root_table, root_id, root_label, pin, refs)
            .await
    }

    pub async fn make_remote_batch(
        &self,
        root_table: &str,
        roots: Vec<crate::MakeRemoteRoot>,
        pin: bool,
    ) -> Result<(), MakeRemoteError> {
        self.sync.make_remote_batch(root_table, roots, pin).await
    }

    #[cfg(test)]
    pub(crate) async fn make_remote_with_discovered_order_for_test(
        &self,
        root_table: &str,
        root_id: &str,
        root_label: &str,
        pin: bool,
    ) -> Result<(), MakeRemoteError> {
        let refs = self
            .blobs
            .row_blob_refs_for_root(root_table, root_id)
            .await?;
        self.make_remote(root_table, root_id, root_label, pin, refs)
            .await
    }

    #[cfg(test)]
    pub(crate) async fn make_remote_batch_with_discovered_order_for_test(
        &self,
        root_table: &str,
        roots: Vec<(String, String)>,
        pin: bool,
    ) -> Result<(), MakeRemoteError> {
        let mut prepared = Vec::with_capacity(roots.len());
        for (id, label) in roots {
            let refs = self.blobs.row_blob_refs_for_root(root_table, &id).await?;
            prepared.push(crate::MakeRemoteRoot { id, label, refs });
        }
        self.make_remote_batch(root_table, prepared, pin).await
    }

    /// Cancel a make-remote transition before it reaches Publishing. The intent
    /// becomes Cancelling; the drain deletes each retained exact object, spool,
    /// and cached copy before retiring its upload record. The last record and
    /// intent retire together. Cleanup failures retain the work for retry, and
    /// the root stays Local. A Publishing transition refuses cancellation.
    /// Cancellation can be recorded offline; the drain needs a provider
    /// connection to carry out the cleanup.
    pub async fn cancel_make_remote(
        &self,
        root_table: &str,
        root_id: &str,
    ) -> Result<(), MakeRemoteError> {
        self.sync.cancel_make_remote(root_table, root_id).await
    }

    /// Make `(root_table, root_id)` Local (Remote → Local): bring each blob back to
    /// a local file durability-first — a user-provided blob to the path named in
    /// `dest` (blob id → destination path), a host-provided blob to coven's local
    /// store (no dest) — then flip the gate false, register the external refs, and
    /// enqueue the cloud deletes in one atomic commit. `cancel` aborts before the
    /// commit (the root stays Remote). Errors with [`MakeLocalError::SyncNotReady`]
    /// when no provider is connected.
    pub async fn make_local(
        &self,
        root_table: &str,
        root_id: &str,
        dest: &HashMap<String, PathBuf>,
        cancel: &watch::Receiver<bool>,
    ) -> Result<(), MakeLocalError> {
        self.sync
            .make_local(root_table, root_id, dest, cancel)
            .await
    }

    /// Every upload the durable queue is holding, oldest first.
    ///
    /// An upload appears here the moment [`make_remote`](Self::make_remote)
    /// enqueues it — before any transfer is attempted, and whether or not sync
    /// is connected — and stays until its publication activates or its
    /// cancellation clears it. The queue is a table in the store database, so
    /// this survives restarts: a host can render "waiting to upload" without
    /// having observed the transfer that will do it.
    ///
    /// This is a read; nothing here starts or advances a transfer. Compare
    /// [`drain_uploads`](Self::drain_uploads), which does the work.
    ///
    /// [`make_remote_progress`](Self::make_remote_progress) distinguishes a root
    /// still uploading from one Publishing or Cancelling. Created uploads remain
    /// queued until publication activates or cancellation cleanup completes.
    pub async fn queued_uploads(&self) -> Result<Vec<crate::QueuedUpload>, crate::DbError> {
        self.blobs.queued_uploads().await
    }

    /// Subscribe to the durable upload queue and make-remote intents as one
    /// committed snapshot. The first [`crate::CloudOutboxLiveQuery::next`] returns
    /// immediately; later calls wake from the same committed-change stream as
    /// row live queries.
    pub fn subscribe_cloud_outbox(&self) -> crate::CloudOutboxLiveQuery {
        self.blobs.subscribe_cloud_outbox()
    }

    /// Read the same committed durable state the cloud-outbox subscription
    /// emits, without waiting for a change.
    pub async fn cloud_outbox_snapshot(
        &self,
    ) -> Result<crate::CloudOutboxSnapshot, crate::DbError> {
        self.blobs.cloud_outbox_snapshot().await
    }

    /// The queued uploads belonging to one gated root.
    ///
    /// The filter runs in SQL, so asking about one root does not decode every
    /// other queued upload in the store. The result includes Created uploads
    /// awaiting publication, so a nonempty result does not imply more bytes must
    /// be sent. Use [`make_remote_progress`](Self::make_remote_progress) for the
    /// root's transition phase, or the cloud-outbox snapshot for both together.
    pub async fn queued_uploads_for_root(
        &self,
        root_table: &str,
        root_id: &str,
    ) -> Result<Vec<crate::QueuedUpload>, crate::DbError> {
        self.blobs
            .queued_uploads_for_root(root_table, root_id)
            .await
    }

    /// Where the user's own file for a row's blob lives on disk, or `None`
    /// when the row has no external registration.
    ///
    /// This is the read that mirrors
    /// [`SqlContext::register_external_blob`](crate::SqlContext::register_external_blob):
    /// a host that needs the original file itself — to re-read its tags, to
    /// find an artifact it produced — asks here rather than reading coven's
    /// copy, because for a user-provided blob there is no copy.
    ///
    /// `None` means no registration, which is an ordinary answer: a row whose
    /// blobs coven copies, or one whose registration was cleared, has no user
    /// file to name. A registration that disagrees with the row it belongs to
    /// is an error, not a `None`.
    pub async fn external_blob(
        &self,
        table: &str,
        row_id: &str,
    ) -> Result<Option<crate::ExternalBlob>, crate::DbError> {
        self.blobs.external_blob(table, row_id).await
    }

    /// Every cloud tombstone the durable queue is holding, oldest first.
    ///
    /// A tombstone is queued by
    /// [`SqlContext::enqueue_blob_delete`](crate::SqlContext::enqueue_blob_delete)
    /// and stays until a sync cycle carries the removal out, so this reports
    /// removals still owed to the cloud across restarts.
    pub async fn queued_deletes(&self) -> Result<Vec<crate::QueuedDelete>, crate::DbError> {
        self.blobs.queued_deletes().await
    }

    /// How far the make-remote for one gated root has got, or `None` when that
    /// root has none running.
    ///
    /// Once every upload is Created, the intent becomes
    /// [`MakeRemoteProgress::Publishing`](crate::MakeRemoteProgress). Its upload
    /// records remain queued until the Store write activates, when the records
    /// and intent are consumed atomically. Cancelling instead retains the intent
    /// until the drain completes the exact upload cleanup.
    pub async fn make_remote_progress(
        &self,
        root_table: &str,
        root_id: &str,
    ) -> Result<Option<crate::MakeRemoteProgress>, crate::DbError> {
        self.blobs.make_remote_progress(root_table, root_id).await
    }

    /// Drain pending blob uploads now: read each local file, seal it under its
    /// scope, write it to the cloud, and keep a `retain_pinned` entry's plaintext
    /// in the protected cache.
    ///
    /// The sync loop drains each cycle; this drives a drain directly off the
    /// connected home, against coven's own register clock and the handle's
    /// observer. Errors when no provider is connected (there is no cloud to write
    /// to).
    ///
    /// The [`DrainOutcome`] says what the pass found, not just how much it moved:
    /// an empty queue, a queue held entirely in retry backoff, and a paused one
    /// are each their own answer rather than a zero count.
    ///
    /// A host that connects with a running loop shares the queue with the
    /// cycle's drain. The two never run at once — the queue is drained under an
    /// exclusive turn, so one entry is never in two uploads — but they do divide
    /// the work: this call may wait for a cycle's drain and then find the
    /// entries it wanted already uploaded, and answer `QueueEmpty`. The outcome
    /// describes this pass, never the queue's whole history, so a host that
    /// needs the latter should watch
    /// [`subscribe_cloud_outbox`](Self::subscribe_cloud_outbox) rather than
    /// count one drain's return. `connect_sync_with_test_home_caller_driven`
    /// (test builds only) connects without a loop, so this call is the only
    /// drain and its count is the whole truth.
    pub async fn drain_uploads(&self) -> Result<DrainOutcome, SyncError> {
        self.sync.drain_uploads().await
    }

    /// Retry every failed upload now, without waiting for its automatic retry
    /// delay. This clears the durable delay only after confirming that a cloud
    /// connection can run the drain, then attempts the queue immediately.
    ///
    /// Provider failures remain in the returned [`DrainOutcome`], with their
    /// updated attempt records available through
    /// [`subscribe_cloud_outbox`](Self::subscribe_cloud_outbox). Connection or
    /// database failures are returned as [`SyncError`].
    pub async fn retry_uploads_now(&self) -> Result<DrainOutcome, SyncError> {
        self.sync.retry_uploads_now().await
    }

    pub async fn get_cache_budget(&self, namespace: &str) -> Result<Option<u64>, crate::DbError> {
        self.blobs.cache_budget(namespace).await
    }

    pub async fn set_cache_budget(
        &self,
        namespace: &str,
        max_bytes: u64,
    ) -> Result<(), crate::DbError> {
        self.blobs.set_cache_budget(namespace, max_bytes).await
    }

    /// Generate a restore code, seeded with the store's current membership-head
    /// floor read from the cloud. Requires a connected provider because minting
    /// a trustworthy floor is a network read, not a pure function of local
    /// config and keyring state — a restore code minted without one would carry
    /// no protection against a storage provider replaying an older, otherwise
    /// validly signed membership state to the device that redeems it.
    pub async fn generate_restore_code(&self) -> Result<String, SyncError> {
        self.recovery.generate_restore_code().await
    }

    pub async fn get_members(&self) -> Result<Vec<MemberInfo>, SyncError> {
        self.membership.members().await
    }

    pub async fn start_device_pairing(
        &self,
    ) -> Result<crate::DevicePairingHost, crate::StartDevicePairingError> {
        self.pairing.start().await
    }

    pub async fn approve_device_pairing(
        &self,
        host: &crate::DevicePairingHost,
        request: &crate::DevicePairingRequest,
        role: MemberRole,
        policy: crate::DeviceJoinApprovalPolicy<'_>,
        access_administrator: Option<&dyn crate::DeviceProviderAccessAdministrator>,
        on_progress: &(dyn Fn(crate::AdmittingDeviceJoinProgress) + Send + Sync),
        cancel: tokio::sync::watch::Receiver<bool>,
    ) -> Result<crate::DeviceJoinDriveOutcome, crate::ApproveDevicePairingError> {
        self.pairing
            .approve(
                host,
                request,
                role,
                policy,
                access_administrator,
                on_progress,
                cancel,
            )
            .await
    }

    pub async fn cancel_device_pairing(
        &self,
        host: &crate::DevicePairingHost,
    ) -> Result<(), crate::ApproveDevicePairingError> {
        self.pairing.cancel(host).await
    }

    pub async fn begin_device_join(
        &self,
        member_pubkey: &str,
    ) -> Result<crate::DeviceJoinOffer, SyncError> {
        self.sync.begin_device_join(member_pubkey).await
    }

    pub async fn abandon_device_join(
        &self,
        offer: crate::DeviceJoinOffer,
    ) -> Result<crate::DeviceJoinAbandonment, SyncError> {
        self.sync.abandon_device_join(offer).await
    }

    pub async fn authorize_device_provider_access(
        &self,
        request: crate::DeviceProviderAccessRequest,
        access_administrator: Option<&dyn crate::DeviceProviderAccessAdministrator>,
    ) -> Result<crate::DeviceProviderAdmissionApproval, SyncError> {
        self.sync
            .authorize_device_provider_access(request, access_administrator)
            .await
    }

    pub async fn accept_device_registration_request(
        &self,
        request: crate::DeviceRegistrationRequest,
    ) -> Result<crate::ProvisionalDeviceBootstrap, SyncError> {
        self.sync.accept_device_registration(request).await
    }

    pub async fn publish_device_provider_challenge(
        &self,
        bootstrap: crate::ProvisionalDeviceBootstrap,
    ) -> Result<crate::ProviderReadyDeviceBootstrap, SyncError> {
        self.sync.publish_device_provider_challenge(bootstrap).await
    }

    pub async fn complete_device_provider_admission(
        &self,
        readiness: crate::DeviceJoinReadiness,
    ) -> Result<crate::DeviceProviderAdmissionCompletion, SyncError> {
        self.sync
            .complete_device_provider_admission(readiness)
            .await
    }

    pub async fn finalize_device_join(
        &self,
        completion: crate::DeviceProviderAdmissionCompletion,
    ) -> Result<crate::DeviceJoinActivation, SyncError> {
        self.sync.finalize_device_join(completion).await
    }

    pub async fn device_join_status(
        &self,
        attempt_id: crate::DeviceJoinAttemptId,
        role: crate::DeviceJoinRole,
    ) -> Result<Option<crate::DeviceJoinStatus>, SyncError> {
        self.joining.status(attempt_id, role).await
    }

    pub async fn resume_device_joins(&self) -> Result<Vec<crate::DeviceJoinAction>, SyncError> {
        self.joining.resumable_actions().await
    }

    pub async fn remove_member(&self, public_key_hex: &str) -> Result<(), SyncError> {
        self.membership.remove(public_key_hex).await
    }

    #[cfg(test)]
    pub(crate) async fn admit_member_for_test(
        &self,
        public_key_hex: &str,
        role: MemberRole,
    ) -> Result<coven_replication::sync::MemberAdmission, SyncError> {
        self.membership.admit(public_key_hex, None, role).await
    }

    /// Propose excluding one Store device and return the code that identifies
    /// the exact activated proposal.
    pub async fn propose_device_exclusion(
        &self,
        device_id: crate::StoreDeviceId,
    ) -> Result<String, SyncError> {
        self.membership.propose_device_exclusion(device_id).await
    }

    /// Cancel the exact Store-device exclusion proposal carried by `proposal_code`.
    pub async fn cancel_device_exclusion(&self, proposal_code: &str) -> Result<(), SyncError> {
        self.membership.cancel_device_exclusion(proposal_code).await
    }

    /// Finalize the exact Store-device exclusion proposal carried by `proposal_code`.
    pub async fn finalize_device_exclusion(&self, proposal_code: &str) -> Result<(), SyncError> {
        self.membership
            .finalize_device_exclusion(proposal_code)
            .await
    }

    /// Begin transferring Store ownership to an active device and return the
    /// request code that device must accept.
    pub async fn begin_owner_promotion(
        &self,
        device_id: crate::StoreDeviceId,
    ) -> Result<String, SyncError> {
        self.membership.begin_owner_promotion(device_id).await
    }

    /// Accept an Owner-promotion request and return the acceptance code the
    /// existing Owner must finalize.
    pub async fn accept_owner_promotion(&self, request_code: &str) -> Result<String, SyncError> {
        self.membership.accept_owner_promotion(request_code).await
    }

    /// Finalize the Owner-promotion acceptance carried by `acceptance_code`.
    pub async fn finalize_owner_promotion(&self, acceptance_code: &str) -> Result<(), SyncError> {
        self.membership
            .finalize_owner_promotion(acceptance_code)
            .await
    }

    /// The Circle application surface: create, lifecycle, inspection, and typed
    /// [`CircleError`](crate::CircleError). A borrowed namespace with no state of
    /// its own.
    pub fn circles(&self) -> crate::Circles<'_> {
        crate::Circles::new(&self.circles)
    }

    // =========================================================================
    // Rows
    // =========================================================================

    #[cfg(test)]
    pub(crate) async fn create_test_store(
        &self,
        store_id: &str,
        signer: coven_keys::keys::UserKeypair,
        home: std::sync::Arc<coven_storage::cloud::test_utils::InMemoryCloudHome>,
    ) -> Result<
        std::sync::Arc<coven_replication::sync::test_helpers::TestStore>,
        coven_replication::sync::test_helpers::TestError,
    > {
        self.sync.create_test_store(store_id, signer, home).await
    }

    #[cfg(test)]
    pub(crate) async fn publish_test_store(
        &self,
        store: &coven_replication::sync::test_helpers::TestStore,
    ) -> Result<bool, coven_replication::sync::test_helpers::TestError> {
        self.sync.publish_test_store(store).await
    }

    #[cfg(test)]
    pub(crate) async fn pull_test_store(
        &self,
        store: &coven_replication::sync::test_helpers::TestStore,
    ) -> (
        std::collections::BTreeMap<String, u64>,
        coven_replication::sync::store::StorePullResult,
    ) {
        self.sync
            .pull_test_store(store)
            .await
            .expect("pull exact test Store")
    }

    #[cfg(test)]
    pub(crate) async fn store_write_partition_for_test(
        &self,
        write_id: &crate::WriteId,
    ) -> Result<Vec<u8>, coven_database::DbError> {
        self.rows.store_write_partition_for_test(write_id).await
    }

    #[cfg(test)]
    pub(crate) async fn write_blob_lease_count_for_test(
        &self,
        write_id: &crate::WriteId,
    ) -> Result<i64, coven_database::DbError> {
        self.rows.write_blob_lease_count_for_test(write_id).await
    }

    #[cfg(test)]
    pub(crate) async fn store_write_journal_counts_for_test(
        &self,
    ) -> Result<(i64, i64), coven_database::DbError> {
        self.rows.store_write_journal_counts_for_test().await
    }

    /// Count cleanup obligations for one blob in integration tests.
    #[cfg(any(test, feature = "test-utils"))]
    pub async fn cleanup_intent_count_for_test(
        &self,
        namespace: &str,
        blob_id: &str,
    ) -> Result<i64, coven_database::DbError> {
        self.rows
            .cleanup_intent_count_for_test(namespace, blob_id)
            .await
    }

    #[cfg(test)]
    pub(crate) async fn coven_table_exists_for_test(
        &self,
        table: coven_database::DatabaseTestTable,
    ) -> Result<bool, coven_database::DbError> {
        self.rows.coven_table_exists_for_test(table).await
    }

    #[cfg(test)]
    pub(crate) async fn install_store_write_failure_trigger_for_test(
        &self,
    ) -> Result<(), coven_database::DbError> {
        self.rows
            .install_store_write_failure_trigger_for_test()
            .await
    }

    #[cfg(test)]
    pub(crate) async fn remove_store_write_failure_trigger_for_test(
        &self,
    ) -> Result<(), coven_database::DbError> {
        self.rows
            .remove_store_write_failure_trigger_for_test()
            .await
    }

    #[cfg(test)]
    pub(crate) async fn write_blob_facts_for_test(
        &self,
        write_id: crate::WriteId,
    ) -> Result<String, coven_database::DbError> {
        self.rows.write_blob_facts_for_test(write_id).await
    }

    #[cfg(test)]
    pub(crate) async fn execute_sql_with_blob_staging_for_test(
        &self,
        blob_staging: Option<Box<dyn coven_database::AudienceBlobMoveStaging>>,
        sql: String,
    ) -> crate::CovenResult<crate::WriteReceipt<()>> {
        self.rows
            .execute_sql_with_blob_staging_for_test(blob_staging, sql)
            .await
    }

    #[cfg(test)]
    pub(crate) async fn latest_materialized_commit_coordinate_for_test(
        &self,
    ) -> Result<(String, u64), coven_database::DbError> {
        self.sync
            .latest_materialized_commit_coordinate_for_test()
            .await
    }

    #[cfg(test)]
    pub(crate) fn arm_pull_after_remote_commit_for_test(
        &self,
        device_id: String,
        sequence: u64,
    ) -> (
        std::sync::Arc<tokio::sync::Notify>,
        std::sync::Arc<tokio::sync::Notify>,
    ) {
        self.sync
            .arm_pull_after_remote_commit_for_test(device_id, sequence)
    }

    #[cfg(test)]
    pub(crate) async fn prepare_test_join_snapshot(
        &self,
        store: &coven_replication::sync::test_helpers::TestStore,
        owner: &coven_keys::keys::UserKeypair,
        snapshot_path: std::path::PathBuf,
    ) -> Result<(), coven_replication::sync::test_helpers::TestError> {
        self.sync
            .prepare_test_join_snapshot(store, owner, snapshot_path)
            .await
    }
}

#[cfg(test)]
#[path = "handle_tests.rs"]
mod tests;