alkcall/src/channels/env.rs

//! `ChannelOperationEnv` — the extension trait (ADR-047 §4, as
//! amended 2026-08-13) that adds the `channel_manager()` accessor to
//! `OperationEnv` without coupling the call crate to channels types.
//!
//! **The open-op path does not use this trait.** Per the ADR-047 §4
//! amendment, open ops are registered per-connection (in the
//! `install_channel_zero` hook) with a per-connection `ChannelCore`,
//! and the wrapper uses `ChannelCore::manager()` directly — no
//! `context.env` downcast. The original §4's dynamic-resolution shape
//! (downcast `context.env` to `&dyn ChannelOperationEnv` at invocation
//! time) was unworkable: `context.env` is a `PeerCompositeEnv`, not a
//! single concrete type that can be downcast to a channels-backed env.
//!
//! The trait and `ChannelsSessionEnv` impl are retained as a two-way-
//! door implementation detail — they are not on the open-op path, but
//! remain available for future per-connection routing (e.g., nested
//! channels where each connection's overlay carries its own manager
//! reference for non-open-op queries).
//!
//! This keeps `alkcall`'s call crate free of any channels types and
//! preserves the layering (ADR-044). The `OperationEnv` is already the
//! integration point for per-connection state (ADR-019); adding a
//! `ChannelManager` accessor via an extension trait is the natural
//! extension.

use std::sync::Arc;

use crate::registry::env::OperationEnv;

use super::manager::ChannelManager;

/// Extension trait that adds the per-connection `ChannelManager`
/// accessor (ADR-047 §4, as amended 2026-08-13). Implemented by the
/// connection overlay in `channels-call` (the per-connection
/// `OperationEnv` that carries a `ChannelManager` reference).
///
/// **Not on the open-op path** — see the module doc. Retained for
/// future per-connection routing.
#[async_trait::async_trait]
pub trait ChannelOperationEnv: OperationEnv {
    /// The per-connection `ChannelManager` for this channels session.
    /// `None` if the env is not channels-backed (a bare call
    /// connection).
    fn channel_manager(&self) -> Option<&ChannelManager>;
}

/// A `ChannelOperationEnv` impl that wraps a base `OperationEnv` and a
/// `ChannelManager`. This is the per-connection overlay
/// `channels-call` installs on channel 0's call registry.
pub struct ChannelsSessionEnv {
    pub base: Arc<dyn OperationEnv + Send + Sync>,
    pub manager: ChannelManager,
}

#[async_trait::async_trait]
impl OperationEnv for ChannelsSessionEnv {
    async fn invoke_with_policy(
        &self,
        namespace: &str,
        operation: &str,
        input: serde_json::Value,
        parent: &crate::registry::context::OperationContext,
        policy: crate::registry::context::AbortPolicy,
    ) -> crate::protocol::wire::ResponseEnvelope {
        self.base
            .invoke_with_policy(namespace, operation, input, parent, policy)
            .await
    }

    fn contains(&self, name: &str) -> bool {
        self.base.contains(name)
    }

    fn peer_ids(&self) -> Vec<crate::registry::env::PeerId> {
        self.base.peer_ids()
    }

    fn peer_contains(&self, peer: &crate::registry::env::PeerId, name: &str) -> bool {
        self.base.peer_contains(peer, name)
    }

    fn peer_operations(&self, peer: &crate::registry::env::PeerId) -> Vec<String> {
        self.base.peer_operations(peer)
    }

    fn list_operation_names(&self) -> Vec<String> {
        self.base.list_operation_names()
    }

    async fn invoke_peer(
        &self,
        peer: &crate::registry::env::PeerRef,
        namespace: &str,
        operation: &str,
        input: serde_json::Value,
        parent: &crate::registry::context::OperationContext,
        policy: crate::registry::context::AbortPolicy,
    ) -> crate::protocol::wire::ResponseEnvelope {
        self.base
            .invoke_peer(peer, namespace, operation, input, parent, policy)
            .await
    }
}

#[async_trait::async_trait]
impl ChannelOperationEnv for ChannelsSessionEnv {
    fn channel_manager(&self) -> Option<&ChannelManager> {
        Some(&self.manager)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::channels::mux::MuxRunner;
    use crate::core::types::Capabilities;
    use crate::protocol::wire::ResponseEnvelope;
    use crate::registry::context::{AbortPolicy, OperationContext, ScopedPeerEnv};
    use crate::registry::env::{LocalOperationEnv, PeerId, PeerRef};
    use parking_lot::Mutex;
    use std::collections::HashMap;
    use tokio::io::duplex;

    struct MockEnv {
        contains_val: bool,
        peer_ids_val: Vec<PeerId>,
        peer_contains_val: bool,
        peer_operations_val: Vec<String>,
        invoke_with_policy_called: Mutex<bool>,
        invoke_peer_called: Mutex<bool>,
    }

    impl MockEnv {
        fn new() -> Self {
            Self {
                contains_val: true,
                peer_ids_val: vec!["peer-a".to_string(), "peer-b".to_string()],
                peer_contains_val: true,
                peer_operations_val: vec!["op1".to_string(), "op2".to_string()],
                invoke_with_policy_called: Mutex::new(false),
                invoke_peer_called: Mutex::new(false),
            }
        }
    }

    #[async_trait::async_trait]
    impl OperationEnv for MockEnv {
        async fn invoke_with_policy(
            &self,
            _namespace: &str,
            _operation: &str,
            _input: serde_json::Value,
            parent: &OperationContext,
            _policy: AbortPolicy,
        ) -> ResponseEnvelope {
            *self.invoke_with_policy_called.lock() = true;
            ResponseEnvelope::ok(
                parent.request_id.clone(),
                serde_json::Value::String("mock".into()),
            )
        }

        fn contains(&self, _name: &str) -> bool {
            self.contains_val
        }

        fn peer_ids(&self) -> Vec<PeerId> {
            self.peer_ids_val.clone()
        }

        fn peer_contains(&self, _peer: &PeerId, _name: &str) -> bool {
            self.peer_contains_val
        }

        fn peer_operations(&self, _peer: &PeerId) -> Vec<String> {
            self.peer_operations_val.clone()
        }

        async fn invoke_peer(
            &self,
            _peer: &PeerRef,
            _namespace: &str,
            _operation: &str,
            _input: serde_json::Value,
            parent: &OperationContext,
            _policy: AbortPolicy,
        ) -> ResponseEnvelope {
            *self.invoke_peer_called.lock() = true;
            ResponseEnvelope::ok(
                parent.request_id.clone(),
                serde_json::Value::String("peer".into()),
            )
        }
    }

    fn make_test_context() -> OperationContext {
        OperationContext {
            request_id: "req-1".to_string(),
            parent_request_id: None,
            identity: None,
            handler_identity: None,
            forwarded_for: None,
            capabilities: Capabilities::new(),
            metadata: HashMap::new(),
            scoped_env: ScopedPeerEnv::empty(),
            env: Arc::new(MockEnv::new()),
            abort_policy: AbortPolicy::default(),
            deadline: None,
            internal: false,
            ownership: None,
        }
    }

    fn make_env() -> ChannelsSessionEnv {
        let mock = Arc::new(MockEnv::new());
        let (_client, server) = duplex(64);
        let (_reader, writer) = tokio::io::split(server);
        let (handle, runner) = MuxRunner::new(Box::new(writer));
        tokio::spawn(async move {
            let _ = runner.run().await;
        });
        let manager = ChannelManager::with_defaults(handle, None);
        ChannelsSessionEnv {
            base: mock,
            manager,
        }
    }

    #[tokio::test]
    async fn channels_session_env_delegates_to_base() {
        let registry = Arc::new(crate::registry::registration::OperationRegistry::new());
        let base: Arc<dyn OperationEnv + Send + Sync> = Arc::new(LocalOperationEnv::new(registry));
        let (_client, server) = duplex(64);
        let (_reader, writer) = tokio::io::split(server);
        let (handle, runner) = MuxRunner::new(Box::new(writer));
        tokio::spawn(async move {
            let _ = runner.run().await;
        });
        let manager = ChannelManager::with_defaults(handle, None);
        let env = ChannelsSessionEnv { base, manager };
        assert!(env.channel_manager().is_some());
    }

    #[test]
    fn channel_operation_env_is_operation_env() {
        fn assert_operation_env<T: OperationEnv>() {}
        assert_operation_env::<ChannelsSessionEnv>();
    }

    #[tokio::test]
    async fn invoke_with_policy_delegates_to_base() {
        let env = make_env();
        let ctx = make_test_context();
        let result = env
            .invoke_with_policy(
                "ns",
                "op",
                serde_json::Value::Null,
                &ctx,
                AbortPolicy::default(),
            )
            .await;
        assert!(result.result.is_ok());
        assert_eq!(
            result.result.unwrap(),
            serde_json::Value::String("mock".into())
        );
    }

    #[tokio::test]
    async fn contains_delegates_to_base() {
        let env = make_env();
        assert!(env.contains("any-op"));
    }

    #[tokio::test]
    async fn peer_ids_delegates_to_base() {
        let env = make_env();
        let ids = env.peer_ids();
        assert_eq!(ids, vec!["peer-a".to_string(), "peer-b".to_string()]);
    }

    #[tokio::test]
    async fn peer_contains_delegates_to_base() {
        let env = make_env();
        assert!(env.peer_contains(&"peer-a".to_string(), "any-op"));
    }

    #[tokio::test]
    async fn peer_operations_delegates_to_base() {
        let env = make_env();
        let ops = env.peer_operations(&"peer-a".to_string());
        assert_eq!(ops, vec!["op1".to_string(), "op2".to_string()]);
    }

    #[tokio::test]
    async fn invoke_peer_delegates_to_base() {
        let env = make_env();
        let ctx = make_test_context();
        let result = env
            .invoke_peer(
                &PeerRef::Any,
                "ns",
                "op",
                serde_json::Value::Null,
                &ctx,
                AbortPolicy::default(),
            )
            .await;
        assert!(result.result.is_ok());
        assert_eq!(
            result.result.unwrap(),
            serde_json::Value::String("peer".into())
        );
    }
}