use std::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,
credentials::{CredentialDirectory, CredentialRead, CredentialWrite},
error::StateError,
id::{CommandId, NamespaceId},
versioned::{VersionedRead, VersionedTransact},
};
use crate::{
admission::{
AudienceBinding, PeerIdentity, check_audience_binding, check_call_context_version,
check_not_draining, check_transport_deadline, state_audience,
},
error::to_connect_error,
trace::adopt_caller_trace,
wire::{DeclaredCall, Kernel, declared_call},
};
use super::wire::{command_from_wire, index_to_wire, record_to_wire, request_to_wire};
pub trait CredentialAuthority: CredentialRead + CredentialWrite {
fn namespace(&self) -> &NamespaceId;
}
impl<V> CredentialAuthority for CredentialDirectory<V>
where
V: VersionedRead + VersionedTransact,
{
fn namespace(&self) -> &NamespaceId {
self.namespace()
}
}
pub struct CredentialSvc {
authority: Arc<dyn CredentialAuthority>,
draining: Arc<AtomicBool>,
binding: Arc<AudienceBinding>,
}
impl CredentialSvc {
#[must_use]
pub const fn new(
authority: Arc<dyn CredentialAuthority>,
draining: Arc<AtomicBool>,
binding: Arc<AudienceBinding>,
) -> Self {
Self {
authority,
draining,
binding,
}
}
#[must_use]
pub fn register_on(self, router: Router) -> Router {
use pb::StateCredentialServiceExt as _;
Arc::new(self).register(router)
}
fn peer(context: &RequestContext) -> PeerIdentity {
PeerIdentity::from_verified_leaf(
context
.peer_certs()
.and_then(<[_]>::first)
.map(|leaf| &**leaf),
)
}
fn admit(
&self,
context: &RequestContext,
wire: impl Into<Option<pb::CallContext>>,
method: &'static str,
) -> Result<(CallContext, tracing::Span), ConnectError> {
check_not_draining(self.draining.load(Ordering::Relaxed))?;
let declared: DeclaredCall =
declared_call(wire).map_err(|error| to_connect_error(&error))?;
let family = polyc_state::versioned::family();
check_call_context_version(declared.version).map_err(|error| to_connect_error(&error))?;
check_audience_binding(
&Self::peer(context),
&declared.audience,
&state_audience(),
&self.binding,
&family,
)
.map_err(|error| to_connect_error(&error))?;
check_transport_deadline(context.time_remaining(), &family)
.map_err(|error| to_connect_error(&error))?;
Ok((
declared.origin_relative_context(),
adopt_caller_trace(context.headers(), method),
))
}
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::versioned::family(),
}))
}
}
}
#[allow(refining_impl_trait)]
impl pb::StateCredentialService for CredentialSvc {
async fn transact(
&self,
context: RequestContext,
request: ServiceRequest<'_, pb::TransactStateCredentialRequest>,
) -> ServiceResult<pb::TransactStateCredentialReply> {
let message = request.to_owned_message();
let (call, span) = self.admit(&context, message.context, "TransactCredential")?;
let _entered = span.enter();
let metadata = message.metadata.into_option().ok_or_else(|| {
to_connect_error(&StateError::Malformed {
field: "metadata".into(),
reason: "credential command carries metadata".into(),
})
})?;
self.namespace(&metadata.namespace)?;
let operation = message.operation.into_option().ok_or_else(|| {
to_connect_error(&StateError::Malformed {
field: "operation".into(),
reason: "credential command carries one operation".into(),
})
})?;
let receipt = self
.authority
.transact(
command_from_wire(metadata, operation, message.request_record.into_option())
.map_err(|error| to_connect_error(&error))?,
&call,
)
.map_err(|error| to_connect_error(&error))?;
Response::ok(pb::TransactStateCredentialReply {
receipt: buffa::MessageField::some(pb::Receipt::from(Kernel(&receipt))),
__buffa_unknown_fields: buffa::UnknownFields::default(),
})
}
async fn get_credential(
&self,
context: RequestContext,
request: ServiceRequest<'_, pb::GetStateCredentialRequest>,
) -> ServiceResult<pb::GetStateCredentialReply> {
let message = request.to_owned_message();
let (call, span) = self.admit(&context, message.context, "GetCredential")?;
let _entered = span.enter();
self.namespace(&message.namespace)?;
let fact = self
.authority
.credential(&message.credential_id, &call)
.map_err(|error| to_connect_error(&error))?;
Response::ok(pb::GetStateCredentialReply {
record: fact
.value()
.as_ref()
.map_or_else(buffa::MessageField::none, |record| {
buffa::MessageField::some(record_to_wire(record))
}),
snapshot_revision: fact.snapshot_revision().get(),
entry_revision: fact
.entry_revision()
.map(polyc_state::revision::Revision::get),
__buffa_unknown_fields: buffa::UnknownFields::default(),
})
}
async fn get_request_record(
&self,
context: RequestContext,
request: ServiceRequest<'_, pb::GetStateCredentialRequestRecordRequest>,
) -> ServiceResult<pb::GetStateCredentialRequestRecordReply> {
let message = request.to_owned_message();
let (call, span) = self.admit(&context, message.context, "GetCredentialRequestRecord")?;
let _entered = span.enter();
self.namespace(&message.namespace)?;
let fact = self
.authority
.request(&message.operation_id, &call)
.map_err(|error| to_connect_error(&error))?;
Response::ok(pb::GetStateCredentialRequestRecordReply {
request_record: fact
.value()
.as_ref()
.map_or_else(buffa::MessageField::none, |value| {
buffa::MessageField::some(request_to_wire(value))
}),
snapshot_revision: fact.snapshot_revision().get(),
entry_revision: fact
.entry_revision()
.map(polyc_state::revision::Revision::get),
__buffa_unknown_fields: buffa::UnknownFields::default(),
})
}
async fn get_index(
&self,
context: RequestContext,
request: ServiceRequest<'_, pb::GetStateCredentialIndexRequest>,
) -> ServiceResult<pb::GetStateCredentialIndexReply> {
let message = request.to_owned_message();
let (call, span) = self.admit(&context, message.context, "GetCredentialIndex")?;
let _entered = span.enter();
self.namespace(&message.namespace)?;
let fact = self
.authority
.index(&call)
.map_err(|error| to_connect_error(&error))?;
Response::ok(pb::GetStateCredentialIndexReply {
index: buffa::MessageField::some(index_to_wire(fact.value())),
snapshot_revision: fact.snapshot_revision().get(),
entry_revision: fact
.entry_revision()
.map(polyc_state::revision::Revision::get),
__buffa_unknown_fields: buffa::UnknownFields::default(),
})
}
async fn list_credentials(
&self,
context: RequestContext,
request: ServiceRequest<'_, pb::ListStateCredentialsRequest>,
) -> ServiceResult<pb::ListStateCredentialsReply> {
let message = request.to_owned_message();
let (call, span) = self.admit(&context, message.context, "ListCredentials")?;
let _entered = span.enter();
self.namespace(&message.namespace)?;
let page = self
.authority
.list(message.after.as_deref(), message.limit, &call)
.map_err(|error| to_connect_error(&error))?;
Response::ok(pb::ListStateCredentialsReply {
records: page.records().iter().map(record_to_wire).collect(),
next_after: page.next_after().map(str::to_owned),
snapshot_revision: page.snapshot_revision().get(),
__buffa_unknown_fields: buffa::UnknownFields::default(),
})
}
async fn get_receipt(
&self,
context: RequestContext,
request: ServiceRequest<'_, pb::GetStateCredentialReceiptRequest>,
) -> ServiceResult<pb::GetStateCredentialReceiptReply> {
let message = request.to_owned_message();
let (_call, span) = self.admit(&context, message.context, "GetCredentialReceipt")?;
let _entered = span.enter();
self.namespace(&message.namespace)?;
let receipt = self
.authority
.committed_receipt(
&NamespaceId::new(message.namespace),
&CommandId::new(message.command_id),
)
.map_err(|error| to_connect_error(&error))?;
Response::ok(pb::GetStateCredentialReceiptReply {
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(),
})
}
}