aves 0.5.0

//! The ScriptNode guest RPC boundary.
//!
//! Operation descriptors are generated by `aves_macros`.  Since that generator
//! omits privileged request variants, accepting only generated operation
//! descriptors makes the exclusion structural rather than a runtime check.

use std::{error::Error as StdError, fmt, future::Future, pin::Pin, sync::Arc};

use serde_json::Value;
use thiserror::Error;

use crate::auth::Jwt;
use crate::bus::{BusIngress, BusRpcContext, OperationDispatchFuture, RequestOperationDescriptor};

/// Errors returned while invoking a generated guest operation.
#[derive(Debug, Error)]
enum GuestRpcError<RootError>
where
    RootError: StdError + 'static,
{
    /// The requested operation is not part of the generated normal surface.
    #[error("guest RPC operation is unavailable: {operation}")]
    OperationUnavailable { operation: String },
    /// The generated operation rejected the request or its transport failed.
    #[error(transparent)]
    Dispatch(#[from] RootError),
}

/// A private transport carrying one Flow-scoped node token.
///
/// The token is installed once when the node starts.  It has no accessor,
/// serialization, or formatting implementation, so callers can use the
/// generated bindings without receiving the credential back.
struct GuestRpcTransport {
    context: BusRpcContext,
}

impl GuestRpcTransport {
    /// Binds the received node token to the normal bus ingress.
    fn from_scoped_token(bus: BusIngress, token: Jwt) -> Self {
        Self {
            context: BusRpcContext::authenticated(bus, token),
        }
    }

    fn dispatch<RootError>(
        &self,
        operation: &'static RequestOperationDescriptor<BusRpcContext, RootError>,
        arguments: Value,
    ) -> OperationDispatchFuture<RootError>
    where
        RootError: Send + 'static,
    {
        (operation.dispatch)(self.context.clone(), arguments)
    }
}

/// Generated normal RPC bindings available to a ScriptNode guest.
///
/// The operation list is composed directly from generated normal RPC operation
/// descriptors. Those descriptors omit privileged request variants, so no
/// privileged request type or dispatcher can be represented here.
struct GuestRpcBindings<RootError: 'static> {
    transport: GuestRpcTransport,
    operations: Arc<[&'static RequestOperationDescriptor<BusRpcContext, RootError>]>,
}

impl<RootError> GuestRpcBindings<RootError>
where
    RootError: StdError + Send + 'static,
{
    /// Creates bindings from the complete generated normal RPC operation surface.
    fn new(
        transport: GuestRpcTransport,
        operations: Arc<[&'static RequestOperationDescriptor<BusRpcContext, RootError>]>,
    ) -> Self {
        Self { transport, operations }
    }

    async fn call_value(&self, qualified_operation: &str, input: Value) -> Result<Value, GuestRpcError<RootError>> {
        let operation = self
            .operations
            .iter()
            .copied()
            .find(|operation| operation.qualified_operation == qualified_operation)
            .ok_or_else(|| GuestRpcError::OperationUnavailable {
                operation: qualified_operation.to_owned(),
            })?;
        Ok(self.transport.dispatch(operation, input).await?)
    }
}

type ErasedCallFuture<'a> = Pin<Box<dyn Future<Output = Result<Value, String>> + Send + 'a>>;

trait ErasedGuestRpcBindings: Send + Sync {
    fn call_value<'a>(&'a self, operation: &'a str, input: Value) -> ErasedCallFuture<'a>;
}

struct TypedGuestRpcBindings<RootError: 'static>(GuestRpcBindings<RootError>);

impl<RootError> ErasedGuestRpcBindings for TypedGuestRpcBindings<RootError>
where
    RootError: StdError + Send + Sync + 'static,
{
    fn call_value<'a>(&'a self, operation: &'a str, input: Value) -> ErasedCallFuture<'a> {
        Box::pin(async move {
            self.0
                .call_value(operation, input)
                .await
                .map_err(|error| error.to_string())
        })
    }
}

trait ErasedGuestRpcFactory: Send + Sync {
    fn bind(&self, bus: BusIngress, token: Jwt) -> GuestRpcBindingsHandle;
}

struct TypedGuestRpcFactory<RootError: 'static> {
    operations: Arc<[&'static RequestOperationDescriptor<BusRpcContext, RootError>]>,
}

impl<RootError> ErasedGuestRpcFactory for TypedGuestRpcFactory<RootError>
where
    RootError: StdError + Send + Sync + 'static,
{
    fn bind(&self, bus: BusIngress, token: Jwt) -> GuestRpcBindingsHandle {
        let bindings = GuestRpcBindings::new(
            GuestRpcTransport::from_scoped_token(bus, token),
            Arc::clone(&self.operations),
        );
        GuestRpcBindingsHandle(Arc::new(TypedGuestRpcBindings(bindings)))
    }
}

/// Embedding-owned generated operation catalog used to mint token-baked ScriptNode bindings.
#[derive(Clone)]
pub struct GuestRpcBindingFactoryHandle(Arc<dyn ErasedGuestRpcFactory>);

impl fmt::Debug for GuestRpcBindingFactoryHandle {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str("GuestRpcBindingFactoryHandle(<generated normal operations>)")
    }
}

impl GuestRpcBindingFactoryHandle {
    pub(super) fn bind(&self, bus: BusIngress, token: Jwt) -> GuestRpcBindingsHandle {
        self.0.bind(bus, token)
    }
}

/// Creates a factory from the embedding application's generated normal RPC operation descriptors.
pub fn generated_guest_rpc_factory<RootError>(
    operation_sets: Vec<&'static [RequestOperationDescriptor<BusRpcContext, RootError>]>,
) -> GuestRpcBindingFactoryHandle
where
    RootError: StdError + Send + Sync + 'static,
{
    let operations = operation_sets
        .into_iter()
        .flat_map(|operations| operations.iter())
        .collect::<Vec<_>>()
        .into();
    GuestRpcBindingFactoryHandle(Arc::new(TypedGuestRpcFactory { operations }))
}

/// Token-baked generated bindings owned by one running ScriptNode.
pub(super) struct GuestRpcBindingsHandle(Arc<dyn ErasedGuestRpcBindings>);

impl GuestRpcBindingsHandle {
    pub(super) async fn call_value(&self, operation: &str, input: Value) -> Result<Value, String> {
        self.0.call_value(operation, input).await
    }
}

#[cfg(test)]
struct FixedTestBindings(Value);

#[cfg(test)]
impl ErasedGuestRpcBindings for FixedTestBindings {
    fn call_value<'a>(&'a self, _operation: &'a str, _input: Value) -> ErasedCallFuture<'a> {
        Box::pin(async move { Ok(self.0.clone()) })
    }
}

#[cfg(test)]
impl GuestRpcBindingsHandle {
    pub(super) fn fixed_test_response(value: Value) -> Self {
        Self(Arc::new(FixedTestBindings(value)))
    }
}