alkhttp/src/gateway/dispatch.rs

//! Shared dispatch spine for the `to_openapi` / `to_mcp` gateway
//! projections.
//!
//! Thin concrete struct (not a trait — the alknet research ruled out a
//! trait with an associated output type). Holds `Arc<OperationRegistry>`
//! and `Arc<dyn IdentityProvider>` and exposes a `resolve_bearer()` and
//! `invoke()` method pair returning the neutral `ResponseEnvelope`. Each
//! gateway maps the envelope to its own wire shape (`to_openapi` → HTTP
//! `Response`, `to_mcp` → `CallToolResult`).
//!
//! # Security invariants
//!
//! - `internal: false` — ACL runs against the caller's `identity`, not a
//!   handler's composition authority (alkcall ADR-017).
//! - `forwarded_for: None` — wire-ingress only.
//!
//! The root `OperationContext` is constructed identically for both
//! gateways, making them provably identical on the security axis (auth,
//! authority, ACL); they diverge only on wire-framing.
//!
//! # Deadline
//!
//! Once-op invokes ([`GatewayDispatch::invoke`]) are bounded by
//! `DEFAULT_TIMEOUT` (30 s): the registry invoke is wrapped in
//! `tokio::time::timeout` and a hung handler surfaces as a `TIMEOUT`
//! error envelope (`504` under the gateway's error mapping), not an
//! indefinitely-held HTTP request. Streaming and sink dispatch set
//! `deadline: None` (subscriptions are unbounded per alkcall ADR-021,
//! and a `/publish` body is bounded by the client's upload, not a
//! fixed window).

use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};

use alkcall::core::auth::{AuthToken, Identity, IdentityProvider};
use alkcall::core::types::Capabilities;
use alkcall::protocol::wire::{CallError, ResponseEnvelope};
use alkcall::registry::context::{AbortPolicy, OperationContext, ScopedPeerEnv};
use alkcall::registry::env::LocalOperationEnv;
use alkcall::registry::registration::OperationRegistry;
use futures::stream::BoxStream;
use serde_json::Value;

const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);

pub struct GatewayDispatch {
    registry: Arc<OperationRegistry>,
    identity_provider: Arc<dyn IdentityProvider>,
}

impl GatewayDispatch {
    pub fn new(
        registry: Arc<OperationRegistry>,
        identity_provider: Arc<dyn IdentityProvider>,
    ) -> Self {
        Self {
            registry,
            identity_provider,
        }
    }

    pub fn registry(&self) -> &Arc<OperationRegistry> {
        &self.registry
    }

    pub fn identity_provider(&self) -> &Arc<dyn IdentityProvider> {
        &self.identity_provider
    }

    pub fn resolve_bearer(&self, token: &AuthToken) -> Option<Identity> {
        self.identity_provider.resolve_from_token(token)
    }

    pub async fn invoke(
        &self,
        identity: Option<Identity>,
        op: &str,
        input: Value,
    ) -> ResponseEnvelope {
        let operation_name = strip_leading_slash(op).to_string();
        let request_id = uuid::Uuid::new_v4().to_string();
        let context = self.build_root_context(&request_id, &operation_name, identity);
        let result = tokio::time::timeout(
            DEFAULT_TIMEOUT,
            self.registry.invoke(&operation_name, input, context),
        )
        .await;
        match result {
            Ok(envelope) => envelope,
            Err(_elapsed) => ResponseEnvelope::error(
                request_id,
                CallError::timeout(format!(
                    "operation did not complete within the {DEFAULT_TIMEOUT:?} gateway deadline"
                )),
            ),
        }
    }

    pub fn invoke_streaming(
        &self,
        identity: Option<Identity>,
        op: &str,
        input: Value,
    ) -> BoxStream<'static, ResponseEnvelope> {
        let operation_name = strip_leading_slash(op).to_string();
        let request_id = uuid::Uuid::new_v4().to_string();
        let context = self.build_root_context_streaming(&request_id, &operation_name, identity);
        self.registry
            .invoke_streaming(&operation_name, input, context)
    }

    /// Dispatch a `Pub` operation (ADR-046, ADR-068): the
    /// `publish_stream` is the initiator's chunk stream (one
    /// `Ok(Value)` per published chunk); the handler's final
    /// `ResponseEnvelope` is the result. Pre-handler failures
    /// (not-found, forbidden, non-Pub op) surface as a single error
    /// envelope — the same envelope the `/publish` route maps to HTTP.
    pub async fn invoke_sink(
        &self,
        identity: Option<Identity>,
        op: &str,
        input: Value,
        publish_stream: alkcall::registry::registration::PublishStream,
    ) -> ResponseEnvelope {
        let operation_name = strip_leading_slash(op).to_string();
        let request_id = uuid::Uuid::new_v4().to_string();
        let context = self.build_root_context_sink(&request_id, &operation_name, identity);
        self.registry
            .invoke_sink(&operation_name, input, publish_stream, context)
            .await
    }

    fn build_root_context_sink(
        &self,
        request_id: &str,
        operation_name: &str,
        identity: Option<Identity>,
    ) -> OperationContext {
        self.build_root_context_inner(request_id, operation_name, identity, false)
    }

    fn build_root_context(
        &self,
        request_id: &str,
        operation_name: &str,
        identity: Option<Identity>,
    ) -> OperationContext {
        self.build_root_context_inner(request_id, operation_name, identity, true)
    }

    fn build_root_context_streaming(
        &self,
        request_id: &str,
        operation_name: &str,
        identity: Option<Identity>,
    ) -> OperationContext {
        self.build_root_context_inner(request_id, operation_name, identity, false)
    }

    fn build_root_context_inner(
        &self,
        request_id: &str,
        operation_name: &str,
        identity: Option<Identity>,
        bounded: bool,
    ) -> OperationContext {
        let registration = self.registry.registration(operation_name);
        let (composition_authority, capabilities, scoped_env) = match registration {
            Some(r) => (
                r.composition_authority.clone(),
                r.capabilities.clone(),
                r.scoped_env.clone().unwrap_or_else(ScopedPeerEnv::empty),
            ),
            None => (None, Capabilities::new(), ScopedPeerEnv::empty()),
        };

        let env: Arc<dyn alkcall::registry::env::OperationEnv + Send + Sync> =
            Arc::new(LocalOperationEnv::new(Arc::clone(&self.registry)));

        OperationContext {
            request_id: request_id.to_string(),
            parent_request_id: None,
            identity,
            handler_identity: composition_authority,
            forwarded_for: None,
            capabilities,
            metadata: HashMap::new(),
            deadline: bounded.then(|| Instant::now() + DEFAULT_TIMEOUT),
            scoped_env,
            env,
            abort_policy: AbortPolicy::default(),
            internal: false,
            ownership: None,
        }
    }
}

fn strip_leading_slash(operation_id: &str) -> &str {
    operation_id.strip_prefix('/').unwrap_or(operation_id)
}

#[cfg(test)]
mod tests {
    use super::*;
    use alkcall::registry::registration::{
        make_handler, make_streaming_handler, HandlerKind, HandlerRegistration, OperationProvenance,
    };
    use alkcall::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
    use futures::StreamExt;
    use std::sync::Mutex as StdMutex;

    // The tests below run against the module's real 30 s deadline; only
    // the elapsed-time bound (< 60 s, anti-flake) is asserted by hand.

    struct StaticIdentityProvider {
        tokens: StdMutex<HashMap<String, Identity>>,
    }

    impl StaticIdentityProvider {
        fn new() -> Self {
            Self {
                tokens: StdMutex::new(HashMap::new()),
            }
        }
    }

    impl IdentityProvider for StaticIdentityProvider {
        fn resolve_from_fingerprint(&self, _fp: &str) -> Option<Identity> {
            None
        }
        fn resolve_from_token(&self, token: &AuthToken) -> Option<Identity> {
            let token_str = String::from_utf8_lossy(&token.raw);
            self.tokens.lock().unwrap().get(token_str.as_ref()).cloned()
        }
    }

    fn spec(name: &str, visibility: Visibility, op_type: OperationType) -> OperationSpec {
        OperationSpec::new(
            name,
            op_type,
            visibility,
            serde_json::json!({}),
            serde_json::json!({}),
            vec![],
            AccessControl::default(),
            None,
        )
    }

    fn echo_registry() -> Arc<OperationRegistry> {
        let mut registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                spec("echo/run", Visibility::External, OperationType::Query),
                HandlerKind::Once(make_handler(|input, ctx| async move {
                    ResponseEnvelope::ok(ctx.request_id, input)
                })),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        Arc::new(registry)
    }

    #[tokio::test]
    async fn invoke_external_op_round_trips() {
        let dispatch =
            GatewayDispatch::new(echo_registry(), Arc::new(StaticIdentityProvider::new()));
        let envelope = dispatch
            .invoke(None, "/echo/run", serde_json::json!({ "x": 1 }))
            .await;
        assert!(envelope.result.is_ok(), "expected ok, got {envelope:?}");
    }

    #[tokio::test]
    async fn invoke_unknown_op_returns_not_found() {
        let dispatch =
            GatewayDispatch::new(echo_registry(), Arc::new(StaticIdentityProvider::new()));
        let envelope = dispatch
            .invoke(None, "/missing/op", serde_json::json!({}))
            .await;
        match envelope.result {
            Err(error) => assert_eq!(error.code, "NOT_FOUND"),
            Ok(v) => panic!("expected error, got {v:?}"),
        }
    }

    #[tokio::test]
    async fn invoke_internal_op_returns_not_found() {
        let mut registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                spec("internal/op", Visibility::Internal, OperationType::Query),
                HandlerKind::Once(make_handler(|_input, ctx| async move {
                    ResponseEnvelope::ok(ctx.request_id, serde_json::json!({}))
                })),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let dispatch =
            GatewayDispatch::new(Arc::new(registry), Arc::new(StaticIdentityProvider::new()));
        let envelope = dispatch
            .invoke(None, "/internal/op", serde_json::json!({}))
            .await;
        match envelope.result {
            Err(error) => assert_eq!(error.code, "NOT_FOUND"),
            Ok(v) => panic!("expected error, got {v:?}"),
        }
    }

    #[tokio::test]
    async fn invoke_enforces_the_default_deadline_on_a_hung_handler() {
        use std::time::Duration;

        let mut registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                spec("hung/op", Visibility::External, OperationType::Query),
                HandlerKind::Once(make_handler(|_input, ctx| async move {
                    tokio::time::sleep(Duration::from_secs(120)).await;
                    ResponseEnvelope::ok(ctx.request_id, serde_json::json!({}))
                })),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let dispatch =
            GatewayDispatch::new(Arc::new(registry), Arc::new(StaticIdentityProvider::new()));
        let started = std::time::Instant::now();
        let envelope = dispatch
            .invoke(None, "/hung/op", serde_json::json!({}))
            .await;
        let elapsed = started.elapsed();
        assert!(
            elapsed < Duration::from_secs(60),
            "the 30 s deadline must fire well before the 120 s handler sleep, took {elapsed:?}"
        );
        match envelope.result {
            Err(error) => {
                assert_eq!(error.code, "TIMEOUT");
                assert!(error.retryable, "the deadline error is retryable");
            }
            Ok(v) => panic!("expected a TIMEOUT error, got {v:?}"),
        }
    }

    #[tokio::test]
    async fn invoke_completes_within_the_deadline_for_a_fast_handler() {
        let dispatch =
            GatewayDispatch::new(echo_registry(), Arc::new(StaticIdentityProvider::new()));
        let envelope = dispatch
            .invoke(None, "/echo/run", serde_json::json!({}))
            .await;
        assert!(envelope.result.is_ok(), "a fast handler must not time out");
    }

    #[tokio::test]
    async fn streaming_sub_op_streams_envelopes() {
        let mut registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                spec("tick/stream", Visibility::External, OperationType::Sub),
                HandlerKind::Stream(make_streaming_handler(|input, ctx| {
                    let count = input.get("count").and_then(|v| v.as_u64()).unwrap_or(2);
                    let request_id = ctx.request_id.clone();
                    futures::stream::iter(0..count).map(move |i| {
                        ResponseEnvelope::ok(request_id.clone(), serde_json::json!({ "tick": i }))
                    })
                })),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let dispatch =
            GatewayDispatch::new(Arc::new(registry), Arc::new(StaticIdentityProvider::new()));
        let mut stream =
            dispatch.invoke_streaming(None, "/tick/stream", serde_json::json!({ "count": 3 }));
        let mut ticks = Vec::new();
        while let Some(envelope) = stream.next().await {
            ticks.push(envelope);
        }
        assert_eq!(ticks.len(), 3);
    }
}