mod file;

use std::io;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};

use avril_caps::{Capabilities, PathTarget};
use avril_core::serde_v8::{JsBuffer, ToJsBuffer};
use avril_core::{Extension, Op};
use avril_http::{Address, Authorize, Engine};
use bytes::Bytes;

use avril_ext_websocket::Sockets;

type RespondArgs = (u64, u16, Vec<(String, String)>, Option<JsBuffer>, bool);

pub fn init(
    runtime: tokio::runtime::Handle,
    caps: Arc<Capabilities>,
    sockets: Arc<Sockets>,
    files: Arc<avril_ext_file::Files>,
    cli: Option<Address>,
) -> Extension {
    let engine = Arc::new(Engine::new(runtime.clone()));
    let file_ops = file::ops(runtime.clone(), caps.clone(), files, engine.clone());
    let cli = Mutex::new(cli);
    let listen = engine.clone();
    let host = engine.clone();
    let next = engine.clone();
    let finished = engine.clone();
    let stop = engine.clone();
    let read = engine.clone();
    let respond = engine.clone();
    let write = engine.clone();
    let wait = engine.clone();
    let upgrade = engine.clone();
    let mut extension = Extension {
        name: "ext:http",
        dependencies: &[
            "ext:http_types",
            "ext:file",
            "ext:web",
            "ext:websocket",
            "ext:webstreams",
        ],
        ops: vec![
            Op::new("op_http_listen", move |mut address: Address| {
                let caps = caps.clone();
                let authorize: Authorize = match &mut address {
                    Address::Tcp { hostname, .. } => {
                        let hostname = hostname.clone();
                        Arc::new(move |address| {
                            let Address::Tcp { port, .. } = address else {
                                return Err("Invalid listener transport".into());
                            };
                            caps.check_net(&hostname, *port).map_err(Into::into)
                        })
                    }
                    Address::Unix { path } => {
                        let label = path
                            .as_ref()
                            .ok_or_else(|| io::Error::other("Missing Unix socket path"))?;
                        if label.is_empty() {
                            return Err(io::Error::other("Empty Unix socket path"));
                        }
                        let target = PathTarget::for_entry(caps.cwd().join(label))?;
                        *path = Some(
                            target
                                .as_path()
                                .to_str()
                                .ok_or_else(|| io::Error::other("Unix socket paths must be UTF-8"))?
                                .to_owned(),
                        );
                        Arc::new(move |_| caps.check_write(&target).map_err(Into::into))
                    }
                };
                listen.listen(address, authorize).map_err(io::Error::other)
            }),
            Op::with_scope("op_http_cli_listen", move |scope, _: ()| {
                let address = cli
                    .lock()
                    .unwrap()
                    .take()
                    .ok_or_else(|| io::Error::other("No CLI listener configured"))?;
                let weak = Arc::downgrade(&host);
                let interrupted = Arc::new(AtomicBool::new(false));
                let stopping = interrupted.clone();
                let isolate = scope.thread_safe_handle();
                let shutdown = Arc::new(move || {
                    if let Some(engine) = weak.upgrade() {
                        let force = stopping.swap(true, Ordering::SeqCst);
                        engine.stop_all(force);
                        if force {
                            avril_ext_process::Tree::global().force_live_processes();
                            isolate.terminate_execution();
                        }
                    }
                });
                let interrupt = shutdown.clone();
                #[cfg(unix)]
                // Serve owns SIGINT fate from here on through the shared
                // chaining registry: top-level listeners observe the
                // shutdown instead of being clobbered by it.
                avril_ext_process::Tree::global().claim_interrupt(&runtime, interrupt)?;
                #[cfg(not(unix))]
                ctrlc::set_handler(move || interrupt()).map_err(io::Error::other)?;
                let bound = host
                    .listen(address, Arc::new(|_| Ok(())))
                    .map_err(io::Error::other)?;
                #[cfg(unix)]
                {
                    // Serve owns SIGTERM fate from here on: the subprocess
                    // supervisor hands first deliveries to this shutdown
                    // instead of stopping children itself.
                    avril_ext_process::Tree::global().claim_term(&runtime, shutdown)?;
                }
                if interrupted.load(Ordering::SeqCst) {
                    host.stop(bound.0, true);
                    return Err(io::Error::other("Server shutdown requested during binding"));
                }
                match &bound.1 {
                    Address::Tcp { hostname, port } => eprintln!(
                        "Listening on http://{}:{port}",
                        if hostname.contains(':') {
                            format!("[{hostname}]")
                        } else {
                            hostname.clone()
                        }
                    ),
                    Address::Unix { path } => {
                        eprintln!("Listening on unix:{}", path.as_deref().unwrap_or_default())
                    }
                }
                Ok::<_, io::Error>(bound)
            }),
            Op::asynchronous("op_http_next", move |id: u64| next.next(id)),
            Op::asynchronous("op_http_finished", move |id: u64| finished.finished(id)),
            Op::new("op_http_stop", move |(id, force): (u64, bool)| {
                stop.stop(id, force);
                Ok::<_, io::Error>(())
            }),
            Op::asynchronous("op_http_read", move |id: u64| {
                let pending = read.read(id);
                async move {
                    // JS needs writable memory independent of Hyper's shared chunks.
                    Ok::<_, io::Error>(pending.await?.map(|bytes| ToJsBuffer::from(bytes.to_vec())))
                }
            }),
            Op::new(
                "op_http_respond",
                move |(id, status, headers, bytes, streaming): RespondArgs| {
                    respond
                        .respond((
                            id,
                            status,
                            headers,
                            bytes.map(|bytes| Bytes::copy_from_slice(&bytes)),
                            streaming,
                        ))
                        .map_err(io::Error::other)
                },
            ),
            Op::asynchronous(
                "op_http_write",
                move |(id, bytes): (u64, Option<JsBuffer>)| {
                    write.write(id, bytes.map(|bytes| Bytes::copy_from_slice(&bytes)))
                },
            ),
            Op::asynchronous("op_http_wait", move |id: u64| wait.wait(id)),
            Op::new(
                "op_http_upgrade",
                move |(id, protocol): (u64, Option<String>)| {
                    let (pending, headers) = upgrade
                        .upgrade(id, protocol.as_deref())
                        .map_err(io::Error::other)?;
                    let socket = sockets.accept(pending, protocol.unwrap_or_default())?;
                    Ok::<_, io::Error>((socket, headers))
                },
            ),
            Op::new("op_http_cancel", move |id: u64| {
                engine.cancel(id);
                Ok::<_, io::Error>(())
            }),
        ],
        modules: &[
            (
                "ext:http/main.js",
                include_str!(concat!(env!("OUT_DIR"), "/main.js")),
            ),
            (
                "ext:http/file.js",
                include_str!(concat!(env!("OUT_DIR"), "/file.js")),
            ),
        ],
        entry_point: None,
    };
    extension.ops.extend(file_ops);
    extension
}