polyc-state-connect 2026.8.3

//! Server adapter for durable ingress.

use std::sync::{
    Arc,
    atomic::{AtomicBool, Ordering},
};

use connectrpc::{ConnectError, RequestContext, Response, Router, ServiceRequest, ServiceResult};
use polyc_proto::proto::polychrome::state::v1 as pb;
use polyc_state::{
    context::CallContext,
    error::StateError,
    id::CommandId,
    ingress::{EdgeId, IngressAuthority, IngressRead, ReadInboxDepth, ReadIngressItem},
    receipt::Receipt,
};

use crate::{
    admission::{
        AudienceBinding, PeerIdentity, check_audience_binding, check_call_context_version,
        check_not_draining, check_transport_deadline, state_audience,
    },
    error::to_connect_error,
    trace::adopt_caller_trace,
    wire::{DeclaredCall, Kernel, declared_call},
};

use super::wire::{
    claim_from_wire, decide_from_wire, depth_to_wire, item_to_wire, receive_from_wire,
    scope_from_wire, source_from_wire,
};

/// Complete durable-ingress authority port.
pub trait StateIngressAuthority: IngressRead + IngressAuthority {}

impl<T> StateIngressAuthority for T where T: IngressRead + IngressAuthority {}

/// State's durable-ingress service.
pub struct IngressSvc {
    authority: Arc<dyn StateIngressAuthority>,
    draining: Arc<AtomicBool>,
    binding: Arc<AudienceBinding>,
}

impl IngressSvc {
    /// Builds the service.
    #[must_use]
    pub const fn new(
        authority: Arc<dyn StateIngressAuthority>,
        draining: Arc<AtomicBool>,
        binding: Arc<AudienceBinding>,
    ) -> Self {
        Self {
            authority,
            draining,
            binding,
        }
    }

    /// Registers the durable-ingress routes.
    #[must_use]
    pub fn register_on(self, router: Router) -> Router {
        use pb::StateIngressServiceExt as _;
        Arc::new(self).register(router)
    }

    fn peer(ctx: &RequestContext) -> PeerIdentity {
        PeerIdentity::from_verified_leaf(
            ctx.peer_certs().and_then(<[_]>::first).map(|leaf| &**leaf),
        )
    }

    fn admit(
        &self,
        ctx: &RequestContext,
        context: impl Into<Option<pb::CallContext>>,
        method: &'static str,
    ) -> Result<(CallContext, tracing::Span), ConnectError> {
        check_not_draining(self.draining.load(Ordering::Relaxed))?;
        let declared: DeclaredCall =
            declared_call(context).map_err(|error| to_connect_error(&error))?;
        let family = polyc_state::ingress::family();
        check_call_context_version(declared.version).map_err(|error| to_connect_error(&error))?;
        check_audience_binding(
            &Self::peer(ctx),
            &declared.audience,
            &state_audience(),
            &self.binding,
            &family,
        )
        .map_err(|error| to_connect_error(&error))?;
        check_transport_deadline(ctx.time_remaining(), &family)
            .map_err(|error| to_connect_error(&error))?;
        Ok((
            declared.origin_relative_context(),
            adopt_caller_trace(ctx.headers(), method),
        ))
    }

    async fn blocking<R: Send + 'static>(
        operation: impl FnOnce() -> Result<R, StateError> + Send + 'static,
    ) -> Result<R, ConnectError> {
        tokio::task::spawn_blocking(operation)
            .await
            .map_err(|_| {
                to_connect_error(&StateError::Unavailable {
                    family: polyc_state::ingress::family(),
                    reach: polyc_state::error::OutageReach::PossiblyApplied,
                })
            })?
            .map_err(|error| to_connect_error(&error))
    }
}

#[allow(refining_impl_trait)]
impl pb::StateIngressService for IngressSvc {
    async fn receive(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, pb::ReceiveIngressRequest>,
    ) -> ServiceResult<pb::ReceiveIngressReply> {
        let message = request.to_owned_message();
        let (context, span) = self.admit(&ctx, message.context, "ReceiveIngress")?;
        let command = receive_from_wire(required_command(message.command, "receive")?)
            .map_err(|error| to_connect_error(&error))?;
        let authority = Arc::clone(&self.authority);
        let receipt = Self::blocking(move || authority.receive(command, &context))
            .instrument(span)
            .await?;
        Response::ok(pb::ReceiveIngressReply {
            receipt: buffa::MessageField::some(pb::Receipt::from(Kernel(&receipt))),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        })
    }

    async fn claim(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, pb::ClaimIngressRequest>,
    ) -> ServiceResult<pb::ClaimIngressReply> {
        let message = request.to_owned_message();
        let (context, span) = self.admit(&ctx, message.context, "ClaimIngress")?;
        let command = claim_from_wire(required_command(message.command, "claim")?)
            .map_err(|error| to_connect_error(&error))?;
        let authority = Arc::clone(&self.authority);
        let claimed = Self::blocking(move || authority.claim(command, &context))
            .instrument(span)
            .await?;
        let (receipt, item) = claimed.map_or_else(
            || (buffa::MessageField::none(), buffa::MessageField::none()),
            |claimed| {
                (
                    buffa::MessageField::some(pb::Receipt::from(Kernel(claimed.receipt()))),
                    buffa::MessageField::some(item_to_wire(claimed.item())),
                )
            },
        );
        Response::ok(pb::ClaimIngressReply {
            receipt,
            item,
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        })
    }

    async fn decide(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, pb::DecideIngressRequest>,
    ) -> ServiceResult<pb::DecideIngressReply> {
        let message = request.to_owned_message();
        let (context, span) = self.admit(&ctx, message.context, "DecideIngress")?;
        let command = decide_from_wire(required_command(message.command, "decision")?)
            .map_err(|error| to_connect_error(&error))?;
        let authority = Arc::clone(&self.authority);
        let receipt = Self::blocking(move || authority.decide(command, &context))
            .instrument(span)
            .await?;
        Response::ok(pb::DecideIngressReply {
            receipt: buffa::MessageField::some(pb::Receipt::from(Kernel(&receipt))),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        })
    }

    async fn get_depth(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, pb::GetIngressDepthRequest>,
    ) -> ServiceResult<pb::GetIngressDepthReply> {
        let message = request.to_owned_message();
        let (context, span) = self.admit(&ctx, message.context, "GetIngressDepth")?;
        let scope = message
            .scope
            .into_option()
            .ok_or_else(|| malformed_connect("scope"))?;
        let request = ReadInboxDepth::new(scope_from_wire(scope), EdgeId::new(message.edge));
        let authority = Arc::clone(&self.authority);
        let depth = Self::blocking(move || authority.depth(request, &context))
            .instrument(span)
            .await?;
        Response::ok(pb::GetIngressDepthReply {
            depth: buffa::MessageField::some(depth_to_wire(depth)),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        })
    }

    async fn get_item(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, pb::GetIngressItemRequest>,
    ) -> ServiceResult<pb::GetIngressItemReply> {
        let message = request.to_owned_message();
        let (context, span) = self.admit(&ctx, message.context, "GetIngressItem")?;
        let scope = message
            .scope
            .into_option()
            .ok_or_else(|| malformed_connect("scope"))?;
        let source = message
            .source
            .into_option()
            .ok_or_else(|| malformed_connect("source"))?;
        let request = ReadIngressItem::new(
            scope_from_wire(scope),
            source_from_wire(source).map_err(|error| to_connect_error(&error))?,
        );
        let authority = Arc::clone(&self.authority);
        let item = Self::blocking(move || authority.item(request, &context))
            .instrument(span)
            .await?;
        Response::ok(pb::GetIngressItemReply {
            item: item
                .as_ref()
                .map_or_else(buffa::MessageField::none, |item| {
                    buffa::MessageField::some(item_to_wire(item))
                }),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        })
    }

    async fn get_receipt(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, pb::GetIngressReceiptRequest>,
    ) -> ServiceResult<pb::GetIngressReceiptReply> {
        let message = request.to_owned_message();
        let (_context, span) = self.admit(&ctx, message.context, "GetIngressReceipt")?;
        let scope = message
            .scope
            .into_option()
            .ok_or_else(|| malformed_connect("scope"))?;
        let authority = Arc::clone(&self.authority);
        let receipt: Option<Receipt> = Self::blocking(move || {
            authority
                .committed_receipt(&scope_from_wire(scope), &CommandId::new(message.command_id))
        })
        .instrument(span)
        .await?;
        Response::ok(pb::GetIngressReceiptReply {
            receipt: receipt
                .as_ref()
                .map_or_else(buffa::MessageField::none, |receipt| {
                    buffa::MessageField::some(pb::Receipt::from(Kernel(receipt)))
                }),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        })
    }
}

// Named over `P: buffa::ProtoBox<T>` rather than `impl Into<Option<T>>`: see
// `crate::wire::required`'s doc comment — `T` is generic per call here too,
// and that bound doesn't infer through the `?` in this file's
// `xxx_from_wire(required_command(...)?)` call sites.
fn required_command<T: Default, P: buffa::ProtoBox<T>>(
    value: buffa::MessageField<T, P>,
    name: &str,
) -> Result<T, ConnectError> {
    value.into_option().ok_or_else(|| {
        to_connect_error(&StateError::Malformed {
            field: "command".to_owned(),
            reason: format!("an ingress request carries its {name} command"),
        })
    })
}

fn malformed_connect(field: &str) -> ConnectError {
    to_connect_error(&StateError::Malformed {
        field: field.to_owned(),
        reason: format!("an ingress request carries {field}"),
    })
}

use tracing::Instrument as _;