#![allow(dead_code)]
// Shared across test targets; each uses only part of this module.
use std::collections::{BTreeMap, BTreeSet};
use std::io::{Read, Write};
use std::net::TcpListener;
use std::path::Path;
use std::process::{Command, Output};
use std::sync::{
    Arc, Mutex,
    atomic::{AtomicBool, AtomicUsize, Ordering},
};
use std::thread::JoinHandle;
use std::time::Duration;

use avril_runtime::registry::{Options, PackageStore, checksum, npm::resolution::NpmResolver};
use serde_json::json;
use url::Url;

/// The directory's `avril.lock` as data: locks are fJSON, so plain JSON parsing misses
/// implicit root braces and other fJSON spellings.
pub fn read_lock(directory: &std::path::Path) -> serde_json::Value {
    let path = directory.join("avril.lock");
    let source = std::fs::read_to_string(&path).unwrap();
    serde_fjson::parse(&source).unwrap()
}

// Downloads must remain driven while tests wait outside Tokio.
pub fn runtime_handle() -> tokio::runtime::Handle {
    static RUNTIME: std::sync::OnceLock<tokio::runtime::Runtime> = std::sync::OnceLock::new();
    RUNTIME
        .get_or_init(|| {
            tokio::runtime::Builder::new_multi_thread()
                .worker_threads(2)
                .enable_all()
                .build()
                .unwrap()
        })
        .handle()
        .clone()
}

pub fn manifest(name: &str, version: &str, extra: serde_json::Value) -> String {
    let mut package = json!({"name": name, "version": version, "type": "module"});
    package
        .as_object_mut()
        .unwrap()
        .extend(extra.as_object().unwrap().clone());
    package.to_string()
}

pub fn resolver(options: Options) -> NpmResolver {
    NpmResolver::new(Arc::new(
        PackageStore::new(runtime_handle(), options).unwrap(),
    ))
}

pub fn files(dir: &Path, entries: &[(&str, &str)]) {
    for (name, source) in entries {
        write(dir, name, source);
    }
}

pub fn write(dir: &Path, name: &str, source: &str) {
    let path = dir.join(name);
    std::fs::create_dir_all(path.parent().unwrap()).unwrap();
    std::fs::write(path, source).unwrap();
}

pub fn success(output: &Output) -> String {
    assert!(
        output.status.success(),
        "{}",
        String::from_utf8_lossy(&output.stderr)
    );
    String::from_utf8(output.stdout.clone()).unwrap()
}

pub fn failure(output: &Output) -> String {
    assert!(
        !output.status.success(),
        "{}",
        String::from_utf8_lossy(&output.stdout)
    );
    String::from_utf8_lossy(&output.stderr).into_owned()
}

struct RegistryState {
    files: Arc<Mutex<BTreeMap<String, Vec<u8>>>>,
    abbreviated: Arc<Mutex<BTreeMap<String, Vec<u8>>>>,
    requests: Arc<Mutex<Vec<String>>>,
    responses: Arc<Mutex<Vec<String>>>,
    stop: Arc<AtomicBool>,
    hold_tarballs: Arc<AtomicBool>,
    hold_paths: Arc<Mutex<BTreeSet<String>>>,
    active: Arc<AtomicUsize>,
    max: Arc<AtomicUsize>,
}

struct ActiveGuard {
    state: Arc<RegistryState>,
}

impl Drop for ActiveGuard {
    fn drop(&mut self) {
        self.state.active.fetch_sub(1, Ordering::SeqCst);
    }
}

pub struct Registry {
    pub url: Url,
    pub files: Arc<Mutex<BTreeMap<String, Vec<u8>>>>,
    pub requests: Arc<Mutex<Vec<String>>>,
    /// When set, tarball responses stall until cleared, so guest code can
    /// revoke authority mid-acquisition deterministically.
    pub hold_tarballs: Arc<AtomicBool>,
    /// Opt-in concurrent serving for overlap/bound tests; default off keeps
    /// the original serial accept loop.
    pub concurrent: Arc<AtomicBool>,
    /// Tarball paths to stall while present; stall loops also watch stop.
    pub hold_paths: Arc<Mutex<BTreeSet<String>>>,
    pub active_tarballs: Arc<AtomicUsize>,
    pub max_tarballs: Arc<AtomicUsize>,
    pub responses: Arc<Mutex<Vec<String>>>,
    abbreviated: Arc<Mutex<BTreeMap<String, Vec<u8>>>>,
    stop: Arc<AtomicBool>,
    worker: Option<JoinHandle<()>>,
    handlers: Arc<Mutex<Vec<JoinHandle<()>>>>,
}

impl Registry {
    pub fn new() -> Self {
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let url = Url::parse(&format!("http://{}/", listener.local_addr().unwrap())).unwrap();
        listener.set_nonblocking(true).unwrap();
        let files = Arc::new(Mutex::new(BTreeMap::<String, Vec<u8>>::new()));
        let requests = Arc::new(Mutex::new(Vec::new()));
        let stop = Arc::new(AtomicBool::new(false));
        let hold_tarballs = Arc::new(AtomicBool::new(false));
        let concurrent = Arc::new(AtomicBool::new(false));
        let hold_paths = Arc::new(Mutex::new(BTreeSet::<String>::new()));
        let active = Arc::new(AtomicUsize::new(0));
        let max = Arc::new(AtomicUsize::new(0));
        let responses = Arc::new(Mutex::new(Vec::<String>::new()));
        let abbreviated = Arc::new(Mutex::new(BTreeMap::<String, Vec<u8>>::new()));
        let handlers = Arc::new(Mutex::new(Vec::<JoinHandle<()>>::new()));
        let state = Arc::new(RegistryState {
            files: files.clone(),
            abbreviated: abbreviated.clone(),
            requests: requests.clone(),
            responses: responses.clone(),
            stop: stop.clone(),
            hold_tarballs: hold_tarballs.clone(),
            hold_paths: hold_paths.clone(),
            active: active.clone(),
            max: max.clone(),
        });
        let accept_concurrent = concurrent.clone();
        let accept_handlers = handlers.clone();
        let worker = std::thread::spawn(move || {
            while !state.stop.load(Ordering::Relaxed) {
                let stream = match listener.accept() {
                    Ok((stream, _)) => stream,
                    Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
                        std::thread::sleep(Duration::from_millis(1));
                        continue;
                    }
                    Err(error) => panic!("{error}"),
                };
                if accept_concurrent.load(Ordering::Relaxed) {
                    let state = state.clone();
                    accept_handlers
                        .lock()
                        .unwrap()
                        .push(std::thread::spawn(move || {
                            serve_stream(state, stream);
                        }));
                    continue;
                }
                serve_stream(state.clone(), stream);
            }
        });
        Self {
            url,
            files,
            abbreviated,
            requests,
            hold_tarballs,
            concurrent,
            hold_paths,
            active_tarballs: active,
            max_tarballs: max,
            responses,
            stop,
            worker: Some(worker),
            handlers,
        }
    }

    pub fn package(
        &self,
        name: &str,
        version: &str,
        files: &[(&str, &str)],
        exports: &[(&str, &str)],
    ) {
        let mut artifacts = self.files.lock().unwrap();
        let index_path = format!("/{name}/meta.json");
        let mut index = artifacts
            .get(&index_path)
            .map(|bytes| serde_json::from_slice::<serde_json::Value>(bytes).unwrap())
            .unwrap_or_else(|| json!({"versions": {}}));
        index["versions"][version] = json!({});
        artifacts.insert(index_path, serde_json::to_vec(&index).unwrap());
        let manifest: BTreeMap<_, _> = files
            .iter()
            .map(|(path, source)| {
                artifacts.insert(
                    format!("/{name}/{version}/{path}"),
                    source.as_bytes().to_vec(),
                );
                (
                    format!("/{path}"),
                    json!({"size": source.len(), "checksum": checksum(source.as_bytes())}),
                )
            })
            .collect();
        let exports: BTreeMap<_, _> = exports.iter().copied().collect();
        artifacts.insert(
            format!("/{name}/{version}_meta.json"),
            serde_json::to_vec(
                &json!({"manifest": manifest, "exports": exports, "moduleGraph2": {}}),
            )
            .unwrap(),
        );
    }

    pub fn tarball_bytes(files: &[(&str, &str)]) -> Vec<u8> {
        use flate2::{Compression, write::GzEncoder};
        let mut tar_data = Vec::new();
        {
            let mut tar = tar::Builder::new(&mut tar_data);
            for (path, source) in files {
                let full = format!("package/{path}");
                let mut header = tar::Header::new_gnu();
                header.set_mode(0o644);
                header.set_size(source.len() as u64);
                header.set_cksum();
                tar.append_data(&mut header, &full, source.as_bytes())
                    .unwrap();
            }
            tar.finish().unwrap();
        }
        let mut encoder = GzEncoder::new(Vec::new(), Compression::default());
        use std::io::Write;
        encoder.write_all(&tar_data).unwrap();
        encoder.finish().unwrap()
    }

    pub fn tarball_integrity(bytes: &[u8]) -> String {
        use base64::Engine;
        use sha2::Digest;
        let digest = sha2::Sha512::digest(bytes);
        format!(
            "sha512-{}",
            base64::engine::general_purpose::STANDARD.encode(digest)
        )
    }

    fn tarball_path(name: &str, version: &str) -> String {
        let leaf = name.rsplit('/').next().unwrap();
        format!("/{name}/-/{leaf}-{version}.tgz")
    }

    pub fn publish_package(
        &self,
        name: &str,
        version: &str,
        extra: serde_json::Value,
        sources: &[(&str, &str)],
    ) {
        let package = manifest(name, version, extra);
        let mut files = vec![("package.json", package.as_str())];
        if !sources.iter().any(|(path, _)| path.starts_with("index.")) {
            files.push(("index.js", "export const value = 1;\nexport default 1;"));
        }
        files.extend_from_slice(sources);
        self.publish_npm(name, version, &files);
    }

    pub fn resolver(&self, dir: &Path) -> NpmResolver {
        resolver(self.options(dir))
    }

    pub fn frozen_resolver(&self, dir: &Path) -> NpmResolver {
        let mut options = self.options(dir);
        options.offline = true;
        options.frozen = true;
        resolver(options)
    }

    pub fn command(&self, dir: &Path, subcommand: &str) -> Command {
        // This fixture also builds in CLI unit tests, which never launch commands.
        let Some(binary) = option_env!("CARGO_BIN_EXE_avril") else {
            panic!("CLI commands require an integration-test target");
        };
        let mut command = Command::new(binary);
        command
            .current_dir(dir)
            .arg(subcommand)
            .arg(format!("--jsr-registry={}", self.url))
            .arg(format!("--npm-registry={}", self.url))
            .arg(format!("--cache-dir={}", dir.join("cache").display()));
        command
    }

    pub fn run_npm(&self, dir: &Path, args: &[&str], entry: &str) -> Output {
        let mut command = self.command(dir, args[0]);
        command.args(&args[1..]);
        // `add` and friends take no entry module.
        if !entry.is_empty() {
            command.arg(entry);
        }
        command.output().unwrap()
    }

    pub fn publish_npm(&self, name: &str, version: &str, files: &[(&str, &str)]) {
        let tarball = Self::tarball_bytes(files);
        let integrity = Self::tarball_integrity(&tarball);
        self.publish_npm_dist(name, version, tarball, &integrity, None, true);
    }

    pub fn publish_npm_dist(
        &self,
        name: &str,
        version: &str,
        tarball: Vec<u8>,
        integrity: &str,
        shasum: Option<&str>,
        latest: bool,
    ) {
        let mut artifacts = self.files.lock().unwrap();
        let tarball_path = Self::tarball_path(name, version);
        let tarball_url = format!("{}{tarball_path}", self.url.as_str().trim_end_matches('/'));
        artifacts.insert(tarball_path, tarball);
        let packument_path = format!("/{name}");
        let mut packument = artifacts
            .get(&packument_path)
            .map(|bytes| serde_json::from_slice::<serde_json::Value>(bytes).unwrap())
            .unwrap_or_else(|| json!({"name": name, "dist-tags": {}, "versions": {}}));
        packument["versions"][version] = json!({
            "name": name,
            "version": version,
            "dist": {
                "tarball": tarball_url,
                "integrity": integrity,
            },
        });
        if let Some(shasum) = shasum {
            packument["versions"][version]["dist"]["shasum"] = json!(shasum);
        }
        if latest {
            packument["dist-tags"]["latest"] = json!(version);
        }
        artifacts.insert(packument_path, serde_json::to_vec(&packument).unwrap());
    }

    /// Abbreviated metadata for Accept-negotiating clients; others get the ordinary response.
    pub fn publish_npm_abbreviated(&self, name: &str, body: Vec<u8>) {
        self.abbreviated
            .lock()
            .unwrap()
            .insert(format!("/{name}"), body);
    }

    pub fn options(&self, directory: &std::path::Path) -> Options {
        Options {
            jsr_registry: self.url.clone(),
            npm_registry: self.url.clone(),
            cache: directory.join("cache"),
            lock: directory.join("avril.lock"),
            project_root: directory.to_owned(),
            offline: false,
            frozen: false,
            update: false,
            ephemeral_lock: false,
            vendor: None,
        }
    }
}

fn serve_stream(state: Arc<RegistryState>, mut stream: std::net::TcpStream) {
    stream
        .set_read_timeout(Some(Duration::from_secs(3)))
        .unwrap();
    let mut request = Vec::new();
    let mut buf = [0; 1024];
    while !request.windows(4).any(|bytes| bytes == b"\r\n\r\n") {
        let length = match stream.read(&mut buf) {
            Ok(0) | Err(_) => break,
            Ok(length) => length,
        };
        request.extend_from_slice(&buf[..length]);
    }
    let request = String::from_utf8_lossy(&request);
    let Some(path) = request
        .lines()
        .next()
        .and_then(|line| line.split_whitespace().nth(1))
    else {
        return;
    };
    assert!(!request.to_ascii_lowercase().contains("accept: text/html"));
    let abbreviated = request.lines().any(|line| {
        line.to_ascii_lowercase().starts_with("accept:")
            && line.contains("application/vnd.npm.install-v1+json")
    });
    let _guard = path.ends_with(".tgz").then(|| {
        let active = state.active.fetch_add(1, Ordering::SeqCst) + 1;
        state.max.fetch_max(active, Ordering::SeqCst);
        ActiveGuard { state: state.clone() }
    });
    state.requests.lock().unwrap().push(path.to_owned());
    while !state.stop.load(Ordering::Relaxed)
        && ((path.ends_with(".tgz") && state.hold_tarballs.load(Ordering::Relaxed))
            || state.hold_paths.lock().unwrap().contains(path))
    {
        std::thread::sleep(Duration::from_millis(10));
    }
    if state.stop.load(Ordering::Relaxed) {
        return;
    }
    let mut bytes = None;
    if abbreviated {
        bytes = state.abbreviated.lock().unwrap().get(path).cloned();
    }
    let bytes = bytes.or_else(|| state.files.lock().unwrap().get(path).cloned());
    let (status, bytes) = match bytes {
        Some(bytes) => ("200 OK", bytes),
        None => ("404 Not Found", b"not found".to_vec()),
    };
    let headers = format!(
        "HTTP/1.1 {status}\r\nContent-Type: application/octet-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
        bytes.len()
    );
    stream
        .set_write_timeout(Some(Duration::from_secs(3)))
        .unwrap();
    let _ = stream
        .write_all(headers.as_bytes())
        .and_then(|_| stream.write_all(&bytes));
    state.responses.lock().unwrap().push(path.to_owned());
}

impl Drop for Registry {
    fn drop(&mut self) {
        self.stop.store(true, Ordering::Relaxed);
        self.worker.take().unwrap().join().unwrap();
        for handler in self.handlers.lock().unwrap().drain(..) {
            if std::thread::panicking() {
                let _ = handler.join();
            } else {
                handler.join().unwrap();
            }
        }
    }
}