use std::fs::File;
use std::io;
#[cfg(not(any(unix, windows)))]
use std::io::{Read, Seek, SeekFrom};
#[cfg(unix)]
use std::os::unix::fs::FileExt;
#[cfg(windows)]
use std::os::windows::fs::FileExt;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use hyper::http::header::{HeaderName, HeaderValue};
use hyper::http::{HeaderMap, Response};
use tokio::runtime::Handle;

pub type CheckFile = Arc<dyn Fn() -> io::Result<()> + Send + Sync>;
pub type HeaderPairs = Vec<(String, String)>;
const SAFE_INTEGER: u64 = (1 << 53) - 1;

struct Source {
    file: File,
    offset: u64,
    remaining: u64,
    check: CheckFile,
}

pub struct FileBody {
    runtime: Handle,
    source: Mutex<Option<Source>>,
    closed: AtomicBool,
    length: u64,
}

impl FileBody {
    pub fn len(&self) -> u64 {
        self.length
    }

    pub fn is_empty(&self) -> bool {
        self.length == 0
    }

    pub async fn read(self: &Arc<Self>) -> io::Result<(Vec<u8>, bool)> {
        let body = self.clone();
        self.runtime
            .spawn_blocking(move || {
                let mut source = body.source.lock().unwrap();
                let result = (|| {
                    if body.closed.load(Ordering::Acquire) {
                        return Err(io::Error::new(
                            io::ErrorKind::NotConnected,
                            "File response is closed",
                        ));
                    }
                    let source = source.as_mut().ok_or_else(|| {
                        io::Error::new(io::ErrorKind::NotConnected, "File response is closed")
                    })?;
                    (source.check)()?;
                    let mut bytes = vec![0; source.remaining.min(65536) as usize];
                    #[cfg(unix)]
                    let count = source.file.read_at(&mut bytes, source.offset)?;
                    #[cfg(windows)]
                    let count = source.file.seek_read(&mut bytes, source.offset)?;
                    #[cfg(not(any(unix, windows)))]
                    let count = {
                        source.file.seek(SeekFrom::Start(source.offset))?;
                        source.file.read(&mut bytes)?
                    };
                    if count == 0 {
                        return Err(io::Error::new(
                            io::ErrorKind::UnexpectedEof,
                            "File ended before Content-Length",
                        ));
                    }
                    if body.closed.load(Ordering::Acquire) {
                        return Err(io::Error::new(
                            io::ErrorKind::NotConnected,
                            "File response is closed",
                        ));
                    }
                    bytes.truncate(count);
                    source.offset += count as u64;
                    source.remaining -= count as u64;
                    Ok((bytes, source.remaining == 0))
                })();
                if result.as_ref().map_or(true, |(_, done)| *done) {
                    body.closed.store(true, Ordering::Release);
                    source.take();
                }
                result
            })
            .await
            .map_err(io::Error::other)?
    }

    pub fn close(self: &Arc<Self>) {
        if !self.closed.swap(true, Ordering::AcqRel) {
            let body = self.clone();
            self.runtime.spawn_blocking(move || {
                body.source.lock().unwrap().take();
            });
        }
    }
}

impl Drop for FileBody {
    fn drop(&mut self) {
        // File descriptors are closed off the guest thread, including abandoned responses.
        let source = self.source.get_mut().unwrap().take();
        if source.is_some() {
            self.runtime.spawn_blocking(move || drop(source));
        }
    }
}

pub fn headers(pairs: HeaderPairs) -> io::Result<HeaderMap> {
    let mut headers = HeaderMap::new();
    for (name, value) in pairs {
        let name = HeaderName::from_bytes(name.as_bytes()).map_err(io::Error::other)?;
        let value = value
            .chars()
            .map(|c| u8::try_from(c as u32))
            .collect::<Result<Vec<_>, _>>()
            .map_err(io::Error::other)?;
        let value = HeaderValue::from_bytes(&value).map_err(io::Error::other)?;
        headers.append(name, value);
    }
    Ok(headers)
}

pub fn header_pairs(headers: &HeaderMap) -> HeaderPairs {
    headers
        .iter()
        .map(|(name, value)| {
            (
                name.to_string(),
                value
                    .as_bytes()
                    .iter()
                    .map(|byte| char::from(*byte))
                    .collect(),
            )
        })
        .collect()
}

fn set(headers: &mut HeaderMap, name: &'static str, value: impl ToString) -> io::Result<()> {
    headers.insert(
        name,
        HeaderValue::from_str(&value.to_string()).map_err(io::Error::other)?,
    );
    Ok(())
}

fn get(headers: &HeaderMap, name: &str) -> Option<Vec<u8>> {
    let mut values = headers.get_all(name).iter();
    let mut value = values.next()?.as_bytes().to_vec();
    for next in values {
        value.extend_from_slice(b", ");
        value.extend_from_slice(next.as_bytes());
    }
    Some(value)
}

fn tags(value: &[u8]) -> Option<Vec<&[u8]>> {
    let mut value = value.trim_ascii();
    let mut tags = Vec::new();
    while !value.is_empty() {
        if value[0] == b',' {
            value = value[1..].trim_ascii_start();
            continue;
        }
        let original = value;
        if value.starts_with(b"W/") {
            value = &value[2..];
        }
        if value.first() != Some(&b'"') {
            return None;
        }
        let end = value[1..].iter().position(|byte| *byte == b'"')? + 1;
        if !value[1..end]
            .iter()
            .all(|byte| *byte == 0x21 || *byte >= 0x23 && *byte != 0x7f)
        {
            return None;
        }
        let length = original.len() - value.len() + end + 1;
        tags.push(&original[..length]);
        value = value[end + 1..].trim_ascii_start();
        if !value.is_empty() && value[0] != b',' {
            return None;
        }
    }
    Some(tags)
}

fn matches(value: &[u8], etag: Option<&[u8]>, weak: bool) -> bool {
    if value == b"*" {
        return true;
    }
    let Some(etag) = etag else {
        return false;
    };
    tags(value).is_some_and(|tags| {
        tags.iter().any(|tag| {
            if weak {
                tag.strip_prefix(b"W/").unwrap_or(tag) == etag.strip_prefix(b"W/").unwrap_or(etag)
            } else {
                !tag.starts_with(b"W/") && *tag == etag
            }
        })
    })
}

fn date(value: &[u8]) -> Option<SystemTime> {
    httpdate::parse_http_date(std::str::from_utf8(value).ok()?).ok()
}

#[derive(Debug, PartialEq)]
enum Range {
    Ignore,
    Unsatisfiable,
    Bytes(u64, u64),
}

fn range(value: &[u8], size: u64) -> Range {
    let Some(equal) = value.iter().position(|byte| *byte == b'=') else {
        return Range::Ignore;
    };
    let (unit, interval) = (&value[..equal], &value[equal + 1..]);
    if !unit.eq_ignore_ascii_case(b"bytes") {
        return Range::Ignore;
    }
    let Some(dash) = interval.iter().position(|byte| *byte == b'-') else {
        return Range::Ignore;
    };
    let (start, end) = (&interval[..dash], &interval[dash + 1..]);
    if (start.is_empty() && end.is_empty()) || !start.iter().chain(end).all(u8::is_ascii_digit) {
        return Range::Ignore;
    }
    let normalize = |value: &[u8]| {
        value
            .iter()
            .position(|byte| *byte != b'0')
            .unwrap_or(value.len())
    };
    let start = &start[normalize(start)..];
    let end = &end[normalize(end)..];
    let parse = |value: &[u8]| {
        value.iter().try_fold(0u64, |number, digit| {
            number.checked_mul(10)?.checked_add((digit - b'0') as u64)
        })
    };
    if interval.starts_with(b"-") {
        let suffix = parse(end).unwrap_or(u64::MAX);
        return if suffix == 0 || size == 0 {
            Range::Unsatisfiable
        } else {
            Range::Bytes(size.saturating_sub(suffix), size - 1)
        };
    }
    // Compare decimal strings before conversion so oversized reversed intervals remain malformed.
    if !interval.ends_with(b"-") && (end.len(), end) < (start.len(), start) {
        return Range::Ignore;
    }
    let Some(start) = parse(start).filter(|start| *start < size) else {
        return Range::Unsatisfiable;
    };
    Range::Bytes(
        start,
        parse(end)
            .filter(|_| !interval.ends_with(b"-"))
            .unwrap_or(u64::MAX)
            .min(size - 1),
    )
}

pub fn serve_file(
    runtime: Handle,
    method: &str,
    request: HeaderMap,
    path: &str,
    headers: HeaderMap,
    open: impl FnOnce() -> io::Result<(File, CheckFile)>,
) -> io::Result<Response<Option<Arc<FileBody>>>> {
    let mut response = Response::new(None);
    *response.headers_mut() = headers;
    let headers = response.headers_mut();
    for name in ["content-length", "content-range", "last-modified"] {
        headers.remove(name);
    }
    if method != "GET" && method != "HEAD" {
        set(headers, "allow", "GET, HEAD")?;
        set(headers, "content-length", 0)?;
        *response.status_mut() = hyper::StatusCode::METHOD_NOT_ALLOWED;
        return Ok(response);
    }
    let (file, check) = match open() {
        Ok(opened) => opened,
        Err(error)
            if matches!(
                error.kind(),
                io::ErrorKind::NotFound | io::ErrorKind::NotADirectory
            ) =>
        {
            set(headers, "content-length", 0)?;
            *response.status_mut() = hyper::StatusCode::NOT_FOUND;
            return Ok(response);
        }
        Err(error) => return Err(error),
    };
    check()?;
    let info = file.metadata()?;
    if !info.is_file() {
        set(headers, "content-length", 0)?;
        *response.status_mut() = hyper::StatusCode::NOT_FOUND;
        return Ok(response);
    }
    if info.len() > SAFE_INTEGER {
        return Err(io::Error::other("Invalid file size"));
    }
    let modified = info.modified().ok();
    // httpdate only formats years 1970 through 9999.
    let modified_seconds = modified
        .and_then(|date| date.duration_since(UNIX_EPOCH).ok())
        .filter(|duration| duration.as_secs() < 253402300800)
        .map(|duration| UNIX_EPOCH + Duration::from_secs(duration.as_secs()));
    if let Some(modified) = modified
        && !headers.contains_key("etag")
    {
        let millis = match modified.duration_since(UNIX_EPOCH) {
            Ok(duration) => duration.as_millis() as i128,
            Err(error) => -(error.duration().as_millis() as i128),
        };
        set(headers, "etag", format!("W/\"{}-{millis}\"", info.len()))?;
    }
    if let Some(modified) = modified_seconds {
        set(headers, "last-modified", httpdate::fmt_http_date(modified))?;
    }
    let etag = get(headers, "etag");
    if !headers.contains_key("content-type") {
        let mime = mime_guess::from_path(path).first_or_octet_stream();
        let text = mime.type_() == "text"
            || matches!(
                mime.essence_str(),
                "application/javascript" | "application/json" | "application/xml" | "image/svg+xml"
            );
        set(
            headers,
            "content-type",
            format!("{mime}{}", if text { "; charset=utf-8" } else { "" }),
        )?;
    }
    set(headers, "accept-ranges", "bytes")?;
    let mut status = 200;
    if let Some(value) = get(&request, "if-match") {
        if !matches(&value, etag.as_deref(), false) {
            status = 412;
        }
    } else if let Some(value) = get(&request, "if-unmodified-since")
        && let (Some(date), Some(modified)) = (date(&value), modified_seconds)
        && modified > date
    {
        status = 412;
    }
    if status == 200 {
        if let Some(value) = get(&request, "if-none-match") {
            if matches(&value, etag.as_deref(), true) {
                status = 304;
            }
        } else if let Some(value) = get(&request, "if-modified-since")
            && let (Some(date), Some(modified)) = (date(&value), modified_seconds)
            && modified <= date
        {
            status = 304;
        }
    }
    if status != 200 {
        if status == 412 {
            set(headers, "content-length", 0)?;
        }
        *response.status_mut() = hyper::StatusCode::from_u16(status).unwrap();
        return Ok(response);
    }

    let mut start = 0;
    let mut length = info.len();
    let if_range = get(&request, "if-range");
    let use_range = if_range.as_ref().is_none_or(|value| {
        if value.starts_with(b"\"") || value.starts_with(b"W/\"") {
            !value.starts_with(b"W/")
                && etag
                    .as_deref()
                    .is_some_and(|etag| !etag.starts_with(b"W/") && etag == value)
        } else {
            date(value).is_some_and(|date| Some(date) == modified_seconds)
        }
    });
    if method == "GET"
        && use_range
        && let Some(value) = get(&request, "range")
    {
        match range(&value, info.len()) {
            Range::Ignore => {}
            Range::Unsatisfiable => {
                status = 416;
                length = 0;
                set(headers, "content-range", format!("bytes */{}", info.len()))?;
            }
            Range::Bytes(first, last) => {
                status = 206;
                start = first;
                length = last - first + 1;
                set(
                    headers,
                    "content-range",
                    format!("bytes {first}-{last}/{}", info.len()),
                )?;
            }
        }
    }
    set(headers, "content-length", length)?;
    if method == "GET" && length != 0 {
        *response.body_mut() = Some(Arc::new(FileBody {
            runtime,
            source: Mutex::new(Some(Source {
                file,
                offset: start,
                remaining: length,
                check,
            })),
            closed: AtomicBool::new(false),
            length,
        }));
    }
    *response.status_mut() = hyper::StatusCode::from_u16(status).unwrap();
    Ok(response)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn ranges_handle_overflow_and_reversed_intervals() {
        assert_eq!(
            range(b"bytes=9007199254740993-9007199254740992", 10),
            Range::Ignore
        );
        assert_eq!(
            range(
                b"bytes=999999999999999999999999-999999999999999999999998",
                10
            ),
            Range::Ignore
        );
        assert_eq!(
            range(b"bytes=999999999999999999999999-", 10),
            Range::Unsatisfiable
        );
        assert_eq!(
            range(b"bytes=8-999999999999999999999999", 10),
            Range::Bytes(8, 9)
        );
        assert_eq!(
            range(b"bytes=-999999999999999999999999", 10),
            Range::Bytes(0, 9)
        );
        assert_eq!(range(b"bytes=0002-0004", 10), Range::Bytes(2, 4));
        assert_eq!(range(b"bytes=0-", 0), Range::Unsatisfiable);
        assert_eq!(range(b"bytes=-0", 10), Range::Unsatisfiable);
        assert_eq!(range(b"bytes=0-1,4-5", 10), Range::Ignore);
    }

    #[test]
    fn entity_tag_lists_preserve_commas_and_comparison_strength() {
        assert!(matches(
            b", \"other,tag\", W/\"known\",",
            Some(b"\"known\""),
            true
        ));
        assert!(!matches(b"W/\"known\"", Some(b"W/\"known\""), false));
        assert!(matches(b"\"known\"", Some(b"\"known\""), false));
        assert!(matches(b"*", None, false));
        assert!(!matches(b"junk, \"known\"", Some(b"\"known\""), true));
    }
}