use std::process::{Command, Stdio};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use core_api::{
    ServiceRequestId, SnapshotIncludeFlags, SnapshotRateLimit, SnapshotConfig,
    UdpServiceClient,
};
use introspection_core::{RuntimeLoadReport, StatusSnapshot};
use serde::{Deserialize, Serialize};
use serde_json::Value;

const STATUS_SERVICE_API_VERSION: &str = "robotrt.gateway.service.v1";
const STATUS_SERVICE_NAME: &str = "/robotrt/gateway/query";

#[derive(Debug, Serialize, Deserialize)]
struct StatusServiceRequest {
    api_version: String,
    op: String,
    snapshot_config: Option<SnapshotConfig>,
    service_call: Option<Value>,
    op_payload: Option<Value>,
    gateway_meta: Option<GatewayRequestMeta>,
}

#[derive(Debug, Serialize, Deserialize)]
struct StatusServiceResponse {
    api_version: String,
    op: String,
    snapshot: Option<StatusSnapshot>,
    runtime: Option<RuntimeLoadReport>,
    snapshot_config: Option<SnapshotConfig>,
    op_result: Option<Value>,
}

#[derive(Debug, Serialize, Deserialize)]
struct GatewayRequestMeta {
    plane: String,
    route_mode: String,
    traffic_kind: String,
    allow_gateway_data_proxy: bool,
}

fn allocate_udp_endpoint() -> String {
    let socket = std::net::UdpSocket::bind("127.0.0.1:0").expect("bind loopback udp port");
    let addr = socket.local_addr().expect("read local addr");
    format!("127.0.0.1:{}", addr.port())
}

#[test]
fn host_serves_demo_runtime_once() {
    let endpoint = allocate_udp_endpoint();

    let mut child = Command::new(env!("CARGO_BIN_EXE_robotrt-host"))
        .args([
            "gateway",
            "--bind",
            endpoint.as_str(),
            "--source",
            "demo",
            "--once",
        ])
        .stdout(Stdio::null())
        .stderr(Stdio::piped())
        .spawn()
        .expect("spawn robotrt-host gateway");

    std::thread::sleep(Duration::from_millis(120));

    let req = StatusServiceRequest {
        api_version: STATUS_SERVICE_API_VERSION.to_string(),
        op: "runtime".to_string(),
        snapshot_config: None,
        service_call: None,
        op_payload: None,
        gateway_meta: None,
    };

    let mut last_err = None;
    let mut resp_opt: Option<StatusServiceResponse> = None;
    for _ in 0..5 {
        let client = UdpServiceClient::new(endpoint.clone(), Duration::from_millis(2_000))
            .expect("create udp service client");
        match client.call_json(STATUS_SERVICE_NAME, ServiceRequestId(2), &req) {
            Ok(resp) => {
                resp_opt = Some(resp);
                break;
            }
            Err(err) => {
                last_err = Some(err.to_string());
                std::thread::sleep(Duration::from_millis(150));
            }
        }
    }

    let resp = resp_opt
        .ok_or_else(|| format!("query runtime report from daemon host failed: {last_err:?}"))
        .expect("query runtime report from daemon host");

    assert_eq!(resp.api_version, STATUS_SERVICE_API_VERSION);
    assert_eq!(resp.op, "runtime");
    let runtime = resp.runtime.expect("runtime payload exists");
    assert!(runtime.queue_depth.is_some());

    let status = child.wait().expect("wait daemon host child");
    assert!(status.success(), "daemon host exited with {status:?}");
}

#[test]
fn host_serves_demo_snapshot_once() {
    let endpoint = allocate_udp_endpoint();

    let mut child = Command::new(env!("CARGO_BIN_EXE_robotrt-host"))
        .args([
            "gateway",
            "--bind",
            endpoint.as_str(),
            "--source",
            "demo",
            "--once",
        ])
        .stdout(Stdio::null())
        .stderr(Stdio::piped())
        .spawn()
        .expect("spawn robotrt-host gateway");

    std::thread::sleep(Duration::from_millis(120));

    let req = StatusServiceRequest {
        api_version: STATUS_SERVICE_API_VERSION.to_string(),
        op: "snapshot".to_string(),
        snapshot_config: None,
        service_call: None,
        op_payload: None,
        gateway_meta: None,
    };

    let mut last_err = None;
    let mut resp_opt: Option<StatusServiceResponse> = None;
    for _ in 0..5 {
        let client = UdpServiceClient::new(endpoint.clone(), Duration::from_millis(2_000))
            .expect("create udp service client");
        match client.call_json(STATUS_SERVICE_NAME, ServiceRequestId(1), &req) {
            Ok(resp) => {
                resp_opt = Some(resp);
                break;
            }
            Err(err) => {
                last_err = Some(err.to_string());
                std::thread::sleep(Duration::from_millis(150));
            }
        }
    }

    let resp = resp_opt
        .ok_or_else(|| format!("query status snapshot from daemon host failed: {last_err:?}"))
        .expect("query status snapshot from daemon host");

    assert_eq!(resp.api_version, STATUS_SERVICE_API_VERSION);
    assert_eq!(resp.op, "snapshot");
    let snapshot = resp.snapshot.expect("snapshot payload exists");
    assert!(snapshot.nodes.iter().any(|n| n.name == "demo_node"));
    assert!(snapshot
        .topics
        .iter()
        .any(|t| t.name == "/demo/camera/raw"));

    let status = child.wait().expect("wait daemon host child");
    assert!(status.success(), "daemon host exited with {status:?}");
}

#[test]
fn host_applies_snapshot_config_online_immediately() {
    let endpoint = allocate_udp_endpoint();

    let mut child = Command::new(env!("CARGO_BIN_EXE_robotrt-host"))
        .args(["gateway", "--bind", endpoint.as_str(), "--source", "demo"])
        .stdout(Stdio::null())
        .stderr(Stdio::piped())
        .spawn()
        .expect("spawn robotrt-host gateway");

    std::thread::sleep(Duration::from_millis(150));

    let config = SnapshotConfig {
        enabled: true,
        interval_ms: 200,
        rate_limit: SnapshotRateLimit {
            max_writes_per_sec: 5,
            burst: 5,
        },
        atomic_write: true,
        include_flags: SnapshotIncludeFlags {
            status: true,
            runtime: false,
            middleware: true,
            resource: true,
        },
        ..SnapshotConfig::default()
    };

    let set_req = StatusServiceRequest {
        api_version: STATUS_SERVICE_API_VERSION.to_string(),
        op: "snapshot_config_set".to_string(),
        snapshot_config: Some(config),
        service_call: None,
        op_payload: None,
        gateway_meta: None,
    };

    let mut last_set_err = None;
    let mut set_resp_opt: Option<StatusServiceResponse> = None;
    for _ in 0..5 {
        let client = UdpServiceClient::new(endpoint.clone(), Duration::from_millis(2_000))
            .expect("create udp service client");
        match client.call_json(STATUS_SERVICE_NAME, ServiceRequestId(21), &set_req) {
            Ok(resp) => {
                set_resp_opt = Some(resp);
                break;
            }
            Err(err) => {
                last_set_err = Some(err.to_string());
                std::thread::sleep(Duration::from_millis(150));
            }
        }
    }

    let set_resp = set_resp_opt
        .ok_or_else(|| format!("set snapshot config failed: {last_set_err:?}"))
        .expect("set snapshot config");
    assert_eq!(set_resp.op, "snapshot_config_set");
    let applied = set_resp
        .snapshot_config
        .expect("snapshot_config payload exists");
    assert!(!applied.include_flags.runtime);

    let runtime_req = StatusServiceRequest {
        api_version: STATUS_SERVICE_API_VERSION.to_string(),
        op: "runtime".to_string(),
        snapshot_config: None,
        service_call: None,
        op_payload: None,
        gateway_meta: None,
    };
    let client = UdpServiceClient::new(endpoint.clone(), Duration::from_millis(2_000))
        .expect("create udp service client");
    let runtime_result: Result<StatusServiceResponse, _> =
        client.call_json(STATUS_SERVICE_NAME, ServiceRequestId(22), &runtime_req);
    let runtime_err = runtime_result
        .expect_err("runtime should be rejected when include_flags.runtime=false");
    assert!(
        runtime_err.to_string().contains("include_flags.runtime"),
        "unexpected runtime error: {runtime_err}"
    );

    child.kill().expect("kill daemon host child");
    let _ = child.wait();
}

#[test]
fn host_rejects_non_governance_gateway_meta() {
    let endpoint = allocate_udp_endpoint();

    let mut child = Command::new(env!("CARGO_BIN_EXE_robotrt-host"))
        .args([
            "gateway",
            "--bind",
            endpoint.as_str(),
            "--source",
            "demo",
            "--once",
        ])
        .stdout(Stdio::null())
        .stderr(Stdio::piped())
        .spawn()
        .expect("spawn robotrt-host gateway");

    std::thread::sleep(Duration::from_millis(120));

    let req = StatusServiceRequest {
        api_version: STATUS_SERVICE_API_VERSION.to_string(),
        op: "runtime".to_string(),
        snapshot_config: None,
        service_call: None,
        op_payload: None,
        gateway_meta: Some(GatewayRequestMeta {
            plane: String::from("data"),
            route_mode: String::from("direct_or_nearby"),
            traffic_kind: String::from("governance"),
            allow_gateway_data_proxy: false,
        }),
    };

    let mut last_err = None;
    for _ in 0..5 {
        let client = UdpServiceClient::new(endpoint.clone(), Duration::from_millis(2_000))
            .expect("create udp service client");
        let result: Result<StatusServiceResponse, _> =
            client.call_json(STATUS_SERVICE_NAME, ServiceRequestId(31), &req);
        match result {
            Ok(_) => panic!("non-governance plane should be rejected"),
            Err(err) => {
                if err.to_string().contains("GW_ERR_INVALID_META") {
                    last_err = Some(err.to_string());
                    break;
                }
                last_err = Some(err.to_string());
                std::thread::sleep(Duration::from_millis(150));
            }
        }
    }
    let err = last_err.unwrap_or_else(|| String::from("missing error from gateway"));
    assert!(
        err.contains("GW_ERR_INVALID_META"),
        "unexpected error: {err}"
    );

    let status = child.wait().expect("wait daemon host child");
    assert!(status.success(), "daemon host exited with {status:?}");
}

#[test]
fn host_rejects_topic_publish_data_plane_payload() {
    let endpoint = allocate_udp_endpoint();

    let mut child = Command::new(env!("CARGO_BIN_EXE_robotrt-host"))
        .args([
            "gateway",
            "--bind",
            endpoint.as_str(),
            "--source",
            "demo",
            "--once",
        ])
        .stdout(Stdio::null())
        .stderr(Stdio::piped())
        .spawn()
        .expect("spawn robotrt-host gateway");

    std::thread::sleep(Duration::from_millis(120));

    let req = StatusServiceRequest {
        api_version: STATUS_SERVICE_API_VERSION.to_string(),
        op: "topic_publish".to_string(),
        snapshot_config: None,
        service_call: None,
        op_payload: Some(serde_json::json!({
            "topic": "/demo/chatter",
            "payload": {"msg": "hello"},
            "packet_kind": "data"
        })),
        gateway_meta: Some(GatewayRequestMeta {
            plane: String::from("governance"),
            route_mode: String::from("direct_or_nearby"),
            traffic_kind: String::from("topic"),
            allow_gateway_data_proxy: false,
        }),
    };

    let mut last_err = None;
    for _ in 0..5 {
        let client = UdpServiceClient::new(endpoint.clone(), Duration::from_millis(2_000))
            .expect("create udp service client");
        let result: Result<StatusServiceResponse, _> =
            client.call_json(STATUS_SERVICE_NAME, ServiceRequestId(32), &req);
        match result {
            Ok(_) => panic!("data packet should be blocked in gateway topic_publish"),
            Err(err) => {
                if err.to_string().contains("GW_ERR_DATA_PLANE_BLOCKED") {
                    last_err = Some(err.to_string());
                    break;
                }
                last_err = Some(err.to_string());
                std::thread::sleep(Duration::from_millis(150));
            }
        }
    }
    let err = last_err.unwrap_or_else(|| String::from("missing error from gateway"));
    assert!(
        err.contains("GW_ERR_DATA_PLANE_BLOCKED"),
        "unexpected error: {err}"
    );

    let status = child.wait().expect("wait daemon host child");
    assert!(status.success(), "daemon host exited with {status:?}");
}

#[test]
fn host_supports_gateway_observe_policy_and_audit_flow() {
    let endpoint = allocate_udp_endpoint();

    let mut child = Command::new(env!("CARGO_BIN_EXE_robotrt-host"))
        .args([
            "gateway",
            "--bind",
            endpoint.as_str(),
            "--source",
            "demo",
        ])
        .stdout(Stdio::null())
        .stderr(Stdio::piped())
        .spawn()
        .expect("spawn robotrt-host gateway");

    std::thread::sleep(Duration::from_millis(150));

    let client = UdpServiceClient::new(endpoint.clone(), Duration::from_millis(2_000))
        .expect("create udp service client");

    let call_with_retry = |request_id: u64, req: &StatusServiceRequest| {
        let mut last_err = None;
        for _ in 0..5 {
            match client.call_json(STATUS_SERVICE_NAME, ServiceRequestId(request_id), req) {
                Ok(resp) => return Ok(resp),
                Err(err) => {
                    last_err = Some(err.to_string());
                    std::thread::sleep(Duration::from_millis(120));
                }
            }
        }
        Err(last_err.unwrap_or_else(|| String::from("missing call error")))
    };

    let set_policy = StatusServiceRequest {
        api_version: STATUS_SERVICE_API_VERSION.to_string(),
        op: "gateway_policy_set".to_string(),
        snapshot_config: None,
        service_call: None,
        op_payload: Some(serde_json::json!({
            "route_preference": "network_first",
            "rate_limit_per_sec": 120,
            "namespace_isolation": "tenant-a",
            "rollback_enabled": true,
            "topic_max_retry": 4,
            "topic_retry_timeout_ms": 250,
            "service_retry_timeout_ms": 320,
            "action_retry_timeout_ms": 350,
            "mission_retry_timeout_ms": 380,
            "topic_max_inflight": 24,
            "topic_replay_window": 32,
            "topic_dedupe_window": 96,
            "topic_replay_strategy": "drop_oldest"
        })),
        gateway_meta: Some(GatewayRequestMeta {
            plane: String::from("governance"),
            route_mode: String::from("direct_or_nearby"),
            traffic_kind: String::from("governance"),
            allow_gateway_data_proxy: false,
        }),
    };
    let set_resp: StatusServiceResponse = call_with_retry(41, &set_policy)
        .expect("set gateway policy");
    assert_eq!(set_resp.op, "gateway_policy_set");
    let set_result = set_resp.op_result.expect("policy set result");
    assert_eq!(
        set_result["policy"]["route_preference"].as_str(),
        Some("network_first")
    );
    assert_eq!(
        set_result["policy"]["namespace_isolation"].as_str(),
        Some("tenant-a")
    );
    assert_eq!(set_result["policy"]["topic_max_retry"].as_u64(), Some(4));
    assert_eq!(
        set_result["policy"]["topic_retry_timeout_ms"].as_u64(),
        Some(250)
    );
    assert_eq!(
        set_result["policy"]["service_retry_timeout_ms"].as_u64(),
        Some(320)
    );
    assert_eq!(
        set_result["policy"]["action_retry_timeout_ms"].as_u64(),
        Some(350)
    );
    assert_eq!(
        set_result["policy"]["mission_retry_timeout_ms"].as_u64(),
        Some(380)
    );
    assert_eq!(
        set_result["policy"]["topic_max_inflight"].as_u64(),
        Some(24)
    );
    assert_eq!(
        set_result["policy"]["topic_replay_window"].as_u64(),
        Some(32)
    );
    assert_eq!(
        set_result["policy"]["topic_dedupe_window"].as_u64(),
        Some(96)
    );
    assert_eq!(
        set_result["policy"]["topic_replay_strategy"].as_str(),
        Some("drop_oldest")
    );

    let observe_req = StatusServiceRequest {
        api_version: STATUS_SERVICE_API_VERSION.to_string(),
        op: "gateway_observe".to_string(),
        snapshot_config: None,
        service_call: None,
        op_payload: Some(serde_json::json!({"audit_limit": 10})),
        gateway_meta: Some(GatewayRequestMeta {
            plane: String::from("governance"),
            route_mode: String::from("direct_or_nearby"),
            traffic_kind: String::from("governance"),
            allow_gateway_data_proxy: false,
        }),
    };
    let observe_resp: StatusServiceResponse = call_with_retry(42, &observe_req)
        .expect("observe via gateway");
    assert_eq!(observe_resp.op, "gateway_observe");
    let observe = observe_resp.op_result.expect("observe result");
    assert_eq!(
        observe["policy"]["route_preference"].as_str(),
        Some("network_first")
    );
    assert!(
        observe["reports"]["snapshot"]["nodes"]
            .as_u64()
            .unwrap_or(0)
            > 0
    );

    let audit_req = StatusServiceRequest {
        api_version: STATUS_SERVICE_API_VERSION.to_string(),
        op: "gateway_audit_list".to_string(),
        snapshot_config: None,
        service_call: None,
        op_payload: Some(serde_json::json!({"limit": 50, "category": "control"})),
        gateway_meta: Some(GatewayRequestMeta {
            plane: String::from("governance"),
            route_mode: String::from("direct_or_nearby"),
            traffic_kind: String::from("governance"),
            allow_gateway_data_proxy: false,
        }),
    };
    let audit_resp: StatusServiceResponse = call_with_retry(43, &audit_req)
        .expect("audit list via gateway");
    assert_eq!(audit_resp.op, "gateway_audit_list");
    let audit = audit_resp.op_result.expect("audit list result");
    assert!(
        audit["entries"]
            .as_array()
            .map(|items| !items.is_empty())
            .unwrap_or(false)
    );

    let shutdown_req = StatusServiceRequest {
        api_version: STATUS_SERVICE_API_VERSION.to_string(),
        op: "gateway_shutdown".to_string(),
        snapshot_config: None,
        service_call: None,
        op_payload: None,
        gateway_meta: Some(GatewayRequestMeta {
            plane: String::from("governance"),
            route_mode: String::from("direct_or_nearby"),
            traffic_kind: String::from("governance"),
            allow_gateway_data_proxy: false,
        }),
    };
    let _: StatusServiceResponse = call_with_retry(44, &shutdown_req)
        .expect("shutdown gateway");

    let status = child.wait().expect("wait daemon host child");
    assert!(status.success(), "daemon host exited with {status:?}");
}

#[test]
fn sidecar_standalone_restarts_and_writes_evidence() {
    let endpoint = allocate_udp_endpoint();
    let evidence = std::env::temp_dir().join(format!(
        "robotrt-sidecar-evidence-{}.json",
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("unix epoch")
            .as_nanos()
    ));

    let mut child = Command::new(env!("CARGO_BIN_EXE_robotrt-host"))
        .args([
            "sidecar",
            "--mode",
            "standalone",
            "--bind",
            endpoint.as_str(),
            "--source",
            "demo",
            "--child-once",
            "--restart-max",
            "1",
            "--health-interval-ms",
            "50",
            "--handshake-retries",
            "20",
            "--evidence",
            evidence.to_str().expect("utf-8 evidence path"),
            "--json",
        ])
        .stdout(Stdio::null())
        .stderr(Stdio::piped())
        .spawn()
        .expect("spawn robotrt-host sidecar");

    let status = child.wait().expect("wait sidecar child");
    assert!(!status.success(), "sidecar should exit after restart budget exhausted");

    let evidence_body = std::fs::read_to_string(&evidence).expect("read sidecar evidence file");
    let payload: Value = serde_json::from_str(&evidence_body).expect("parse sidecar evidence");
    assert_eq!(
        payload["api_version"].as_str(),
        Some("robotrt.gateway.sidecar.evidence.v1")
    );
    assert!(
        payload["restart_count"].as_u64().unwrap_or(0) >= 1,
        "evidence should record restart_count >= 1"
    );
    assert!(
        payload["health_checks"]
            .as_array()
            .map(|items| !items.is_empty())
            .unwrap_or(false),
        "evidence should include health checks"
    );
    assert!(
        payload["last_error"].as_str().is_some(),
        "evidence should include last_error"
    );

    let _ = std::fs::remove_file(evidence);
}
