polyc-state-connect 2026.8.3

//! Capability-specific durable-ingress client.

use connectrpc::client::{ClientConfig, ClientTransport};
use polyc_proto::proto::polychrome::state::v1 as pb;
use polyc_state::{
    error::StateError,
    id::CommandId,
    ingress::{
        ClaimIngress, ClaimedIngress, InboxDepth, IngressItem, ReadInboxDepth, ReadIngressItem,
        ReceiveIngress, RecordIngressDecision,
    },
    receipt::Receipt,
};

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

use super::wire::{
    claim_to_wire, decide_to_wire, depth_from_wire, depth_request_to_wire, item_from_wire,
    item_request_to_wire, receive_to_wire, scope_to_wire,
};

/// Client bound to the durable-ingress capability.
pub struct IngressClient<T> {
    inner: pb::StateIngressServiceClient<T>,
}

impl<T> IngressClient<T>
where
    T: ClientTransport,
    <T::ResponseBody as connectrpc::http_body::Body>::Error: std::fmt::Display,
{
    /// Builds a durable-ingress client.
    #[must_use]
    pub fn new(transport: T, config: ClientConfig) -> Self {
        Self {
            inner: pb::StateIngressServiceClient::new(
                transport,
                config.with_default_max_message_size(MAX_INGRESS_WIRE_MESSAGE_BYTES),
            ),
        }
    }

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

    /// Durably receives one external message.
    ///
    /// # Errors
    ///
    /// Returns a typed State or transport failure.
    pub async fn receive(
        &self,
        declared: &DeclaredCall,
        command: &ReceiveIngress,
    ) -> Result<Receipt, StateError> {
        let request = pb::ReceiveIngressRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            command: buffa::MessageField::some(receive_to_wire(command)),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .receive_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(|| {
            malformed(
                "receipt",
                "a successful receive returns its durable receipt",
            )
        })?)
        .map(Kernel::into_inner)
    }

    /// Claims the next available item, if the inbox has one.
    ///
    /// # Errors
    ///
    /// Returns a typed State or transport failure.
    pub async fn claim(
        &self,
        declared: &DeclaredCall,
        command: &ClaimIngress,
    ) -> Result<Option<ClaimedIngress>, StateError> {
        let request = pb::ClaimIngressRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            command: buffa::MessageField::some(claim_to_wire(command)),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .claim_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        match (reply.receipt.into_option(), reply.item.into_option()) {
            (None, None) => Ok(None),
            (Some(receipt), Some(item)) => Ok(Some(ClaimedIngress::new(
                Kernel::<Receipt>::try_from(receipt)?.into_inner(),
                item_from_wire(item)?,
            ))),
            _ => Err(malformed(
                "claim",
                "a claim reply carries both receipt and item, or neither",
            )),
        }
    }

    /// Records the terminal policy decision for one claimed item.
    ///
    /// # Errors
    ///
    /// Returns a typed State or transport failure.
    pub async fn decide(
        &self,
        declared: &DeclaredCall,
        command: &RecordIngressDecision,
    ) -> Result<Receipt, StateError> {
        let request = pb::DecideIngressRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            command: buffa::MessageField::some(decide_to_wire(command)),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .decide_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(|| {
            malformed(
                "receipt",
                "a successful decision returns its durable receipt",
            )
        })?)
        .map(Kernel::into_inner)
    }

    /// Reads one inbox's bounded depth counters.
    ///
    /// # Errors
    ///
    /// Returns a typed State or transport failure.
    pub async fn depth(
        &self,
        declared: &DeclaredCall,
        request: &ReadInboxDepth,
    ) -> Result<InboxDepth, StateError> {
        let (scope, edge) = depth_request_to_wire(request);
        let request = pb::GetIngressDepthRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            scope: buffa::MessageField::some(scope),
            edge,
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .get_depth_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        depth_from_wire(
            reply.depth.into_option().ok_or_else(|| {
                malformed("depth", "a successful depth read returns its counters")
            })?,
        )
    }

    /// Reads one named ingress item.
    ///
    /// # Errors
    ///
    /// Returns a typed State or transport failure.
    pub async fn item(
        &self,
        declared: &DeclaredCall,
        request: &ReadIngressItem,
    ) -> Result<Option<IngressItem>, StateError> {
        let (scope, source) = item_request_to_wire(request);
        let request = pb::GetIngressItemRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            scope: buffa::MessageField::some(scope),
            source: buffa::MessageField::some(source),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .get_item_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        reply.item.into_option().map(item_from_wire).transpose()
    }

    /// Settles an ambiguous ingress command from its durable receipt.
    ///
    /// # Errors
    ///
    /// Returns a typed State or transport failure.
    pub async fn committed_receipt(
        &self,
        declared: &DeclaredCall,
        scope: &polyc_state::command::CommandScope,
        command_id: &CommandId,
    ) -> Result<Option<Receipt>, StateError> {
        let request = pb::GetIngressReceiptRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            scope: buffa::MessageField::some(scope_to_wire(scope)),
            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(|receipt| Kernel::<Receipt>::try_from(receipt).map(Kernel::into_inner))
            .transpose()
    }
}

fn malformed(field: &str, reason: &str) -> StateError {
    StateError::Malformed {
        field: field.to_owned(),
        reason: reason.to_owned(),
    }
}