polyc-state-connect 2026.10.2

//! Capability-specific credential metadata client.

use connectrpc::client::{ClientConfig, ClientTransport};
use polyc_proto::proto::polychrome::state::v1 as pb;
use polyc_state::{
    credentials::{
        CredentialChangePage, CredentialCommand, CredentialFact, CredentialIndex, CredentialPage,
        CredentialRecord, CredentialRequestRecord,
    },
    error::StateError,
    id::{CommandId, NamespaceId},
    page::{Cursor, PageCompleteness},
    receipt::Receipt,
    revision::{JournalPosition, PartitionIncarnation, Revision},
};

use crate::{
    MAX_CREDENTIAL_WIRE_MESSAGE_BYTES,
    error::{TransportFallback, from_connect_error},
    trace::bounded_traced_options,
    wire::{DeclaredCall, Kernel, fixed_bytes, malformed},
};

use super::wire::{
    change_from_wire, index_from_wire, metadata_to_wire, operation_to_wire, record_from_wire,
    request_from_wire, request_to_wire,
};

/// Client bound to one credential authority namespace.
pub struct CredentialClient<T> {
    inner: pb::StateCredentialServiceClient<T>,
    namespace: NamespaceId,
}

impl<T> CredentialClient<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::StateCredentialServiceClient::new(
                transport,
                config.with_default_max_message_size(MAX_CREDENTIAL_WIRE_MESSAGE_BYTES),
            ),
            namespace,
        }
    }

    fn fallback(declared: &DeclaredCall, attempted: usize) -> TransportFallback {
        TransportFallback::new(
            polyc_state::versioned::family(),
            MAX_CREDENTIAL_WIRE_MESSAGE_BYTES as u64,
            attempted as u64,
            declared.budget,
        )
    }

    /// Commits one typed credential transition.
    ///
    /// # Errors
    ///
    /// Returns the State authority's typed refusal or a bounded transport failure.
    pub async fn transact(
        &self,
        declared: &DeclaredCall,
        command: &CredentialCommand,
    ) -> Result<Receipt, StateError> {
        if command.metadata().scope().namespace() != &self.namespace {
            return Err(StateError::Denied {
                family: polyc_state::versioned::family(),
            });
        }
        let request = pb::TransactStateCredentialRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            metadata: buffa::MessageField::some(metadata_to_wire(command)),
            operation: buffa::MessageField::some(operation_to_wire(command.operation())),
            request_record: command
                .request()
                .map_or_else(buffa::MessageField::none, |request| {
                    buffa::MessageField::some(request_to_wire(request))
                }),
            __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(declared, attempted)))?
            .into_owned();
        Kernel::<Receipt>::try_from(reply.receipt.into_option().ok_or_else(|| {
            StateError::Malformed {
                field: "receipt".into(),
                reason: "successful credential mutation returns a receipt".into(),
            }
        })?)
        .map(Kernel::into_inner)
    }

    /// Reads one current record, including a tombstone.
    ///
    /// # Errors
    ///
    /// Returns the State authority's typed refusal or a bounded transport failure.
    pub async fn credential(
        &self,
        declared: &DeclaredCall,
        credential_id: &str,
    ) -> Result<CredentialFact<Option<CredentialRecord>>, StateError> {
        let request = pb::GetStateCredentialRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: self.namespace.as_str().to_owned(),
            credential_id: credential_id.to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .get_credential_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(declared, attempted)))?
            .into_owned();
        Ok(CredentialFact::observed(
            reply
                .record
                .into_option()
                .map(record_from_wire)
                .transpose()?,
            Revision::new(reply.snapshot_revision),
            reply.entry_revision.map(Revision::new),
        ))
    }

    /// Reads one durable public lifecycle request result.
    ///
    /// # Errors
    ///
    /// Returns the State authority's typed refusal or a bounded transport failure.
    pub async fn request(
        &self,
        declared: &DeclaredCall,
        operation_id: &str,
    ) -> Result<CredentialFact<Option<CredentialRequestRecord>>, StateError> {
        let request = pb::GetStateCredentialRequestRecordRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: self.namespace.as_str().to_owned(),
            operation_id: operation_id.to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .get_request_record_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(declared, attempted)))?
            .into_owned();
        Ok(CredentialFact::observed(
            reply
                .request_record
                .into_option()
                .map(request_from_wire)
                .transpose()?,
            Revision::new(reply.snapshot_revision),
            reply.entry_revision.map(Revision::new),
        ))
    }

    /// Reads the canonical id index.
    ///
    /// # Errors
    ///
    /// Returns the State authority's typed refusal or a bounded transport failure.
    pub async fn index(
        &self,
        declared: &DeclaredCall,
    ) -> Result<CredentialFact<CredentialIndex>, StateError> {
        let request = pb::GetStateCredentialIndexRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: self.namespace.as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .get_index_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(declared, attempted)))?
            .into_owned();
        Ok(CredentialFact::observed(
            index_from_wire(
                reply
                    .index
                    .into_option()
                    .ok_or_else(|| StateError::Malformed {
                        field: "index".into(),
                        reason: "credential index reply carries a value".into(),
                    })?,
            ),
            Revision::new(reply.snapshot_revision),
            reply.entry_revision.map(Revision::new),
        ))
    }

    /// Reads one bounded page.
    ///
    /// # Errors
    ///
    /// Returns the State authority's typed refusal or a bounded transport failure.
    pub async fn list(
        &self,
        declared: &DeclaredCall,
        after: Option<&str>,
        limit: u32,
    ) -> Result<CredentialPage, StateError> {
        let request = pb::ListStateCredentialsRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: self.namespace.as_str().to_owned(),
            after: after.map(str::to_owned),
            limit,
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .list_credentials_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(declared, attempted)))?
            .into_owned();
        Ok(CredentialPage::observed(
            reply
                .records
                .into_iter()
                .map(record_from_wire)
                .collect::<Result<_, _>>()?,
            reply.next_after,
            Revision::new(reply.snapshot_revision),
        ))
    }

    /// Reads one bounded, ordered page of lifecycle changes.
    ///
    /// # Errors
    ///
    /// Returns the State authority's typed refusal or a bounded transport
    /// failure. A position the authority never issued is refused there, and
    /// this neither retries it as a fresh read nor clamps it.
    pub async fn changes(
        &self,
        declared: &DeclaredCall,
        after: Option<JournalPosition>,
        expected_lineage: Option<PartitionIncarnation>,
        limit: u32,
    ) -> Result<CredentialChangePage, StateError> {
        let request = pb::ListStateCredentialChangesRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: self.namespace.as_str().to_owned(),
            after: after.map(JournalPosition::get),
            expected_lineage: expected_lineage.map(|l| l.as_bytes().to_vec()),
            limit,
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .list_changes_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(declared, attempted)))?
            .into_owned();
        let completeness = match reply.completeness {
            buffa::EnumValue::Known(pb::PageCompleteness::Complete) => PageCompleteness::Complete,
            buffa::EnumValue::Known(pb::PageCompleteness::Truncated) => PageCompleteness::Truncated,
            buffa::EnumValue::Known(pb::PageCompleteness::Unspecified)
            | buffa::EnumValue::Unknown(_) => {
                return Err(malformed(
                    "page.completeness",
                    "a lifecycle page declares whether the bound cut it short",
                ));
            }
        };
        Ok(CredentialChangePage::observed(
            reply
                .changes
                .into_iter()
                .map(change_from_wire)
                .collect::<Result<_, _>>()?,
            reply
                .next_after
                .map(|position| Cursor::at(JournalPosition::new(position))),
            completeness,
            Revision::new(reply.frontier),
            PartitionIncarnation::from_bytes(fixed_bytes::<32>("lineage", &reply.lineage)?),
        ))
    }

    /// Retrieves a prior durable receipt.
    ///
    /// # Errors
    ///
    /// Returns the State authority's typed refusal or a bounded transport failure.
    pub async fn committed_receipt(
        &self,
        declared: &DeclaredCall,
        command_id: &CommandId,
    ) -> Result<Option<Receipt>, StateError> {
        let request = pb::GetStateCredentialReceiptRequest {
            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(declared, attempted)))?
            .into_owned();
        reply
            .receipt
            .into_option()
            .map(|value| Kernel::<Receipt>::try_from(value).map(Kernel::into_inner))
            .transpose()
    }
}