polyc-state-connect 2026.8.3

//! Capability-specific client for Query's audit trail.

use connectrpc::client::{ClientConfig, ClientTransport};
use polyc_proto::proto::polychrome::state::v1 as pb;
use polyc_state::{
    error::StateError,
    page::Page,
    query_audit::{
        AuditPhase, BeginQueryAudit, CompleteQueryAudit, ExecutionPermit, ListUnmatchedIntents,
        QueryAudit, QueryAuditError, QueryId, ReadQueryAudit,
    },
    receipt::Receipt,
};

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

/// A client that can write and inspect query audit records, and nothing else.
pub struct QueryAuditClient<T> {
    inner: pb::StateQueryAuditServiceClient<T>,
}

impl<T> QueryAuditClient<T>
where
    T: ClientTransport,
    <T::ResponseBody as connectrpc::http_body::Body>::Error: std::fmt::Display,
{
    /// Builds a client under the query-audit wire bound.
    pub fn new(transport: T, config: ClientConfig) -> Self {
        Self {
            inner: pb::StateQueryAuditServiceClient::new(
                transport,
                config.with_default_max_message_size(MAX_WIRE_MESSAGE_BYTES),
            ),
        }
    }

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

    /// Durably records an intent and returns the only permit under which that
    /// query may execute.
    ///
    /// # Errors
    ///
    /// Returns the exact query-audit outcome carried by State. Any error means
    /// no permit and therefore no authorized execution.
    pub async fn begin(
        &self,
        declared: &DeclaredCall,
        command: &BeginQueryAudit,
    ) -> Result<ExecutionPermit, QueryAuditError> {
        let metadata = command.metadata();
        let request = pb::BeginQueryAuditRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            query: command.query().as_str().to_owned(),
            namespace: command.namespace().as_str().to_owned(),
            requester: command.requester().as_str().to_owned(),
            shape: command.shape().as_bytes().to_vec(),
            digest: metadata.digest().as_bytes().to_vec(),
            purpose: metadata.envelope().purpose().as_str().to_owned(),
            command_audience: metadata.envelope().audience().as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .begin_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        let receipt = receipt(reply.receipt)?;
        Ok(ExecutionPermit::granted(
            command.query().clone(),
            command.namespace().clone(),
            receipt,
        ))
    }

    /// Durably records the outcome of an already-permitted query.
    ///
    /// # Errors
    ///
    /// Returns the exact family outcome. A dropped or ambiguous call is
    /// settled by retrying this same command or calling [`Self::receipt`].
    pub async fn complete(
        &self,
        declared: &DeclaredCall,
        command: &CompleteQueryAudit,
    ) -> Result<Receipt, QueryAuditError> {
        let metadata = command.metadata();
        let request = pb::CompleteQueryAuditRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            query: command.query().as_str().to_owned(),
            namespace: command.namespace().as_str().to_owned(),
            completion: buffa::MessageField::some(pb::QueryCompletion::from(Kernel(
                command.completion(),
            ))),
            digest: metadata.digest().as_bytes().to_vec(),
            purpose: metadata.envelope().purpose().as_str().to_owned(),
            command_audience: metadata.envelope().audience().as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .complete_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        receipt(reply.receipt)
    }

    /// Reads one query's whole audit trail.
    ///
    /// # Errors
    ///
    /// Returns the exact family outcome or a malformed-response outcome if a
    /// successful listener sent an invalid record.
    pub async fn audit(
        &self,
        declared: &DeclaredCall,
        request: &ReadQueryAudit,
    ) -> Result<Option<QueryAudit>, QueryAuditError> {
        let wire = pb::GetQueryAuditRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            query: request.query().as_str().to_owned(),
            namespace: request.namespace().as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&wire) as usize;
        let reply = self
            .inner
            .get_audit_with_options(wire, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        reply
            .audit
            .into_option()
            .map(|audit| {
                Kernel::<QueryAudit>::try_from(audit)
                    .map(Kernel::into_inner)
                    .map_err(QueryAuditError::from)
            })
            .transpose()
    }

    /// Lists one bounded page of unmatched intents.
    ///
    /// # Errors
    ///
    /// Returns the exact family outcome, including the page bound.
    pub async fn unmatched(
        &self,
        declared: &DeclaredCall,
        request: &ListUnmatchedIntents,
    ) -> Result<Page<QueryAudit>, QueryAuditError> {
        let wire = pb::ListUnmatchedQueryAuditsRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: request.namespace().as_str().to_owned(),
            page: buffa::MessageField::some(Kernel(request.page()).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&wire) as usize;
        let reply = self
            .inner
            .list_unmatched_with_options(wire, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        let page = reply.page.into_option().ok_or_else(|| {
            QueryAuditError::from(StateError::Malformed {
                field: "page".to_owned(),
                reason: "a successful listing carries its page".to_owned(),
            })
        })?;
        Kernel::<Page<QueryAudit>>::try_from(page)
            .map(Kernel::into_inner)
            .map_err(QueryAuditError::from)
    }

    /// Retrieves the durable receipt for one phase.
    ///
    /// # Errors
    ///
    /// Returns the exact family outcome. Absence means that phase never
    /// committed, not that State could not determine its outcome.
    pub async fn receipt(
        &self,
        declared: &DeclaredCall,
        query: &QueryId,
        namespace: &polyc_state::id::NamespaceId,
        phase: AuditPhase,
    ) -> Result<Option<Receipt>, QueryAuditError> {
        let wire = pb::GetQueryAuditReceiptRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            query: query.as_str().to_owned(),
            namespace: namespace.as_str().to_owned(),
            completion: matches!(phase, AuditPhase::Completion),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&wire) as usize;
        let reply = self
            .inner
            .get_receipt_with_options(wire, 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)
                    .map_err(QueryAuditError::from)
            })
            .transpose()
    }
}

fn receipt(field: impl Into<Option<pb::Receipt>>) -> Result<Receipt, QueryAuditError> {
    let value = field.into().ok_or_else(|| {
        QueryAuditError::from(StateError::Malformed {
            field: "receipt".to_owned(),
            reason: "a successful audit write carries its durable receipt".to_owned(),
        })
    })?;
    Kernel::<Receipt>::try_from(value)
        .map(Kernel::into_inner)
        .map_err(QueryAuditError::from)
}