polyc-state-connect 2026.10.2

//! Server adapter for the durable wallet-link and spend-policy authority.

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

use connectrpc::{ConnectError, RequestContext, Response, Router, ServiceRequest, ServiceResult};
use polyc_proto::proto::polychrome::state::v1 as pb;
use polyc_state::{
    context::CallContext,
    error::StateError,
    id::{CommandId, NamespaceId},
    versioned::{VersionedRead, VersionedTransact},
    wallet::{PersonaId, WalletAddress, WalletDirectory, WalletRead, WalletWrite},
};

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},
};

use super::wire::{command_from_wire, link_to_wire, pointer_to_wire, policy_to_wire};

/// Complete, capability-specific wallet authority port.
pub trait WalletAuthority: WalletRead + WalletWrite {
    /// Namespace this authority owns.
    fn namespace(&self) -> &NamespaceId;
}

impl<V> WalletAuthority for WalletDirectory<V>
where
    V: VersionedRead + VersionedTransact,
{
    fn namespace(&self) -> &NamespaceId {
        self.namespace()
    }
}

/// State's wallet-link and spend-policy service.
pub struct WalletSvc {
    authority: Arc<dyn WalletAuthority>,
    draining: Arc<AtomicBool>,
    binding: Arc<AudienceBinding>,
}

impl WalletSvc {
    /// Builds the service.
    #[must_use]
    pub const fn new(
        authority: Arc<dyn WalletAuthority>,
        draining: Arc<AtomicBool>,
        binding: Arc<AudienceBinding>,
    ) -> Self {
        Self {
            authority,
            draining,
            binding,
        }
    }

    /// Registers the capability-specific route family.
    #[must_use]
    pub fn register_on(self, router: Router) -> Router {
        use pb::StateWalletServiceExt as _;
        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<pb::CallContext>>,
        method: &'static str,
    ) -> Result<(CallContext, 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 = polyc_state::wallet::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))?;
        Ok((
            declared.origin_relative_context(),
            adopt_caller_trace(ctx.headers(), method),
            slot_wait(ctx.time_remaining(), &declared),
        ))
    }

    fn namespace(&self, value: &str) -> Result<(), ConnectError> {
        if value == self.authority.namespace().as_str() {
            Ok(())
        } else {
            Err(to_connect_error(&StateError::Denied {
                family: polyc_state::wallet::family(),
            }))
        }
    }
}

#[allow(refining_impl_trait)]
impl pb::StateWalletService for WalletSvc {
    fn transact(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, pb::TransactStateWalletRequest>,
    ) -> impl Future<Output = ServiceResult<pb::TransactStateWalletReply>> {
        let message = request.to_owned_message();
        async move {
            let (context, span, wait) = self.admit(&ctx, message.context, "TransactStateWallet")?;
            let metadata = message.metadata.into_option().ok_or_else(|| {
                malformed_connect("metadata", "a wallet command carries metadata")
            })?;
            self.namespace(&metadata.namespace)?;
            let operation = message.operation.into_option().ok_or_else(|| {
                malformed_connect("operation", "a wallet command carries one operation")
            })?;
            let command = command_from_wire(metadata, message.persona_id, operation)
                .map_err(|error| to_connect_error(&error))?;
            let authority = Arc::clone(&self.authority);
            let receipt =
                blocking::mutation(span, polyc_state::wallet::family(), wait, move || {
                    authority.transact(command, &context)
                })
                .await?;
            Response::ok(pb::TransactStateWalletReply {
                receipt: buffa::MessageField::some(pb::Receipt::from(Kernel(&receipt))),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        }
    }

    fn get_link(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, pb::GetStateWalletLinkRequest>,
    ) -> impl Future<Output = ServiceResult<pb::GetStateWalletLinkReply>> {
        let message = request.to_owned_message();
        async move {
            let (context, span, wait) = self.admit(&ctx, message.context, "GetStateWalletLink")?;
            self.namespace(&message.namespace)?;
            let persona_id = PersonaId::new(message.persona_id);
            let authority = Arc::clone(&self.authority);
            let fact = blocking::read(span, polyc_state::wallet::family(), wait, move || {
                authority.link(&persona_id, &context)
            })
            .await?;
            Response::ok(pb::GetStateWalletLinkReply {
                link: fact
                    .value()
                    .as_ref()
                    .map_or_else(buffa::MessageField::none, |link| {
                        buffa::MessageField::some(link_to_wire(link))
                    }),
                snapshot_revision: fact.snapshot_revision().get(),
                entry_revision: fact
                    .entry_revision()
                    .map(polyc_state::revision::Revision::get),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        }
    }

    fn get_policy(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, pb::GetStateWalletPolicyRequest>,
    ) -> impl Future<Output = ServiceResult<pb::GetStateWalletPolicyReply>> {
        let message = request.to_owned_message();
        async move {
            let (context, span, wait) =
                self.admit(&ctx, message.context, "GetStateWalletPolicy")?;
            self.namespace(&message.namespace)?;
            let persona_id = PersonaId::new(message.persona_id);
            let authority = Arc::clone(&self.authority);
            let fact = blocking::read(span, polyc_state::wallet::family(), wait, move || {
                authority.policy(&persona_id, &context)
            })
            .await?;
            Response::ok(pb::GetStateWalletPolicyReply {
                policy: fact
                    .value()
                    .as_ref()
                    .map_or_else(buffa::MessageField::none, |policy| {
                        buffa::MessageField::some(policy_to_wire(policy))
                    }),
                snapshot_revision: fact.snapshot_revision().get(),
                entry_revision: fact
                    .entry_revision()
                    .map(polyc_state::revision::Revision::get),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        }
    }

    fn get_view(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, pb::GetStateWalletViewRequest>,
    ) -> impl Future<Output = ServiceResult<pb::GetStateWalletViewReply>> {
        let message = request.to_owned_message();
        async move {
            let (context, span, wait) = self.admit(&ctx, message.context, "GetStateWalletView")?;
            self.namespace(&message.namespace)?;
            let persona_id = PersonaId::new(message.persona_id);
            let authority = Arc::clone(&self.authority);
            let view = blocking::read(span, polyc_state::wallet::family(), wait, move || {
                authority.view(&persona_id, &context)
            })
            .await?;
            Response::ok(pb::GetStateWalletViewReply {
                link: view.link().map_or_else(buffa::MessageField::none, |link| {
                    buffa::MessageField::some(link_to_wire(link))
                }),
                policy: view
                    .policy()
                    .map_or_else(buffa::MessageField::none, |policy| {
                        buffa::MessageField::some(policy_to_wire(policy))
                    }),
                snapshot_revision: view.snapshot_revision().get(),
                link_revision: revision_of(view.expected_link()),
                policy_revision: revision_of(view.expected_policy()),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        }
    }

    fn resolve_address(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, pb::ResolveStateWalletAddressRequest>,
    ) -> impl Future<Output = ServiceResult<pb::ResolveStateWalletAddressReply>> {
        let message = request.to_owned_message();
        async move {
            let (context, span, wait) =
                self.admit(&ctx, message.context, "ResolveStateWalletAddress")?;
            self.namespace(&message.namespace)?;
            let address = WalletAddress::new(message.wallet_address);
            let authority = Arc::clone(&self.authority);
            let fact = blocking::read(span, polyc_state::wallet::family(), wait, move || {
                authority.resolve_address(&address, &context)
            })
            .await?;
            Response::ok(pb::ResolveStateWalletAddressReply {
                pointer: fact
                    .value()
                    .as_ref()
                    .map_or_else(buffa::MessageField::none, |pointer| {
                        buffa::MessageField::some(pointer_to_wire(pointer))
                    }),
                snapshot_revision: fact.snapshot_revision().get(),
                entry_revision: fact
                    .entry_revision()
                    .map(polyc_state::revision::Revision::get),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        }
    }

    fn get_receipt(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, pb::GetStateWalletReceiptRequest>,
    ) -> impl Future<Output = ServiceResult<pb::GetStateWalletReceiptReply>> {
        let message = request.to_owned_message();
        async move {
            let (_context, span, wait) =
                self.admit(&ctx, message.context, "GetStateWalletReceipt")?;
            self.namespace(&message.namespace)?;
            let namespace = NamespaceId::new(message.namespace);
            let command_id = CommandId::new(message.command_id);
            let authority = Arc::clone(&self.authority);
            let receipt = blocking::read(span, polyc_state::wallet::family(), wait, move || {
                authority.committed_receipt(&namespace, &command_id)
            })
            .await?;
            Response::ok(pb::GetStateWalletReceiptReply {
                receipt: receipt
                    .as_ref()
                    .map_or_else(buffa::MessageField::none, |value| {
                        buffa::MessageField::some(pb::Receipt::from(Kernel(value)))
                    }),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        }
    }
}

/// Renders a row premise as the optional revision the reply carries.
const fn revision_of(expected: polyc_state::versioned::EntryExpectation) -> Option<u64> {
    match expected {
        polyc_state::versioned::EntryExpectation::Absent => None,
        polyc_state::versioned::EntryExpectation::Revision(revision) => Some(revision.get()),
    }
}

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