//! Full-surface integration suite (phase 4): one in-process `HttpAdapter`
//! serving the gateway endpoints, the `/openapi.json` projection, and the
//! WS channels session over real I/O; `from_openapi` imported against a
//! local HTTP echo server; `to_openapi`/`to_mcp` projections consumed
//! back. Exercises the composition the assembly layer performs.

#![cfg(feature = "mcp")]

use std::collections::HashMap;
use std::sync::Arc;

use alkcall::core::auth::{AuthContext, Identity, IdentityProvider};
use alkcall::core::types::{Capabilities, Connection};
use alkcall::protocol::wire::{EventEnvelope, ResponseEnvelope};
use alkcall::registry::discovery::{
    services_list_handler, services_list_spec, services_schema_handler, services_schema_spec,
};
use alkcall::registry::registration::{
    make_handler, make_streaming_handler, HandlerKind, HandlerRegistration, OperationProvenance,
    OperationRegistry,
};
use alkcall::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
use alkhttp::adapters::FromOpenAPI;
use alkhttp::client::HttpClientConfig;
use alkhttp::server::HttpAdapter;
use alkhttp::websocket::{frame_channel0_chunk, ChunkAssembler, FrameAssembler, WsClient};

fn identity(id: &str, scopes: &[&str]) -> Identity {
    Identity {
        id: id.to_string(),
        scopes: scopes.iter().map(|s| s.to_string()).collect(),
        resources: HashMap::new(),
    }
}

struct StaticTokens {
    tokens: std::sync::Mutex<HashMap<String, Identity>>,
}

impl IdentityProvider for StaticTokens {
    fn resolve_from_fingerprint(&self, _: &str) -> Option<Identity> {
        None
    }
    fn resolve_from_token(&self, token: &alkcall::core::auth::AuthToken) -> Option<Identity> {
        let s = String::from_utf8_lossy(&token.raw).to_string();
        self.tokens
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .get(&s)
            .cloned()
    }
}

fn provider_with(tokens: Vec<(&str, Identity)>) -> Arc<dyn IdentityProvider> {
    let map: HashMap<String, Identity> = tokens
        .into_iter()
        .map(|(t, i)| (t.to_string(), i))
        .collect();
    Arc::new(StaticTokens {
        tokens: std::sync::Mutex::new(map),
    })
}

/// The local operation registry the `HttpAdapter` serves: an echo op (open
/// and echo-restricted variants), a streaming sub op, and the discovery
/// ops the adapters need.
fn local_registry() -> Arc<OperationRegistry> {
    let inner = OperationRegistry::new();
    inner
        .register(HandlerRegistration::new(
            OperationSpec::new(
                "echo/run",
                OperationType::Query,
                Visibility::External,
                serde_json::json!({}),
                serde_json::json!({}),
                vec![],
                AccessControl::default(),
                None,
            ),
            HandlerKind::Once(make_handler(|input, ctx| async move {
                ResponseEnvelope::ok(ctx.request_id, input)
            })),
            OperationProvenance::Local,
            None,
            None,
            Capabilities::new(),
        ))
        .unwrap();
    inner
        .register(HandlerRegistration::new(
            OperationSpec::new(
                "events/tick",
                OperationType::Sub,
                Visibility::External,
                serde_json::json!({}),
                serde_json::json!({}),
                vec![],
                AccessControl::default(),
                None,
            ),
            HandlerKind::Stream(make_streaming_handler(|input, ctx| {
                futures::stream::iter(vec![
                    ResponseEnvelope::ok(
                        ctx.request_id.clone(),
                        serde_json::json!({ "n": 1, "input": input }),
                    ),
                    ResponseEnvelope::ok(
                        ctx.request_id.clone(),
                        serde_json::json!({ "n": 2, "input": input }),
                    ),
                ])
            })),
            OperationProvenance::Local,
            None,
            None,
            Capabilities::new(),
        ))
        .unwrap();
    let inner = Arc::new(inner);

    let registry = OperationRegistry::new();
    registry
        .register(HandlerRegistration::new(
            services_list_spec(),
            HandlerKind::Once(services_list_handler(Arc::clone(&inner))),
            OperationProvenance::Local,
            None,
            None,
            Capabilities::new(),
        ))
        .unwrap();
    registry
        .register(HandlerRegistration::new(
            services_schema_spec(),
            HandlerKind::Once(services_schema_handler(Arc::clone(&inner))),
            OperationProvenance::Local,
            None,
            None,
            Capabilities::new(),
        ))
        .unwrap();
    for spec in inner.list_operations() {
        let name = spec.name.clone();
        let reg = inner.registration(&name).unwrap();
        registry
            .register(HandlerRegistration::new(
                reg.spec.clone(),
                reg.handler.clone(),
                reg.provenance,
                reg.composition_authority.clone(),
                reg.scoped_env.clone(),
                reg.capabilities.clone(),
            ))
            .unwrap();
    }
    Arc::new(registry)
}

/// Serve the full adapter surface over a real TCP listener. Each accepted
/// TCP connection is wrapped as an alkcall `Connection` (single-stream,
/// `http/1.1` ALPN) and handed to the adapter's `ProtocolHandler::handle`
/// — the same path a production endpoint drives. Returns the base URL.
async fn spawn_full_server(
    registry: Arc<OperationRegistry>,
    provider: Arc<dyn IdentityProvider>,
) -> String {
    let adapter = std::sync::Arc::new(HttpAdapter::new(Arc::clone(&provider), registry));
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(async move {
        loop {
            let Ok((sock, _)) = listener.accept().await else {
                break;
            };
            let conn = Connection::from_bidi(sock, b"http/1.1".to_vec(), None);
            // The bearer middleware resolves the identity per request from
            // the Authorization header; the transport-level AuthContext
            // carries no identity (TLS-identity binding is the endpoint's
            // job, not this test's).
            let auth = AuthContext::anonymous(b"http/1.1");
            let a = std::sync::Arc::clone(&adapter);
            tokio::spawn(async move {
                let _ =
                    alkcall::core::types::ProtocolHandler::handle(a.as_ref(), conn, &auth).await;
            });
        }
    });
    format!("http://{addr}")
}

#[tokio::test]
async fn full_surface_gateway_over_http() {
    // A minimal registry with the discovery ops; the gateway endpoints
    // must serve search/schema/call/subscribe against it over HTTP.
    let registry = local_registry();
    let provider = provider_with(vec![("tok-1", identity("alice", &["user"]))]);
    let base = spawn_full_server(Arc::clone(&registry), Arc::clone(&provider)).await;
    let client = reqwest::Client::new();

    // /healthz — no auth.
    let resp = client.get(format!("{base}/healthz")).send().await.unwrap();
    assert_eq!(resp.status(), 200);

    // /search — ACL-filtered discovery.
    let resp = client
        .get(format!("{base}/search"))
        .header("Authorization", "Bearer tok-1")
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    let names: Vec<&str> = body["output"]["operations"]
        .as_array()
        .unwrap()
        .iter()
        .filter_map(|o| o["name"].as_str())
        .collect();
    assert!(names.contains(&"echo/run"), "got {names:?}");
    assert!(names.contains(&"events/tick"), "got {names:?}");

    // /call — request/response round trip.
    let resp = client
        .post(format!("{base}/call"))
        .header("Authorization", "Bearer tok-1")
        .json(&serde_json::json!({ "operation": "echo/run", "input": { "v": 42 } }))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["output"]["v"], 42);

    // /subscribe — SSE stream of the Sub op.
    let resp = client
        .post(format!("{base}/subscribe"))
        .header("Authorization", "Bearer tok-1")
        .json(&serde_json::json!({ "operation": "events/tick", "input": {} }))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let text = resp.text().await.unwrap();
    assert!(text.contains("\"n\":1"), "first chunk in SSE: {text}");
    assert!(text.contains("\"n\":2"), "second chunk in SSE: {text}");

    // /schema — the full spec (GET with a name query param).
    let resp = client
        .get(format!("{base}/schema?name=echo%2Frun"))
        .header("Authorization", "Bearer tok-1")
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["output"]["name"], "echo/run");

    // /openapi.json — the 6-endpoint projection including /publish.
    let resp = client
        .get(format!("{base}/openapi.json"))
        .header("Authorization", "Bearer tok-1")
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["info"]["version"], "1.4.0");
    assert!(body["paths"].get("/publish").is_some());
    assert!(body["paths"].get("/call").is_some());
}

#[tokio::test]
async fn full_surface_ws_call_round_trip() {
    let registry = local_registry();
    let provider = provider_with(vec![("tok-1", identity("alice", &["user"]))]);
    let base = spawn_full_server(registry, provider).await;
    // WS endpoint rides the same TCP listener.
    let ws_base = base.replacen("http://", "ws://", 1);

    let mut ws = WsClient::connect_authorized(&format!("{ws_base}/alk/channels"), "tok-1")
        .await
        .unwrap();
    let frame = EventEnvelope::requested(
        "ws-full-1",
        serde_json::json!({ "operationId": "echo/run", "input": { "v": 7 } }),
    );
    ws.send_binary(frame_channel0_chunk(&frame)).await;

    let mut chunks = ChunkAssembler::new();
    let mut frames = FrameAssembler::new();
    let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
    loop {
        assert!(tokio::time::Instant::now() < deadline, "timed out");
        if let Some(env) = frames.next_frame() {
            assert_eq!(env.r#type, "call.responded");
            assert_eq!(env.id, "ws-full-1");
            assert_eq!(env.payload["output"]["v"], 7);
            break;
        }
        let bin = ws.next_binary(std::time::Duration::from_millis(500)).await;
        match bin {
            Some(bytes) => {
                chunks.push(&bytes);
                while let Some((channel_id, payload)) = chunks.next_chunk() {
                    assert_eq!(channel_id, 0);
                    frames.push(&payload);
                }
            }
            None => panic!("ws closed unexpectedly"),
        }
    }
    ws.close().await;
}

#[tokio::test]
async fn from_openapi_import_then_gateway_call() {
    // A local HTTP service the adapter imports; the gateway dispatch then
    // reaches it through the imported forwarding handler.
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let upstream = format!("http://{addr}");
    tokio::spawn(async move {
        let app = axum::Router::new().route(
            "/widgets",
            axum::routing::get(|| async {
                axum::Json(serde_json::json!({ "widgets": ["a", "b"] }))
            }),
        );
        axum::serve(listener, app).await.unwrap();
    });

    let doc = r#"{"openapi":"3.0.0","info":{"title":"T","version":"1"},"paths":{"/widgets":{"get":{"operationId":"listWidgets","responses":{"200":{"content":{"application/json":{"schema":{}}}}}}}}}"#;
    let spec = alkhttp::adapters::OpenAPISpec::from_json(doc).unwrap();
    let config = alkhttp::adapters::HttpServiceConfig {
        namespace: "upstream".to_string(),
        base_url: upstream.clone(),
        auth: None,
        default_headers: HashMap::new(),
    };
    let http_client =
        Arc::new(alkhttp::client::SharedHttpClient::new(HttpClientConfig::default()).unwrap());
    let adapter = FromOpenAPI::new(spec, config, http_client);
    let bundles = alkcall::client::OperationAdapter::import(&adapter)
        .await
        .expect("import succeeds");
    assert_eq!(bundles.len(), 1);
    assert_eq!(bundles[0].spec.name, "upstream/listWidgets");

    // The imported bundles are Internal (ADR-015 — composition material);
    // a wire call to them is NOT_FOUND (ADR-015 §2). The assembly layer
    // composes them under an External facade. Verify Internal-not-callable
    // through the gateway, then compose the External facade and call that.
    let registry = OperationRegistry::new();
    for b in bundles {
        registry.register(b).unwrap();
    }
    registry
        .register(HandlerRegistration::new(
            OperationSpec::new(
                "widgets/list",
                OperationType::Query,
                Visibility::External,
                serde_json::json!({}),
                serde_json::json!({}),
                vec![],
                AccessControl::default(),
                None,
            ),
            HandlerKind::Once(make_handler(|_input, ctx| {
                // The facade composes the imported leaf (env.invoke —
                // composition-only per ADR-015). Its scoped_env declares
                // the imported leaf as the reachable set.
                async move {
                    let response = ctx
                        .env
                        .invoke("upstream", "listWidgets", serde_json::json!({}), &ctx)
                        .await;
                    ResponseEnvelope {
                        request_id: ctx.request_id,
                        result: response.result,
                    }
                }
            })),
            OperationProvenance::Local,
            None,
            Some(alkcall::registry::context::ScopedPeerEnv::new([
                "upstream/listWidgets",
            ])),
            Capabilities::new(),
        ))
        .unwrap();
    let provider = provider_with(vec![("tok-1", identity("alice", &[]))]);
    let base = spawn_full_server(Arc::new(registry), provider).await;

    let client = reqwest::Client::new();

    // Internal op from the wire → NOT_FOUND (does not leak existence).
    let resp = client
        .post(format!("{base}/call"))
        .header("Authorization", "Bearer tok-1")
        .json(&serde_json::json!({ "operation": "upstream/listWidgets", "input": {} }))
        .send()
        .await
        .unwrap();
    assert_eq!(
        resp.status(),
        404,
        "Internal imported op is invisible from the wire"
    );

    // The External facade composes it: external HTTP API → from_openapi
    // forwarding handler → upstream HTTP API.
    let resp = client
        .post(format!("{base}/call"))
        .header("Authorization", "Bearer tok-1")
        .json(&serde_json::json!({ "operation": "widgets/list", "input": {} }))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200, "facade composes the imported op");
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["output"]["widgets"], serde_json::json!(["a", "b"]));
}

#[tokio::test]
async fn to_openapi_and_to_mcp_projections_over_served_registry() {
    use rmcp::model::{CallToolRequestParams, ClientCapabilities, ClientInfo, Implementation};
    use rmcp::service::RoleClient;
    use rmcp::transport::streamable_http_client::{
        StreamableHttpClientTransport, StreamableHttpClientTransportConfig,
    };
    use rmcp::{Peer, ServiceExt};

    let registry = local_registry();
    let provider = provider_with(vec![("tok-1", identity("alice", &["user"]))]);

    // /openapi.json served from the same registry: the to_openapi
    // projection sees the local ops through services/list.
    let base = spawn_full_server(Arc::clone(&registry), Arc::clone(&provider)).await;
    let client = reqwest::Client::new();
    let resp = client
        .get(format!("{base}/openapi.json"))
        .header("Authorization", "Bearer tok-1")
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let doc: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(doc["info"]["title"], "alk gateway");
    assert_eq!(doc["paths"].as_object().unwrap().len(), 6);
    // Validates against openapiv3 (ADR-042 contract).
    let text = serde_json::to_string(&doc).unwrap();
    let _: openapiv3::OpenAPI = serde_json::from_str(&text).unwrap();

    // /mcp served by the same adapter: an MCP client connects, lists the
    // 4 gateway tools, calls search — the tool-gateway pattern (ADR-041).
    let url = format!("{base}/mcp");
    let transport = StreamableHttpClientTransport::from_config(
        StreamableHttpClientTransportConfig::with_uri(url),
    );
    let client_info = ClientInfo::new(
        ClientCapabilities::default(),
        Implementation::new("integration-test", "0.1.0"),
    );
    let running = client_info.serve(transport).await.expect("initialize");
    let peer: Peer<RoleClient> = running.peer().clone();

    let tools = peer
        .list_tools(Default::default())
        .await
        .expect("tools/list");
    let names: Vec<String> = tools.tools.iter().map(|t| t.name.to_string()).collect();
    assert_eq!(names.len(), 4);
    assert!(names.contains(&"search".to_string()));
    assert!(names.contains(&"schema".to_string()));
    assert!(names.contains(&"call".to_string()));
    assert!(names.contains(&"batch".to_string()));

    let mut args = serde_json::Map::new();
    args.insert("query".to_string(), serde_json::Value::Null);
    let params = CallToolRequestParams::new("search".to_string()).with_arguments(args);
    let result = peer.call_tool(params).await.expect("search call");
    assert_eq!(result.is_error, Some(false));
    let structured = result.structured_content.expect("structured present");
    let ops = structured
        .get("operations")
        .and_then(serde_json::Value::as_array)
        .expect("operations array");
    let names: Vec<&str> = ops
        .iter()
        .filter_map(|o| o.get("name").and_then(|v| v.as_str()))
        .collect();
    assert!(names.contains(&"echo/run"), "got {names:?}");
    assert!(
        !names.contains(&"events/tick"),
        "Sub ops excluded from search"
    );
}

#[tokio::test]
async fn gateway_error_fidelity_end_to_end() {
    // Unknown op through /call → 404 NOT_FOUND; internal op → 404.
    let registry = local_registry();
    let provider = provider_with(vec![("tok-1", identity("alice", &["user"]))]);
    let base = spawn_full_server(registry, provider).await;
    let client = reqwest::Client::new();

    let resp = client
        .post(format!("{base}/call"))
        .header("Authorization", "Bearer tok-1")
        .json(&serde_json::json!({ "operation": "no/such", "input": {} }))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 404);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["code"], "NOT_FOUND");
}

use http::header::AUTHORIZATION;

#[tokio::test]
async fn gateway_endpoints_exist_with_bearer_enforcement() {
    let registry = local_registry();
    let provider = provider_with(vec![("tok-1", identity("alice", &["user"]))]);
    let base = spawn_full_server(registry, provider).await;
    let client = reqwest::Client::new();

    // GET endpoints: /search, /schema (schema needs a known name).
    let resp = client
        .get(format!("{base}/search"))
        .header(AUTHORIZATION, "Bearer tok-1")
        .send()
        .await
        .unwrap();
    assert_ne!(
        resp.status(),
        404,
        "/search must exist on the gateway surface"
    );
    let resp = client
        .get(format!("{base}/schema?name=echo/run"))
        .header(AUTHORIZATION, "Bearer tok-1")
        .send()
        .await
        .unwrap();
    assert_ne!(
        resp.status(),
        404,
        "/schema must exist on the gateway surface"
    );
    for path in ["/call", "/batch", "/subscribe", "/publish"] {
        let resp = client
            .post(format!("{base}{path}"))
            .header(AUTHORIZATION, "Bearer tok-1")
            .header("Content-Type", "application/json")
            .body("{}")
            .send()
            .await
            .unwrap();
        assert_ne!(
            resp.status(),
            404,
            "{path} must exist on the gateway surface"
        );
    }

    // Unauthenticated call to an op with no restrictions: allowed
    // (AccessControl::default() passes for any identity, including none).
    let resp = client
        .post(format!("{base}/call"))
        .header("Content-Type", "application/json")
        .body(r#"{"operation": "no/such", "input": {}}"#)
        .send()
        .await
        .unwrap();
    assert_eq!(
        resp.status(),
        404,
        "unknown op is NOT_FOUND regardless of auth"
    );
}

/// COV-11b: the rmcp-entered `ServerHandler::call_tool` routing shell is
/// what production runs; the dispatch-level tests below it bypass it.
/// One real rmcp-protocol `peer.call_tool` round-trip per gateway tool
/// through the served `/mcp` mount.
#[tokio::test]
async fn to_mcp_call_tool_production_dispatch_round_trips_all_tools() {
    use rmcp::model::{CallToolRequestParams, ClientCapabilities, ClientInfo, Implementation};
    use rmcp::service::RoleClient;
    use rmcp::transport::streamable_http_client::{
        StreamableHttpClientTransport, StreamableHttpClientTransportConfig,
    };
    use rmcp::{Peer, ServiceExt};

    let registry = local_registry();
    let provider = provider_with(vec![("tok-1", identity("alice", &["user"]))]);
    let base = spawn_full_server(registry, provider).await;

    let mut default_headers = reqwest::header::HeaderMap::new();
    default_headers.insert(
        reqwest::header::AUTHORIZATION,
        reqwest::header::HeaderValue::from_static("Bearer tok-1"),
    );
    let http = reqwest::Client::builder()
        .default_headers(default_headers)
        .build()
        .unwrap();
    let url = format!("{base}/mcp");
    let transport = StreamableHttpClientTransport::with_client(
        http,
        StreamableHttpClientTransportConfig::with_uri(url),
    );
    let client_info = ClientInfo::new(
        ClientCapabilities::default(),
        Implementation::new("integration-test", "0.1.0"),
    );
    let running = client_info.serve(transport).await.expect("initialize");
    let peer: Peer<RoleClient> = running.peer().clone();

    let schema_params = CallToolRequestParams::new("schema".to_string()).with_arguments(
        serde_json::json!({ "name": "echo/run" })
            .as_object()
            .unwrap()
            .clone(),
    );
    let schema = peer.call_tool(schema_params).await.expect("schema call");
    assert_eq!(schema.is_error, Some(false));
    let structured = schema.structured_content.expect("structured present");
    assert_eq!(structured["name"], "echo/run");
    assert!(structured.get("input_schema").is_some());

    let call_params = CallToolRequestParams::new("call".to_string()).with_arguments(
        serde_json::json!({ "operation": "echo/run", "input": { "v": 7 } })
            .as_object()
            .unwrap()
            .clone(),
    );
    let call_result = peer.call_tool(call_params).await.expect("call call");
    assert_eq!(call_result.is_error, Some(false));
    assert_eq!(
        call_result.structured_content,
        Some(serde_json::json!({ "v": 7 }))
    );

    let batch_params = CallToolRequestParams::new("batch".to_string()).with_arguments(
        serde_json::json!({ "calls": [
            { "operation": "echo/run", "input": { "n": 1 } },
            { "operation": "echo/run", "input": { "n": 2 } }
        ] })
        .as_object()
        .unwrap()
        .clone(),
    );
    let batch = peer.call_tool(batch_params).await.expect("batch call");
    assert_eq!(batch.is_error, Some(false));
    let results = batch
        .structured_content
        .and_then(|v| v.get("results").cloned())
        .expect("results array");
    assert_eq!(
        results,
        serde_json::json!([
            { "isError": false, "output": { "n": 1 } },
            { "isError": false, "output": { "n": 2 } }
        ])
    );

    let unknown = peer
        .call_tool(CallToolRequestParams::new("bogus".to_string()))
        .await
        .expect("unknown tool call resolves, not errors");
    assert_eq!(unknown.is_error, Some(true));
    let err = unknown
        .structured_content
        .expect("structured error present");
    assert_eq!(err["code"], "NOT_FOUND");
    assert!(err["message"]
        .as_str()
        .unwrap_or_default()
        .contains("unknown gateway tool"));

    let _ = running.cancel().await;
}