polyc-state-connect 2026.10.2

//! 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},
    page::{Cursor, PageCompleteness},
    persona::{
        IdentityLink, LinkAttemptKey, LinkAttempts, LinkChallenge, LinkCodeDigest, Participation,
        ParticipationScope, PersonaChangePage, PersonaFact, PersonaProfile, PersonaSnapshot,
        PersonaTransaction, SearchVisibility,
    },
    receipt::Receipt,
    revision::{JournalPosition, PartitionIncarnation, Revision},
    versioned::authority::{ExternalIdentity, PersonaId, ScopeId},
};

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

use super::wire::{
    attempt_key_to_wire, attempts_from_wire, challenge_from_wire, change_from_wire,
    command_to_wire, identity_to_wire, link_from_wire, merge_id_ticket_from_wire,
    merge_request_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(declared: &DeclaredCall, attempted: usize) -> TransportFallback {
        TransportFallback::new(
            polyc_state::versioned::family(),
            MAX_WIRE_MESSAGE_BYTES as u64,
            attempted as u64,
            declared.budget,
        )
    }

    /// 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(declared, 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(declared, 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(declared, 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(declared, 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(declared, 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(declared, 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(declared, 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(declared, 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(declared, 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(declared, 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(declared, 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(declared, attempted)))?
            .into_owned();
        Ok(PersonaFact::observed(
            reply.digest.map(LinkCodeDigest::new),
            Revision::new(reply.snapshot_revision),
            reply.entry_revision.map(Revision::new),
        ))
    }

    /// Reads one digest-addressed dual-approval merge request.
    ///
    /// # Errors
    ///
    /// Returns typed transport, deadline, availability, and malformed response
    /// errors from the State contract.
    pub async fn merge_request(
        &self,
        declared: &DeclaredCall,
        digest: &LinkCodeDigest,
    ) -> Result<PersonaFact<Option<polyc_state::persona::MergeRequest>>, StateError> {
        let request = pb::GetPersonaMergeRequestRequest {
            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_merge_request_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(declared, attempted)))?
            .into_owned();
        Ok(PersonaFact::observed(
            reply
                .request
                .into_option()
                .map(merge_request_from_wire)
                .transpose()?,
            Revision::new(reply.snapshot_revision),
            reply.entry_revision.map(Revision::new),
        ))
    }

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

    /// Reads one digest-addressed merge-id ticket.
    ///
    /// # Errors
    ///
    /// Returns typed transport, deadline, availability, and malformed response
    /// errors from the State contract.
    pub async fn merge_id_ticket(
        &self,
        declared: &DeclaredCall,
        digest: &LinkCodeDigest,
    ) -> Result<PersonaFact<Option<polyc_state::persona::MergeIdTicket>>, StateError> {
        let request = pb::GetPersonaMergeIdTicketRequest {
            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_merge_id_ticket_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(declared, attempted)))?
            .into_owned();
        Ok(PersonaFact::observed(
            reply
                .ticket
                .into_option()
                .map(merge_id_ticket_from_wire)
                .transpose()?,
            Revision::new(reply.snapshot_revision),
            reply.entry_revision.map(Revision::new),
        ))
    }

    /// Reads the live merge-id ticket digest for one persona.
    ///
    /// # Errors
    ///
    /// Returns typed transport, deadline, availability, and malformed response
    /// errors from the State contract.
    pub async fn outgoing_merge_id(
        &self,
        declared: &DeclaredCall,
        persona: &PersonaId,
    ) -> Result<PersonaFact<Option<LinkCodeDigest>>, StateError> {
        let request = pb::GetPersonaOutgoingMergeIdRequest {
            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_outgoing_merge_id_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(declared, attempted)))?
            .into_owned();
        Ok(PersonaFact::observed(
            reply.digest.map(LinkCodeDigest::new),
            Revision::new(reply.snapshot_revision),
            reply.entry_revision.map(Revision::new),
        ))
    }

    /// Reads one bounded, ordered page of directory changes.
    ///
    /// # Errors
    ///
    /// Returns the State authority's typed refusal or a bounded transport
    /// failure. A position the authority never issued is refused there, and
    /// this neither retries it as a fresh read nor clamps it.
    pub async fn changes(
        &self,
        declared: &DeclaredCall,
        after: Option<JournalPosition>,
        expected_lineage: Option<PartitionIncarnation>,
        limit: u32,
    ) -> Result<PersonaChangePage, StateError> {
        let request = pb::ListStatePersonaChangesRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: self.namespace.as_str().to_owned(),
            after: after.map(JournalPosition::get),
            expected_lineage: expected_lineage.map(|lineage| lineage.as_bytes().to_vec()),
            limit,
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .list_changes_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(declared, attempted)))?
            .into_owned();
        let completeness = match reply.completeness {
            buffa::EnumValue::Known(pb::PageCompleteness::Complete) => PageCompleteness::Complete,
            buffa::EnumValue::Known(pb::PageCompleteness::Truncated) => PageCompleteness::Truncated,
            buffa::EnumValue::Known(pb::PageCompleteness::Unspecified)
            | buffa::EnumValue::Unknown(_) => {
                return Err(malformed(
                    "page.completeness",
                    "a change page declares whether the bound cut it short",
                ));
            }
        };
        Ok(PersonaChangePage::observed(
            reply
                .changes
                .into_iter()
                .map(change_from_wire)
                .collect::<Result<_, _>>()?,
            reply
                .next_after
                .map(|position| Cursor::at(JournalPosition::new(position))),
            completeness,
            Revision::new(reply.frontier),
            PartitionIncarnation::from_bytes(fixed_bytes::<32>("lineage", &reply.lineage)?),
        ))
    }

    /// 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(declared, attempted)))?
            .into_owned();
        reply
            .receipt
            .into_option()
            .map(|value| Kernel::<Receipt>::try_from(value).map(Kernel::into_inner))
            .transpose()
    }
}