polyc-state-connect 2026.8.3

//! Server adapter for immutable-object metadata.

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

use connectrpc::{ConnectError, RequestContext, Response, Router, ServiceRequest, ServiceResult};
use polyc_proto::proto::polychrome::state::v1::{
    GetObjectGenerationReply, GetObjectGenerationRequest, GetObjectHeadReply, GetObjectHeadRequest,
    GetObjectReceiptReply, GetObjectReceiptRequest, ListObjectGenerationsReply,
    ListObjectGenerationsRequest, PruneObjectGenerationReply, PruneObjectGenerationRequest,
    RecordObjectGenerationReply, RecordObjectGenerationRequest, StateImmutableMetadataService,
    StateImmutableMetadataServiceExt, UpdateObjectHeadReply, UpdateObjectHeadRequest,
};
use polyc_state::{
    context::CallContext,
    error::StateError,
    id::{CommandId, OwnerId},
    immutable::{
        self, Generation, ImmutableObjectRead, ImmutableObjectWrite, ListObjectGenerations,
        ObjectId, PruneObjectGeneration, ReadObjectGeneration, ReadObjectHead,
        RecordObjectGeneration, UpdateObjectHead,
    },
};

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

/// The complete immutable-metadata authority port.
pub trait ImmutableMetadataAuthority: ImmutableObjectRead + ImmutableObjectWrite {}

impl<T> ImmutableMetadataAuthority for T where T: ImmutableObjectRead + ImmutableObjectWrite {}

/// State's metadata authority for immutable object generations.
pub struct ImmutableMetadataSvc {
    metadata: Arc<dyn ImmutableMetadataAuthority>,
    draining: Arc<AtomicBool>,
    binding: Arc<AudienceBinding>,
}

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

    /// Registers the capability-specific service.
    #[must_use]
    pub fn register_on(self, router: Router) -> Router {
        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<polyc_proto::proto::polychrome::state::v1::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| state_to_connect_error(&error))?;
        let family = immutable::family();
        check_call_context_version(declared.version)
            .map_err(|error| state_to_connect_error(&error))?;
        check_audience_binding(
            &Self::peer(ctx),
            &declared.audience,
            &state_audience(),
            &self.binding,
            &family,
        )
        .map_err(|error| state_to_connect_error(&error))?;
        check_transport_deadline(ctx.time_remaining(), &family)
            .map_err(|error| state_to_connect_error(&error))?;
        Ok((
            declared.origin_relative_context(),
            adopt_caller_trace(ctx.headers(), method),
        ))
    }
}

fn required_field(field: &str, reason: &str) -> ConnectError {
    to_connect_error(
        &StateError::Malformed {
            field: field.to_owned(),
            reason: reason.to_owned(),
        }
        .into(),
    )
}

#[allow(refining_impl_trait)]
impl StateImmutableMetadataService for ImmutableMetadataSvc {
    async fn record(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, RecordObjectGenerationRequest>,
    ) -> ServiceResult<RecordObjectGenerationReply> {
        let message = request.to_owned_message();
        let (context, span) = self.admit(&ctx, message.context, "RecordObjectGeneration")?;
        let _entered = span.enter();
        let metadata =
            Kernel::try_from(message.metadata.into_option().ok_or_else(|| {
                required_field("metadata", "an object command carries its metadata")
            })?)
            .map_err(|error: StateError| to_connect_error(&error.into()))?
            .into_inner();
        let descriptor = Kernel::try_from(message.descriptor.into_option().ok_or_else(|| {
            required_field("descriptor", "a record carries its object descriptor")
        })?)
        .map_err(|error: StateError| to_connect_error(&error.into()))?
        .into_inner();
        let receipt = self
            .metadata
            .record(
                RecordObjectGeneration::new(metadata, descriptor, OwnerId::new(message.caller)),
                &context,
            )
            .map_err(|error| to_connect_error(&error))?;
        Response::ok(RecordObjectGenerationReply {
            receipt: buffa::MessageField::some(Kernel(&receipt).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        })
    }

    async fn update_head(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, UpdateObjectHeadRequest>,
    ) -> ServiceResult<UpdateObjectHeadReply> {
        let message = request.to_owned_message();
        let (context, span) = self.admit(&ctx, message.context, "UpdateObjectHead")?;
        let _entered = span.enter();
        let metadata =
            Kernel::try_from(message.metadata.into_option().ok_or_else(|| {
                required_field("metadata", "an object command carries its metadata")
            })?)
            .map_err(|error: StateError| to_connect_error(&error.into()))?
            .into_inner();
        let receipt = self
            .metadata
            .update_head(
                UpdateObjectHead::new(
                    metadata,
                    ObjectId::new(message.object),
                    Generation::new(message.expected),
                    Generation::new(message.publish),
                    OwnerId::new(message.caller),
                ),
                &context,
            )
            .map_err(|error| to_connect_error(&error))?;
        Response::ok(UpdateObjectHeadReply {
            receipt: buffa::MessageField::some(Kernel(&receipt).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        })
    }

    async fn prune(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, PruneObjectGenerationRequest>,
    ) -> ServiceResult<PruneObjectGenerationReply> {
        let message = request.to_owned_message();
        let (context, span) = self.admit(&ctx, message.context, "PruneObjectGeneration")?;
        let _entered = span.enter();
        let metadata =
            Kernel::try_from(message.metadata.into_option().ok_or_else(|| {
                required_field("metadata", "an object command carries its metadata")
            })?)
            .map_err(|error: StateError| to_connect_error(&error.into()))?
            .into_inner();
        let receipt = self
            .metadata
            .prune(
                PruneObjectGeneration::new(
                    metadata,
                    ObjectId::new(message.object),
                    Generation::new(message.generation),
                    OwnerId::new(message.caller),
                ),
                &context,
            )
            .map_err(|error| to_connect_error(&error))?;
        Response::ok(PruneObjectGenerationReply {
            receipt: buffa::MessageField::some(Kernel(&receipt).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        })
    }

    async fn get_head(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, GetObjectHeadRequest>,
    ) -> ServiceResult<GetObjectHeadReply> {
        let message = request.to_owned_message();
        let (context, span) = self.admit(&ctx, message.context, "GetObjectHead")?;
        let _entered = span.enter();
        let head = self
            .metadata
            .head(
                ReadObjectHead::new(ObjectId::new(message.object), OwnerId::new(message.caller)),
                &context,
            )
            .map_err(|error| to_connect_error(&error))?;
        Response::ok(GetObjectHeadReply {
            head: head
                .as_ref()
                .map_or_else(buffa::MessageField::default, |head| {
                    buffa::MessageField::some(Kernel(head).into())
                }),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        })
    }

    async fn get_generation(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, GetObjectGenerationRequest>,
    ) -> ServiceResult<GetObjectGenerationReply> {
        let message = request.to_owned_message();
        let (context, span) = self.admit(&ctx, message.context, "GetObjectGeneration")?;
        let _entered = span.enter();
        let metadata = self
            .metadata
            .generation(
                ReadObjectGeneration::new(
                    ObjectId::new(message.object),
                    Generation::new(message.generation),
                    OwnerId::new(message.caller),
                ),
                &context,
            )
            .map_err(|error| to_connect_error(&error))?;
        Response::ok(GetObjectGenerationReply {
            metadata: metadata
                .as_ref()
                .map_or_else(buffa::MessageField::default, |metadata| {
                    buffa::MessageField::some(Kernel(metadata).into())
                }),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        })
    }

    async fn list_generations(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, ListObjectGenerationsRequest>,
    ) -> ServiceResult<ListObjectGenerationsReply> {
        let message = request.to_owned_message();
        let (context, span) = self.admit(&ctx, message.context, "ListObjectGenerations")?;
        let _entered = span.enter();
        let page =
            Kernel::try_from(message.page.into_option().ok_or_else(|| {
                required_field("page", "a bounded listing carries its page request")
            })?)
            .map_err(|error: StateError| to_connect_error(&error.into()))?
            .into_inner();
        let result = self
            .metadata
            .generations(
                ListObjectGenerations::new(
                    ObjectId::new(message.object),
                    page,
                    OwnerId::new(message.caller),
                ),
                &context,
            )
            .map_err(|error| to_connect_error(&error))?;
        Response::ok(ListObjectGenerationsReply {
            page: buffa::MessageField::some(Kernel(&result).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        })
    }

    async fn get_receipt(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, GetObjectReceiptRequest>,
    ) -> ServiceResult<GetObjectReceiptReply> {
        let message = request.to_owned_message();
        let (_context, span) = self.admit(&ctx, message.context, "GetObjectReceipt")?;
        let _entered = span.enter();
        let receipt = self
            .metadata
            .recorded_receipt(
                &ObjectId::new(message.object),
                &CommandId::new(message.command_id),
            )
            .map_err(|error| to_connect_error(&error))?;
        Response::ok(GetObjectReceiptReply {
            receipt: receipt
                .as_ref()
                .map_or_else(buffa::MessageField::default, |receipt| {
                    buffa::MessageField::some(Kernel(receipt).into())
                }),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        })
    }
}