//! MC-5 (store/WS level) — per-channel multi-select over a real `run_dispatch`
//! WebSocket session, mirroring `messaging_config_gate.rs`'s dispatcher harness.
//!
//! Proves the per-channel WS round-trip (Units 2-3, with the server-side
//! optional `channel` field): a fresh store shows BOTH channels disabled;
//! `messaging.config.set { channel: "slack", enabled: true }` flips ONLY Slack;
//! enabling iMessage too leaves both on; and a `set` with NO `channel` field
//! defaults to iMessage (back-compat). All host-gated — a non-host caller is
//! rejected, identical to the single-channel gate.

use car_memgine::MemgineEngine;
use car_server_core::{run_dispatch, ServerState, ServerStateConfig};
use futures::{SinkExt, StreamExt};
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
use std::sync::Arc;
use tempfile::TempDir;
use tokio::net::TcpListener;
use tokio::sync::Mutex;
use tokio_tungstenite::{accept_async, connect_async, tungstenite::Message};

type Ws =
    tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>;

const HOST_TOKEN: &str = "test-host-token-bbbbbbbbbbbbbbbbbbbbbbbbbbb";

/// These tests mutate the process-global `HOME` (the handlers resolve
/// `~/.car/messaging.json` from it at call time), so they must not run their
/// `HOME`-dependent body concurrently. Each test holds this guard for its whole
/// body. A `tokio::sync::Mutex` guard is `Send`, so it may be held across
/// `.await`. (The existing `messaging_config_gate.rs` is a single test, so it
/// never raced; this file has several.)
static HOME_GUARD: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());

fn gated_state(journal_dir: std::path::PathBuf) -> Arc<ServerState> {
    let engine = Arc::new(Mutex::new(MemgineEngine::new(None)));
    let cfg = ServerStateConfig::new(journal_dir).with_shared_memgine(engine);
    let state = Arc::new(ServerState::with_config(cfg));
    state
        .install_host_token(HOST_TOKEN.to_string())
        .expect("install host token");
    state
}

async fn spawn_dispatcher(state: Arc<ServerState>) -> SocketAddr {
    let listener = TcpListener::bind(SocketAddr::V4(SocketAddrV4::new(
        Ipv4Addr::new(127, 0, 0, 1),
        0,
    )))
    .await
    .expect("bind loopback");
    let addr = listener.local_addr().expect("local_addr");
    tokio::spawn(async move {
        let (stream, peer) = listener.accept().await.expect("accept");
        let ws = accept_async(stream).await.expect("ws handshake");
        let (write, read) = ws.split();
        let _ = run_dispatch(read, Box::pin(write), peer.to_string(), state).await;
    });
    addr
}

async fn call(
    ws: &mut Ws,
    id: &str,
    method: &str,
    params: serde_json::Value,
) -> serde_json::Value {
    ws.send(Message::Text(
        serde_json::json!({ "jsonrpc": "2.0", "id": id, "method": method, "params": params })
            .to_string()
            .into(),
    ))
    .await
    .expect("send");
    let text = ws
        .next()
        .await
        .expect("frame")
        .expect("frame ok")
        .into_text()
        .expect("text")
        .to_string();
    serde_json::from_str(&text).expect("parse")
}

async fn auth_host(ws: &mut Ws) {
    let auth = call(
        ws,
        "auth",
        "session.auth",
        serde_json::json!({ "host_token": HOST_TOKEN }),
    )
    .await;
    assert_eq!(auth["result"]["role"], "host", "host auth should succeed: {auth}");
}

/// MC-5: both channels default off; each enables independently; both-at-once
/// works; a channel-less `set` defaults to iMessage.
#[tokio::test]
async fn per_channel_multi_select_round_trips_via_ws() {
    let _home = HOME_GUARD.lock().await;
    let journal = TempDir::new().unwrap();
    let fake_home = TempDir::new().unwrap();
    std::env::set_var("HOME", fake_home.path());

    let state = gated_state(journal.path().to_path_buf());
    let addr = spawn_dispatcher(state).await;
    let (mut ws, _) = connect_async(format!("ws://{addr}"))
        .await
        .expect("connect");
    auth_host(&mut ws).await;

    // 1. Fresh store: BOTH channels default disabled.
    let get_imsg = call(
        &mut ws,
        "g0",
        "messaging.config.get",
        serde_json::json!({ "channel": "imessage" }),
    )
    .await;
    assert_eq!(get_imsg["result"]["enabled"], false, "iMessage defaults off");
    assert_eq!(get_imsg["result"]["channel"], "imessage");

    let get_slack = call(
        &mut ws,
        "g1",
        "messaging.config.get",
        serde_json::json!({ "channel": "slack" }),
    )
    .await;
    assert_eq!(get_slack["result"]["enabled"], false, "Slack defaults off");
    assert_eq!(get_slack["result"]["channel"], "slack");

    // 2. Enable Slack ONLY → Slack on, iMessage still off (independent flags).
    let set_slack = call(
        &mut ws,
        "s0",
        "messaging.config.set",
        serde_json::json!({ "channel": "slack", "enabled": true }),
    )
    .await;
    assert!(set_slack.get("error").is_none(), "set slack should succeed: {set_slack}");
    assert_eq!(set_slack["result"]["enabled"], true);
    assert_eq!(set_slack["result"]["channel"], "slack");

    let get_imsg = call(
        &mut ws,
        "g2",
        "messaging.config.get",
        serde_json::json!({ "channel": "imessage" }),
    )
    .await;
    assert_eq!(
        get_imsg["result"]["enabled"], false,
        "enabling Slack must NOT flip iMessage"
    );
    let get_slack = call(
        &mut ws,
        "g3",
        "messaging.config.get",
        serde_json::json!({ "channel": "slack" }),
    )
    .await;
    assert_eq!(get_slack["result"]["enabled"], true, "Slack is now on");

    // 3. Enable iMessage too (the "both at once" multi-select, C2).
    let set_imsg = call(
        &mut ws,
        "s1",
        "messaging.config.set",
        serde_json::json!({ "channel": "imessage", "enabled": true }),
    )
    .await;
    assert_eq!(set_imsg["result"]["enabled"], true);
    // Both on now.
    let get_imsg = call(
        &mut ws,
        "g4",
        "messaging.config.get",
        serde_json::json!({ "channel": "imessage" }),
    )
    .await;
    let get_slack = call(
        &mut ws,
        "g5",
        "messaging.config.get",
        serde_json::json!({ "channel": "slack" }),
    )
    .await;
    assert_eq!(get_imsg["result"]["enabled"], true);
    assert_eq!(get_slack["result"]["enabled"], true);

    // 4. Per-channel allowlists are independent.
    let add_slack = call(
        &mut ws,
        "s2",
        "messaging.config.set",
        serde_json::json!({ "channel": "slack", "add_handles": ["U012ABCDEF"] }),
    )
    .await;
    assert_eq!(add_slack["result"]["allowlisted_handles"][0], "U012ABCDEF");
    let get_imsg = call(
        &mut ws,
        "g6",
        "messaging.config.get",
        serde_json::json!({ "channel": "imessage" }),
    )
    .await;
    assert_eq!(
        get_imsg["result"]["allowlisted_handles"]
            .as_array()
            .unwrap()
            .len(),
        0,
        "Slack allowlist add must NOT touch iMessage allowlist"
    );
}

/// MC-5 back-compat: a `set`/`get` with NO `channel` field defaults to iMessage
/// (the #403 surface and the FFI bindings, which have no channel field, keep
/// working unchanged).
#[tokio::test]
async fn channel_absent_defaults_to_imessage() {
    let _home = HOME_GUARD.lock().await;
    let journal = TempDir::new().unwrap();
    let fake_home = TempDir::new().unwrap();
    std::env::set_var("HOME", fake_home.path());

    let state = gated_state(journal.path().to_path_buf());
    let addr = spawn_dispatcher(state).await;
    let (mut ws, _) = connect_async(format!("ws://{addr}"))
        .await
        .expect("connect");
    auth_host(&mut ws).await;

    // set with NO channel field → mutates iMessage (the default).
    let set = call(
        &mut ws,
        "s0",
        "messaging.config.set",
        serde_json::json!({ "enabled": true, "add_handles": ["+15551234567"] }),
    )
    .await;
    assert!(set.get("error").is_none(), "channel-less set should succeed: {set}");
    assert_eq!(set["result"]["channel"], "imessage", "absent channel ⇒ iMessage");
    assert_eq!(set["result"]["enabled"], true);

    // get with NO channel field reflects the iMessage mutation.
    let get = call(&mut ws, "g0", "messaging.config.get", serde_json::json!({})).await;
    assert_eq!(get["result"]["channel"], "imessage");
    assert_eq!(get["result"]["enabled"], true);
    assert_eq!(get["result"]["allowlisted_handles"][0], "+15551234567");

    // The Slack channel is untouched (still off, empty allowlist).
    let get_slack = call(
        &mut ws,
        "g1",
        "messaging.config.get",
        serde_json::json!({ "channel": "slack" }),
    )
    .await;
    assert_eq!(get_slack["result"]["enabled"], false);
    assert_eq!(
        get_slack["result"]["allowlisted_handles"]
            .as_array()
            .unwrap()
            .len(),
        0
    );
}

/// An unrecognized `channel` string is rejected (fail-closed on a typo), not
/// silently defaulted.
#[tokio::test]
async fn unknown_channel_is_rejected() {
    let _home = HOME_GUARD.lock().await;
    let journal = TempDir::new().unwrap();
    let fake_home = TempDir::new().unwrap();
    std::env::set_var("HOME", fake_home.path());

    let state = gated_state(journal.path().to_path_buf());
    let addr = spawn_dispatcher(state).await;
    let (mut ws, _) = connect_async(format!("ws://{addr}"))
        .await
        .expect("connect");
    auth_host(&mut ws).await;

    let resp = call(
        &mut ws,
        "x0",
        "messaging.config.set",
        serde_json::json!({ "channel": "telegram", "enabled": true }),
    )
    .await;
    assert!(
        resp.get("error").is_some(),
        "an unknown channel must be rejected, got: {resp}"
    );
}
