//! Shared test wiring for the channels-based tests (the producer
//! harness in `channels.rs::tests` and the consumer end-to-end tests
//! in `session.rs::tests` both need the same `ChannelClient` ↔
//! server wiring). `#[cfg(test)]`-only — never part of the public
//! API.
use std::collections::HashMap;
use std::sync::Arc;

use alkcall::channels::client::ChannelClient;
use alkcall::channels::operations::ChannelCore;
use alkcall::channels::policy::default_policy;
use alkcall::core::auth::{AuthContext, Identity};
use alkcall::core::types::Connection as CoreConnection;
use alkcall::registry::registration::OperationRegistry;

use crate::backend::TtyBackend;
use crate::channels::register_openable;

/// Build a `ChannelClient` and a server-side `ChannelCore` + registry
/// with `channels/tty/sub` registered, wired over a `tokio::io::duplex`
/// carrying the channels 8-byte chunk header wire format. Returns the
/// client. The server-side dispatch loop is spawned and runs until the
/// client drops or the test ends.
///
/// This mirrors the alkcall `channel_0_end_to_end_register_openable`
/// test's wiring pattern but uses alktty's `register_openable` helper
/// so the registration is the code under test.
pub(crate) async fn wire_client_and_server(
    backends: Arc<HashMap<String, Arc<dyn TtyBackend>>>,
    ownership: Option<Arc<dyn alkcall::core::ownership::OwnershipProvider>>,
    identity: Option<Identity>,
) -> ChannelClient {
    use alkcall::channels::adapter::{ChannelsAdapter, InstallChannelZero};
    use alkcall::channels::policy::NoCap;
    use alkcall::core::auth::IdentityProvider;
    use alkcall::protocol::connection::split_single_stream;
    use alkcall::protocol::dispatch::Dispatcher;

    struct NoopIdProvider;
    impl IdentityProvider for NoopIdProvider {
        fn resolve_from_fingerprint(&self, _: &str) -> Option<Identity> {
            None
        }
        fn resolve_from_token(&self, _: &alkcall::core::auth::AuthToken) -> Option<Identity> {
            None
        }
    }

    let policy = default_policy();
    let policy_for_hook = Arc::clone(&policy);
    let identity_for_conn = identity.clone();

    let install_hook: InstallChannelZero = Arc::new(move |manager, channel0_conn, auth| {
        let backends = Arc::clone(&backends);
        let ownership = ownership.clone();
        let _identity = identity.clone();
        let policy = Arc::clone(&policy_for_hook);
        tokio::spawn(async move {
            let channel0_bidi = match channel0_conn.accept_bi().await {
                Ok(s) => s,
                Err(_) => return,
            };
            let (writer, reader) = split_single_stream(channel0_bidi);

            // Propagate the identity to the channel-0
            // connection so the call dispatch's
            // `resolve_identity` sees it. The
            // `ChannelsAdapter` builds `channel0_conn` fresh
            // from `channel_source` and does NOT inherit the
            // outer connection's identity — we set it here
            // so the `AccessControl` scope-gate can check it.
            if let Some(id) = _identity {
                let _ = channel0_conn.set_identity(id);
            }

            let core = ChannelCore::new(manager, policy);
            let mut registry = OperationRegistry::new();
            register_openable(
                &core,
                Arc::clone(&backends),
                ownership.clone(),
                &mut registry,
                auth.clone(),
            )
            .expect("register_openable");
            let registry = Arc::new(registry);
            let provider: Arc<dyn IdentityProvider> = Arc::new(NoopIdProvider);
            let call_connection = Arc::new(
                alkcall::protocol::connection::CallConnection::new_single_stream(
                    channel0_conn,
                    Arc::clone(&writer),
                ),
            );
            let dp = Dispatcher::new(registry, provider);
            dp.run_loop_single_stream(call_connection, reader, writer)
                .await;
        })
    });

    let (client_end, server_end) = tokio::io::duplex(64 * 1024);
    let client_conn = CoreConnection::from_bidi(client_end, b"alk/channels".to_vec(), None);
    let server_conn = CoreConnection::from_bidi(server_end, b"alk/channels".to_vec(), None);
    // Set the identity on the server connection so the call
    // dispatch sees it (the ACL check runs against the
    // connection's identity, not the `install_channel_zero`
    // hook's `auth`).
    if let Some(id) = &identity_for_conn {
        let _ = server_conn.set_identity(id.clone());
    }

    let adapter = ChannelsAdapter::new(install_hook, Arc::new(NoCap));
    let auth = AuthContext::anonymous(b"alk/channels");
    let _server_handle = tokio::spawn(async move {
        let _ = alkcall::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await;
    });

    ChannelClient::from_connection(client_conn)
        .await
        .expect("channel client init")
}

/// An identity with the `tty:open` scope (the gate
/// `channels/tty/sub` requires).
pub(crate) fn tty_identity(id: &str) -> Identity {
    Identity {
        id: id.to_string(),
        scopes: vec![crate::adapter::TTY_OPEN_SCOPE.to_string()],
        resources: HashMap::new(),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::backend::MockBackend;
    use crate::channels::OP_TTY_OPEN;

    #[test]
    fn tty_identity_has_tty_open_scope() {
        let id = tty_identity("carol");
        assert_eq!(id.id, "carol");
        assert!(id
            .scopes
            .iter()
            .any(|s| s == crate::adapter::TTY_OPEN_SCOPE));
    }

    #[tokio::test]
    async fn harness_open_op_returns_channel_id() {
        let mut backends: HashMap<String, Arc<dyn TtyBackend>> = HashMap::new();
        backends.insert("mock".to_string(), Arc::new(MockBackend::with_exit_code(0)));
        let client =
            wire_client_and_server(Arc::new(backends), None, Some(tty_identity("alice"))).await;
        let response = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            client.call_open_op(
                OP_TTY_OPEN,
                serde_json::json!({ "carriage": "raw", "backend": "mock", "cmd": ["true"] }),
            ),
        )
        .await
        .expect("open op timed out");
        assert!(
            response.result.is_ok(),
            "open op failed: {:?}",
            response.result
        );
    }

    #[tokio::test]
    async fn harness_handler_to_client_data_flows() {
        use tokio::io::AsyncReadExt;

        let mut backends: HashMap<String, Arc<dyn TtyBackend>> = HashMap::new();
        backends.insert("mock".to_string(), Arc::new(MockBackend::with_exit_code(0)));
        let client =
            wire_client_and_server(Arc::new(backends), None, Some(tty_identity("alice"))).await;

        let (channel_id, send, mut recv) = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            client.open_channel(
                OP_TTY_OPEN,
                serde_json::json!({ "carriage": "raw", "backend": "mock", "cmd": ["true"] }),
                crate::channels::TTY_ALPN,
            ),
        )
        .await
        .expect("open_channel timed out")
        .expect("open_channel");

        assert!(channel_id > 0);

        // The producer writes a 5-byte TTY header (stdout sentinel)
        // almost immediately after the open (MockBackend resolves
        // exit right away) — a push-first producer whose first write
        // races the consumer's adopt. Read it: alkcall 0.4.1's
        // early-arrival park guarantees the first chunks are not lost.
        let mut first = [0u8; 5];
        tokio::time::timeout(
            std::time::Duration::from_secs(5),
            recv.read_exact(&mut first),
        )
        .await
        .expect("no first chunk from producer (handler→client direction)")
        .expect("read first chunk header");
        assert_eq!(first[0], crate::wire::STREAM_STDOUT);
        drop(send);
    }

    /// End-to-end (R4): `input` that passes the registry's (partial)
    /// schema but fails the handler's full `NegotiateRequest` parse —
    /// `cwd` typed as a number. The open op succeeds (the schema is
    /// deliberately partial), the producer-side handler writes a
    /// `0x00`-prefixed `malformed_negotiation` error frame on the
    /// channel stream, and the consumer-side post-open read sequence
    /// (`from_halves_raw`'s peek + error-frame parse) observes it — the
    /// same failure class as the post-open semantic failures, not a
    /// silent teardown. Uses `ChannelClient::open_channel` directly
    /// (the consumer's local fail-fast parse in `open_via_channels`
    /// would reject these params before the open op runs — R5's
    /// fail-fast path).
    #[tokio::test]
    async fn schema_valid_but_unparseable_input_gets_malformed_negotiation_frame() {
        use tokio::io::AsyncReadExt;

        let mut backends: HashMap<String, Arc<dyn TtyBackend>> = HashMap::new();
        backends.insert("mock".to_string(), Arc::new(MockBackend::with_exit_code(0)));
        let client =
            wire_client_and_server(Arc::new(backends), None, Some(tty_identity("alice"))).await;

        let (_channel_id, _send, mut recv) = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            client.open_channel(
                OP_TTY_OPEN,
                serde_json::json!({
                    "carriage": "raw",
                    "backend": "mock",
                    "cmd": ["true"],
                    "cwd": 42
                }),
                crate::channels::TTY_ALPN,
            ),
        )
        .await
        .expect("open_channel timed out")
        .expect("open_channel — the partial schema accepts cwd: 42");

        // The consumer-side post-open sequence (`from_halves_raw`):
        // peek the first byte; `0x00` = negotiation error frame.
        let mut first = [0u8; 1];
        tokio::time::timeout(
            std::time::Duration::from_secs(5),
            recv.read_exact(&mut first),
        )
        .await
        .expect("no response byte from producer")
        .expect("read first byte");
        assert_eq!(first[0], 0x00, "error frame length prefix starts with 0x00");

        let mut len_rest = [0u8; 3];
        recv.read_exact(&mut len_rest).await.expect("read len rest");
        let len = u32::from_be_bytes([first[0], len_rest[0], len_rest[1], len_rest[2]]) as usize;
        let mut body = vec![0u8; len];
        recv.read_exact(&mut body).await.expect("read error body");
        let v: serde_json::Value = serde_json::from_slice(&body).expect("parse error frame");
        assert_eq!(v["error"], "malformed_negotiation");
        assert!(v["message"].as_str().is_some_and(|m| !m.is_empty()));
    }
}