polyc-state-connect 2026.8.3

//! Capability-specific persona-memory snapshot client.
//!
//! Async end to end, like every typed State family: the generated client is
//! async and every method here awaits it. Nothing in this family reaches for a
//! blocking bridge or a blocking pool.
//!
//! Every call carries the caller's remaining budget. A snapshot write is an
//! optimization that must never hold up the append it follows, so a caller
//! that supplies a real deadline is what keeps a slow State from turning into
//! a slow memory write.

use connectrpc::client::{ClientConfig, ClientTransport};
use polyc_proto::proto::polychrome::state::v1 as pb;
use polyc_state::{
    error::StateError,
    id::NamespaceId,
    persona_memory::{
        MemoryPartitionId, MemorySnapshot,
        store::{SnapshotCommand, SnapshotFact},
    },
    receipt::Receipt,
    revision::Revision,
};

use crate::{
    MAX_PERSONA_MEMORY_WIRE_MESSAGE_BYTES,
    error::{TransportFallback, from_connect_error},
    trace::bounded_traced_options,
    wire::{DeclaredCall, Kernel},
};

use super::wire::{metadata_to_wire, operation_to_wire};

/// Client bound to one snapshot namespace and no other State capability.
pub struct PersonaMemorySnapshotClient<T> {
    inner: pb::StatePersonaMemorySnapshotServiceClient<T>,
    namespace: NamespaceId,
}

impl<T> PersonaMemorySnapshotClient<T>
where
    T: ClientTransport,
    <T::ResponseBody as connectrpc::http_body::Body>::Error: std::fmt::Display,
{
    /// Builds a namespace-bound client.
    #[must_use]
    pub fn new(transport: T, config: ClientConfig, namespace: NamespaceId) -> Self {
        Self {
            inner: pb::StatePersonaMemorySnapshotServiceClient::new(
                transport,
                config.with_default_max_message_size(MAX_PERSONA_MEMORY_WIRE_MESSAGE_BYTES),
            ),
            namespace,
        }
    }

    fn fallback(attempted: usize) -> TransportFallback {
        TransportFallback::new(
            polyc_state::persona_memory::store::family(),
            MAX_PERSONA_MEMORY_WIRE_MESSAGE_BYTES as u64,
            attempted as u64,
        )
    }

    /// Stores or discards one partition's snapshot.
    ///
    /// # Errors
    ///
    /// Returns a typed State or transport failure. Every one of them is a lost
    /// optimization rather than a lost fact — the journal the snapshot derives
    /// from is untouched by anything that happens here.
    pub async fn put(
        &self,
        declared: &DeclaredCall,
        command: &SnapshotCommand,
    ) -> Result<Receipt, StateError> {
        if command.metadata().scope().namespace() != &self.namespace {
            return Err(StateError::Denied {
                family: polyc_state::persona_memory::store::family(),
            });
        }
        let request = pb::PutStatePersonaMemorySnapshotRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            metadata: buffa::MessageField::some(metadata_to_wire(command)),
            partition: command.partition().as_str().to_owned(),
            operation: buffa::MessageField::some(operation_to_wire(command.operation())),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .put_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        Kernel::<Receipt>::try_from(reply.receipt.into_option().ok_or_else(|| {
            StateError::Malformed {
                field: "receipt".into(),
                reason: "a successful snapshot mutation returns a receipt".into(),
            }
        })?)
        .map(Kernel::into_inner)
    }

    /// Reads one partition's stored snapshot.
    ///
    /// [`None`] is the honest answer for a partition nobody has snapshotted,
    /// and it means a full replay rather than a failure.
    ///
    /// # Errors
    ///
    /// Returns a typed State or transport failure.
    pub async fn get(
        &self,
        declared: &DeclaredCall,
        partition: &MemoryPartitionId,
    ) -> Result<SnapshotFact<Option<MemorySnapshot>>, StateError> {
        let request = pb::GetStatePersonaMemorySnapshotRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: self.namespace.as_str().to_owned(),
            partition: partition.as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .get_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        let stored = reply
            .record
            .map(|bytes| MemorySnapshot::decode(&bytes))
            .transpose()?;
        Ok(SnapshotFact::observed(
            stored,
            Revision::new(reply.snapshot_revision),
            reply.entry_revision.map(Revision::new),
        ))
    }
}