spikard_http/grpc/

mod.rs

//! gRPC runtime support for Spikard
//!
//! This module provides gRPC server infrastructure using Tonic, enabling
//! Spikard to handle both HTTP/1.1 REST requests and HTTP/2 gRPC requests.
//!
//! # Architecture
//!
//! The gRPC support follows the same language-agnostic pattern as the HTTP handler:
//!
//! 1. **GrpcHandler trait**: Language-agnostic interface for handling gRPC requests
//! 2. **Service bridge**: Converts between Tonic's types and our internal representation
//! 3. **Streaming support**: Utilities for handling streaming RPCs
//! 4. **Server integration**: Multiplexes HTTP/1.1 and HTTP/2 traffic
//!
//! # Example
//!
//! ```ignore
//! use spikard_http::grpc::{GrpcHandler, GrpcRequestData, GrpcResponseData};
//! use std::sync::Arc;
//!
//! // Implement GrpcHandler for your language binding
//! struct MyGrpcHandler;
//!
//! impl GrpcHandler for MyGrpcHandler {
//!     fn call(&self, request: GrpcRequestData) -> Pin<Box<dyn Future<Output = GrpcHandlerResult> + Send>> {
//!         Box::pin(async move {
//!             // Handle the gRPC request
//!             Ok(GrpcResponseData {
//!                 payload: bytes::Bytes::from("response"),
//!                 metadata: tonic::metadata::MetadataMap::new(),
//!             })
//!         })
//!     }
//!
//!     fn service_name(&self) -> &str {
//!         "mypackage.MyService"
//!     }
//! }
//!
//! // Register with the server
//! let handler = Arc::new(MyGrpcHandler);
//! let config = GrpcConfig::default();
//! ```

pub(crate) mod framing;
pub(crate) mod handler;
pub(crate) mod service;
pub(crate) mod streaming;

// Re-export types used from integration tests and binding crates
pub use handler::{GrpcHandler, GrpcHandlerResult, GrpcRequestData, GrpcResponseData};
// Internal implementation details only
pub(crate) use handler::RpcMode;
pub(crate) use service::{GenericGrpcService, parse_grpc_path};

use serde::{Deserialize, Serialize};
use std::collections::HashMap;
#[cfg(test)]
use std::collections::HashSet;
use std::sync::Arc;

/// Configuration for gRPC support
///
/// Controls how the server handles gRPC requests, including compression,
/// timeouts, and protocol settings.
///
/// # Stream Limits
///
/// This configuration enforces message-level size limits but delegates
/// concurrent stream limiting to the HTTP/2 transport layer:
///
/// - **Message Size Limits**: The `max_message_size` field is enforced per
///   individual message (request or response) in both unary and streaming RPCs.
///   When a single message exceeds this limit, the request is rejected with
///   `PAYLOAD_TOO_LARGE` (HTTP 413).
///
/// - **Concurrent Stream Limits**: The `max_concurrent_streams` is an advisory
///   configuration passed to the HTTP/2 layer for connection-level stream
///   negotiation. The HTTP/2 transport automatically enforces this limit and
///   returns GOAWAY frames when exceeded. Applications should not rely on
///   custom enforcement of this limit.
///
/// - **Stream Response Size Limits**: The `max_stream_response_bytes` field caps the
///   total encoded bytes emitted across a server-streaming or bidi-streaming response.
///   When the cumulative size exceeds the limit, the stream is terminated with
///   `tonic::Status::resource_exhausted`. Defaults to `None` (unbounded).
///
/// # Example
///
/// ```ignore
/// let mut config = GrpcConfig::default();
/// config.max_message_size = 10 * 1024 * 1024; // 10MB per message
/// config.max_concurrent_streams = 50; // Advised to HTTP/2 layer
/// ```
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GrpcConfig {
    /// Enable gRPC support
    #[serde(default = "default_true")]
    pub enabled: bool,

    /// Maximum message size in bytes (for both sending and receiving)
    ///
    /// This limit applies to individual messages in both unary and streaming RPCs.
    /// When a single message exceeds this size, the request is rejected with HTTP 413
    /// (Payload Too Large).
    ///
    /// Default: 4MB (4194304 bytes)
    ///
    /// # Note
    /// This limit does NOT apply to the total response size in streaming RPCs.
    /// For multi-message streams, the total response can exceed this limit as long
    /// as each individual message stays within the limit.
    #[serde(default = "default_max_message_size")]
    pub max_message_size: usize,

    /// Enable gzip compression for gRPC messages
    #[serde(default = "default_true")]
    pub enable_compression: bool,

    /// Timeout for gRPC requests in seconds (None = no timeout)
    #[serde(default)]
    pub request_timeout: Option<u64>,

    /// Maximum number of concurrent streams per connection (HTTP/2 advisory)
    ///
    /// This value is communicated to HTTP/2 clients as the server's flow control limit.
    /// The HTTP/2 transport layer enforces this limit automatically via SETTINGS frames
    /// and GOAWAY responses. Applications should NOT implement custom enforcement.
    ///
    /// Default: 100 streams per connection
    ///
    /// # Stream Limiting Strategy
    /// - **Per Connection**: This limit applies per HTTP/2 connection, not globally
    /// - **Transport Enforcement**: HTTP/2 handles all stream limiting; applications
    ///   need not implement custom checks
    /// - **Streaming Requests**: In server streaming or bidi streaming, each logical
    ///   RPC consumes one stream slot. Message ordering within a stream follows
    ///   HTTP/2 frame ordering.
    #[serde(default = "default_max_concurrent_streams")]
    pub max_concurrent_streams: u32,

    /// Enable HTTP/2 keepalive
    #[serde(default = "default_true")]
    pub enable_keepalive: bool,

    /// HTTP/2 keepalive interval in seconds
    #[serde(default = "default_keepalive_interval")]
    pub keepalive_interval: u64,

    /// HTTP/2 keepalive timeout in seconds
    #[serde(default = "default_keepalive_timeout")]
    pub keepalive_timeout: u64,

    /// Total byte cap across an entire streaming response.
    ///
    /// When `Some(n)`, the streaming adapter aborts the stream with
    /// `tonic::Status::resource_exhausted` once the cumulative encoded message
    /// bytes exceed `n`. The stream yields the error item and then terminates.
    ///
    /// Per-message cap remains `max_message_size`. This limit applies to
    /// server-streaming and bidirectional-streaming RPCs only; unary RPCs are
    /// governed solely by `max_message_size`.
    ///
    /// Default: `None` (unbounded total response size).
    #[serde(default)]
    pub max_stream_response_bytes: Option<usize>,
}

impl Default for GrpcConfig {
    fn default() -> Self {
        Self {
            enabled: true,
            max_message_size: default_max_message_size(),
            enable_compression: true,
            request_timeout: None,
            max_concurrent_streams: default_max_concurrent_streams(),
            enable_keepalive: true,
            keepalive_interval: default_keepalive_interval(),
            keepalive_timeout: default_keepalive_timeout(),
            max_stream_response_bytes: None,
        }
    }
}

const fn default_true() -> bool {
    true
}

const fn default_max_message_size() -> usize {
    4 * 1024 * 1024 // 4MB
}

const fn default_max_concurrent_streams() -> u32 {
    100
}

const fn default_keepalive_interval() -> u64 {
    75 // seconds
}

const fn default_keepalive_timeout() -> u64 {
    20 // seconds
}

/// Registry for gRPC handlers
///
/// Maps service and method names to their handlers and RPC modes. Used by the
/// server to route incoming gRPC requests to the appropriate handler method based
/// on the parsed gRPC path.
///
/// # Example
///
/// ```ignore
/// use spikard_http::grpc::{GrpcRegistry, RpcMode};
/// use std::sync::Arc;
///
/// let mut registry = GrpcRegistry::new();
/// registry.register("mypackage.UserService", "GetUser", Arc::new(user_handler), RpcMode::Unary);
/// registry.register(
///     "mypackage.StreamService",
///     "StreamUsers",
///     Arc::new(stream_handler),
///     RpcMode::ServerStreaming,
/// );
/// ```
type GrpcHandlerEntry = (Arc<dyn GrpcHandler>, RpcMode);
const WILDCARD_METHOD: &str = "*";

#[derive(Clone)]
pub(crate) struct GrpcRegistry {
    handlers: Arc<HashMap<(String, String), GrpcHandlerEntry>>,
}

impl GrpcRegistry {
    /// Create a new empty gRPC handler registry
    pub fn new() -> Self {
        Self {
            handlers: Arc::new(HashMap::new()),
        }
    }

    /// Get a handler and its RPC mode by service and method name
    ///
    /// Returns both the handler and the RPC mode it was registered with. Exact
    /// method matches take precedence over service-level wildcard handlers.
    pub fn get(&self, service_name: &str, method_name: &str) -> Option<(Arc<dyn GrpcHandler>, RpcMode)> {
        self.handlers
            .get(&(service_name.to_owned(), method_name.to_owned()))
            .or_else(|| {
                self.handlers
                    .get(&(service_name.to_owned(), WILDCARD_METHOD.to_owned()))
            })
            .cloned()
    }

    /// Check if a service has any registered handlers.
    pub fn contains_service(&self, service_name: &str) -> bool {
        self.handlers
            .keys()
            .any(|(registered_service, _)| registered_service == service_name)
    }

    /// Check if the registry is empty
    pub fn is_empty(&self) -> bool {
        self.handlers.is_empty()
    }
}

#[cfg(test)]
impl GrpcRegistry {
    /// Register a gRPC handler for a specific service method (test helper)
    pub fn register(
        &mut self,
        service_name: impl Into<String>,
        method_name: impl Into<String>,
        handler: Arc<dyn GrpcHandler>,
        rpc_mode: RpcMode,
    ) {
        let handlers = Arc::make_mut(&mut self.handlers);
        handlers.insert((service_name.into(), method_name.into()), (handler, rpc_mode));
    }

    /// Register a gRPC handler for an entire service (test helper)
    pub fn register_service(
        &mut self,
        service_name: impl Into<String>,
        handler: Arc<dyn GrpcHandler>,
        rpc_mode: RpcMode,
    ) {
        self.register(service_name, WILDCARD_METHOD, handler, rpc_mode);
    }

    /// Get all registered service names (test helper)
    pub fn service_names(&self) -> Vec<String> {
        self.handlers
            .keys()
            .map(|(service_name, _)| service_name.clone())
            .collect::<HashSet<_>>()
            .into_iter()
            .collect()
    }

    /// Get all explicitly registered method names for a service (test helper)
    pub fn method_names(&self, service_name: &str) -> Vec<String> {
        self.handlers
            .keys()
            .filter(|(registered_service, method_name)| {
                registered_service == service_name && method_name.as_str() != WILDCARD_METHOD
            })
            .map(|(_, method_name)| method_name.clone())
            .collect()
    }

    /// Check if a specific service method is registered (test helper)
    pub fn contains(&self, service_name: &str, method_name: &str) -> bool {
        self.handlers
            .contains_key(&(service_name.to_owned(), method_name.to_owned()))
    }

    /// Get the number of registered services (test helper)
    pub fn len(&self) -> usize {
        self.handlers.len()
    }
}

impl Default for GrpcRegistry {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::grpc::handler::{GrpcHandler, GrpcHandlerResult, GrpcRequestData};
    use std::future::Future;
    use std::pin::Pin;

    struct TestHandler;

    impl GrpcHandler for TestHandler {
        fn call(&self, _request: GrpcRequestData) -> Pin<Box<dyn Future<Output = GrpcHandlerResult> + Send>> {
            Box::pin(async {
                Ok(GrpcResponseData {
                    payload: bytes::Bytes::new(),
                    metadata: tonic::metadata::MetadataMap::new(),
                })
            })
        }

        fn service_name(&self) -> &'static str {
            // Since we can't return a reference to self.0 with 'static lifetime,
            // we need to use a workaround. In real usage, service names should be static.
            "test.Service"
        }
    }

    #[test]
    fn test_grpc_config_default() {
        let config = GrpcConfig::default();
        assert!(config.enabled);
        assert_eq!(config.max_message_size, 4 * 1024 * 1024);
        assert!(config.enable_compression);
        assert!(config.request_timeout.is_none());
        assert_eq!(config.max_concurrent_streams, 100);
        assert!(config.enable_keepalive);
        assert_eq!(config.keepalive_interval, 75);
        assert_eq!(config.keepalive_timeout, 20);
        assert!(config.max_stream_response_bytes.is_none());
    }

    #[test]
    fn test_grpc_config_serialization() {
        let config = GrpcConfig::default();
        let json = serde_json::to_string(&config).unwrap();
        let deserialized: GrpcConfig = serde_json::from_str(&json).unwrap();

        assert_eq!(config.enabled, deserialized.enabled);
        assert_eq!(config.max_message_size, deserialized.max_message_size);
        assert_eq!(config.enable_compression, deserialized.enable_compression);
    }

    #[test]
    fn test_grpc_registry_new() {
        let registry = GrpcRegistry::new();
        assert!(registry.is_empty());
        assert_eq!(registry.len(), 0);
    }

    #[test]
    fn test_grpc_registry_register() {
        let mut registry = GrpcRegistry::new();
        let handler = Arc::new(TestHandler);

        registry.register("test.Service", "TestMethod", handler, RpcMode::Unary);

        assert!(!registry.is_empty());
        assert_eq!(registry.len(), 1);
        assert!(registry.contains("test.Service", "TestMethod"));
    }

    #[test]
    fn test_grpc_registry_get() {
        let mut registry = GrpcRegistry::new();
        let handler = Arc::new(TestHandler);

        registry.register("test.Service", "TestMethod", handler, RpcMode::Unary);

        let retrieved = registry.get("test.Service", "TestMethod");
        assert!(retrieved.is_some());
        let (handler, rpc_mode) = retrieved.unwrap();
        assert_eq!(handler.service_name(), "test.Service");
        assert_eq!(rpc_mode, RpcMode::Unary);
    }

    #[test]
    fn test_grpc_registry_get_nonexistent() {
        let registry = GrpcRegistry::new();
        let result = registry.get("nonexistent.Service", "MissingMethod");
        assert!(result.is_none());
    }

    #[test]
    fn test_grpc_registry_service_names() {
        let mut registry = GrpcRegistry::new();

        registry.register("service1", "Method1", Arc::new(TestHandler), RpcMode::Unary);
        registry.register("service2", "Method2", Arc::new(TestHandler), RpcMode::ServerStreaming);
        registry.register("service3", "Method3", Arc::new(TestHandler), RpcMode::Unary);

        let mut names = registry.service_names();
        names.sort();

        assert_eq!(names, vec!["service1", "service2", "service3"]);
    }

    #[test]
    fn test_grpc_registry_contains() {
        let mut registry = GrpcRegistry::new();
        registry.register("test.Service", "TestMethod", Arc::new(TestHandler), RpcMode::Unary);

        assert!(registry.contains("test.Service", "TestMethod"));
        assert!(!registry.contains("other.Service", "TestMethod"));
    }

    #[test]
    fn test_grpc_registry_multiple_services() {
        let mut registry = GrpcRegistry::new();

        registry.register("user.Service", "GetUser", Arc::new(TestHandler), RpcMode::Unary);
        registry.register(
            "post.Service",
            "ListPosts",
            Arc::new(TestHandler),
            RpcMode::ServerStreaming,
        );

        assert_eq!(registry.len(), 2);
        assert!(registry.contains("user.Service", "GetUser"));
        assert!(registry.contains("post.Service", "ListPosts"));
    }

    #[test]
    fn test_grpc_registry_clone() {
        let mut registry = GrpcRegistry::new();
        registry.register("test.Service", "TestMethod", Arc::new(TestHandler), RpcMode::Unary);

        let cloned = registry.clone();

        assert_eq!(cloned.len(), 1);
        assert!(cloned.contains("test.Service", "TestMethod"));
    }

    #[test]
    fn test_grpc_registry_default() {
        let registry = GrpcRegistry::default();
        assert!(registry.is_empty());
    }

    #[test]
    fn test_grpc_registry_rpc_mode_storage() {
        let mut registry = GrpcRegistry::new();

        registry.register("unary.Service", "UnaryMethod", Arc::new(TestHandler), RpcMode::Unary);
        registry.register(
            "server_stream.Service",
            "StreamMethod",
            Arc::new(TestHandler),
            RpcMode::ServerStreaming,
        );
        registry.register(
            "client_stream.Service",
            "UploadMethod",
            Arc::new(TestHandler),
            RpcMode::ClientStreaming,
        );
        registry.register(
            "bidi.Service",
            "ChatMethod",
            Arc::new(TestHandler),
            RpcMode::BidirectionalStreaming,
        );

        let (_, mode) = registry.get("unary.Service", "UnaryMethod").unwrap();
        assert_eq!(mode, RpcMode::Unary);

        let (_, mode) = registry.get("server_stream.Service", "StreamMethod").unwrap();
        assert_eq!(mode, RpcMode::ServerStreaming);

        let (_, mode) = registry.get("client_stream.Service", "UploadMethod").unwrap();
        assert_eq!(mode, RpcMode::ClientStreaming);

        let (_, mode) = registry.get("bidi.Service", "ChatMethod").unwrap();
        assert_eq!(mode, RpcMode::BidirectionalStreaming);
    }

    #[test]
    fn test_grpc_registry_service_fallback() {
        let mut registry = GrpcRegistry::new();
        registry.register_service("test.Service", Arc::new(TestHandler), RpcMode::Unary);

        assert!(registry.contains_service("test.Service"));
        assert!(registry.get("test.Service", "AnyMethod").is_some());
        assert!(registry.method_names("test.Service").is_empty());
    }

    #[test]
    fn test_grpc_registry_prefers_method_specific_handler() {
        struct MethodSpecificHandler;

        impl GrpcHandler for MethodSpecificHandler {
            fn call(&self, _request: GrpcRequestData) -> Pin<Box<dyn Future<Output = GrpcHandlerResult> + Send>> {
                Box::pin(async {
                    Ok(GrpcResponseData {
                        payload: bytes::Bytes::from("method-specific"),
                        metadata: tonic::metadata::MetadataMap::new(),
                    })
                })
            }

            fn service_name(&self) -> &str {
                "test.Service"
            }
        }

        let mut registry = GrpcRegistry::new();
        registry.register_service("test.Service", Arc::new(TestHandler), RpcMode::Unary);
        registry.register(
            "test.Service",
            "GetThing",
            Arc::new(MethodSpecificHandler),
            RpcMode::ServerStreaming,
        );

        let (_, mode) = registry.get("test.Service", "GetThing").unwrap();
        assert_eq!(mode, RpcMode::ServerStreaming);
        let (_, fallback_mode) = registry.get("test.Service", "OtherThing").unwrap();
        assert_eq!(fallback_mode, RpcMode::Unary);
    }
}