polyc-state-connect 2026.9.0

//! Capability-specific projection catalog client.

use connectrpc::client::{ClientConfig, ClientTransport};
use polyc_proto::proto::polychrome::state::v1 as pb;
use polyc_state::{
    error::StateError,
    projection::{
        ProjectionCatalogError, ProjectionGeneration, ProjectionHead, ProjectionManifest,
        ProjectionResolution, PublishManifest, PublisherFence, ReadProjectionReceipt,
        RecordManifest, RegisterPublisher, ResolveManifest,
    },
    receipt::Receipt,
    revision::JournalSource,
};

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

/// A client that can access only projection manifest authority.
pub struct ProjectionCatalogClient<T> {
    inner: pb::StateProjectionCatalogServiceClient<T>,
}

impl<T> ProjectionCatalogClient<T>
where
    T: ClientTransport,
    <T::ResponseBody as connectrpc::http_body::Body>::Error: std::fmt::Display,
{
    /// Builds a client under the State metadata wire bound.
    pub fn new(transport: T, config: ClientConfig) -> Self {
        Self {
            inner: pb::StateProjectionCatalogServiceClient::new(
                transport,
                config.with_default_max_message_size(MAX_WIRE_MESSAGE_BYTES),
            ),
        }
    }
    fn fallback(attempted: usize) -> TransportFallback {
        TransportFallback::new(
            polyc_state::projection::family(),
            MAX_WIRE_MESSAGE_BYTES as u64,
            attempted as u64,
        )
    }

    /// Registers publisher authority.
    ///
    /// # Errors
    /// Returns the exact catalog or transport outcome.
    pub async fn register(
        &self,
        declared: &DeclaredCall,
        command: &RegisterPublisher,
    ) -> Result<(PublisherFence, Receipt), ProjectionCatalogError> {
        let wire = pb::RegisterProjectionPublisherRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            metadata: buffa::MessageField::some(Kernel(command.metadata()).into()),
            key: buffa::MessageField::some(Kernel(command.key()).into()),
            source: buffa::MessageField::some(Kernel(command.source()).into()),
            publisher: command.publisher().as_str().to_owned(),
            caller: command.caller().as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&wire) as usize;
        let reply = self
            .inner
            .register_publisher_with_options(wire, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        let fence: PublisherFence = required(reply.fence, "fence")?;
        if fence.key() != command.key() || fence.incarnation() != command.source().incarnation() {
            return Err(malformed_binding("fence"));
        }
        Ok((fence, receipt(reply.receipt)?))
    }

    /// Records one manifest without publishing it.
    ///
    /// # Errors
    /// Returns the exact catalog or transport outcome.
    pub async fn record(
        &self,
        declared: &DeclaredCall,
        command: &RecordManifest,
    ) -> Result<Receipt, ProjectionCatalogError> {
        let wire = pb::RecordProjectionManifestRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            metadata: buffa::MessageField::some(Kernel(command.metadata()).into()),
            manifest: buffa::MessageField::some(Kernel(command.manifest()).into()),
            caller: command.caller().as_str().to_owned(),
            signed_artifact: command.artifact().to_vec(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&wire) as usize;
        let reply = self
            .inner
            .record_manifest_with_options(wire, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        receipt(reply.receipt)
    }

    /// Publishes one exactly recorded manifest.
    ///
    /// # Errors
    /// Returns the exact catalog or transport outcome.
    pub async fn publish(
        &self,
        declared: &DeclaredCall,
        command: &PublishManifest,
    ) -> Result<Receipt, ProjectionCatalogError> {
        let wire = pb::PublishProjectionManifestRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            metadata: buffa::MessageField::some(Kernel(command.metadata()).into()),
            key: buffa::MessageField::some(Kernel(command.key()).into()),
            source: buffa::MessageField::some(Kernel(command.source()).into()),
            expected: command.expected().get(),
            publish: command.publish().get(),
            publisher: command.publisher().as_str().to_owned(),
            fence: buffa::MessageField::some(Kernel(command.fence()).into()),
            caller: command.caller().as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&wire) as usize;
        let reply = self
            .inner
            .publish_manifest_with_options(wire, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        receipt(reply.receipt)
    }

    /// Resolves the current exact-source manifest.
    ///
    /// # Errors
    /// Returns the exact catalog or transport outcome.
    pub async fn resolve(
        &self,
        declared: &DeclaredCall,
        request: &ResolveManifest,
    ) -> Result<ProjectionResolution, ProjectionCatalogError> {
        let wire = pb::ResolveProjectionManifestRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            key: buffa::MessageField::some(Kernel(request.key()).into()),
            source: buffa::MessageField::some(Kernel(request.source()).into()),
            caller: request.caller().as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&wire) as usize;
        let reply = self
            .inner
            .resolve_manifest_with_options(wire, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        let pb::ResolveProjectionManifestReply {
            manifest,
            superseded_generation,
            superseded_source,
            pending_manifest,
            highest_recorded_generation,
            __buffa_unknown_fields: _,
        } = reply;
        let manifest = manifest
            .into_option()
            .map(|value| {
                Kernel::<ProjectionManifest>::try_from(value)
                    .map(Kernel::into_inner)
                    .map_err(ProjectionCatalogError::from)
            })
            .transpose()?;
        let superseded_source = superseded_source
            .into_option()
            .map(|value| {
                Kernel::<JournalSource>::try_from(value)
                    .map(Kernel::into_inner)
                    .map_err(ProjectionCatalogError::from)
            })
            .transpose()?;
        let pending = pending_manifest
            .into_option()
            .map(|value| {
                Kernel::<ProjectionManifest>::try_from(value)
                    .map(Kernel::into_inner)
                    .map_err(ProjectionCatalogError::from)
            })
            .transpose()?;

        let head = match (manifest, superseded_generation, superseded_source) {
            (None, None, None) => ProjectionHead::Absent,
            (Some(manifest), None, None) => {
                // A manifest for another key or another lineage is not an
                // answer to this request, whatever the server meant by it.
                if manifest.key() != request.key()
                    || manifest.checkpoint().source() != request.source()
                    || manifest.object_descriptor().owner() != request.caller()
                {
                    return Err(malformed_binding("manifest"));
                }
                ProjectionHead::Current(Box::new(manifest))
            }
            (None, Some(generation), Some(source)) => {
                // Superseded stays in the requested partition but names a
                // different incarnation. Any crossed scope contradicts it.
                if source.partition() != request.source().partition()
                    || source.partition() != request.key().source()
                    || source.incarnation() == request.source().incarnation()
                    || generation == 0
                {
                    return Err(malformed_binding("superseded_head"));
                }
                ProjectionHead::Superseded {
                    generation: ProjectionGeneration::new(generation),
                    source: Box::new(source),
                }
            }
            // Any other combination is a reply that claims two answers at once
            // or half of one. Neither is admitted.
            _ => return Err(malformed_binding("resolved_head")),
        };
        let highest = ProjectionGeneration::new(highest_recorded_generation);
        if highest.get() < head.generation().get() {
            return Err(malformed_binding("highest_recorded_generation"));
        }
        if let Some(pending) = pending.as_ref()
            && (pending.key() != request.key()
                || pending.checkpoint().source() != request.source()
                || pending.object_descriptor().owner() != request.caller()
                || pending.generation().get() <= head.generation().get()
                || pending.generation().get() > highest.get())
        {
            return Err(malformed_binding("pending_manifest"));
        }
        Ok(ProjectionResolution::new(
            head,
            pending.map(Box::new),
            highest,
        ))
    }

    /// Reads one durable catalog receipt.
    ///
    /// # Errors
    /// Returns the exact catalog or transport outcome.
    pub async fn receipt(
        &self,
        declared: &DeclaredCall,
        request: &ReadProjectionReceipt,
    ) -> Result<Option<Receipt>, ProjectionCatalogError> {
        let wire = pb::GetProjectionReceiptRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            key: buffa::MessageField::some(Kernel(request.key()).into()),
            source: buffa::MessageField::some(Kernel(request.source()).into()),
            command_id: request.command().as_str().to_owned(),
            caller: request.caller().as_str().to_owned(),
            __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(|value| {
                Kernel::<Receipt>::try_from(value)
                    .map(Kernel::into_inner)
                    .map_err(ProjectionCatalogError::from)
            })
            .transpose()
    }
}

fn malformed(field: &str) -> ProjectionCatalogError {
    StateError::Malformed {
        field: field.to_owned(),
        reason: "a successful projection reply carries its required field".to_owned(),
    }
    .into()
}
fn malformed_binding(field: &str) -> ProjectionCatalogError {
    StateError::Malformed {
        field: field.to_owned(),
        reason: "a successful projection reply binds the exact request identity".to_owned(),
    }
    .into()
}
fn required<T, W, P>(
    field: buffa::MessageField<W, P>,
    name: &str,
) -> Result<T, ProjectionCatalogError>
where
    W: Default,
    P: buffa::ProtoBox<W>,
    Kernel<T>: TryFrom<W, Error = StateError>,
{
    Kernel::try_from(field.into_option().ok_or_else(|| malformed(name))?)
        .map(Kernel::into_inner)
        .map_err(ProjectionCatalogError::from)
}
fn receipt(
    field: buffa::MessageField<pb::Receipt, buffa::Inline<pb::Receipt>>,
) -> Result<Receipt, ProjectionCatalogError> {
    required(field, "receipt")
}

#[cfg(test)]
mod response_binding_tests {
    #![allow(clippy::unwrap_used, missing_docs)]

    use std::pin::Pin;
    use std::sync::Arc;
    use std::task::{Context, Poll};
    use std::time::Duration;

    use bytes::Bytes;
    use connectrpc::client::{BoxFuture, ClientBody};
    use connectrpc::http_body::{Body, Frame};
    use polyc_state::digest::ContentDigest;
    use polyc_state::feed::SourceCheckpoint;
    use polyc_state::id::{OwnerId, PartitionId};
    use polyc_state::immutable::{
        Classification, ContentReference, Generation, ObjectDescriptor, Retention,
    };
    use polyc_state::journal::JournalAttestation;
    use polyc_state::projection::artifact::{ExactObjectRef, ObjectNamespace};
    use polyc_state::projection::{FamilyId, ProjectionKey, PublisherId};
    use polyc_state::revision::{CommitRoot, JournalPosition, PartitionIncarnation};

    use super::*;
    use crate::state_audience;

    struct CannedBody(std::vec::IntoIter<Bytes>);

    impl Body for CannedBody {
        type Data = Bytes;
        type Error = std::io::Error;

        fn poll_frame(
            mut self: Pin<&mut Self>,
            _context: &mut Context<'_>,
        ) -> Poll<Option<Result<Frame<Bytes>, Self::Error>>> {
            Poll::Ready(self.0.next().map(|bytes| Ok(Frame::data(bytes))))
        }
    }

    #[derive(Clone)]
    struct CannedTransport {
        frames: Arc<Vec<Bytes>>,
    }

    impl ClientTransport for CannedTransport {
        type ResponseBody = CannedBody;
        type Error = std::io::Error;

        fn send(
            &self,
            _request: http::Request<ClientBody>,
        ) -> BoxFuture<'static, Result<http::Response<Self::ResponseBody>, Self::Error>> {
            let frames = self.frames.as_ref().clone();
            Box::pin(async move {
                Ok(http::Response::builder()
                    .status(http::StatusCode::OK)
                    .header(http::header::CONTENT_TYPE, "application/proto")
                    .body(CannedBody(frames.into_iter()))
                    .unwrap())
            })
        }
    }

    fn source() -> JournalSource {
        JournalSource::new(
            PartitionId::new("conv-a"),
            PartitionIncarnation::from_bytes([3; PartitionIncarnation::LEN]),
        )
    }

    fn crossed_owner_manifest() -> ProjectionManifest {
        let source = source();
        let key = ProjectionKey::new(
            FamilyId::new("conversation-core/v1"),
            source.partition().clone(),
        );
        let reference = ContentReference::try_new("projection/conv-a/manifest").unwrap();
        ProjectionManifest::new(
            key.clone(),
            ProjectionGeneration::new(1),
            SourceCheckpoint::try_new(
                source.clone(),
                JournalPosition::new(8),
                JournalPosition::new(9),
                9,
                JournalAttestation::new(
                    CommitRoot::from_bytes([4; CommitRoot::LEN]),
                    10,
                    vec![5; 64],
                    vec![6; 32],
                ),
            )
            .unwrap(),
            1,
            1,
            ObjectDescriptor::new(
                key.object(),
                Generation::new(1),
                ContentDigest::from_bytes([7; ContentDigest::LEN]),
                OwnerId::new("crossed-owner"),
                Classification::Confidential,
                Retention::For(Duration::from_mins(1)),
                128,
                reference.clone(),
            ),
            ExactObjectRef::try_new(
                ObjectNamespace::try_new("conversation-visible").unwrap(),
                reference,
                10,
            )
            .unwrap(),
            PublisherId::new("projector-a"),
            PublisherFence::new(key, source.incarnation(), 1),
        )
    }

    #[tokio::test]
    async fn resolve_refuses_a_manifest_admitted_for_another_owner() {
        let manifest = crossed_owner_manifest();
        let reply = pb::ResolveProjectionManifestReply {
            manifest: buffa::MessageField::some(Kernel(&manifest).into()),
            superseded_generation: None,
            superseded_source: buffa::MessageField::default(),
            pending_manifest: buffa::MessageField::default(),
            highest_recorded_generation: 1,
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let frames = vec![Bytes::from(buffa::Message::encode_to_vec(&reply))];
        let client = ProjectionCatalogClient::new(
            CannedTransport {
                frames: Arc::new(frames),
            },
            ClientConfig::new("http://projection.invalid".parse().unwrap()),
        );
        let request =
            ResolveManifest::new(manifest.key().clone(), source(), OwnerId::new("projector"));
        let error = client
            .resolve(
                &DeclaredCall::live(state_audience(), Duration::MAX),
                &request,
            )
            .await
            .expect_err("the successful response crossed the admitted owner");

        assert!(
            matches!(
                error,
                ProjectionCatalogError::State(StateError::Malformed { ref field, .. })
                    if field == "manifest"
            ),
            "got {error:?}"
        );
    }

    #[tokio::test]
    async fn resolve_refuses_a_pending_manifest_admitted_for_another_owner() {
        let manifest = crossed_owner_manifest();
        let reply = pb::ResolveProjectionManifestReply {
            manifest: buffa::MessageField::default(),
            superseded_generation: None,
            superseded_source: buffa::MessageField::default(),
            pending_manifest: buffa::MessageField::some(Kernel(&manifest).into()),
            highest_recorded_generation: 1,
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let client = ProjectionCatalogClient::new(
            CannedTransport {
                frames: Arc::new(vec![Bytes::from(buffa::Message::encode_to_vec(&reply))]),
            },
            ClientConfig::new("http://projection.invalid".parse().unwrap()),
        );
        let request =
            ResolveManifest::new(manifest.key().clone(), source(), OwnerId::new("projector"));
        let error = client
            .resolve(
                &DeclaredCall::live(state_audience(), Duration::MAX),
                &request,
            )
            .await
            .expect_err("the pending response crossed the admitted owner");

        assert!(
            matches!(
                error,
                ProjectionCatalogError::State(StateError::Malformed { ref field, .. })
                    if field == "pending_manifest"
            ),
            "got {error:?}"
        );
    }

    #[tokio::test]
    async fn resolve_refuses_a_superseded_source_for_another_partition() {
        let crossed = JournalSource::new(
            PartitionId::new("conv-b"),
            PartitionIncarnation::from_bytes([9; PartitionIncarnation::LEN]),
        );
        let reply = pb::ResolveProjectionManifestReply {
            manifest: buffa::MessageField::default(),
            superseded_generation: Some(1),
            superseded_source: buffa::MessageField::some(Kernel(&crossed).into()),
            pending_manifest: buffa::MessageField::default(),
            highest_recorded_generation: 1,
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let client = ProjectionCatalogClient::new(
            CannedTransport {
                frames: Arc::new(vec![Bytes::from(buffa::Message::encode_to_vec(&reply))]),
            },
            ClientConfig::new("http://projection.invalid".parse().unwrap()),
        );
        let requested_source = source();
        let request = ResolveManifest::new(
            ProjectionKey::new(
                FamilyId::new("conversation-core/v1"),
                requested_source.partition().clone(),
            ),
            requested_source,
            OwnerId::new("projector"),
        );
        let error = client
            .resolve(
                &DeclaredCall::live(state_audience(), Duration::MAX),
                &request,
            )
            .await
            .expect_err("a superseded response cannot cross partitions");

        assert!(
            matches!(
                error,
                ProjectionCatalogError::State(StateError::Malformed { ref field, .. })
                    if field == "superseded_head"
            ),
            "got {error:?}"
        );
    }
}