use anyhow::{Result, bail};
use async_trait::async_trait;
use roentgen::{
    controller::{Caller, Controller, Operation},
    model::Trojan,
    runtime::{ProbeConnection, ProbeSession, Runtime},
    subscription::SubscriptionSource,
    xray::{ProbeListener, probe_configuration},
};
use std::{path::PathBuf, process::Stdio, time::Duration};
use tempfile::TempDir;

struct Source {
    port: u16,
}
#[async_trait]
impl SubscriptionSource for Source {
    async fn fetch(&self, _: &str) -> Result<Vec<u8>> {
        Ok(format!(
            "trojan://fixture-password@127.0.0.1:{}?sni=localhost#Fixture\n",
            self.port
        )
        .into_bytes())
    }
}

#[derive(Clone)]
enum ProbeMode {
    Available,
    Unavailable,
    Impostor,
}
struct FixtureRuntime {
    mode: std::sync::Arc<std::sync::Mutex<ProbeMode>>,
    impostor: String,
}
struct FixtureProbe {
    child: Option<tokio::process::Child>,
    _directory: TempDir,
    proxy: String,
    certificate: String,
}
#[async_trait]
impl ProbeSession for FixtureProbe {
    fn connection(&self) -> ProbeConnection<'_> {
        ProbeConnection {
            proxy_url: &self.proxy,
            certificate: &self.certificate,
        }
    }
    async fn stop(mut self: Box<Self>) {
        if let Some(child) = &mut self.child {
            let _ = child.kill().await;
            let _ = child.wait().await;
        }
    }
}
#[async_trait]
impl Runtime for FixtureRuntime {
    async fn arm(&mut self) -> Result<()> {
        bail!("Host activation is outside this diagnostic fixture")
    }
    async fn apply(&mut self, _: Option<&Trojan>, _: u64) -> Result<()> {
        bail!("No TUN in diagnostic fixture")
    }
    async fn disable(&mut self) -> Result<()> {
        Ok(())
    }
    fn routing(&self) -> &str {
        "direct"
    }
    async fn probe(&mut self, connection: &Trojan, _: u64) -> Result<Box<dyn ProbeSession>> {
        let mode = self.mode.lock().unwrap().clone();
        if matches!(mode, ProbeMode::Unavailable) {
            bail!("Fixture probe could not start");
        }
        let socket = std::net::TcpListener::bind("127.0.0.1:0")?;
        let port = socket.local_addr()?.port();
        let certificate = rcgen::generate_simple_self_signed(vec!["127.0.0.1".into()])?;
        let certificate_pem = certificate.cert.pem();
        let key = certificate.key_pair.serialize_pem();
        let directory = tempfile::tempdir()?;
        let path = directory.path().join("probe.json");
        if matches!(mode, ProbeMode::Impostor) {
            return Ok(Box::new(FixtureProbe {
                child: None,
                _directory: directory,
                proxy: self.impostor.clone(),
                certificate: certificate_pem,
            }));
        }
        let config = probe_configuration(
            connection,
            &ProbeListener {
                port,
                password: "private-probe-password",
                certificate: &certificate_pem,
                private_key: &key,
            },
        );
        std::fs::write(&path, serde_json::to_vec(&config)?)?;
        drop(socket);
        // The older distribution core is used only as a non-TUN fixture. Product
        // host activation retains its independent compatibility requirement.
        let child = tokio::process::Command::new("/usr/bin/xray")
            .args(["run", "-config"])
            .arg(path)
            .kill_on_drop(true)
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .spawn()?;
        let probe = FixtureProbe {
            child: Some(child),
            _directory: directory,
            proxy: format!("https://roentgen:private-probe-password@127.0.0.1:{port}"),
            certificate: certificate_pem,
        };
        for _ in 0..100 {
            if tokio::net::TcpStream::connect(("127.0.0.1", port))
                .await
                .is_ok()
            {
                return Ok(Box::new(probe));
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        bail!("Fixture Xray proxy did not start")
    }
}

#[derive(serde::Deserialize)]
struct Fixture {
    origin: String,
    bad_origin: String,
    trojan_port: u16,
    access_log: PathBuf,
    requests_log: PathBuf,
}

struct Context {
    _home: TempDir,
    controller: Controller,
    fixture: Fixture,
    id: String,
    mode: std::sync::Arc<std::sync::Mutex<ProbeMode>>,
}
impl Context {
    async fn open() -> Self {
        let hostname = std::fs::read_to_string("/etc/hostname").unwrap();
        assert_eq!(
            hostname.trim(),
            "altctlrs",
            "Run diagnostic acceptance only on the authorized VM"
        );
        let path = std::env::var("ROENTGEN_DIAGNOSTIC_FIXTURE")
            .expect("Use tests/vm_diagnostics_acceptance.py on the supplied VM");
        let fixture: Fixture = serde_json::from_slice(&std::fs::read(path).unwrap()).unwrap();
        let home = tempfile::tempdir().unwrap();
        let mode = std::sync::Arc::new(std::sync::Mutex::new(ProbeMode::Available));
        let mut impostor = url::Url::parse(&fixture.bad_origin).unwrap();
        impostor.set_username("roentgen").unwrap();
        impostor
            .set_password(Some("private-probe-password"))
            .unwrap();
        let runtime = FixtureRuntime {
            mode: mode.clone(),
            impostor: impostor.to_string(),
        };
        let mut controller = Controller::open(
            home.path(),
            Box::new(runtime),
            Box::new(Source {
                port: fixture.trojan_port,
            }),
        )
        .unwrap();
        let root = Caller::root();
        controller
            .execute(
                &root,
                Operation::AddSubscription {
                    name: "Fixture".into(),
                    url: "https://subscription.invalid/fixture".into(),
                },
            )
            .await
            .unwrap();
        let id = controller.configurations(None).unwrap()[0].id.clone();
        controller
            .execute(&root, Operation::Select { id: id.clone() })
            .await
            .unwrap();
        let mut result = Self {
            _home: home,
            controller,
            fixture,
            id,
            mode,
        };
        result.settings("/empty", "/large", 1, 1).await;
        result
    }
    async fn settings(&mut self, ping: &str, speed: &str, mib: u64, seconds: u64) {
        let mut settings = self.controller.settings();
        settings.ping_url = format!("{}{}", self.fixture.origin, ping);
        settings.speed_url = format!("{}{}", self.fixture.origin, speed);
        settings.maximum_download_mib = mib;
        settings.maximum_duration_seconds = seconds;
        self.controller
            .execute(&Caller::root(), Operation::SaveSettings { settings })
            .await
            .unwrap();
    }
    fn requests(&self) -> Vec<serde_json::Value> {
        std::fs::read_to_string(&self.fixture.requests_log)
            .unwrap()
            .lines()
            .map(|line| serde_json::from_str(line).unwrap())
            .collect()
    }
}

#[tokio::test]
async fn controller_ping_uses_the_selected_trojan_path_with_verified_tls_without_enabling() {
    let mut context = Context::open().await;
    let accepted_before = std::fs::read_to_string(&context.fixture.access_log)
        .unwrap_or_default()
        .lines()
        .filter(|line| line.contains("accepted"))
        .count();
    let observation = context
        .controller
        .diagnostic(&Caller::root(), None, false)
        .await
        .unwrap();
    assert!(observation.success);
    assert_eq!(observation.bytes, 0);
    assert_eq!(observation.stop_reason, "complete");
    assert_eq!(observation.configuration_id, context.id);
    assert!(!context.controller.status().unwrap().requested_enabled);
    assert_eq!(context.controller.status().unwrap().routing, "direct");
    let accepted_after = std::fs::read_to_string(&context.fixture.access_log)
        .unwrap()
        .lines()
        .filter(|line| line.contains("accepted"))
        .count();
    assert!(
        accepted_after > accepted_before,
        "The fixture Trojan endpoint must carry this diagnostic connection"
    );
    assert!(context.requests().iter().all(|request| {
        request["user_agent"]
            .as_str()
            .is_some_and(|value| value.starts_with("Roentgen/"))
    }));
    assert!(
        context
            .requests()
            .iter()
            .all(|request| request["proxy_authorization"].is_null()),
        "Local proxy credentials must not reach the target"
    );
}

#[tokio::test]
async fn controller_speed_stops_at_the_first_byte_or_duration_limit() {
    let mut context = Context::open().await;
    let bounded = context
        .controller
        .diagnostic(&Caller::root(), None, true)
        .await
        .unwrap();
    assert!(bounded.success);
    assert_eq!(bounded.bytes, 1_048_576);
    assert_eq!(bounded.stop_reason, "byte-limit");
    assert!(bounded.bytes_per_second.unwrap() > 0.0);
    context.settings("/empty", "/slow", 1, 1).await;
    let timed = context
        .controller
        .diagnostic(&Caller::root(), None, true)
        .await
        .unwrap();
    assert!(timed.success);
    assert_eq!(timed.bytes, 16384);
    assert_eq!(timed.stop_reason, "duration-limit");
    assert!(timed.elapsed_ms >= 900 && timed.elapsed_ms < 4000);
    assert!(!context.controller.status().unwrap().requested_enabled);
}

async fn private_bus(controller: Controller) -> (tokio::process::Child, zbus::Connection, String) {
    use tokio::io::{AsyncBufReadExt, BufReader};
    let mut bus = tokio::process::Command::new("/bin/dbus-daemon")
        .args(["--session", "--nofork", "--print-address=1"])
        .stdout(Stdio::piped())
        .stderr(Stdio::null())
        .kill_on_drop(true)
        .spawn()
        .unwrap();
    let mut address = String::new();
    BufReader::new(bus.stdout.take().unwrap())
        .read_line(&mut address)
        .await
        .unwrap();
    let address = address.trim().to_owned();
    let connection = zbus::connection::Builder::address(address.as_str())
        .unwrap()
        .name(roentgen::bus::NAME)
        .unwrap()
        .serve_at(
            roentgen::bus::PATH,
            roentgen::bus::Service {
                controller: std::sync::Arc::new(tokio::sync::Mutex::new(controller)),
                cancellation: std::sync::Arc::new(tokio::sync::Notify::new()),
            },
        )
        .unwrap()
        .build()
        .await
        .unwrap();
    (bus, connection, address)
}

#[tokio::test]
async fn cli_exits_unsuccessfully_when_the_provider_rate_limits_a_diagnostic() {
    let mut context = Context::open().await;
    context.settings("/rate", "/large", 1, 1).await;
    let (_bus, _service, address) = private_bus(context.controller).await;
    let binary = std::env::var("ROENTGEN_TEST_BINARY")
        .expect("Set ROENTGEN_TEST_BINARY to the VM candidate executable");
    let output = tokio::process::Command::new(binary)
        .arg("ping")
        .env("DBUS_SYSTEM_BUS_ADDRESS", address)
        .stdin(Stdio::null())
        .output()
        .await
        .unwrap();
    assert!(
        !output.status.success(),
        "A failed diagnostic must not produce a successful CLI exit: {}",
        String::from_utf8_lossy(&output.stdout)
    );
    let error = String::from_utf8_lossy(&output.stderr);
    assert!(
        error.contains("provider-rate-limit") && error.contains("60"),
        "The failure should report the provider delay: {error}"
    );
}

#[tokio::test]
async fn controller_records_a_failed_check_when_a_previously_working_probe_cannot_start() {
    let mut context = Context::open().await;
    context
        .controller
        .diagnostic(&Caller::root(), None, false)
        .await
        .unwrap();
    assert_eq!(context.controller.status().unwrap().health, "reachable");
    *context.mode.lock().unwrap() = ProbeMode::Unavailable;
    assert!(
        context
            .controller
            .diagnostic(&Caller::root(), None, false)
            .await
            .is_err()
    );
    assert_eq!(context.controller.status().unwrap().health, "check-failed");
    assert!(
        context.controller.configurations(None).unwrap()[0]
            .ping_ms
            .is_none()
    );
    assert!(!context.controller.status().unwrap().requested_enabled);
}

#[tokio::test]
async fn controller_rejects_redirects_oversized_ping_and_error_downloads() {
    let mut context = Context::open().await;
    context
        .controller
        .diagnostic(&Caller::root(), None, false)
        .await
        .unwrap();
    for (path, speed) in [
        ("/html", true),
        ("/partial", true),
        ("/empty", true),
        ("/oversize", false),
        ("/redirect", false),
        ("/error", true),
        ("/json-error", true),
    ] {
        context.settings(path, path, 1, 1).await;
        let before = context.requests();
        assert!(
            context
                .controller
                .diagnostic(&Caller::root(), None, speed)
                .await
                .is_err(),
            "Diagnostic unexpectedly accepted {path}"
        );
        let after = context.requests();
        assert!(
            after[before.len()..]
                .iter()
                .any(|request| request["path"] == path),
            "The response fixture was not reached for {path}"
        );
        if path == "/redirect" {
            assert!(
                !after[before.len()..]
                    .iter()
                    .any(|request| request["path"] == "/empty"),
                "Redirects must not be followed"
            );
        }
        assert!(!context.controller.status().unwrap().requested_enabled);
    }
}

#[tokio::test]
async fn controller_rejects_an_untrusted_target_or_local_proxy_without_direct_fallback() {
    let mut context = Context::open().await;
    context
        .controller
        .diagnostic(&Caller::root(), None, false)
        .await
        .unwrap();
    let before = context.requests().len();
    let mut settings = context.controller.settings();
    settings.ping_url = format!("{}/empty", context.fixture.bad_origin);
    context
        .controller
        .execute(&Caller::root(), Operation::SaveSettings { settings })
        .await
        .unwrap();
    assert!(
        context
            .controller
            .diagnostic(&Caller::root(), None, false)
            .await
            .is_err()
    );
    assert_eq!(
        context.requests().len(),
        before,
        "An untrusted target must not receive an HTTP request"
    );
    context.settings("/empty", "/large", 1, 1).await;
    *context.mode.lock().unwrap() = ProbeMode::Impostor;
    assert!(
        context
            .controller
            .diagnostic(&Caller::root(), None, false)
            .await
            .is_err()
    );
    assert_eq!(
        context.requests().len(),
        before,
        "An impostor proxy must not receive credentials or cause direct target fallback"
    );
    assert!(!context.controller.status().unwrap().requested_enabled);
}

#[tokio::test]
async fn provider_retry_after_blocks_manual_ping_and_speed_for_both_header_forms() {
    for path in ["/rate", "/rate-date"] {
        let mut context = Context::open().await;
        context.settings(path, "/large", 1, 1).await;
        let observation = context
            .controller
            .diagnostic(&Caller::root(), None, false)
            .await
            .unwrap();
        assert!(!observation.success);
        assert_eq!(observation.stop_reason, "provider-rate-limit");
        assert!(observation.retry_after_seconds.unwrap() >= 60);
        let before = context.requests().len();
        for speed in [false, true] {
            let error = context
                .controller
                .diagnostic(&Caller::root(), None, speed)
                .await
                .unwrap_err();
            assert!(error.to_string().contains("Retry-After"));
        }
        assert_eq!(
            context.requests().len(),
            before,
            "Cooldown must prevent all further provider requests"
        );
    }
}

#[tokio::test]
async fn scheduled_checks_use_only_selection_never_run_speed_and_do_not_overlap() {
    let mut context = Context::open().await;
    context
        .controller
        .execute(
            &Caller::root(),
            Operation::AddSubscription {
                name: "Other".into(),
                url: "https://subscription.invalid/other".into(),
            },
        )
        .await
        .unwrap();
    let before = context.requests().len();
    context.controller.scheduled_check().await.unwrap();
    context.controller.scheduled_check().await.unwrap();
    let after = context.requests();
    assert_eq!(after.len(), before + 1, "Only one selected Ping is due");
    assert_eq!(after.last().unwrap()["path"], "/empty");
    let entries = context.controller.configurations(None).unwrap();
    assert!(
        entries
            .iter()
            .find(|entry| entry.id == context.id)
            .unwrap()
            .ping_ms
            .is_some()
    );
    assert!(
        entries
            .iter()
            .filter(|entry| entry.id != context.id)
            .all(|entry| entry.ping_ms.is_none())
    );
    assert!(
        entries
            .iter()
            .all(|entry| entry.speed_bytes_per_second.is_none())
    );
    assert!(!context.controller.status().unwrap().requested_enabled);
}

#[tokio::test]
async fn cancelling_a_bus_diagnostic_stops_it_without_changing_proxy_intent() {
    let mut context = Context::open().await;
    context.settings("/empty", "/slow", 1, 10).await;
    let before = context.requests().len();
    let requests = context.fixture.requests_log.clone();
    let (_bus, _service, address) = private_bus(context.controller).await;
    let connection = zbus::connection::Builder::address(address.as_str())
        .unwrap()
        .build()
        .await
        .unwrap();
    let request_connection = connection.clone();
    let request = tokio::spawn(async move {
        let proxy = zbus::Proxy::new(
            &request_connection,
            roentgen::bus::NAME,
            roentgen::bus::PATH,
            roentgen::bus::INTERFACE,
        )
        .await
        .unwrap();
        proxy
            .call::<_, _, String>("Diagnostic", &(String::new(), true, false))
            .await
    });
    let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
    loop {
        let count = std::fs::read_to_string(&requests).unwrap().lines().count();
        if count > before {
            break;
        }
        assert!(
            tokio::time::Instant::now() < deadline,
            "Diagnostic did not reach its fixture"
        );
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
    let proxy = zbus::Proxy::new(
        &connection,
        roentgen::bus::NAME,
        roentgen::bus::PATH,
        roentgen::bus::INTERFACE,
    )
    .await
    .unwrap();
    let _: () = proxy.call("CancelDiagnostic", &(false,)).await.unwrap();
    let result = tokio::time::timeout(Duration::from_secs(2), request)
        .await
        .unwrap()
        .unwrap();
    assert!(result.unwrap_err().to_string().contains("cancelled"));
    let status: String = proxy.call("Status", &()).await.unwrap();
    let status: roentgen::model::Status = serde_json::from_str(&status).unwrap();
    assert!(!status.requested_enabled);
    assert_eq!(status.selected_configuration, Some(context.id));
    let rows: String = proxy
        .call("Configurations", &(String::new(),))
        .await
        .unwrap();
    let rows: Vec<roentgen::model::ConfigurationSummary> = serde_json::from_str(&rows).unwrap();
    assert!(
        rows.iter()
            .all(|entry| entry.speed_bytes_per_second.is_none())
    );
}