polyc-state-connect 2026.8.3

//! Capability-specific persona-directory client.

use connectrpc::client::{ClientConfig, ClientTransport};
use polyc_proto::proto::polychrome::state::v1 as pb;
use polyc_state::{
    error::StateError,
    id::{CommandId, NamespaceId},
    persona::{
        IdentityLink, LinkAttemptKey, LinkAttempts, LinkChallenge, LinkCodeDigest, Participation,
        ParticipationScope, PersonaFact, PersonaProfile, PersonaSnapshot, PersonaTransaction,
        SearchVisibility,
    },
    receipt::Receipt,
    revision::Revision,
    versioned::authority::{ExternalIdentity, PersonaId, ScopeId},
};

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

use super::wire::{
    attempt_key_to_wire, attempts_from_wire, challenge_from_wire, command_to_wire,
    identity_to_wire, link_from_wire, mutation_to_wire, participation_from_wire, profile_from_wire,
    receipt, visibility_from_wire,
};

/// A client that can access one persona-directory namespace and nothing else.
pub struct PersonaClient<T> {
    inner: pb::StatePersonaServiceClient<T>,
    namespace: NamespaceId,
}

impl<T> PersonaClient<T>
where
    T: ClientTransport,
    <T::ResponseBody as connectrpc::http_body::Body>::Error: std::fmt::Display,
{
    /// Builds a client bound to one directory namespace.
    #[must_use]
    pub fn new(transport: T, config: ClientConfig, namespace: NamespaceId) -> Self {
        Self {
            inner: pb::StatePersonaServiceClient::new(
                transport,
                config.with_default_max_message_size(MAX_WIRE_MESSAGE_BYTES),
            ),
            namespace,
        }
    }

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

    /// Commits one typed persona transaction.
    ///
    /// # Errors
    ///
    /// Returns the exact State or transport outcome.
    pub async fn transact(
        &self,
        declared: &DeclaredCall,
        command: &PersonaTransaction,
    ) -> Result<Receipt, StateError> {
        if command.metadata().scope().namespace() != &self.namespace {
            return Err(StateError::Denied {
                family: polyc_state::versioned::family(),
            });
        }
        let request = pb::TransactPersonaRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            metadata: buffa::MessageField::some(command_to_wire(command)),
            mutations: command.mutations().iter().map(mutation_to_wire).collect(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .transact_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        receipt(reply.receipt)
    }

    /// Resolves an external identity without provisioning.
    ///
    /// # Errors
    ///
    /// Returns the exact State or transport outcome.
    pub async fn resolve(
        &self,
        declared: &DeclaredCall,
        identity: &ExternalIdentity,
    ) -> Result<PersonaFact<Option<IdentityLink>>, StateError> {
        let request = pb::ResolvePersonaIdentityRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: self.namespace.as_str().to_owned(),
            identity: buffa::MessageField::some(identity_to_wire(identity)),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .resolve_identity_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        Ok(PersonaFact::observed(
            reply.link.into_option().map(link_from_wire).transpose()?,
            Revision::new(reply.snapshot_revision),
            reply.entry_revision.map(Revision::new),
        ))
    }

    /// Reads one current profile.
    ///
    /// # Errors
    ///
    /// Returns the exact State or transport outcome.
    pub async fn profile(
        &self,
        declared: &DeclaredCall,
        persona: &PersonaId,
    ) -> Result<PersonaFact<Option<PersonaProfile>>, StateError> {
        let request = pb::GetPersonaProfileRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: self.namespace.as_str().to_owned(),
            persona: persona.as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .get_profile_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        Ok(PersonaFact::observed(
            reply
                .profile
                .into_option()
                .map(profile_from_wire)
                .transpose()?,
            Revision::new(reply.snapshot_revision),
            reply.entry_revision.map(Revision::new),
        ))
    }

    /// Reads a profile and its complete identity presentation at one revision.
    ///
    /// # Errors
    ///
    /// Returns typed transport, deadline, bounds, availability, and malformed
    /// response errors from the State contract.
    pub async fn snapshot(
        &self,
        declared: &DeclaredCall,
        persona: &PersonaId,
    ) -> Result<PersonaFact<PersonaSnapshot>, StateError> {
        let request = pb::GetPersonaSnapshotRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: self.namespace.as_str().to_owned(),
            persona: persona.as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .get_snapshot_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        Ok(PersonaFact::observed(
            PersonaSnapshot {
                profile: reply
                    .profile
                    .into_option()
                    .map(profile_from_wire)
                    .transpose()?,
                identity_links: reply
                    .identity_links
                    .into_iter()
                    .map(link_from_wire)
                    .collect::<Result<_, _>>()?,
            },
            Revision::new(reply.snapshot_revision),
            reply.entry_revision.map(Revision::new),
        ))
    }

    /// Reads one current participation.
    ///
    /// # Errors
    ///
    /// Returns the exact State or transport outcome.
    pub async fn participation(
        &self,
        declared: &DeclaredCall,
        persona: &PersonaId,
        scope: &ScopeId,
    ) -> Result<PersonaFact<Option<Participation>>, StateError> {
        let request = pb::GetPersonaParticipationRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: self.namespace.as_str().to_owned(),
            persona: persona.as_str().to_owned(),
            scope: scope.as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .get_participation_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        Ok(PersonaFact::observed(
            reply
                .participation
                .into_option()
                .map(participation_from_wire)
                .transpose()?,
            Revision::new(reply.snapshot_revision),
            reply.entry_revision.map(Revision::new),
        ))
    }

    /// Reads one current search-visibility decision.
    ///
    /// # Errors
    ///
    /// Returns the exact State or transport outcome.
    pub async fn search_visibility(
        &self,
        declared: &DeclaredCall,
        persona: &PersonaId,
        scope: &ScopeId,
    ) -> Result<PersonaFact<Option<SearchVisibility>>, StateError> {
        let request = pb::GetPersonaSearchVisibilityRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: self.namespace.as_str().to_owned(),
            persona: persona.as_str().to_owned(),
            scope: scope.as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .get_search_visibility_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        Ok(PersonaFact::observed(
            reply.visibility.into_option().map(visibility_from_wire),
            Revision::new(reply.snapshot_revision),
            reply.entry_revision.map(Revision::new),
        ))
    }

    /// Lists a bounded coherent participation set.
    ///
    /// # Errors
    ///
    /// Returns the exact State or transport outcome.
    pub async fn participations(
        &self,
        declared: &DeclaredCall,
        persona: &PersonaId,
        cap: u32,
    ) -> Result<PersonaFact<Vec<Participation>>, StateError> {
        let request = pb::ListPersonaParticipationsRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: self.namespace.as_str().to_owned(),
            persona: persona.as_str().to_owned(),
            cap,
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .list_participations_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        Ok(PersonaFact::observed(
            reply
                .participations
                .into_iter()
                .map(participation_from_wire)
                .collect::<Result<_, _>>()?,
            Revision::new(reply.snapshot_revision),
            reply.entry_revision.map(Revision::new),
        ))
    }

    /// Lists the bounded canonical persona-id index.
    ///
    /// # Errors
    ///
    /// Returns the exact State or transport outcome.
    pub async fn personas(
        &self,
        declared: &DeclaredCall,
        cap: u32,
    ) -> Result<PersonaFact<Vec<PersonaId>>, StateError> {
        let request = pb::ListPersonasRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: self.namespace.as_str().to_owned(),
            cap,
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .list_personas_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        Ok(PersonaFact::observed(
            reply.personas.into_iter().map(PersonaId::new).collect(),
            Revision::new(reply.snapshot_revision),
            reply.entry_revision.map(Revision::new),
        ))
    }

    /// Resolves visible participation at one authority snapshot.
    ///
    /// # Errors
    ///
    /// Returns the exact State or transport outcome.
    pub async fn participation_scope(
        &self,
        declared: &DeclaredCall,
        persona: &PersonaId,
        cap: u32,
    ) -> Result<PersonaFact<ParticipationScope>, StateError> {
        let request = pb::ResolvePersonaParticipationScopeRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: self.namespace.as_str().to_owned(),
            persona: persona.as_str().to_owned(),
            cap,
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .resolve_participation_scope_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        let scope = reply.refused_count.map_or_else(
            || {
                ParticipationScope::Resolved(
                    reply
                        .visible_scopes
                        .into_iter()
                        .map(polyc_state::versioned::authority::ScopeId::new)
                        .collect(),
                )
            },
            |count| ParticipationScope::RefusedOverCap {
                count: usize::try_from(count).unwrap_or(usize::MAX),
            },
        );
        Ok(PersonaFact::observed(
            scope,
            Revision::new(reply.snapshot_revision),
            reply.entry_revision.map(Revision::new),
        ))
    }

    /// Reads one digest-addressed generic link challenge.
    ///
    /// # Errors
    ///
    /// Returns typed transport, deadline, availability, and malformed response
    /// errors from the State contract.
    pub async fn link_challenge(
        &self,
        declared: &DeclaredCall,
        digest: &LinkCodeDigest,
    ) -> Result<PersonaFact<Option<LinkChallenge>>, StateError> {
        let request = pb::GetPersonaLinkChallengeRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: self.namespace.as_str().to_owned(),
            digest: digest.as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .get_link_challenge_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        Ok(PersonaFact::observed(
            reply.challenge.into_option().map(challenge_from_wire),
            Revision::new(reply.snapshot_revision),
            reply.entry_revision.map(Revision::new),
        ))
    }

    /// Reads one fixed-window generic-link abuse counter.
    ///
    /// # Errors
    ///
    /// Returns typed transport, deadline, availability, and malformed response
    /// errors from the State contract.
    pub async fn link_attempts(
        &self,
        declared: &DeclaredCall,
        key: &LinkAttemptKey,
    ) -> Result<PersonaFact<Option<LinkAttempts>>, StateError> {
        let request = pb::GetPersonaLinkAttemptsRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: self.namespace.as_str().to_owned(),
            key: buffa::MessageField::some(attempt_key_to_wire(key)),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .get_link_attempts_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        Ok(PersonaFact::observed(
            reply
                .attempts
                .into_option()
                .map(|value| attempts_from_wire(&value)),
            Revision::new(reply.snapshot_revision),
            reply.entry_revision.map(Revision::new),
        ))
    }

    /// Reads the pending invite pointer for one target identity.
    ///
    /// # Errors
    ///
    /// Returns typed transport, deadline, availability, and malformed response
    /// errors from the State contract.
    pub async fn pending_invite(
        &self,
        declared: &DeclaredCall,
        identity: &ExternalIdentity,
    ) -> Result<PersonaFact<Option<LinkCodeDigest>>, StateError> {
        let request = pb::GetPersonaPendingInviteRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: self.namespace.as_str().to_owned(),
            identity: buffa::MessageField::some(identity_to_wire(identity)),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .get_pending_invite_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        Ok(PersonaFact::observed(
            reply.digest.map(LinkCodeDigest::new),
            Revision::new(reply.snapshot_revision),
            reply.entry_revision.map(Revision::new),
        ))
    }

    /// Retrieves a committed persona transaction receipt.
    ///
    /// # Errors
    ///
    /// Returns the exact State or transport outcome.
    pub async fn committed_receipt(
        &self,
        declared: &DeclaredCall,
        command_id: &CommandId,
    ) -> Result<Option<Receipt>, StateError> {
        let request = pb::GetPersonaReceiptRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: self.namespace.as_str().to_owned(),
            command_id: command_id.as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .get_receipt_with_options(request, 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))
            .transpose()
    }
}