polyc-state-connect 2026.10.2

use connectrpc::client::{ClientConfig, ClientTransport};
use polyc_proto::proto::polychrome::state::v1 as pb;
use polyc_state::{
    command::CommandScope,
    error::StateError,
    revision::{JournalHead, JournalPosition, PartitionIncarnation},
    versioned::{
        ListVersionedDirectorySnapshot, VersionedDirectoryPage, VersionedDirectorySnapshot,
        VersionedDirectorySnapshotId, VersionedSourceHead,
    },
};

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

/// Capability-specific State client for Versioned source coordinates.
pub struct VersionedSourceClient<T> {
    inner: pb::StateVersionedSourceServiceClient<T>,
}

impl<T> VersionedSourceClient<T>
where
    T: ClientTransport,
    <T::ResponseBody as connectrpc::http_body::Body>::Error: std::fmt::Display,
{
    /// Builds a client for one State listener.
    #[must_use]
    pub fn new(transport: T, config: ClientConfig) -> Self {
        Self {
            inner: pb::StateVersionedSourceServiceClient::new(
                transport,
                config.with_default_max_message_size(MAX_WIRE_MESSAGE_BYTES),
            ),
        }
    }

    fn fallback(declared: &DeclaredCall, attempted: usize) -> TransportFallback {
        TransportFallback::new(
            polyc_state::versioned::family(),
            MAX_WIRE_MESSAGE_BYTES as u64,
            attempted as u64,
            declared.budget,
        )
    }

    /// Reads which lineage the authority is on and how far it has issued.
    ///
    /// # Errors
    ///
    /// Returns a typed State or transport refusal. A reply whose incarnation is
    /// not exactly 32 bytes is malformed, never truncated into a shorter
    /// lineage that would compare equal to nothing.
    pub async fn source_head(
        &self,
        declared: &DeclaredCall,
        scope: &CommandScope,
    ) -> Result<VersionedSourceHead, StateError> {
        let message = pb::GetVersionedSourceHeadRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            aggregate: scope.aggregate().to_string(),
            partition: scope.partition().to_string(),
            namespace: scope.namespace().to_string(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&message) as usize;
        let reply = self
            .inner
            .get_source_head_with_options(message, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(declared, attempted)))?
            .into_owned();
        // A successful answer binds the aggregate it was asked about. The
        // lineage identifies the log, not the aggregate, so every aggregate in
        // one log reports the same bytes: without this check a reply about
        // another aggregate is indistinguishable from the right one.
        if reply.aggregate != scope.aggregate().as_str()
            || reply.partition != scope.partition().as_str()
            || reply.namespace != scope.namespace().as_str()
        {
            return Err(StateError::Malformed {
                field: "scope".to_owned(),
                reason: "a source-head reply binds the aggregate it was asked about".to_owned(),
            });
        }
        let incarnation: [u8; PartitionIncarnation::LEN] =
            reply
                .incarnation
                .try_into()
                .map_err(|_| StateError::Malformed {
                    field: "incarnation".to_owned(),
                    reason: format!("a lineage is exactly {} bytes", PartitionIncarnation::LEN),
                })?;
        let head = reply
            .head
            .into_option()
            .ok_or_else(|| StateError::Malformed {
                field: "head".to_owned(),
                reason: "a source-head read returns its head".to_owned(),
            })?;
        Ok(VersionedSourceHead::new(
            PartitionIncarnation::from_bytes(incarnation),
            JournalHead::new(JournalPosition::new(head.position), None),
        ))
    }

    /// Retains the namespaces in which `family` has an aggregate.
    ///
    /// # Errors
    ///
    /// Returns a typed State or transport refusal. A reply about another
    /// family, one carrying no identity, or one whose incarnation is not
    /// exactly 32 bytes, is malformed: a caller reconciling keys must not take
    /// any of them for the answer it asked for.
    pub async fn create_directory_snapshot(
        &self,
        declared: &DeclaredCall,
        family: &str,
    ) -> Result<VersionedDirectorySnapshot, StateError> {
        let message = pb::CreateVersionedDirectorySnapshotRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            family: family.to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&message) as usize;
        let reply = self
            .inner
            .create_directory_snapshot_with_options(message, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(declared, attempted)))?
            .into_owned();
        if reply.family != family {
            return Err(StateError::Malformed {
                field: "family".to_owned(),
                reason: "a capture reply binds the family it was asked about".to_owned(),
            });
        }
        if reply.snapshot.is_empty() {
            return Err(StateError::Malformed {
                field: "snapshot".to_owned(),
                reason: "a capture reply carries the identity its pages name".to_owned(),
            });
        }
        Ok(VersionedDirectorySnapshot::new(
            VersionedDirectorySnapshotId::new(reply.snapshot),
            reply.namespace_count,
            PartitionIncarnation::from_bytes(lineage(reply.incarnation)?),
        ))
    }

    /// Reads one bounded page from a retained directory.
    ///
    /// # Errors
    ///
    /// Returns a typed State or transport refusal. A page that names another
    /// capture, is not strictly ascending, does not start past its cursor, or
    /// disagrees with itself about having more to give is malformed: a caller
    /// assembling a directory must not take any of them for a page.
    pub async fn directory_page(
        &self,
        declared: &DeclaredCall,
        request: &ListVersionedDirectorySnapshot,
    ) -> Result<VersionedDirectoryPage, StateError> {
        let message = pb::ListVersionedDirectorySnapshotRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            snapshot: request.snapshot().as_str().to_owned(),
            after: request.start_after().map(str::to_owned),
            limit: request.limit(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&message) as usize;
        let reply = self
            .inner
            .directory_page_with_options(message, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(declared, attempted)))?
            .into_owned();
        let completeness = if reply.complete {
            polyc_state::page::PageCompleteness::Complete
        } else {
            polyc_state::page::PageCompleteness::Truncated
        };
        let page = VersionedDirectoryPage::new(
            VersionedDirectorySnapshotId::new(reply.snapshot),
            reply.namespaces,
            reply.next_after,
            completeness,
            PartitionIncarnation::from_bytes(lineage(reply.incarnation)?),
        );
        // Every boundary rule in one place, on the kernel type, so the client
        // and any other consumer hold the reply to the same contract.
        page.validate(request)?;
        Ok(page)
    }

    /// Releases a retained directory. Idempotent.
    ///
    /// # Errors
    ///
    /// Returns a typed State or transport refusal.
    pub async fn release_directory_snapshot(
        &self,
        declared: &DeclaredCall,
        snapshot: &VersionedDirectorySnapshotId,
    ) -> Result<(), StateError> {
        let message = pb::ReleaseVersionedDirectorySnapshotRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            snapshot: snapshot.as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&message) as usize;
        self.inner
            .release_directory_snapshot_with_options(message, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(declared, attempted)))?;
        Ok(())
    }
}

/// Reads a lineage out of a reply, refusing any width but the exact one.
fn lineage(bytes: Vec<u8>) -> Result<[u8; PartitionIncarnation::LEN], StateError> {
    bytes.try_into().map_err(|_| StateError::Malformed {
        field: "incarnation".to_owned(),
        reason: format!("a lineage is exactly {} bytes", PartitionIncarnation::LEN),
    })
}

#[cfg(test)]
mod tests {
    use std::{
        pin::Pin,
        sync::Arc,
        task::{Context, Poll},
        time::Duration,
    };

    use connectrpc::{
        client::{ClientBody, ClientConfig, ClientTransport},
        http_body::{Body, Frame},
    };
    use futures::future::BoxFuture;
    use polyc_proto::proto::polychrome::state::v1 as pb;
    use polyc_state::{
        command::CommandScope,
        error::StateError,
        id::{AggregateId, NamespaceId, PartitionId},
        versioned::{ListVersionedDirectorySnapshot, VersionedDirectorySnapshotId},
    };

    use super::VersionedSourceClient;
    use crate::{state_audience, wire::DeclaredCall};

    type Bytes = bytes::Bytes;

    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 scope(aggregate: &str) -> CommandScope {
        CommandScope::new(
            AggregateId::new(aggregate),
            PartitionId::new("credentials"),
            NamespaceId::new("polychrome"),
        )
    }

    fn client(reply: &pb::GetVersionedSourceHeadReply) -> VersionedSourceClient<CannedTransport> {
        VersionedSourceClient::new(
            CannedTransport {
                frames: Arc::new(vec![Bytes::from(buffa::Message::encode_to_vec(reply))]),
            },
            ClientConfig::new("http://versioned.invalid".parse().unwrap()),
        )
    }

    fn reply(aggregate: &str) -> pb::GetVersionedSourceHeadReply {
        pb::GetVersionedSourceHeadReply {
            incarnation: vec![7; 32],
            head: buffa::MessageField::some(pb::JournalHead {
                position: 9,
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            }),
            aggregate: aggregate.to_owned(),
            partition: "credentials".to_owned(),
            namespace: "polychrome".to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }

    /// A successful reply about another aggregate is refused.
    ///
    /// The lineage identifies the log, so every aggregate in one log reports
    /// the same 32 bytes. Without the echo, a reply about another aggregate
    /// carries an incarnation that compares equal and is indistinguishable
    /// from the answer the reader asked for.
    #[tokio::test]
    async fn a_crossed_scope_reply_is_refused() {
        let refusal = client(&reply("another-aggregate"))
            .source_head(
                &DeclaredCall::live(state_audience(), Duration::MAX),
                &scope("credentials"),
            )
            .await
            .expect_err("the reply crossed the requested aggregate");
        let named_scope =
            matches!(&refusal, StateError::Malformed { field, .. } if field == "scope");
        assert!(named_scope, "expected a scope refusal, got {refusal:?}");
    }

    /// The reply the reader asked for is accepted.
    ///
    /// Without this, the test above would also pass for a client that refused
    /// every reply.
    #[tokio::test]
    async fn the_asked_scope_is_accepted() {
        let head = client(&reply("credentials"))
            .source_head(
                &DeclaredCall::live(state_audience(), Duration::MAX),
                &scope("credentials"),
            )
            .await
            .expect("the reply binds the asked aggregate");
        assert_eq!(head.head().position().get(), 9);
    }

    fn capture(family: &str) -> pb::CreateVersionedDirectorySnapshotReply {
        pb::CreateVersionedDirectorySnapshotReply {
            snapshot: "versioned-directory:test:1".to_owned(),
            namespace_count: 3,
            incarnation: vec![7; 32],
            family: family.to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }

    fn page(
        snapshot: &str,
        namespaces: &[&str],
        complete: bool,
        next_after: Option<&str>,
    ) -> pb::ListVersionedDirectorySnapshotReply {
        pb::ListVersionedDirectorySnapshotReply {
            snapshot: snapshot.to_owned(),
            namespaces: namespaces.iter().map(|n| (*n).to_owned()).collect(),
            complete,
            incarnation: vec![7; 32],
            next_after: next_after.map(str::to_owned),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }

    fn canned<M: buffa::Message>(reply: &M) -> VersionedSourceClient<CannedTransport> {
        VersionedSourceClient::new(
            CannedTransport {
                frames: Arc::new(vec![Bytes::from(buffa::Message::encode_to_vec(reply))]),
            },
            ClientConfig::new("http://versioned.invalid".parse().unwrap()),
        )
    }

    fn live() -> DeclaredCall {
        DeclaredCall::live(state_audience(), Duration::MAX)
    }

    fn request(limit: u32) -> ListVersionedDirectorySnapshot {
        ListVersionedDirectorySnapshot::new(
            VersionedDirectorySnapshotId::new("versioned-directory:test:1"),
            limit,
        )
    }

    /// A capture about another family is refused.
    #[tokio::test]
    async fn a_crossed_family_capture_is_refused() {
        let refusal = canned(&capture("other"))
            .create_directory_snapshot(&live(), "credentials")
            .await
            .expect_err("the reply crossed the requested family");
        let named = matches!(&refusal, StateError::Malformed { field, .. } if field == "family");
        assert!(named, "expected a family refusal, got {refusal:?}");
    }

    /// A capture carrying no identity is refused.
    ///
    /// A page request needs one. Accepting an empty identity would send a
    /// request the authority can only refuse, one round trip later.
    #[tokio::test]
    async fn a_capture_without_an_identity_is_refused() {
        let mut reply = capture("credentials");
        reply.snapshot = String::new();
        let refusal = canned(&reply)
            .create_directory_snapshot(&live(), "credentials")
            .await
            .expect_err("a capture without an identity must refuse");
        let named = matches!(&refusal, StateError::Malformed { field, .. } if field == "snapshot");
        assert!(named, "expected a snapshot refusal, got {refusal:?}");
    }

    /// The capture the reader asked for is accepted.
    #[tokio::test]
    async fn the_asked_capture_is_accepted() {
        let snapshot = canned(&capture("credentials"))
            .create_directory_snapshot(&live(), "credentials")
            .await
            .expect("the reply binds the asked family");
        assert_eq!(snapshot.namespace_count(), 3);
        assert_eq!(snapshot.id().as_str(), "versioned-directory:test:1");
    }

    /// A page naming another capture is refused.
    ///
    /// A caller assembling one directory must not fold a page taken from a
    /// different membership into it.
    #[tokio::test]
    async fn a_page_from_another_capture_is_refused() {
        let refusal = canned(&page("versioned-directory:test:2", &["a"], true, None))
            .directory_page(&live(), &request(8))
            .await
            .expect_err("the page crossed the requested capture");
        let named = matches!(&refusal, StateError::Malformed { field, .. } if field == "snapshot");
        assert!(named, "expected a snapshot refusal, got {refusal:?}");
    }

    /// A page that is not strictly ascending is refused, never sorted.
    ///
    /// A resume cursor is an exclusive namespace bound, so out-of-order
    /// namespaces would let a resume skip one. Sorting here would hide that
    /// the authority broke its own contract.
    #[tokio::test]
    async fn an_unordered_page_is_refused() {
        for namespaces in [&["b", "a"][..], &["a", "a"][..]] {
            let refusal = canned(&page("versioned-directory:test:1", namespaces, true, None))
                .directory_page(&live(), &request(8))
                .await
                .expect_err("an unordered page must refuse");
            let named =
                matches!(&refusal, StateError::Malformed { field, .. } if field == "namespaces");
            assert!(
                named,
                "expected an order refusal for {namespaces:?}, got {refusal:?}"
            );
        }
    }

    /// A page that disagrees with itself about having more to give is refused.
    ///
    /// A truncated page with no cursor strands a traversal part-way; a
    /// complete page with one would make it read a page that cannot exist.
    #[tokio::test]
    async fn a_page_that_contradicts_its_own_cursor_is_refused() {
        for reply in [
            page("versioned-directory:test:1", &["a"], false, None),
            page("versioned-directory:test:1", &["a"], true, Some("a")),
            page("versioned-directory:test:1", &["a"], false, Some("z")),
        ] {
            let refusal = canned(&reply)
                .directory_page(&live(), &request(8))
                .await
                .expect_err("a self-contradicting page must refuse");
            let named =
                matches!(&refusal, StateError::Malformed { field, .. } if field == "next_after");
            assert!(named, "expected a cursor refusal, got {refusal:?}");
        }
    }

    /// A page that does not start past its cursor is refused.
    #[tokio::test]
    async fn a_page_that_repeats_its_cursor_is_refused() {
        let refusal = canned(&page("versioned-directory:test:1", &["b"], true, Some("b")))
            .directory_page(&live(), &request(8).after("b"))
            .await
            .expect_err("a page must start past its cursor");
        let named =
            matches!(&refusal, StateError::Malformed { field, .. } if field == "namespaces");
        assert!(named, "expected an order refusal, got {refusal:?}");
    }

    /// The page the reader asked for is accepted, in order.
    #[tokio::test]
    async fn an_ordered_page_is_accepted() {
        let page = canned(&page(
            "versioned-directory:test:1",
            &["a", "b", "c"],
            true,
            None,
        ))
        .directory_page(&live(), &request(8))
        .await
        .expect("an ascending page is accepted");
        assert_eq!(
            page.namespaces(),
            ["a".to_owned(), "b".to_owned(), "c".to_owned()]
        );
        assert_eq!(page.next_after(), None);
    }
}