use std::fs;
use std::io::{Read, Write};
use std::net::{TcpListener, TcpStream};
use std::path::{Path, PathBuf};
use std::process::Command;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::thread;
use std::time::{Duration, Instant};
use thirtyfour::prelude::*;

const WEBDRIVER_ADDR: &str = "127.0.0.1:4451";
const STARTUP_TIMEOUT: Duration = Duration::from_secs(12);

#[tokio::test]
async fn client_handler_applies_effect_batch_without_network() -> WebDriverResult<()> {
    // req: client_local/005 req: client_local/009 req: client_local/010
    // req: client_local/011 req: client_local/012
    // test: client_local/014
    let workspace = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
        .parent()
        .expect("workspace root")
        .to_owned();
    let (package, bootstrap, rendered) = build_browser_artifact(&workspace);
    let runtime = workspace.join("hemx-js/runtime/hemx.js");
    let server = StaticServer::start(package, runtime, bootstrap, rendered);

    let mut webdriver = Command::new("geckodriver");
    webdriver.arg("--port").arg("4451");
    let _webdriver = ProcessGuard::start(webdriver, WEBDRIVER_ADDR);

    let mut caps = DesiredCapabilities::firefox();
    caps.set_headless()?;
    let driver = WebDriver::new(&format!("http://{WEBDRIVER_ADDR}"), caps).await?;
    let result = async {
        driver.goto(&server.url()).await?;
        wait_until(
            &driver,
            "return document.querySelector('[data-hemx-root]').hasAttribute('data-hemx-client-ready')",
        )
        .await?;
        driver
            .execute(
                "window.__clientErrors = []; document.querySelector('[data-hemx-root]').addEventListener('hemx:client-error', (event) => window.__clientErrors.push(event.detail)); return true",
                Vec::new(),
            )
            .await?;
        let network_before = resource_count(&driver).await?;
        driver
            .execute(
                r#"
                window.__resolveClientRuns = [];
                const actual = window.hemx.registerClientHandler('increment', (...args) => new Promise((resolve) => {
                  window.__resolveClientRuns.push(() => resolve(actual(...args)));
                }));
                window.__actualClientHandler = actual;
                return true;
                "#,
                Vec::new(),
            )
            .await?;
        let button = driver
            .find(By::Css("[data-hemx-client='increment']"))
            .await?;
        button.click().await?;
        button.click().await?;
        wait_until(&driver, "return window.__resolveClientRuns.length === 2").await?;
        driver
            .execute(
                "window.__resolveClientRuns[0](); return true",
                Vec::new(),
            )
            .await?;
        tokio::time::sleep(Duration::from_millis(100)).await;
        assert_eq!(
            driver
                .find(By::Css("[data-sid]"))
                .await?
                .prop("textContent")
                .await?
                .unwrap_or_default(),
            "idle",
            "superseded completion applied stale effects"
        );
        driver
            .execute(
                "window.__resolveClientRuns[1](); return true",
                Vec::new(),
            )
            .await?;
        wait_until(
            &driver,
            "return document.querySelector('[data-sid]').textContent.includes('updated by Rust/WASM (click, count=3)')",
        )
        .await?;

        assert_eq!(
            resource_count(&driver).await?,
            network_before,
            "client handler made a network request"
        );

        driver
            .execute(
                "window.hemx.registerClientHandler('increment', window.__actualClientHandler); return true",
                Vec::new(),
            )
            .await?;
        driver
            .execute(
                "document.querySelector('[data-hemx-root]').setAttribute('data-hemx-client-state-version', '2'); return true",
                Vec::new(),
            )
            .await?;
        driver
            .find(By::Css("[data-hemx-client='increment']"))
            .await?
            .click()
            .await?;
        wait_until(&driver, "return window.__clientErrors.length === 1").await?;
        assert!(
            driver
                .execute("return window.__clientErrors[0].message", Vec::new())
                .await?
                .json()
                .as_str()
                .unwrap_or_default()
                .contains("unsupported client-local state ABI version 2; expected 1"),
            "invalid state must produce an actionable client-local diagnostic"
        );
        assert_eq!(
            resource_count(&driver).await?,
            network_before + 1,
            "declared server fallback was not requested"
        );
        assert!(
            driver
                .execute(
                    "return !document.querySelector('[data-hemx-client]').classList.contains('is-pending')",
                    Vec::new(),
                )
                .await?
                .json()
                .as_bool()
                .unwrap_or(false),
            "invalid input must restore pending UI"
        );

        driver
            .execute(
                r#"
                window.__resolveUnmount = null;
                const actual = window.hemx.registerClientHandler('increment', (...args) => new Promise((resolve) => {
                  window.__resolveUnmount = () => resolve(actual(...args));
                }));
                document.querySelector('[data-hemx-root]').setAttribute('data-hemx-client-state-version', '1');
                return true;
                "#,
                Vec::new(),
            )
            .await?;
        driver
            .find(By::Css("[data-hemx-client='increment']"))
            .await?
            .click()
            .await?;
        driver
            .execute(
                "const root = document.querySelector('[data-hemx-root]'); window.__removedRoot = root; root.remove(); window.__resolveUnmount(); return true",
                Vec::new(),
            )
            .await?;
        tokio::time::sleep(Duration::from_millis(100)).await;
        assert!(
            driver
                .execute(
                    "return !window.__removedRoot.textContent.includes('updated by Rust/WASM')",
                    Vec::new(),
                )
                .await?
                .json()
                .as_bool()
                .unwrap_or(false),
            "unmounted root accepted a late effect"
        );
        Ok::<(), WebDriverError>(())
    }
    .await;
    let quit = driver.quit().await;
    result.and(quit)
}

fn build_browser_artifact(workspace: &Path) -> (PathBuf, PathBuf, String) {
    let status = Command::new("cargo")
        .current_dir(workspace)
        .args([
            "build",
            "-p",
            "hemx-client-local-example",
            "--target",
            "wasm32-unknown-unknown",
        ])
        .status()
        .expect("run wasm cargo build");
    assert!(status.success(), "WASM build failed");

    let output = workspace.join("target/client-local-bindgen");
    fs::create_dir_all(&output).expect("create wasm-bindgen output");
    let status = Command::new("wasm-bindgen")
        .current_dir(workspace)
        .arg("--target")
        .arg("web")
        .arg("--out-name")
        .arg("client_local")
        .arg("--out-dir")
        .arg(&output)
        .arg(workspace.join("target/wasm32-unknown-unknown/debug/hemx_client_local_example.wasm"))
        .status()
        .expect("run wasm-bindgen");
    assert!(status.success(), "wasm-bindgen failed");

    let rendered = Command::new("cargo")
        .current_dir(workspace)
        .args([
            "run",
            "-q",
            "-p",
            "hemx-client-local-example",
            "--features",
            "fixture",
            "--bin",
            "fixture",
        ])
        .output()
        .expect("render generated client fixture");
    assert!(rendered.status.success(), "generated fixture render failed");
    let bootstrap =
        newest_generated_bootstrap(&workspace.join("target/wasm32-unknown-unknown/debug/build"));
    (
        output,
        bootstrap,
        String::from_utf8(rendered.stdout).expect("fixture is UTF-8"),
    )
}

fn newest_generated_bootstrap(build_dir: &Path) -> PathBuf {
    fs::read_dir(build_dir)
        .expect("read wasm build directory")
        .filter_map(Result::ok)
        .filter(|entry| {
            entry
                .file_name()
                .to_string_lossy()
                .starts_with("hemx-client-local-example-")
        })
        .map(|entry| entry.path().join("out/hemx.client.js"))
        .filter(|path| path.is_file())
        .max_by_key(|path| path.metadata().and_then(|meta| meta.modified()).ok())
        .expect("generated hemx client bootstrap")
}

struct StaticServer {
    address: String,
    stop: Arc<AtomicBool>,
}

impl StaticServer {
    fn start(package: PathBuf, runtime: PathBuf, bootstrap: PathBuf, rendered: String) -> Self {
        let listener = TcpListener::bind("127.0.0.1:0").expect("bind browser fixture");
        listener.set_nonblocking(true).expect("nonblocking fixture");
        let address = listener.local_addr().expect("fixture address").to_string();
        let stop = Arc::new(AtomicBool::new(false));
        let thread_stop = Arc::clone(&stop);
        thread::spawn(move || {
            while !thread_stop.load(Ordering::Relaxed) {
                match listener.accept() {
                    Ok((stream, _)) => serve(stream, &package, &runtime, &bootstrap, &rendered),
                    Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
                        thread::sleep(Duration::from_millis(10))
                    }
                    Err(error) => panic!("fixture accept failed: {error}"),
                }
            }
        });
        Self { address, stop }
    }

    fn url(&self) -> String {
        format!("http://{}", self.address)
    }
}

impl Drop for StaticServer {
    fn drop(&mut self) {
        self.stop.store(true, Ordering::Relaxed);
    }
}

fn serve(mut stream: TcpStream, package: &Path, runtime: &Path, bootstrap: &Path, rendered: &str) {
    let mut request = [0_u8; 2048];
    let length = stream.read(&mut request).unwrap_or(0);
    let first = String::from_utf8_lossy(&request[..length]);
    let path = first.split_whitespace().nth(1).unwrap_or("/");
    let (content_type, body) = match path {
        "/" => (
            "text/html; charset=utf-8",
            fixture_html(rendered).into_bytes(),
        ),
        "/hemx.js" => (
            "text/javascript; charset=utf-8",
            fs::read(runtime).expect("read runtime"),
        ),
        "/client_local.js" => (
            "text/javascript; charset=utf-8",
            fs::read(package.join("client_local.js")).expect("read bindings"),
        ),
        "/client_local_bg.wasm" => (
            "application/wasm",
            fs::read(package.join("client_local_bg.wasm")).expect("read wasm"),
        ),
        "/hemx.client.js" => (
            "text/javascript; charset=utf-8",
            fs::read(bootstrap).expect("read generated client bootstrap"),
        ),
        _ => ("text/plain", b"not found".to_vec()),
    };
    let status = if path == "/"
        || path == "/hemx.js"
        || path == "/hemx.client.js"
        || path.starts_with("/client_local")
    {
        "200 OK"
    } else {
        "404 Not Found"
    };
    write!(stream, "HTTP/1.1 {status}\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", body.len()).expect("write fixture headers");
    stream.write_all(&body).expect("write fixture body");
}

fn fixture_html(rendered: &str) -> String {
    format!(
        "<!doctype html><html><body>{rendered}<script src=\"/hemx.js\"></script><script type=\"module\" src=\"/hemx.client.js\"></script></body></html>"
    )
}

async fn wait_until(driver: &WebDriver, script: &str) -> WebDriverResult<()> {
    let deadline = Instant::now() + STARTUP_TIMEOUT;
    loop {
        if driver
            .execute(script, Vec::new())
            .await?
            .json()
            .as_bool()
            .unwrap_or(false)
        {
            return Ok(());
        }
        if Instant::now() >= deadline {
            panic!("timed out waiting for browser fixture");
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
}

async fn resource_count(driver: &WebDriver) -> WebDriverResult<u64> {
    Ok(driver
        .execute(
            "return performance.getEntriesByType('resource').length",
            Vec::new(),
        )
        .await?
        .json()
        .as_u64()
        .unwrap_or_default())
}

struct ProcessGuard(std::process::Child);

impl ProcessGuard {
    fn start(mut command: Command, address: &str) -> Self {
        let child = command.spawn().expect("start geckodriver");
        let deadline = Instant::now() + STARTUP_TIMEOUT;
        while std::net::TcpStream::connect(address).is_err() {
            assert!(Instant::now() < deadline, "timed out starting geckodriver");
            thread::sleep(Duration::from_millis(50));
        }
        Self(child)
    }
}

impl Drop for ProcessGuard {
    fn drop(&mut self) {
        let _ = self.0.kill();
        let _ = self.0.wait();
    }
}