distributed 1.5.6

//! gRPC transport for microsvc — maps gRPC requests to command dispatch.
//!
//! Requires the `grpc` feature. Uses tonic for the gRPC server and prost
//! for message serialization (standard protobuf wire format, no `.proto` file).
//!
//! ## RPCs
//!
//! - `Dispatch` — dispatch a command. Input = `GrpcRequest`, output = `GrpcResponse`.
//! - `Health` — health check returning available commands.
//!
//! ## Example
//!
//! ```ignore
//! use std::sync::Arc;
//! use distributed::{microsvc, HashMapRepository};
//!
//! let service = Arc::new(
//!     microsvc::Service::new().with_repo(HashMapRepository::new())
//!         .command("counter.create")
//!         .handle(|ctx| { /* ... */ })
//! );
//!
//! // Get the server to compose with other tonic routes
//! let grpc_svc = microsvc::grpc_server(service.clone());
//!
//! // Or serve directly
//! microsvc::serve_grpc(service, "[::1]:50051").await?;
//! ```

use std::collections::HashMap;
use std::error::Error;
use std::fmt;
use std::net::{AddrParseError, SocketAddr};
use std::sync::Arc;

use serde_json::json;
use tonic::{Request, Response, Status};

use super::service::Service;
use super::session::Session;

// ---------------------------------------------------------------------------
// Message types (prost — standard protobuf wire format)
// ---------------------------------------------------------------------------

#[derive(Clone, prost::Message)]
pub struct GrpcRequest {
    #[prost(string, tag = "1")]
    pub command: String,
    #[prost(string, tag = "2")]
    pub input: String, // JSON string
    #[prost(map = "string, string", tag = "3")]
    pub session_variables: HashMap<String, String>,
}

#[derive(Clone, prost::Message)]
pub struct GrpcResponse {
    #[prost(uint32, tag = "1")]
    pub status: u32,
    #[prost(string, tag = "2")]
    pub body: String, // JSON string
}

#[derive(Clone, prost::Message)]
pub struct HealthRequest {}

#[derive(Clone, prost::Message)]
pub struct HealthResponse {
    #[prost(bool, tag = "1")]
    pub ok: bool,
    #[prost(string, repeated, tag = "2")]
    pub commands: Vec<String>,
}

// ---------------------------------------------------------------------------
// Generated service trait + server/client
// ---------------------------------------------------------------------------

include!(concat!(
    env!("OUT_DIR"),
    "/sourced.microsvc.CommandService.rs"
));

pub use command_service_client::CommandServiceClient;
pub use command_service_server::{CommandService, CommandServiceServer};

/// Error returned when serving the gRPC transport fails.
#[derive(Debug)]
pub enum GrpcServeError {
    /// The supplied bind address could not be parsed as a socket address.
    InvalidAddress {
        /// Original bind address string supplied by the caller.
        addr: String,
        /// Address parser error.
        source: AddrParseError,
    },
    /// The tonic transport server failed while serving.
    Transport(tonic::transport::Error),
}

impl fmt::Display for GrpcServeError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            GrpcServeError::InvalidAddress { addr, source } => {
                write!(f, "invalid gRPC bind address `{addr}`: {source}")
            }
            GrpcServeError::Transport(source) => write!(f, "gRPC transport error: {source}"),
        }
    }
}

impl Error for GrpcServeError {
    fn source(&self) -> Option<&(dyn Error + 'static)> {
        match self {
            GrpcServeError::InvalidAddress { source, .. } => Some(source),
            GrpcServeError::Transport(source) => Some(source),
        }
    }
}

impl From<tonic::transport::Error> for GrpcServeError {
    fn from(source: tonic::transport::Error) -> Self {
        GrpcServeError::Transport(source)
    }
}

// ---------------------------------------------------------------------------
// Handler implementation
// ---------------------------------------------------------------------------

/// gRPC handler that wraps a `Service<D>` and implements the generated
/// `CommandService` trait. Mirrors the HTTP transport pattern.
pub struct GrpcHandler<D> {
    service: Arc<Service<D>>,
}

impl<D> GrpcHandler<D> {
    pub fn new(service: Arc<Service<D>>) -> Self {
        Self { service }
    }
}

#[tonic::async_trait]
impl<D: Send + Sync + 'static> CommandService for GrpcHandler<D> {
    async fn dispatch(
        &self,
        request: Request<GrpcRequest>,
    ) -> Result<Response<GrpcResponse>, Status> {
        let metadata = request.metadata().clone();
        let req = request.into_inner();

        // Parse JSON input (strict; invalid payloads return 400)
        let input: serde_json::Value = match serde_json::from_str(&req.input) {
            Ok(value) => value,
            Err(e) => {
                return Ok(Response::new(GrpcResponse {
                    status: 400,
                    body: json!({ "error": format!("invalid JSON input: {e}") }).to_string(),
                }));
            }
        };

        // Build session: start with metadata headers, then overlay payload values
        let session = build_session(&metadata, req.session_variables);

        match self.service.dispatch(&req.command, input, session).await {
            Ok(value) => Ok(Response::new(GrpcResponse {
                status: 200,
                body: value.to_string(),
            })),
            Err(e) => Ok(Response::new(GrpcResponse {
                status: e.status_code() as u32,
                body: json!({ "error": e.to_string() }).to_string(),
            })),
        }
    }

    async fn health(
        &self,
        _request: Request<HealthRequest>,
    ) -> Result<Response<HealthResponse>, Status> {
        let commands: Vec<String> = self
            .service
            .command_names()
            .into_iter()
            .map(|s| s.to_string())
            .collect();

        Ok(Response::new(HealthResponse { ok: true, commands }))
    }
}

// ---------------------------------------------------------------------------
// Session building (same pattern as HTTP transport)
// ---------------------------------------------------------------------------

/// Build a session from gRPC metadata and payload session variables.
///
/// 1. Start with gRPC metadata headers (lowercased key → value)
/// 2. Overlay payload `session_variables` (payload takes precedence)
fn build_session(
    metadata: &tonic::metadata::MetadataMap,
    payload_vars: HashMap<String, String>,
) -> Session {
    let mut vars = HashMap::new();

    // Metadata headers
    for kv in metadata.iter() {
        if let tonic::metadata::KeyAndValueRef::Ascii(key, value) = kv {
            if let Ok(v) = value.to_str() {
                vars.insert(key.as_str().to_string(), v.to_string());
            }
        }
    }

    // Payload overrides
    for (k, v) in payload_vars {
        vars.insert(k, v);
    }

    Session::from_map(vars)
}

// ---------------------------------------------------------------------------
// Convenience constructors
// ---------------------------------------------------------------------------

/// Create a `CommandServiceServer` from a shared `Service<D>`.
pub fn grpc_server<D: Send + Sync + 'static>(
    service: Arc<Service<D>>,
) -> CommandServiceServer<GrpcHandler<D>> {
    CommandServiceServer::new(GrpcHandler::new(service))
}

/// Bind and serve the gRPC transport at the given address (e.g. `"[::1]:50051"`).
///
/// Returns [`GrpcServeError::InvalidAddress`] when `addr` is not a valid socket
/// address, and [`GrpcServeError::Transport`] for errors returned by tonic while
/// serving.
pub async fn serve_grpc<D: Send + Sync + 'static>(
    service: Arc<Service<D>>,
    addr: &str,
) -> Result<(), GrpcServeError> {
    let addr: SocketAddr = addr
        .parse()
        .map_err(|source| GrpcServeError::InvalidAddress {
            addr: addr.to_string(),
            source,
        })?;
    tonic::transport::Server::builder()
        .add_service(grpc_server(service))
        .serve(addr)
        .await?;
    Ok(())
}