openhuman/src/core/dispatch.rs

//! Central dispatcher for RPC requests.
//!
//! This module coordinates the routing of incoming requests to either the
//! core subsystem or the OpenHuman domain-specific handlers.

use crate::core::legacy_aliases::resolve_legacy;
use crate::core::rpc_log;
use crate::core::types::{AppState, InvocationResult};
use serde_json::{json, Map, Value};

/// Dispatches an RPC method call to the appropriate subsystem.
///
/// This is the primary entry point for all RPC calls. It uses a tiered routing
/// strategy:
/// 1. **Core Subsystem**: Checks for internal methods like `core.ping` or `core.version`.
/// 2. **Domain-Specific Handlers**: Delegates to the `openhuman` domain dispatcher
///    which handles all registered controllers (memory, skills, etc.).
///
/// # Arguments
///
/// * `state` - The current application state (e.g., core version).
/// * `method` - The name of the RPC method to invoke (e.g., `core.ping`).
/// * `params` - The parameters for the method call as a JSON value.
///
/// # Returns
///
/// A `Result` containing the JSON-formatted response or an error message if
/// the method is unknown or invocation fails.
pub async fn dispatch(
    state: AppState,
    method: &str,
    params: serde_json::Value,
) -> Result<serde_json::Value, String> {
    log::trace!(
        "[rpc:dispatch] enter method={} params={}",
        method,
        rpc_log::redact_params_for_log(&params)
    );

    // Tier 0: Rewrite legacy method names to their canonical form before
    // any subsystem lookup. Symmetric with the frontend's
    // `normalizeRpcMethod` (`app/src/services/rpcMethods.ts`): the
    // frontend rewrites outgoing names for clients that just updated, the
    // core rewrites incoming names for clients that haven't yet. See
    // `crate::core::legacy_aliases` for the shared table.
    let resolved = resolve_legacy(method);
    if resolved != method {
        // Per-rewrite log at debug to keep the dispatcher hot path quiet
        // at scale (per graycyrus review on PR #1544). Aggregate
        // visibility belongs in the observability layer, not here.
        log::debug!(
            "[rpc-legacy-alias] rewrite method={} -> canonical={}",
            method,
            resolved
        );
    }
    let method = resolved;

    // Tier 1: Internal core methods.
    // These are handled directly within the core module and don't require
    // a separate controller registration.
    if let Some(result) = try_core_dispatch(&state, method, params.clone()) {
        log::debug!("[rpc:dispatch] routed method={} subsystem=core", method);
        return result.map(crate::core::types::invocation_to_rpc_json);
    }

    // Tier 2: Registered domain controllers.
    if let Some(result) = try_registry_dispatch(method, params.clone()).await {
        log::debug!(
            "[rpc:dispatch] routed method={} subsystem=controller_registry",
            method
        );
        return result;
    }

    // Tier 3: Legacy domain-specific dispatcher.
    if let Some(result) = crate::rpc::try_dispatch(method, params).await {
        log::debug!(
            "[rpc:dispatch] routed method={} subsystem=openhuman",
            method
        );
        return result;
    }

    log::warn!("[rpc:dispatch] unknown_method method={}", method);
    Err(format!("unknown method: {method}"))
}

/// Handles internal core-level RPC methods.
///
/// These methods provide basic information about the server and its version.
///
/// Currently supported methods:
/// - `core.ping`: A simple liveness check. Returns `{ "ok": true }`.
/// - `core.version`: Returns the version of the running core binary.
fn try_core_dispatch(
    state: &AppState,
    method: &str,
    params: serde_json::Value,
) -> Option<Result<InvocationResult, String>> {
    match method {
        "core.ping" => Some(InvocationResult::ok(json!({ "ok": true }))),
        "core.version" => Some(InvocationResult::ok(
            json!({ "version": state.core_version }),
        )),
        "core.events_subscribe_token" => Some(handle_events_subscribe_token(params)),
        _ => None,
    }
}

/// Mint a single-shot bind token for the SSE `/events` stream.
///
/// Browser `EventSource` cannot attach an `Authorization` header, so an
/// authenticated holder of the per-process RPC bearer first asks for a
/// short-lived token here (this RPC is gated by the same bearer-token
/// middleware as the rest of `/rpc`) and then opens
/// `/events?client_id=<id>&token=<bind>`. The `/events` handler removes
/// the token from the store on first use, so a leaked URL cannot be
/// replayed by a second subscriber.
fn handle_events_subscribe_token(params: serde_json::Value) -> Result<InvocationResult, String> {
    let obj = params.as_object();
    let client_id = obj
        .and_then(|m| m.get("client_id"))
        .and_then(|v| v.as_str())
        .map(|s| s.trim())
        .filter(|s| !s.is_empty())
        .ok_or_else(|| {
            log::warn!(
                "[events-bind] reject mint: missing or empty client_id (param_keys={:?})",
                obj.map(|m| m.keys().collect::<Vec<_>>())
            );
            "missing or empty 'client_id' parameter".to_string()
        })?;
    let ttl = obj
        .and_then(|m| m.get("ttl_secs"))
        .and_then(|v| v.as_u64())
        .map(std::time::Duration::from_secs);

    let issued =
        crate::core::event_bind_tokens::issue(client_id.to_string(), ttl).ok_or_else(|| {
            log::warn!(
                "[events-bind] reject mint: store at capacity (client_id_len={} ttl_secs={:?})",
                client_id.len(),
                ttl.map(|d| d.as_secs())
            );
            "events bind-token store at capacity; try again shortly".to_string()
        })?;

    let ttl_remaining_secs = issued
        .valid_until
        .checked_duration_since(std::time::Instant::now())
        .unwrap_or_default()
        .as_secs();

    log::debug!(
        "[events-bind] minted token for client_id_len={} ttl_secs={}",
        client_id.len(),
        ttl_remaining_secs
    );

    InvocationResult::ok(json!({
        "token": issued.token,
        "ttl_secs": ttl_remaining_secs,
    }))
}

async fn try_registry_dispatch(
    method: &str,
    params: Value,
) -> Option<Result<serde_json::Value, String>> {
    let schema = crate::core::all::schema_for_rpc_method(method)?;
    let params_obj = match params_to_object(params) {
        Ok(params_obj) => params_obj,
        Err(err) => return Some(Err(err)),
    };
    if let Err(err) = crate::core::all::validate_params(&schema, &params_obj) {
        return Some(Err(err));
    }
    crate::core::all::try_invoke_registered_rpc(method, params_obj).await
}

fn params_to_object(params: Value) -> Result<Map<String, Value>, String> {
    match params {
        Value::Object(map) => Ok(map),
        Value::Null => Ok(Map::new()),
        other => Err(format!(
            "invalid params: expected object or null, got {}",
            type_name(&other)
        )),
    }
}

fn type_name(value: &Value) -> &'static str {
    match value {
        Value::Null => "null",
        Value::Bool(_) => "bool",
        Value::Number(_) => "number",
        Value::String(_) => "string",
        Value::Array(_) => "array",
        Value::Object(_) => "object",
    }
}

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

    fn test_state() -> AppState {
        AppState {
            core_version: "9.9.9-test".to_string(),
        }
    }

    #[tokio::test]
    async fn dispatch_core_ping_returns_ok_true() {
        let out = dispatch(test_state(), "core.ping", json!({}))
            .await
            .expect("core.ping should succeed");
        assert_eq!(out, json!({ "ok": true }));
    }

    #[tokio::test]
    async fn dispatch_core_version_returns_state_version() {
        let out = dispatch(test_state(), "core.version", json!({}))
            .await
            .expect("core.version should succeed");
        assert_eq!(out, json!({ "version": "9.9.9-test" }));
    }

    #[tokio::test]
    async fn dispatch_core_ignores_params() {
        // Params must be tolerated even when the method takes none.
        let out = dispatch(test_state(), "core.ping", json!({ "extra": 1 }))
            .await
            .expect("core.ping should ignore extra params");
        assert_eq!(out, json!({ "ok": true }));
    }

    #[tokio::test]
    async fn dispatch_rewrites_legacy_alias_before_lookup() {
        // `openhuman.ping` is a legacy alias for `core.ping` in the shared
        // alias table. Going through the dispatcher must rewrite it and
        // route successfully to Tier 1 instead of falling through to the
        // unknown-method error path.
        let out = dispatch(test_state(), "openhuman.ping", json!({}))
            .await
            .expect("legacy alias openhuman.ping must resolve to core.ping");
        assert_eq!(out, json!({ "ok": true }));
    }

    #[tokio::test]
    async fn dispatch_unknown_method_returns_error() {
        let err = dispatch(test_state(), "does.not.exist", json!({}))
            .await
            .expect_err("unknown methods must error");
        assert!(err.contains("unknown method"));
        assert!(err.contains("does.not.exist"));
    }

    #[tokio::test]
    async fn dispatch_empty_method_returns_unknown_method_error() {
        let err = dispatch(test_state(), "", json!({}))
            .await
            .expect_err("empty method must error");
        assert!(err.contains("unknown method"));
    }

    #[tokio::test]
    async fn dispatch_delegates_to_tier2_for_domain_method() {
        // Tier 2 dispatcher handles `openhuman.security_policy_info`, so
        // it must succeed and return a policy object.
        let out = dispatch(test_state(), "openhuman.security_policy_info", json!({}))
            .await
            .expect("security_policy_info should route via tier 2");
        // With logs present, payload is wrapped as { result, logs }.
        assert!(out.get("result").is_some() || out.get("autonomy").is_some());
    }

    #[test]
    fn try_core_dispatch_returns_none_for_non_core_namespace() {
        let state = test_state();
        assert!(try_core_dispatch(&state, "openhuman.memory_list_namespaces", json!({})).is_none());
        assert!(try_core_dispatch(&state, "corez.ping", json!({})).is_none());
    }

    #[test]
    fn try_core_dispatch_matches_exact_ping_and_version() {
        let state = test_state();
        assert!(try_core_dispatch(&state, "core.ping", json!({})).is_some());
        assert!(try_core_dispatch(&state, "core.version", json!({})).is_some());
        // Prefix match alone must not count.
        assert!(try_core_dispatch(&state, "core.pingz", json!({})).is_none());
        assert!(try_core_dispatch(&state, "core", json!({})).is_none());
    }

    #[test]
    fn try_core_dispatch_version_reflects_appstate() {
        let state = AppState {
            core_version: "0.0.0-abc".into(),
        };
        let result = try_core_dispatch(&state, "core.version", json!({}))
            .expect("core.version must be routed")
            .expect("core.version must produce InvocationResult");
        assert_eq!(result.value, json!({ "version": "0.0.0-abc" }));
        assert!(result.logs.is_empty());
    }

    #[tokio::test]
    async fn dispatch_legacy_ping_rewrites_and_succeeds() {
        let out = dispatch(test_state(), "openhuman.ping", json!({}))
            .await
            .expect("openhuman.ping should be rewritten to core.ping and succeed");
        assert_eq!(out, json!({ "ok": true }));
    }

    #[tokio::test]
    async fn dispatch_legacy_alias_routes_to_registry() {
        // openhuman.get_analytics_settings should rewrite to openhuman.config_get_analytics_settings.
        // This is a read-only call and should succeed if the registry is wired up.
        let out = dispatch(test_state(), "openhuman.get_analytics_settings", json!({}))
            .await
            .expect("openhuman.get_analytics_settings should be rewritten and succeed");

        // The registry-wrapped payload has a "result" field.
        assert!(
            out.get("enabled").is_some() || out.get("result").is_some(),
            "Payload should have 'enabled' or 'result', got: {}",
            out
        );
    }
}