synapse_sdk/

client.rs

//! Unified RPC client with connection pooling
//!
//! SynClient provides a connection pool for all outgoing RPC traffic.
//! Generated service clients use this pool for efficient communication.

use anyhow::{Context, Result, bail};
use bytes::Bytes;
use std::{path::Path, sync::Arc};
use synapse_primitives::{InterfaceId, MethodId};
use synapse_proto::RpcRequest;
use synapse_proto::RpcStatus;
#[cfg(feature = "otlp")]
use synapse_proto::{HeaderEntry, header_entry};
use synapse_rpc::HttpRpcClient;
use tracing::{debug, warn};

/// Unified RPC client with connection pooling
///
/// This is the central client that manages all outgoing RPC connections.
/// Generated service clients use this pool rather than creating their own connections.
///
/// The underlying HTTP client (reqwest) automatically pools connections,
/// so SynClient can be cloned cheaply and shared across the application.
#[derive(Clone)]
pub struct SynClient {
    inner: Arc<SynClientInner>,
}

struct SynClientInner {
    http_client: HttpRpcClient,
}

impl SynClient {
    /// Create a new unified RPC client
    ///
    /// # Arguments
    /// * `gateway_url` - URL of the gateway (e.g., "http://localhost:8080")
    ///
    /// # Example
    /// ```ignore
    /// let client = SynClient::new("http://localhost:8080");
    /// ```
    pub fn new(gateway_url: impl Into<String>) -> Self {
        Self {
            inner: Arc::new(SynClientInner {
                http_client: HttpRpcClient::json(gateway_url),
            }),
        }
    }

    /// Create a client with protobuf encoding (more efficient)
    pub fn with_protobuf(gateway_url: impl Into<String>) -> Self {
        Self {
            inner: Arc::new(SynClientInner {
                http_client: HttpRpcClient::protobuf(gateway_url),
            }),
        }
    }

    /// Create a client with mTLS authentication (production)
    pub fn with_mtls(
        gateway_url: impl Into<String>,
        cert_path: impl AsRef<Path>,
        key_path: impl AsRef<Path>,
        ca_cert_path: impl AsRef<Path>,
    ) -> Result<Self> {
        Ok(Self {
            inner: Arc::new(SynClientInner {
                http_client: HttpRpcClient::protobuf_mtls(
                    gateway_url,
                    cert_path,
                    key_path,
                    ca_cert_path,
                )?,
            }),
        })
    }

    /// Set the request timeout
    pub fn with_timeout(self, timeout: std::time::Duration) -> Self {
        Self {
            inner: Arc::new(SynClientInner {
                http_client: HttpRpcClient::new(
                    self.gateway_url(),
                    self.inner.http_client.content_type(),
                )
                .with_timeout(timeout),
            }),
        }
    }

    /// Call a service method with raw bytes
    ///
    /// # Arguments
    /// * `interface` - The interface name or ID
    /// * `method` - The method name or ID
    /// * `payload` - The request payload (serialized)
    pub async fn call(
        &self,
        interface: impl Into<InterfaceId>,
        method: impl Into<MethodId>,
        payload: Bytes,
    ) -> Result<Bytes> {
        let interface_id = interface.into();
        let method_id = method.into();

        debug!(
            "Calling {}.{} ({} bytes)",
            u32::from(interface_id),
            u32::from(method_id),
            payload.len()
        );

        // Build RPC request (request_id is in the SynapseMessage envelope, not here)
        #[allow(unused_mut)]
        let mut headers = Vec::new();

        // Inject trace context if OpenTelemetry is enabled
        #[cfg(feature = "otlp")]
        {
            use opentelemetry::trace::TraceContextExt;
            use tracing_opentelemetry::OpenTelemetrySpanExt;

            let span = tracing::Span::current();
            let otel_ctx = span.context();
            let span_ref = otel_ctx.span();
            let span_ctx = span_ref.span_context();

            if span_ctx.is_valid() {
                headers.push(HeaderEntry {
                    key: u32::from(*synapse_primitives::id::well_known::TRACE_ID),
                    value: Some(header_entry::Value::StringValue(format!(
                        "{:032x}",
                        span_ctx.trace_id()
                    ))),
                });
                headers.push(HeaderEntry {
                    key: u32::from(*synapse_primitives::id::well_known::SPAN_ID),
                    value: Some(header_entry::Value::StringValue(format!(
                        "{:016x}",
                        span_ctx.span_id()
                    ))),
                });
            }
        }

        let request = RpcRequest {
            interface_id: interface_id.into(),
            method_id: method_id.into(),
            headers,
            payload,
            sent_at_unix_ms: chrono::Utc::now().timestamp_millis(),
        };

        // Make the call
        let response = self
            .inner
            .http_client
            .call(request)
            .await
            .context("RPC call failed")?;

        // Check status
        if response.status != RpcStatus::Ok as i32 {
            let error = response.error.unwrap_or_else(|| synapse_proto::RpcError {
                code: response.status as u32,
                message: format!("RPC failed with status {}", response.status),
                details: vec![],
            });

            warn!("RPC call failed: {} - {}", error.code, error.message);
            bail!("RPC error {}: {}", error.code, error.message);
        }

        debug!("RPC call succeeded ({} bytes)", response.payload.len());
        Ok(response.payload)
    }

    /// Call with typed request and response (JSON serialization)
    ///
    /// This is a convenience method for JSON-serializable types.
    pub async fn call_json<TReq, TResp>(
        &self,
        interface: impl Into<InterfaceId>,
        method: impl Into<MethodId>,
        request: &TReq,
    ) -> Result<TResp>
    where
        TReq: serde::Serialize,
        TResp: serde::de::DeserializeOwned,
    {
        // Serialize request
        let payload = serde_json::to_vec(request).context("Failed to serialize request")?;

        // Make call
        let response_bytes = self.call(interface, method, Bytes::from(payload)).await?;

        // Deserialize response
        let response =
            serde_json::from_slice(&response_bytes).context("Failed to deserialize response")?;

        Ok(response)
    }

    /// Call with typed request and response (Protobuf serialization)
    ///
    /// This is the efficient method for protobuf types.
    /// Generated service clients use this method.
    pub async fn call_proto<TReq, TResp>(
        &self,
        interface: impl Into<InterfaceId>,
        method: impl Into<MethodId>,
        request: &TReq,
    ) -> Result<TResp>
    where
        TReq: prost::Message,
        TResp: prost::Message + Default,
    {
        // Serialize request with prost
        let payload = request.encode_to_vec();

        // Make call
        let response_bytes = self.call(interface, method, Bytes::from(payload)).await?;

        // Deserialize response with prost
        let response =
            TResp::decode(response_bytes.as_ref()).context("Failed to deserialize response")?;

        Ok(response)
    }

    /// Get the gateway URL
    pub fn gateway_url(&self) -> &str {
        self.inner.http_client.gateway_url()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_client_creation() {
        let client = SynClient::new("http://localhost:8080");
        assert_eq!(client.gateway_url(), "http://localhost:8080");

        // with_timeout creates a new client with the specified duration
        let _client =
            SynClient::new("http://gateway:5000").with_timeout(std::time::Duration::from_secs(60));
    }

    #[test]
    fn test_client_clone() {
        let client1 = SynClient::new("http://localhost:8080");
        let client2 = client1.clone();

        // Both should point to same inner Arc
        assert_eq!(client1.gateway_url(), client2.gateway_url());
    }

    #[tokio::test]
    #[ignore] // Requires running gateway
    async fn test_call() {
        let client = SynClient::new("http://localhost:8080");

        let interface = InterfaceId::from_name("test.Service");
        let method = MethodId::from_name("Echo");
        let payload = Bytes::from("hello");

        let _response = client.call(interface, method, payload).await;
    }
}