use std::collections::HashMap;
use std::convert::Infallible;
use std::io;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};

use avril_caps::Capabilities;
use avril_core::Op;
use avril_core::serde_v8::ToJsBuffer;
use avril_http::Engine;
use avril_http::file::{CheckFile, FileBody, HeaderPairs, header_pairs, headers, serve_file};
use serde::Deserialize;
use tokio::runtime::Handle;

use avril_ext_file::{Files as Handles, response_file};
use avril_host::io::response;

#[derive(Deserialize)]
#[serde(untagged)]
enum Source {
    Path(String),
    Handle(u64),
}

type HandleFile = (String, std::fs::File, CheckFile);
type Prepare = (String, HeaderPairs, Source, HeaderPairs);
type Prepared = (u16, HeaderPairs, Option<u64>);

const MAX_ID: u64 = 9_007_199_254_740_991;

struct Files {
    runtime: Handle,
    caps: Arc<Capabilities>,
    bodies: Mutex<HashMap<u64, Arc<FileBody>>>,
    sources: Mutex<HashMap<u64, HandleFile>>,
    next: AtomicU64,
}

impl Files {
    fn next_id(&self) -> io::Result<u64> {
        self.next
            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |id| {
                (id <= MAX_ID).then_some(id + 1)
            })
            .map_err(|_| io::Error::other("File response identifiers exhausted"))
    }

    fn take(&self, id: u64) -> io::Result<Arc<FileBody>> {
        self.bodies
            .lock()
            .unwrap()
            .remove(&id)
            .ok_or_else(|| io::Error::new(io::ErrorKind::NotConnected, "File response is closed"))
    }

    fn prepare(
        self: &Arc<Self>,
        (method, request, source, extra): Prepare,
    ) -> impl Future<Output = io::Result<Prepared>> + use<> {
        let files = self.clone();
        let source = match source {
            Source::Path(path) => Ok((path, None)),
            Source::Handle(id) => self
                .sources
                .lock()
                .unwrap()
                .remove(&id)
                .map(|(name, file, check)| (name, Some((file, check))))
                .ok_or_else(|| {
                    io::Error::new(io::ErrorKind::NotConnected, "File source is closed")
                }),
        };
        let pending = self.runtime.spawn_blocking(move || {
            let (path, opened) = source?;
            let response = serve_file(
                files.runtime.clone(),
                &method,
                headers(request)?,
                &path,
                headers(extra)?,
                || match opened {
                    Some(opened) => Ok(opened),
                    None => response_file(files.caps.clone(), path.clone()),
                },
            )?;
            let status = response.status().as_u16();
            let headers = header_pairs(response.headers());
            let id = if let Some(body) = response.into_body() {
                let id = files.next_id()?;
                files.bodies.lock().unwrap().insert(id, body);
                Some(id)
            } else {
                None
            };
            Ok((status, headers, id))
        });
        async move { pending.await.map_err(io::Error::other)? }
    }
}

impl Drop for Files {
    fn drop(&mut self) {
        for (_, body) in self.bodies.get_mut().unwrap().drain() {
            body.close();
        }
        let sources = std::mem::take(self.sources.get_mut().unwrap());
        if !sources.is_empty() {
            self.runtime.spawn_blocking(move || drop(sources));
        }
    }
}

pub fn ops(
    runtime: Handle,
    caps: Arc<Capabilities>,
    handles: Arc<Handles>,
    engine: Arc<Engine>,
) -> Vec<Op> {
    let files = Arc::new(Files {
        runtime,
        caps,
        bodies: Mutex::new(HashMap::new()),
        sources: Mutex::new(HashMap::new()),
        next: AtomicU64::new(1),
    });
    let acquire = files.clone();
    let prepare = files.clone();
    let read = files.clone();
    let send = files.clone();
    vec![
        Op::new("op_http_file_handle", move |(id, consume): (u64, bool)| {
            let acquired = (|| {
                let token = acquire.next_id()?;
                let source = handles.response_handle(id, consume)?;
                acquire.sources.lock().unwrap().insert(token, source);
                Ok(token)
            })();
            Ok::<_, Infallible>(response(acquired))
        }),
        Op::asynchronous("op_http_serve_file", move |args: Prepare| {
            let pending = prepare.prepare(args);
            async move { Ok::<_, Infallible>(response(pending.await)) }
        }),
        Op::asynchronous("op_http_file_read", move |id: u64| {
            let body = read.bodies.lock().unwrap().get(&id).cloned();
            async move {
                let result = match body {
                    Some(body) => body
                        .read()
                        .await
                        .map(|(bytes, done)| (ToJsBuffer::from(bytes), done)),
                    None => Err(io::Error::new(
                        io::ErrorKind::NotConnected,
                        "File response is closed",
                    )),
                };
                Ok::<_, Infallible>(response(result))
            }
        }),
        Op::new("op_http_file_close", move |id: u64| {
            if let Ok(body) = files.take(id) {
                body.close();
            }
            Ok::<_, Infallible>(())
        }),
        Op::new(
            "op_http_respond_file",
            move |(id, status, headers, body): (u64, u16, HeaderPairs, u64)| {
                let file = send.take(body)?;
                engine
                    .respond_file(id, status, headers, file)
                    .map_err(io::Error::other)
            },
        ),
    ]
}