polyc-state-connect 2026.8.3

//! Capability-specific durable-session client.

use buffa::EnumValue;
use connectrpc::client::{ClientConfig, ClientTransport};
use polyc_proto::proto::polychrome::state::v1 as pb;
use polyc_state::{
    error::StateError,
    id::{CommandId, NamespaceId},
    receipt::Receipt,
    revision::Revision,
    sessions::{
        ExpirationShard, RefreshPreflight, SessionCommand, SessionFact, SessionFamily,
        SessionPrincipal,
    },
    versioned::authority::{AuthorizationEpoch, SessionAuthorization, SessionId},
};

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

use super::wire::{
    expiration_from_wire, family_from_wire, metadata_to_wire, operation_to_wire, principal_to_wire,
    record_from_wire,
};

/// Client bound to one State session namespace.
pub struct SessionClient<T> {
    inner: pb::StateSessionServiceClient<T>,
    namespace: NamespaceId,
}

impl<T> SessionClient<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::StateSessionServiceClient::new(
                transport,
                config.with_default_max_message_size(MAX_WIRE_MESSAGE_BYTES),
            ),
            namespace,
        }
    }

    fn fallback(attempted: usize) -> TransportFallback {
        TransportFallback::new(
            polyc_state::versioned::family(),
            MAX_WIRE_MESSAGE_BYTES as u64,
            attempted as u64,
        )
    }

    /// Commits one typed session mutation.
    ///
    /// # Errors
    ///
    /// Returns a typed State or transport failure.
    pub async fn transact(
        &self,
        declared: &DeclaredCall,
        command: &SessionCommand,
    ) -> Result<Receipt, StateError> {
        if command.metadata().scope().namespace() != &self.namespace {
            return Err(StateError::Denied {
                family: polyc_state::versioned::family(),
            });
        }
        let request = pb::TransactSessionRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            metadata: buffa::MessageField::some(metadata_to_wire(command)),
            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
            .transact_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 session mutation returns a receipt".into(),
            }
        })?)
        .map(Kernel::into_inner)
    }

    /// Reads one long-lived family.
    ///
    /// # Errors
    ///
    /// Returns a typed State or transport failure.
    pub async fn family(
        &self,
        declared: &DeclaredCall,
        id: &SessionId,
    ) -> Result<SessionFact<Option<SessionFamily>>, StateError> {
        let request = pb::GetSessionFamilyRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: self.namespace.as_str().to_owned(),
            family_id: id.as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .get_family_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        Ok(SessionFact::observed(
            reply
                .family
                .into_option()
                .map(family_from_wire)
                .transpose()?,
            Revision::new(reply.snapshot_revision),
            reply.entry_revision.map(Revision::new),
        ))
    }

    /// Classifies a verified family grant without writing.
    ///
    /// # Errors
    ///
    /// Returns a typed State or transport failure.
    pub async fn inspect_refresh(
        &self,
        declared: &DeclaredCall,
        id: &SessionId,
        generation: u64,
        now_ms: u64,
    ) -> Result<RefreshPreflight, StateError> {
        let request = pb::InspectSessionRefreshRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: self.namespace.as_str().to_owned(),
            family_id: id.as_str().to_owned(),
            presented_generation: generation,
            now_ms,
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .inspect_refresh_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        let classification = match reply.classification {
            EnumValue::Known(value) => value,
            EnumValue::Unknown(_) => return Err(malformed_class()),
        };
        let family = reply.family.into_option();
        if classification == pb::SessionRefreshClass::UnknownFamily {
            if family.is_some() || reply.entry_revision.is_some() {
                return Err(malformed_class());
            }
            return Ok(RefreshPreflight::UnknownFamily(SessionFact::observed(
                None,
                Revision::new(reply.snapshot_revision),
                None,
            )));
        }
        let family = family_from_wire(family.ok_or_else(malformed_class)?)?;
        let fact = SessionFact::observed(
            family,
            Revision::new(reply.snapshot_revision),
            reply.entry_revision.map(Revision::new),
        );
        match classification {
            pb::SessionRefreshClass::Rotate => Ok(RefreshPreflight::Rotate(fact)),
            pb::SessionRefreshClass::Replay => Ok(RefreshPreflight::Replay(fact)),
            pb::SessionRefreshClass::ReuseDetected => Ok(RefreshPreflight::ReuseDetected(fact)),
            pb::SessionRefreshClass::Expired => Ok(RefreshPreflight::Expired(fact)),
            pb::SessionRefreshClass::Unspecified | pb::SessionRefreshClass::UnknownFamily => {
                Err(malformed_class())
            }
        }
    }

    /// Reads one bounded expiry shard.
    ///
    /// # Errors
    ///
    /// Returns a typed State or transport failure.
    pub async fn expiration_shard(
        &self,
        declared: &DeclaredCall,
        shard: u16,
    ) -> Result<SessionFact<ExpirationShard>, StateError> {
        let request = pb::GetSessionExpirationShardRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: self.namespace.as_str().to_owned(),
            shard: u32::from(shard),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .get_expiration_shard_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        Ok(SessionFact::observed(
            expiration_from_wire(reply.expiration.into_option().ok_or_else(|| {
                StateError::Malformed {
                    field: "expiration".into(),
                    reason: "expiry reply carries a shard".into(),
                }
            })?)?,
            Revision::new(reply.snapshot_revision),
            reply.entry_revision.map(Revision::new),
        ))
    }

    /// Reads bearer authorization at one coherent revision.
    ///
    /// # Errors
    ///
    /// Returns a typed State or transport failure.
    pub async fn bearer_authorization(
        &self,
        declared: &DeclaredCall,
        principal: &SessionPrincipal,
        session: &SessionId,
    ) -> Result<SessionFact<SessionAuthorization>, StateError> {
        let request = pb::GetBearerAuthorizationRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: self.namespace.as_str().to_owned(),
            principal: buffa::MessageField::some(principal_to_wire(principal)),
            session: session.as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .get_bearer_authorization_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        let record =
            record_from_wire(
                &reply
                    .record
                    .into_option()
                    .ok_or_else(|| StateError::Malformed {
                        field: "record".into(),
                        reason: "authorization reply carries a bearer record".into(),
                    })?,
            );
        Ok(SessionFact::observed(
            SessionAuthorization::new(record, AuthorizationEpoch::new(reply.current_epoch)),
            Revision::new(reply.snapshot_revision),
            reply.entry_revision.map(Revision::new),
        ))
    }

    /// Reads one raw bearer record for bounded expiry reaping.
    ///
    /// Authorization paths must use [`Self::bearer_authorization`].
    ///
    /// # Errors
    ///
    /// Returns a typed State failure when the record cannot be read or decoded.
    pub async fn bearer_record(
        &self,
        declared: &DeclaredCall,
        session: &SessionId,
    ) -> Result<SessionFact<polyc_state::versioned::authority::SessionRecord>, StateError> {
        let request = pb::GetBearerRecordRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: self.namespace.as_str().to_owned(),
            session: session.as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .get_bearer_record_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        let record =
            record_from_wire(
                &reply
                    .record
                    .into_option()
                    .ok_or_else(|| StateError::Malformed {
                        field: "record".into(),
                        reason: "bearer-record reply carries a record".into(),
                    })?,
            );
        Ok(SessionFact::observed(
            record,
            Revision::new(reply.snapshot_revision),
            reply.entry_revision.map(Revision::new),
        ))
    }

    /// Reads one principal epoch.
    ///
    /// # Errors
    ///
    /// Returns a typed State or transport failure.
    pub async fn authorization_epoch(
        &self,
        declared: &DeclaredCall,
        principal: &SessionPrincipal,
    ) -> Result<SessionFact<AuthorizationEpoch>, StateError> {
        let request = pb::GetSessionAuthorizationEpochRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: self.namespace.as_str().to_owned(),
            principal: buffa::MessageField::some(principal_to_wire(principal)),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .get_authorization_epoch_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        Ok(SessionFact::observed(
            AuthorizationEpoch::new(reply.epoch),
            Revision::new(reply.snapshot_revision),
            reply.entry_revision.map(Revision::new),
        ))
    }

    /// Retrieves one committed mutation receipt.
    ///
    /// # Errors
    ///
    /// Returns a typed State or transport failure.
    pub async fn committed_receipt(
        &self,
        declared: &DeclaredCall,
        command_id: &CommandId,
    ) -> Result<Option<Receipt>, StateError> {
        let request = pb::GetSessionReceiptRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: self.namespace.as_str().to_owned(),
            command_id: command_id.as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .get_receipt_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        reply
            .receipt
            .into_option()
            .map(|value| Kernel::<Receipt>::try_from(value).map(Kernel::into_inner))
            .transpose()
    }
}

fn malformed_class() -> StateError {
    StateError::Malformed {
        field: "classification".into(),
        reason: "refresh reply carries one coherent known classification".into(),
    }
}