synapse_sdk/

service.rs

//! Service development helpers

use anyhow::{Context, Result};
use async_trait::async_trait;
use bytes::Bytes;
use std::{
    path::PathBuf,
    sync::{
        Arc,
        atomic::{AtomicBool, Ordering},
    },
};
use synapse_primitives::{InstanceId, Uuid};
use synapse_proto::{
    HttpEndpointRegister, HttpMethod, HttpRoute, InstanceCapabilities, InterfaceRegister,
    MessageKind, RpcRequest, RpcResponse, SynapseMessage, synapse_message,
};
use synapse_rpc::{
    HttpRpcClient, InterfaceRegistration, PROTOCOL_VERSION, RpcHandler, RpcServer,
    extract_http_registration_ack, extract_registration_ack,
};
use tokio::sync::RwLock;
use tracing::{debug, error, info, warn};

/// mTLS configuration for connecting to gateway
#[derive(Clone)]
pub struct MtlsClientConfig {
    pub cert_path: PathBuf,
    pub key_path: PathBuf,
    pub ca_cert_path: PathBuf,
}

/// Service builder for creating Synapse services
pub struct ServiceBuilder {
    service_name: String,
    instance_id: InstanceId,
    http_endpoint: Option<String>,
    gateway_url: Option<String>,
    mtls_config: Option<MtlsClientConfig>,
}

impl ServiceBuilder {
    /// Create a new service builder
    pub fn new(service_name: impl Into<String>) -> Self {
        Self {
            service_name: service_name.into(),
            instance_id: InstanceId::new_random(),
            http_endpoint: None,
            gateway_url: None,
            mtls_config: None,
        }
    }

    /// Set a specific instance ID (useful for testing)
    pub fn with_instance_id(mut self, instance_id: InstanceId) -> Self {
        self.instance_id = instance_id;
        self
    }

    /// Set the HTTP endpoint where this instance receives RPC traffic
    pub fn with_http_endpoint(mut self, http_endpoint: impl Into<String>) -> Self {
        self.http_endpoint = Some(http_endpoint.into());
        self
    }

    /// Set the gateway URL for registration (plain HTTP, dev mode only)
    pub fn with_gateway(mut self, gateway_url: impl Into<String>) -> Self {
        self.gateway_url = Some(gateway_url.into());
        self
    }

    /// Set the gateway URL with mTLS (production)
    pub fn with_gateway_mtls(
        mut self,
        gateway_url: impl Into<String>,
        cert_path: impl Into<PathBuf>,
        key_path: impl Into<PathBuf>,
        ca_cert_path: impl Into<PathBuf>,
    ) -> Self {
        self.gateway_url = Some(gateway_url.into());
        self.mtls_config = Some(MtlsClientConfig {
            cert_path: cert_path.into(),
            key_path: key_path.into(),
            ca_cert_path: ca_cert_path.into(),
        });
        self
    }

    /// Build the service with an RPC server (plain HTTP gateway client)
    pub fn build(self) -> Service {
        let gateway_client = match (&self.gateway_url, &self.mtls_config) {
            (Some(url), Some(mtls)) => {
                match HttpRpcClient::json_mtls(
                    url,
                    &mtls.cert_path,
                    &mtls.key_path,
                    &mtls.ca_cert_path,
                ) {
                    Ok(client) => Some(client),
                    Err(e) => {
                        error!("Failed to create mTLS gateway client: {}", e);
                        None
                    }
                }
            }
            (Some(url), None) => Some(HttpRpcClient::json(url)),
            _ => None,
        };

        Service {
            service_name: self.service_name,
            instance_id: self.instance_id,
            http_endpoint: self.http_endpoint,
            rpc_server: Arc::new(RpcServer::new()),
            gateway_client,
            health_check: None,
            start_time: std::time::Instant::now(),
            registered_interfaces: RwLock::new(Vec::new()),
            registered_http_routes: RwLock::new(Vec::new()),
            registered_http_route_groups: RwLock::new(Vec::new()),
            gateway_connected: AtomicBool::new(false),
        }
    }
}

/// Health check callback that services implement to report their status
pub type HealthCheckFn = Arc<dyn Fn() -> HealthStatus + Send + Sync>;

/// Health status that a service can report
#[derive(Debug, Clone)]
pub struct HealthStatus {
    pub status: synapse_proto::HealthStatus,
    pub message: Option<String>,
}

impl HealthStatus {
    pub fn healthy() -> Self {
        Self {
            status: synapse_proto::HealthStatus::Healthy,
            message: None,
        }
    }

    pub fn degraded(message: impl Into<String>) -> Self {
        Self {
            status: synapse_proto::HealthStatus::Degraded,
            message: Some(message.into()),
        }
    }

    pub fn unhealthy(message: impl Into<String>) -> Self {
        Self {
            status: synapse_proto::HealthStatus::Unhealthy,
            message: Some(message.into()),
        }
    }

    pub fn draining(message: impl Into<String>) -> Self {
        Self {
            status: synapse_proto::HealthStatus::Draining,
            message: Some(message.into()),
        }
    }
}

/// Response type for error pages on this route
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum RouteResponseType {
    /// JSON error responses (for APIs)
    #[default]
    Json,
    /// HTML error pages (for user-facing web routes)
    Html,
}

/// Configuration for an HTTP route to register with the gateway
#[derive(Debug, Clone)]
pub struct HttpRouteConfig {
    /// Path prefix for routing (e.g., "/api/users")
    pub path: String,
    /// HTTP methods (empty = all methods)
    pub methods: Vec<HttpMethod>,
    /// Subdomain pattern: "" (default, bare domain), "any", "exact:domain", "prefix:x-", "suffix:.x"
    pub subdomain: String,
    /// Middleware names to apply (resolved from gateway config definitions)
    pub middleware: Vec<String>,
    /// If true, this route is a metrics endpoint (registered for Prometheus discovery, not public routing)
    pub metrics: bool,
    /// Error response format for this route (default: JSON)
    pub response_type: RouteResponseType,
}

impl HttpRouteConfig {
    /// Create a new route config for the given path (all methods, bare domain)
    pub fn new(path: impl Into<String>) -> Self {
        Self {
            path: path.into(),
            methods: vec![],
            subdomain: String::new(),
            middleware: vec![],
            metrics: false,
            response_type: RouteResponseType::Json,
        }
    }

    /// Restrict to specific HTTP methods
    pub fn with_methods(mut self, methods: Vec<HttpMethod>) -> Self {
        self.methods = methods;
        self
    }

    /// Set subdomain pattern
    pub fn with_subdomain(mut self, subdomain: impl Into<String>) -> Self {
        self.subdomain = subdomain.into();
        self
    }

    /// Set middleware names to apply to this route
    pub fn with_middleware(mut self, middleware: Vec<String>) -> Self {
        self.middleware = middleware;
        self
    }

    /// Mark this route as a metrics endpoint for Prometheus discovery
    pub fn as_metrics(mut self) -> Self {
        self.metrics = true;
        self
    }

    /// Mark this route as serving HTML (error pages will be HTML instead of JSON)
    pub fn as_html(mut self) -> Self {
        self.response_type = RouteResponseType::Html;
        self
    }
}

/// A batch of HTTP routes targeting a specific host:port.
/// Used to register routes that proxy to a different backend than the service itself
/// (e.g., a frontend dev server, static file server, or external API).
///
/// Each group needs its own `service_name` because the gateway maps
/// service_name → endpoint URL (one URL per service). Routes targeting
/// different backends must use different service names.
#[derive(Debug, Clone)]
pub struct HttpRouteGroup {
    /// Service name for this group (used by the gateway for endpoint lookup)
    pub service_name: String,
    /// Target host (e.g., "127.0.0.1")
    pub host: String,
    /// Target port (e.g., 3222)
    pub port: u32,
    /// Routes to register for this target
    pub routes: Vec<HttpRouteConfig>,
}

impl HttpRouteGroup {
    pub fn new(
        service_name: impl Into<String>,
        host: impl Into<String>,
        port: u32,
        routes: Vec<HttpRouteConfig>,
    ) -> Self {
        Self {
            service_name: service_name.into(),
            host: host.into(),
            port,
            routes,
        }
    }
}

/// Stored registration info for reconnection
#[derive(Clone)]
struct RegisteredInterface {
    registration: InterfaceRegistration,
    #[allow(dead_code)]
    handler: Arc<dyn RpcHandler>,
}

/// A Synapse service instance
pub struct Service {
    service_name: String,
    instance_id: InstanceId,
    http_endpoint: Option<String>,
    rpc_server: Arc<RpcServer>,
    gateway_client: Option<HttpRpcClient>,
    health_check: Option<HealthCheckFn>,
    start_time: std::time::Instant,
    /// Track registered interfaces for re-registration on reconnect
    registered_interfaces: RwLock<Vec<RegisteredInterface>>,
    /// Track registered HTTP routes for re-registration on reconnect
    registered_http_routes: RwLock<Vec<HttpRouteConfig>>,
    /// Track registered HTTP route groups (arbitrary host:port) for re-registration
    registered_http_route_groups: RwLock<Vec<HttpRouteGroup>>,
    /// Whether we're currently connected to gateway
    gateway_connected: AtomicBool,
}

impl Service {
    /// Create a new service builder
    pub fn builder(service_name: impl Into<String>) -> ServiceBuilder {
        ServiceBuilder::new(service_name)
    }

    /// Get the service name
    pub fn name(&self) -> &str {
        &self.service_name
    }

    /// Get the instance ID
    pub fn instance_id(&self) -> InstanceId {
        self.instance_id
    }

    /// Get access to the RPC server
    pub fn rpc_server(&self) -> &Arc<RpcServer> {
        &self.rpc_server
    }

    /// Register a codegen-generated interface
    ///
    /// Takes the `(InterfaceRegistration, Arc<dyn RpcHandler>)` tuple returned by
    /// `FooServiceRouter::create(impl)` and registers it locally + with the gateway.
    ///
    /// This is the preferred way to register interfaces — no need to repeat
    /// interface names or method names manually.
    ///
    /// # Example
    /// ```ignore
    /// let impl_ = MyServiceImpl::new();
    /// let (reg, handler) = MyServiceRouter::create(impl_);
    /// service.register(reg, handler).await?;
    /// ```
    pub async fn register(
        &self,
        mut registration: InterfaceRegistration,
        handler: Arc<dyn RpcHandler>,
    ) -> Result<()> {
        // Override instance_id and service_name with this service's values
        registration.instance_id = self.instance_id;
        registration.service_name = self.service_name.clone();

        self.register_inner(registration, handler).await
    }

    /// Register an interface implementation from string parameters
    ///
    /// Prefer [`register`] when using codegen-generated routers.
    /// This method is useful for dynamic/manual interface registration.
    pub async fn register_interface(
        &self,
        interface_name: &str,
        method_names: &[&str],
        handler: Arc<dyn RpcHandler>,
    ) -> Result<()> {
        let registration = InterfaceRegistration::new(
            interface_name,
            &self.service_name,
            method_names,
            self.instance_id,
        );

        self.register_inner(registration, handler).await
    }

    /// Internal: register an interface with a fully-built registration
    async fn register_inner(
        &self,
        registration: InterfaceRegistration,
        handler: Arc<dyn RpcHandler>,
    ) -> Result<()> {
        info!(
            service = %self.service_name,
            interface = %registration.interface_name,
            interface_id = ?registration.interface_id,
            "Registering interface locally"
        );

        // Register locally
        self.rpc_server
            .registry()
            .register(registration.clone(), handler.clone())
            .await
            .map_err(|e| anyhow::anyhow!("Failed to register interface locally: {}", e))?;

        // Store for reconnection
        {
            let mut interfaces = self.registered_interfaces.write().await;
            interfaces.push(RegisteredInterface {
                registration: registration.clone(),
                handler,
            });
        }

        // Register with gateway if configured
        if let Some(ref client) = self.gateway_client {
            match self.register_with_gateway(client, &registration).await {
                Ok(()) => {
                    self.gateway_connected.store(true, Ordering::SeqCst);
                }
                Err(e) => {
                    self.gateway_connected.store(false, Ordering::SeqCst);
                    // Don't fail - we'll retry via the reconnection task
                    warn!(
                        service = %self.service_name,
                        error = %e,
                        "Failed to register with gateway (will retry)"
                    );
                }
            }
        }

        Ok(())
    }

    /// Register HTTP routes with the gateway, proxying to this service's own endpoint.
    ///
    /// Sends an `HttpEndpointRegister` message to the gateway, which will
    /// add the routes to the gateway's HTTP router. The gateway must have
    /// appropriate `http_registration_permissions` configured to allow this.
    ///
    /// Routes are stored and automatically re-registered if the gateway
    /// connection is lost and re-established.
    pub async fn register_http_routes(&self, routes: Vec<HttpRouteConfig>) -> Result<()> {
        let Some(ref client) = self.gateway_client else {
            return Err(anyhow::anyhow!(
                "Cannot register HTTP routes: no gateway client configured"
            ));
        };

        // Store routes for reconnection
        {
            let mut stored = self.registered_http_routes.write().await;
            stored.extend(routes.iter().cloned());
        }

        let endpoint = self.http_endpoint.clone().unwrap_or_default();
        let (host, port) = parse_host_port(&endpoint)?;
        self.send_http_registration(client, &routes, &host, port, &self.service_name, false)
            .await
    }

    /// Register HTTP routes with the gateway, proxying to an arbitrary host:port.
    ///
    /// Use this to register routes for services that aren't Synapse-connected
    /// (e.g., a frontend dev server, static file server, or external API).
    /// The gateway will proxy matching requests to the specified target.
    ///
    /// Each group has its own `service_name` because the gateway maps
    /// service_name → endpoint URL. Different backends need different names.
    ///
    /// Routes are stored and automatically re-registered on reconnect.
    pub async fn register_http_route_group(&self, group: HttpRouteGroup) -> Result<()> {
        let Some(ref client) = self.gateway_client else {
            return Err(anyhow::anyhow!(
                "Cannot register HTTP routes: no gateway client configured"
            ));
        };

        // Store for reconnection
        {
            let mut stored = self.registered_http_route_groups.write().await;
            stored.push(group.clone());
        }

        self.send_http_registration(
            client,
            &group.routes,
            &group.host,
            group.port,
            &group.service_name,
            true,
        )
        .await
    }

    /// Internal: send HTTP route registration to gateway with explicit host:port and service name
    async fn send_http_registration(
        &self,
        client: &HttpRpcClient,
        routes: &[HttpRouteConfig],
        host: &str,
        port: u32,
        service_name: &str,
        delegated: bool,
    ) -> Result<()> {
        let proto_routes: Vec<HttpRoute> = routes
            .iter()
            .map(|r| HttpRoute {
                path_prefix: r.path.clone(),
                methods: r.methods.iter().map(|m| *m as i32).collect(),
                options: None,
                subdomain: r.subdomain.clone(),
                middleware: r.middleware.clone(),
                metrics: r.metrics,
                response_type: match r.response_type {
                    RouteResponseType::Json => synapse_proto::ResponseType::Json as i32,
                    RouteResponseType::Html => synapse_proto::ResponseType::Html as i32,
                },
            })
            .collect();

        let register = HttpEndpointRegister {
            instance_id: Bytes::copy_from_slice(&self.instance_id.as_bytes()),
            routes: proto_routes,
            host: host.to_string(),
            port,
            health_check_path: String::new(),
            health_check_interval_ms: 0,
            service_name: service_name.to_string(),
            delegated,
        };

        let request_id = Uuid::new_v4();
        let message = SynapseMessage {
            protocol_version: PROTOCOL_VERSION.as_u32(),
            kind: MessageKind::HttpEndpointRegister as i32,
            request_id: Bytes::copy_from_slice(request_id.as_bytes()),
            message: Some(synapse_message::Message::HttpEndpointRegister(register)),
        };

        let response = client
            .send(&message)
            .await
            .context("Failed to send HTTP registration to gateway")?;

        if let Some((_, ack)) = extract_http_registration_ack(response) {
            if ack.success {
                info!(
                    service = %service_name,
                    host = %host,
                    port = %port,
                    routes = routes.len(),
                    "HTTP routes registered with gateway"
                );
                Ok(())
            } else {
                error!(
                    service = %service_name,
                    error = %ack.error_message,
                    "Gateway rejected HTTP route registration"
                );
                Err(anyhow::anyhow!(
                    "Gateway rejected HTTP route registration: {}",
                    ack.error_message
                ))
            }
        } else {
            Err(anyhow::anyhow!(
                "Unexpected response from gateway during HTTP registration"
            ))
        }
    }

    /// Register an interface with the gateway
    async fn register_with_gateway(
        &self,
        client: &HttpRpcClient,
        registration: &InterfaceRegistration,
    ) -> Result<()> {
        let http_endpoint = self.http_endpoint.clone().unwrap_or_default();

        info!(
            service = %self.service_name,
            interface = %registration.interface_name,
            endpoint = %http_endpoint,
            "Registering interface with gateway"
        );

        // Build the InterfaceRegister proto message
        let register = InterfaceRegister {
            interface_id: registration.interface_id.into(),
            interface_version: registration.interface_version,
            method_ids: registration
                .method_ids
                .iter()
                .map(|m| (*m).into())
                .collect(),
            instance_id: Bytes::copy_from_slice(&self.instance_id.as_bytes()),
            capabilities: Some(InstanceCapabilities::default()),
            service_name: self.service_name.clone(),
            interface_name: registration.interface_name.clone(),
            http_endpoint,
            method_names: registration.method_names.clone(),
        };

        // Create the SynapseMessage
        let request_id = Uuid::new_v4();
        let message = SynapseMessage {
            protocol_version: PROTOCOL_VERSION.as_u32(),
            kind: MessageKind::InterfaceRegister as i32,
            request_id: Bytes::copy_from_slice(request_id.as_bytes()),
            message: Some(synapse_message::Message::InterfaceRegister(register)),
        };

        // Send to gateway
        let response = client
            .send(&message)
            .await
            .context("Failed to send registration to gateway")?;

        // Check the response
        if let Some((_, ack)) = extract_registration_ack(response) {
            if ack.success {
                info!(
                    service = %self.service_name,
                    interface = %registration.interface_name,
                    "Successfully registered with gateway"
                );
                Ok(())
            } else {
                error!(
                    service = %self.service_name,
                    interface = %registration.interface_name,
                    error = %ack.error_message,
                    "Gateway rejected registration"
                );
                Err(anyhow::anyhow!(
                    "Gateway rejected registration: {}",
                    ack.error_message
                ))
            }
        } else {
            warn!(
                service = %self.service_name,
                "Unexpected response from gateway during registration"
            );
            Err(anyhow::anyhow!("Unexpected response from gateway"))
        }
    }

    /// Handle an RPC request
    pub async fn handle_request(&self, request: RpcRequest) -> RpcResponse {
        self.rpc_server.handle_request(request).await
    }

    /// Set a custom health check function
    pub fn set_health_check<F>(&mut self, check: F)
    where
        F: Fn() -> HealthStatus + Send + Sync + 'static,
    {
        self.health_check = Some(Arc::new(check));
    }

    /// Get current health status
    fn get_health_status(&self) -> HealthStatus {
        if let Some(ref health_check) = self.health_check {
            health_check()
        } else {
            HealthStatus::healthy()
        }
    }

    /// Create a HealthResponse message
    pub fn health_response(&self) -> synapse_proto::HealthResponse {
        let health = self.get_health_status();
        let uptime_ms = self.start_time.elapsed().as_millis() as i64;

        synapse_proto::HealthResponse {
            instance_id: Bytes::copy_from_slice(&self.instance_id.as_bytes()),
            status: health.status as i32,
            version: env!("CARGO_PKG_VERSION").to_string(),
            uptime_ms,
            message: health.message.unwrap_or_default(),
        }
    }

    /// Check if currently connected to gateway
    pub fn is_gateway_connected(&self) -> bool {
        self.gateway_connected.load(Ordering::SeqCst)
    }

    /// Re-register all interfaces and HTTP routes with the gateway
    async fn reregister_all(&self) -> Result<()> {
        let Some(ref client) = self.gateway_client else {
            return Ok(());
        };

        // Re-register RPC interfaces
        let interfaces = self.registered_interfaces.read().await;
        if !interfaces.is_empty() {
            info!(
                service = %self.service_name,
                count = interfaces.len(),
                "Re-registering interfaces with gateway"
            );

            for iface in interfaces.iter() {
                match self
                    .register_with_gateway(client, &iface.registration)
                    .await
                {
                    Ok(()) => {
                        info!(
                            service = %self.service_name,
                            interface = %iface.registration.interface_name,
                            "Re-registered interface"
                        );
                    }
                    Err(e) => {
                        error!(
                            service = %self.service_name,
                            interface = %iface.registration.interface_name,
                            error = %e,
                            "Failed to re-register interface"
                        );
                        return Err(e);
                    }
                }
            }
        }
        drop(interfaces);

        // Re-register HTTP routes (own endpoint)
        let http_routes = self.registered_http_routes.read().await;
        if !http_routes.is_empty() {
            info!(
                service = %self.service_name,
                count = http_routes.len(),
                "Re-registering HTTP routes with gateway"
            );
            let routes: Vec<HttpRouteConfig> = http_routes.clone();
            drop(http_routes);

            let endpoint = self.http_endpoint.clone().unwrap_or_default();
            let (host, port) = parse_host_port(&endpoint)?;
            if let Err(e) = self
                .send_http_registration(client, &routes, &host, port, &self.service_name, false)
                .await
            {
                error!(
                    service = %self.service_name,
                    error = %e,
                    "Failed to re-register HTTP routes"
                );
                return Err(e);
            }
        } else {
            drop(http_routes);
        }

        // Re-register HTTP route groups (delegated services)
        let route_groups = self.registered_http_route_groups.read().await;
        if !route_groups.is_empty() {
            let groups: Vec<HttpRouteGroup> = route_groups.clone();
            drop(route_groups);

            for group in &groups {
                info!(
                    service = %self.service_name,
                    target = %format!("{}:{}", group.host, group.port),
                    count = group.routes.len(),
                    "Re-registering delegated HTTP service with gateway"
                );
                if let Err(e) = self
                    .send_http_registration(
                        client,
                        &group.routes,
                        &group.host,
                        group.port,
                        &group.service_name,
                        true,
                    )
                    .await
                {
                    error!(
                        service = %self.service_name,
                        error = %e,
                        "Failed to re-register HTTP route group"
                    );
                    return Err(e);
                }
            }
        }

        Ok(())
    }

    /// Start background task that pushes health status and handles reconnection
    ///
    /// This task:
    /// - Sends health status to gateway periodically
    /// - Detects when gateway connection is lost
    /// - Automatically re-registers all interfaces and HTTP routes when gateway comes back
    ///
    /// `interval_ms` controls the health push frequency.
    /// `initial_delay_ms` controls the initial startup delay (default 500ms if 0).
    pub fn start_gateway_connection_task(
        self: Arc<Self>,
        interval_ms: u64,
    ) -> tokio::task::JoinHandle<()> {
        self.start_gateway_connection_task_with_delay(interval_ms, 500)
    }

    /// Like `start_gateway_connection_task` but with a custom initial delay
    pub fn start_gateway_connection_task_with_delay(
        self: Arc<Self>,
        interval_ms: u64,
        initial_delay_ms: u64,
    ) -> tokio::task::JoinHandle<()> {
        tokio::spawn(async move {
            if initial_delay_ms > 0 {
                tokio::time::sleep(std::time::Duration::from_millis(initial_delay_ms)).await;
            }

            let mut interval = tokio::time::interval(std::time::Duration::from_millis(interval_ms));
            interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
            let mut consecutive_failures = 0u32;

            loop {
                interval.tick().await;

                if let Some(ref client) = self.gateway_client {
                    let health = self.health_response();
                    let request_id = Uuid::new_v4();

                    // Create health response message to push to gateway
                    let message = SynapseMessage {
                        protocol_version: PROTOCOL_VERSION.as_u32(),
                        kind: MessageKind::HealthResponse as i32,
                        request_id: Bytes::copy_from_slice(request_id.as_bytes()),
                        message: Some(synapse_message::Message::HealthResponse(health.clone())),
                    };

                    match client.send(&message).await {
                        Ok(response) => {
                            // Check if gateway says we need to re-register
                            let needs_reregister =
                                if let Some((_, ack)) = extract_registration_ack(response) {
                                    !ack.success
                                        && ack.error_message.contains("INSTANCE_NOT_REGISTERED")
                                } else {
                                    false
                                };

                            if needs_reregister {
                                // Gateway doesn't know us - re-register
                                info!(
                                    service = %self.service_name,
                                    "Gateway requested re-registration, re-registering interfaces"
                                );

                                if let Err(e) = self.reregister_all().await {
                                    error!(
                                        service = %self.service_name,
                                        error = %e,
                                        "Failed to re-register interfaces"
                                    );
                                    self.gateway_connected.store(false, Ordering::SeqCst);
                                } else {
                                    info!(
                                        service = %self.service_name,
                                        "Successfully re-registered with gateway"
                                    );
                                    self.gateway_connected.store(true, Ordering::SeqCst);
                                }
                            } else {
                                // Normal health ack
                                self.gateway_connected.store(true, Ordering::SeqCst);
                                consecutive_failures = 0;
                                debug!(
                                    service = %self.service_name,
                                    status = health.status,
                                    uptime_ms = health.uptime_ms,
                                    "Health status pushed to gateway"
                                );
                            }
                        }
                        Err(e) => {
                            let was_connected =
                                self.gateway_connected.swap(false, Ordering::SeqCst);
                            consecutive_failures += 1;

                            if was_connected {
                                warn!(
                                    service = %self.service_name,
                                    error = %e,
                                    "Lost connection to gateway, will attempt to reconnect"
                                );
                            } else if consecutive_failures % 10 == 1 {
                                // Log every 10th failure to avoid spam
                                debug!(
                                    service = %self.service_name,
                                    failures = consecutive_failures,
                                    "Gateway still unavailable, retrying..."
                                );
                            }
                        }
                    }
                } else {
                    // No gateway configured, just log locally
                    let health = self.get_health_status();
                    debug!(
                        service = %self.service_name,
                        status = ?health.status,
                        "Health check (no gateway configured)"
                    );
                }
            }
        })
    }

    /// Start background health push task (legacy alias for start_gateway_connection_task)
    pub fn start_health_pusher(self: Arc<Self>, interval_ms: u64) -> tokio::task::JoinHandle<()> {
        self.start_gateway_connection_task(interval_ms)
    }

    /// Get access to the gateway client (if configured)
    pub fn gateway_client(&self) -> Option<&HttpRpcClient> {
        self.gateway_client.as_ref()
    }
}

/// Parse host and port from a URL string like "http://127.0.0.1:9001/rpc"
fn parse_host_port(endpoint: &str) -> Result<(String, u32)> {
    let stripped = endpoint
        .strip_prefix("http://")
        .or_else(|| endpoint.strip_prefix("https://"))
        .context("Endpoint must start with http:// or https://")?;

    // Take everything before the first '/'
    let authority = stripped.split('/').next().unwrap_or(stripped);

    if authority.is_empty() {
        anyhow::bail!("Empty host in endpoint URL");
    }

    // Handle IPv6 addresses like [::1]:9001
    if authority.starts_with('[') {
        // Find closing bracket
        let bracket_end = authority
            .find(']')
            .context("Invalid IPv6 address: missing closing bracket")?;
        let host = &authority[..bracket_end + 1];
        let rest = &authority[bracket_end + 1..];
        if let Some(port_str) = rest.strip_prefix(':') {
            let port: u32 = port_str.parse().context("Invalid port in endpoint URL")?;
            return Ok((host.to_string(), port));
        }
        return Ok((host.to_string(), 80));
    }

    // Standard host:port
    if let Some((host, port_str)) = authority.rsplit_once(':') {
        if host.is_empty() {
            anyhow::bail!("Empty host in endpoint URL");
        }
        let port: u32 = port_str.parse().context("Invalid port in endpoint URL")?;
        Ok((host.to_string(), port))
    } else {
        Ok((authority.to_string(), 80))
    }
}

/// Trait for implementing RPC interfaces
#[async_trait]
pub trait RpcInterface: Send + Sync {
    /// Get the interface name (e.g., "mensa.user.v1.UserInterface")
    fn interface_name(&self) -> &str;

    /// Get the list of method names
    fn method_names(&self) -> &[&'static str];

    /// Handle an RPC request
    async fn handle(&self, request: RpcRequest) -> RpcResponse;
}

/// Helper to wrap an RpcInterface as an RpcHandler
pub struct InterfaceHandler<T: RpcInterface> {
    interface: Arc<T>,
}

impl<T: RpcInterface> InterfaceHandler<T> {
    pub fn new(interface: T) -> Self {
        Self {
            interface: Arc::new(interface),
        }
    }

    pub fn into_arc(self) -> Arc<Self> {
        Arc::new(self)
    }
}

#[async_trait]
impl<T: RpcInterface + 'static> RpcHandler for InterfaceHandler<T> {
    async fn handle(&self, request: RpcRequest) -> RpcResponse {
        self.interface.handle(request).await
    }
}

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

    // ========== ServiceBuilder ==========

    #[test]
    fn test_builder_sets_name() {
        let svc = ServiceBuilder::new("my-service").build();
        assert_eq!(svc.name(), "my-service");
    }

    #[test]
    fn test_builder_random_instance_id() {
        let s1 = ServiceBuilder::new("svc").build();
        let s2 = ServiceBuilder::new("svc").build();
        assert_ne!(s1.instance_id(), s2.instance_id());
    }

    #[test]
    fn test_builder_custom_instance_id() {
        let id = InstanceId::new_random();
        let svc = ServiceBuilder::new("svc").with_instance_id(id).build();
        assert_eq!(svc.instance_id(), id);
    }

    #[test]
    fn test_builder_with_http_endpoint() {
        let svc = ServiceBuilder::new("svc")
            .with_http_endpoint("http://localhost:9001/rpc")
            .build();
        assert_eq!(
            svc.http_endpoint.as_deref(),
            Some("http://localhost:9001/rpc")
        );
    }

    #[test]
    fn test_builder_with_gateway() {
        let svc = ServiceBuilder::new("svc")
            .with_gateway("http://localhost:8080")
            .build();
        assert!(svc.gateway_client().is_some());
    }

    #[test]
    fn test_builder_no_gateway() {
        let svc = ServiceBuilder::new("svc").build();
        assert!(svc.gateway_client().is_none());
    }

    #[test]
    fn test_builder_chaining() {
        let id = InstanceId::new_random();
        let svc = ServiceBuilder::new("svc")
            .with_instance_id(id)
            .with_http_endpoint("http://localhost:9001/rpc")
            .with_gateway("http://localhost:8080")
            .build();
        assert_eq!(svc.name(), "svc");
        assert_eq!(svc.instance_id(), id);
        assert!(svc.gateway_client().is_some());
    }

    // ========== Service ==========

    #[test]
    fn test_service_builder_static() {
        let builder = Service::builder("test");
        let svc = builder.build();
        assert_eq!(svc.name(), "test");
    }

    #[test]
    fn test_is_gateway_connected_default_false() {
        let svc = ServiceBuilder::new("svc").build();
        assert!(!svc.is_gateway_connected());
    }

    #[test]
    fn test_health_response_defaults_healthy() {
        let svc = ServiceBuilder::new("svc").build();
        let resp = svc.health_response();
        assert_eq!(resp.status, synapse_proto::HealthStatus::Healthy as i32);
    }

    #[test]
    fn test_health_response_has_uptime() {
        let svc = ServiceBuilder::new("svc").build();
        std::thread::sleep(std::time::Duration::from_millis(10));
        let resp = svc.health_response();
        assert!(resp.uptime_ms >= 10);
    }

    #[test]
    fn test_health_response_has_instance_id() {
        let svc = ServiceBuilder::new("svc").build();
        let resp = svc.health_response();
        assert_eq!(resp.instance_id.len(), 16);
    }

    #[test]
    fn test_set_health_check_custom() {
        let mut svc = ServiceBuilder::new("svc").build();
        svc.set_health_check(|| HealthStatus::degraded("high load"));
        let resp = svc.health_response();
        assert_eq!(resp.status, synapse_proto::HealthStatus::Degraded as i32);
        assert_eq!(resp.message, "high load");
    }

    // ========== HealthStatus ==========

    #[test]
    fn test_health_status_healthy() {
        let h = HealthStatus::healthy();
        assert_eq!(h.status, synapse_proto::HealthStatus::Healthy);
        assert!(h.message.is_none());
    }

    #[test]
    fn test_health_status_degraded() {
        let h = HealthStatus::degraded("slow");
        assert_eq!(h.status, synapse_proto::HealthStatus::Degraded);
        assert_eq!(h.message.as_deref(), Some("slow"));
    }

    #[test]
    fn test_health_status_unhealthy() {
        let h = HealthStatus::unhealthy("broken");
        assert_eq!(h.status, synapse_proto::HealthStatus::Unhealthy);
    }

    #[test]
    fn test_health_status_draining() {
        let h = HealthStatus::draining("shutting down");
        assert_eq!(h.status, synapse_proto::HealthStatus::Draining);
    }

    // ========== register_interface + handle_request ==========

    #[tokio::test]
    async fn test_register_interface_locally() {
        let svc = ServiceBuilder::new("svc").build();
        let handler = Arc::new(FunctionHandler::new(move |_req| {
            Box::pin(async move {
                RpcResponse {
                    status: synapse_proto::RpcStatus::Ok as i32,
                    payload: Bytes::from("handled"),
                    error: None,
                    headers: vec![],
                    responded_at_unix_ms: 0,
                }
            })
        }));

        svc.register_interface("test.Echo", &["Echo"], handler)
            .await
            .unwrap();

        // Verify we can route to it
        let req = RpcRequest {
            interface_id: synapse_primitives::InterfaceId::from_name("test.Echo").into(),
            method_id: synapse_primitives::MethodId::from_name("Echo").into(),
            headers: vec![],
            payload: Bytes::new(),
            sent_at_unix_ms: 0,
        };
        let resp = svc.handle_request(req).await;
        assert_eq!(resp.status, synapse_proto::RpcStatus::Ok as i32);
        assert_eq!(resp.payload, Bytes::from("handled"));
    }

    #[tokio::test]
    async fn test_handle_request_unknown_interface() {
        let svc = ServiceBuilder::new("svc").build();
        let req = RpcRequest {
            interface_id: 99999,
            method_id: 1,
            headers: vec![],
            payload: Bytes::new(),
            sent_at_unix_ms: 0,
        };
        let resp = svc.handle_request(req).await;
        assert_eq!(
            resp.status,
            synapse_proto::RpcStatus::InterfaceNotFound as i32
        );
    }
}