polyc-state-connect 2026.10.2

//! Server adapter for the observation authority (7-O).
//!
//! The only writer is the control-plane observer; the only readers are the
//! projector's observed-source port and the Fleet Query plane.

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

use connectrpc::{ConnectError, RequestContext, Response, Router, ServiceRequest, ServiceResult};
use polyc_proto::proto::polychrome::state::v1::{
    ListObservedCollectionsReply, ListObservedCollectionsRequest, ObservationHeadMessage,
    ObservationRecordMessage, ReadLatestObservationReply, ReadLatestObservationRequest,
    ReadObservationReply, ReadObservationRequest, RecordObservationReply, RecordObservationRequest,
    StateObservationService, StateObservationServiceExt,
};
use polyc_state::{
    command::CommandEnvelope,
    error::StateError,
    id::{Audience, Purpose},
    observation::{
        self, CollectionId, CollectionKind, ObservationAuthority, ObservationDraft,
        ObservationOrdinal, ObservationPayload, ObservationSource, ObserverId, RecordObservation,
        ResourceVersion,
    },
};

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

/// Observation's sole durable-write and read surface.
pub struct ObservationSvc {
    authority: Arc<dyn ObservationAuthority>,
    draining: Arc<AtomicBool>,
    binding: Arc<AudienceBinding>,
}

impl ObservationSvc {
    /// Builds the service over one observation authority.
    #[must_use]
    pub const fn new(
        authority: Arc<dyn ObservationAuthority>,
        draining: Arc<AtomicBool>,
        binding: Arc<AudienceBinding>,
    ) -> Self {
        Self {
            authority,
            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<(DeclaredCall, tracing::Span, std::time::Duration), ConnectError> {
        check_not_draining(self.draining.load(Ordering::Relaxed))?;
        let declared: DeclaredCall =
            declared_call(context).map_err(|error| to_connect_error(&error))?;
        let family = observation::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))?;
        let wait = slot_wait(ctx.time_remaining(), &declared);
        Ok((declared, adopt_caller_trace(ctx.headers(), method), wait))
    }
}

fn envelope() -> CommandEnvelope {
    CommandEnvelope::new(
        Purpose::new("observation"),
        Audience::new("state"),
        observation::command_bounds(),
    )
}

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

fn collection_kind(
    field: &str,
    wire: buffa::EnumValue<polyc_proto::proto::polychrome::state::v1::ObservedCollectionKind>,
) -> Result<CollectionKind, ConnectError> {
    use polyc_proto::proto::polychrome::state::v1::ObservedCollectionKind as Wire;
    match known(field, wire).map_err(|error| to_connect_error(&error))? {
        Wire::OBSERVED_COLLECTION_KIND_ROUTINES => Ok(CollectionKind::Routines),
        Wire::OBSERVED_COLLECTION_KIND_UNSPECIFIED => Err(required_field(
            field,
            "an observed collection names a known kind",
        )),
    }
}

fn head_to_wire(head: &polyc_state::observation::ObservationHead) -> ObservationHeadMessage {
    ObservationHeadMessage {
        source: buffa::MessageField::some(
            polyc_proto::proto::polychrome::state::v1::ObservationSource::from(Kernel(
                head.source(),
            )),
        ),
        ordinal: head.ordinal().as_journal_position().get(),
        payload_digest: head.payload_digest().as_bytes().to_vec(),
        resource_version: head.resource_version().as_str().to_owned(),
        observed_at_nanos: head.observed_at().as_nanos(),
        recorded_at_nanos: head.recorded_at().as_nanos(),
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

fn record_to_wire(
    record: &polyc_state::observation::ObservationRecord,
) -> ObservationRecordMessage {
    ObservationRecordMessage {
        source: buffa::MessageField::some(
            polyc_proto::proto::polychrome::state::v1::ObservationSource::from(Kernel(
                record.source(),
            )),
        ),
        ordinal: record.ordinal().as_journal_position().get(),
        resource_version: record.resource_version().as_str().to_owned(),
        observed_at_nanos: record.observed_at().as_nanos(),
        recorded_at_nanos: record.recorded_at().as_nanos(),
        observer: record.observer().as_str().to_owned(),
        row_count: record.row_count(),
        payload_digest: record.payload_digest().as_bytes().to_vec(),
        payload: record.payload().as_bytes().to_vec(),
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

#[allow(refining_impl_trait)]
impl StateObservationService for ObservationSvc {
    fn record(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, RecordObservationRequest>,
    ) -> impl Future<Output = ServiceResult<RecordObservationReply>> {
        let message = request.to_owned_message();
        async move {
            let (_declared, span, wait) = self.admit(&ctx, message.context, "RecordObservation")?;
            let kind = collection_kind("collection_kind", message.collection_kind)?;
            let collection = CollectionId::try_new(kind, message.collection_namespace)
                .map_err(|error| to_connect_error(&error))?;
            let payload = ObservationPayload::try_new(message.payload)
                .map_err(|error| to_connect_error(&error))?;
            let draft = ObservationDraft::try_new(
                ResourceVersion::try_new(message.resource_version)
                    .map_err(|error| to_connect_error(&error))?,
                polyc_state::deadline::MonotonicInstant::from_nanos(message.observed_at_nanos),
                ObserverId::try_new(message.observer).map_err(|error| to_connect_error(&error))?,
                message.row_count,
                payload,
            )
            .map_err(|error| to_connect_error(&error))?;
            let expected = ObservationOrdinal::new(polyc_state::revision::JournalPosition::new(
                message.expected_ordinal,
            ));
            let command = RecordObservation::for_relist(collection, expected, draft, envelope());
            let authority = Arc::clone(&self.authority);
            let receipt = blocking::mutation(span, observation::family(), wait, move || {
                authority.record(command)
            })
            .await?;
            Response::ok(RecordObservationReply {
                receipt: buffa::MessageField::some(Kernel(&receipt).into()),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        }
    }

    fn read_latest(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, ReadLatestObservationRequest>,
    ) -> impl Future<Output = ServiceResult<ReadLatestObservationReply>> {
        let message = request.to_owned_message();
        async move {
            let (_declared, span, wait) =
                self.admit(&ctx, message.context, "ReadLatestObservation")?;
            let kind = collection_kind("collection_kind", message.collection_kind)?;
            let collection = CollectionId::try_new(kind, message.collection_namespace)
                .map_err(|error| to_connect_error(&error))?;
            let authority = Arc::clone(&self.authority);
            let head = blocking::read(span, observation::family(), wait, move || {
                authority.latest(&collection)
            })
            .await?;
            Response::ok(ReadLatestObservationReply {
                head: head
                    .as_ref()
                    .map_or_else(buffa::MessageField::default, |head| {
                        buffa::MessageField::some(head_to_wire(head))
                    }),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        }
    }

    fn read(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, ReadObservationRequest>,
    ) -> impl Future<Output = ServiceResult<ReadObservationReply>> {
        let message = request.to_owned_message();
        async move {
            let (_declared, span, wait) = self.admit(&ctx, message.context, "ReadObservation")?;
            let source =
                Kernel::<ObservationSource>::try_from(message.source.into_option().ok_or_else(
                    || required_field("source", "a read names the exact collection and lineage"),
                )?)
                .map_err(|error: StateError| to_connect_error(&error))?
                .into_inner();
            let ordinal = ObservationOrdinal::new(polyc_state::revision::JournalPosition::new(
                message.ordinal,
            ));
            let authority = Arc::clone(&self.authority);
            let record = blocking::read(span, observation::family(), wait, move || {
                authority.read(&source, ordinal)
            })
            .await?;
            Response::ok(ReadObservationReply {
                record: buffa::MessageField::some(record_to_wire(&record)),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        }
    }

    fn list_collections(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, ListObservedCollectionsRequest>,
    ) -> impl Future<Output = ServiceResult<ListObservedCollectionsReply>> {
        let message = request.to_owned_message();
        async move {
            let (_declared, span, wait) =
                self.admit(&ctx, message.context, "ListObservedCollections")?;
            let kind = collection_kind("collection_kind", message.collection_kind)?;
            let authority = Arc::clone(&self.authority);
            let listing = blocking::read(span, observation::family(), wait, move || {
                authority.collections(kind)
            })
            .await?;
            Response::ok(ListObservedCollectionsReply {
                collection_namespaces: listing
                    .collections()
                    .iter()
                    .map(|collection| collection.namespace().to_owned())
                    .collect(),
                complete: listing.complete(),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        }
    }
}