heddle-thread-api 0.28.9

//! Local SQLite adapter. Blocking work stays off the stream runtime.
use std::{collections::BTreeSet, path::PathBuf, sync::Arc};

use heddle_object_model::object::{
    ContentHash,
    thread_replication::{Admission, ThreadFacet},
};
use objects::store::ObjectStore;
use repo::thread_replication::ThreadReplica;

use super::store::{ReceivedOperation, ReplicaStore};

mod hosted;
pub use hosted::HostedReplica;

#[derive(Debug, thiserror::Error)]
pub enum Error {
    #[error(transparent)]
    Store(#[from] repo::thread_replication::Error),
    #[error("local replica worker: {0}")]
    Worker(#[from] tokio::task::JoinError),
    #[error("HYBRID original requires independently selected hosted trust")]
    HostedTrustRequired,
}
pub struct LocalReplica<S> {
    replica: ThreadReplica,
    objects: Arc<S>,
    authority_home: Option<PathBuf>,
}
impl<S> Clone for LocalReplica<S> {
    fn clone(&self) -> Self {
        Self {
            replica: self.replica.clone(),
            objects: self.objects.clone(),
            authority_home: self.authority_home.clone(),
        }
    }
}
impl<S: ObjectStore + Send + Sync + 'static> LocalReplica<S> {
    pub fn new(replica: ThreadReplica, objects: Arc<S>) -> Self {
        Self {
            replica,
            objects,
            authority_home: None,
        }
    }
    /// Use independently enrolled local account authority for original metadata
    /// authors. Without this binding, new metadata admission fails closed.
    pub fn with_device_authority(mut self, home: PathBuf) -> Self {
        self.authority_home = Some(home);
        self
    }
    async fn execute<T: Send + 'static>(
        &self,
        operation: impl FnOnce(&ThreadReplica, &S) -> repo::thread_replication::Result<T>
        + Send
        + 'static,
    ) -> Result<T, Error> {
        let local = self.clone();
        Ok(
            tokio::task::spawn_blocking(move || operation(&local.replica, &local.objects))
                .await??,
        )
    }
}
impl<S: ObjectStore + Send + Sync + 'static> ReplicaStore for LocalReplica<S> {
    type Error = Error;
    fn thread_id(&self) -> ContentHash {
        self.replica.thread_id()
    }
    async fn generation(&self) -> Result<i64, Error> {
        self.execute(|replica, _| replica.generation()).await
    }
    async fn sharing(&self, destination: [u8; 32]) -> Result<BTreeSet<ThreadFacet>, Error> {
        self.execute(move |replica, _| replica.sharing(&destination).map(|(facets, _)| facets))
            .await
    }
    async fn frontier_page(
        &self,
        facet: ThreadFacet,
        after: Option<ContentHash>,
        limit: usize,
    ) -> Result<Vec<ContentHash>, Error> {
        self.execute(move |replica, _| replica.frontier_page(facet, after, limit))
            .await
    }
    async fn operation(
        &self,
        id: ContentHash,
    ) -> Result<Option<(ReceivedOperation, Admission)>, Error> {
        let stored = self
            .execute(move |replica, _| replica.operation_with_authority_admission(&id))
            .await?;
        let Some(stored) = stored else {
            return Ok(None);
        };
        if stored.authority_admission.is_some()
            || self
                .execute(|replica, _| replica.hybrid_import_bundle())
                .await?
                .is_some()
        {
            return Err(Error::HostedTrustRequired);
        }
        Ok(Some((
            ReceivedOperation {
                native_authority: None,
                original: stored.original,
                authority_admission: None,
                import_authority: None,
            },
            stored.status,
        )))
    }
    async fn receive(&self, received: ReceivedOperation) -> Result<Admission, Error> {
        if received.import_authority.is_some() || received.authority_admission.is_some() {
            return Err(Error::HostedTrustRequired);
        }
        let authority_home = self.authority_home.clone();
        self.execute(move |replica, objects| {
            let operation = received.original;
            replica.receive(&operation, objects, |native| {
                use heddle_object_model::object::thread_replication::ThreadOperationBody;
                if !matches!(native.body, ThreadOperationBody::Metadata(_)) && native.source_author()?.is_none() {
                    return Ok(());
                }
                // Durable original-author receipt remains valid while causal
                // parents arrive later. Neither the envelope nor claimed time
                // can synthesize the atomically retained admission marker.
                if replica.original_authority_admitted(&operation)? {
                    return Ok(());
                }
                let genesis = replica.genesis()?;
                if matches!(native.source_author()?, Some(heddle_object_model::object::thread_replication::SourceAuthor::LocalKey)) {
                    return replica.verify_local_source_owner(native);
                }
                let home = authority_home.as_ref().ok_or_else(|| {
                    repo::thread_replication::Error::Invalid(
                        "original operation requires independently enrolled account authority"
                            .into(),
                    )
                })?;
                let now = chrono::Utc::now().timestamp();
                let authority = repo::device_authority::load(home, now).map_err(authority_error)?;
                if let ThreadOperationBody::Metadata(bytes) = &native.body {
                    heddle_object_model::object::thread_replication::metadata::ThreadControl::decode(bytes)?.validate_parents(&genesis, &[])?;
                }
                let spool = genesis.spool.parse().map_err(authority_error)?;
                let registered =
                    repo::device_catalog::load(home, spool).map_err(authority_error)?;
                if native.source_author()?.is_some() {
                    replica.verify_source_authority(native, &authority, &registered.capability_path, now)
                } else {
                    repo::thread_replication::metadata::verify_control_authority(native, &authority, &registered.capability_path, now)
                }

            })
        })
        .await
    }
    async fn remember_peer_heads(
        &self,
        peer: [u8; 32],
        heads: Vec<(ThreadFacet, ContentHash)>,
    ) -> Result<(), Error> {
        self.execute(move |replica, _| replica.remember_peer_heads(peer, &heads))
            .await
    }
    async fn record_peer_receipt(
        &self,
        peer: [u8; 32],
        id: ContentHash,
        admission: Admission,
    ) -> Result<(), Error> {
        self.execute(move |replica, _| replica.record_peer_receipt(peer, id, &admission))
            .await
    }
    async fn settled_peer_heads(
        &self,
        peer: [u8; 32],
        facets: BTreeSet<ThreadFacet>,
        limit: usize,
    ) -> Result<Vec<(ContentHash, Admission)>, Error> {
        self.execute(move |replica, _| replica.settled_peer_heads(peer, &facets, limit))
            .await
    }
    async fn needed_from_peer(
        &self,
        peer: [u8; 32],
        facets: BTreeSet<ThreadFacet>,
        limit: usize,
    ) -> Result<Vec<ContentHash>, Error> {
        self.execute(move |replica, _| replica.needed_from_peer(peer, &facets, limit))
            .await
    }
}

fn authority_error(error: impl std::fmt::Display) -> repo::thread_replication::Error {
    repo::thread_replication::Error::Invalid(error.to_string())
}