use crate::{error::Error, handler::{Handler,Next}, request::Request, response::Response, Result};
use async_trait::async_trait;
use enumflags2::{bitflags, BitFlags};
use futures::{ready, Stream};
use headers::{
    AcceptRanges, ContentLength, ContentRange, ContentType, ETag, HeaderMapExt, IfMatch,
    IfModifiedSince, IfNoneMatch, IfUnmodifiedSince, LastModified,
};
use hyper::{
    body::Bytes,
    http::{HeaderMap, HeaderValue},
    StatusCode,
};
#[cfg(unix)]
use std::os::unix::fs::MetadataExt;
use std::{
    borrow::Cow,
    cmp,
    fs::Metadata,
    future::Future,
    io::{Error as IoError, ErrorKind, Read, Seek},
    ops::{Deref, DerefMut},
    path::{Path, PathBuf},
    pin::Pin,
    task::{Context, Poll},
    time::{SystemTime, UNIX_EPOCH},
};
use tokio::fs::File;

const CHUNK_SIZE: u64 = 1024 * 1024;

#[bitflags(default = Etag | LastModified | ContentDisposition)]
#[repr(u8)]
#[derive(Copy, Clone, Debug, PartialEq)]
pub(crate) enum Flag {
    Etag = 0b0001,
    LastModified = 0b0010,
    ContentDisposition = 0b0100,
}

/// A file with an associated name.
#[derive(Debug)]
pub struct NamedFile {
    path: PathBuf,
    file: File,
    modified: Option<SystemTime>,
    buffer_size: u64,
    metadata: Metadata,
    flags: BitFlags<Flag>,
    content_type: mime::Mime,
    content_disposition: Option<HeaderValue>,
    content_encoding: Option<HeaderValue>,
}

/// Builder for build [`NamedFile`].
#[derive(Clone)]
pub struct StaticFile {
    path: PathBuf,
    attached_name: Option<String>,
    disposition_type: Option<String>,
    content_type: Option<mime::Mime>,
    content_encoding: Option<String>,
    buffer_size: Option<u64>,
    flags: BitFlags<Flag>,
}

impl StaticFile {
    /// Create a new `StaticFile`.
    #[inline]
    pub fn new(path: impl Into<PathBuf>) -> Self {
        NamedFile::builder(path)
    }

    /// Sets attached filename and returns `Self`.
    #[inline]
    pub fn attached_name<T: Into<String>>(mut self, attached_name: T) -> Self {
        self.attached_name = Some(attached_name.into());
        self.flags.insert(Flag::ContentDisposition);
        self
    }

    /// Sets disposition encoding and returns `Self`.
    #[inline]
    pub fn disposition_type<T: Into<String>>(mut self, disposition_type: T) -> Self {
        self.disposition_type = Some(disposition_type.into());
        self.flags.insert(Flag::ContentDisposition);
        self
    }

    /// Disable `Content-Disposition` header.
    ///
    /// By default Content-Disposition` header is enabled.
    #[inline]
    pub fn disable_content_disposition(&mut self) {
        self.flags.remove(Flag::ContentDisposition);
    }

    /// Sets content type and returns `Self`.
    #[inline]
    pub fn content_type<T: Into<mime::Mime>>(mut self, content_type: T) -> Self {
        self.content_type = Some(content_type.into());
        self
    }

    /// Sets content encoding and returns `Self`.
    #[inline]
    pub fn content_encoding<T: Into<String>>(mut self, content_encoding: T) -> Self {
        self.content_encoding = Some(content_encoding.into());
        self
    }

    /// Sets buffer size and returns `Self`.
    /// During the file chunk read, the maximum read size at one time will affect the
    /// access experience and the demand for server memory.
    ///
    /// Please set it according to your own situation.
    ///
    /// The default is 1M.
    #[inline]
    pub fn buffer_size(mut self, buffer_size: u64) -> Self {
        self.buffer_size = Some(buffer_size);
        self
    }

    ///Specifies whether to use ETag or not.
    ///
    /// Default is true.
    #[inline]
    pub fn use_etag(mut self, value: bool) -> Self {
        if value {
            self.flags.insert(Flag::Etag);
        } else {
            self.flags.remove(Flag::Etag);
        }
        self
    }

    ///Specifies whether to use Last-Modified or not.
    ///
    ///Default is true.
    #[inline]
    pub fn use_last_modified(mut self, value: bool) -> Self {
        if value {
            self.flags.insert(Flag::LastModified);
        } else {
            self.flags.remove(Flag::LastModified);
        }
        self
    }

    /// Build a new `NamedFile` and send it.
    pub async fn send(self, req_headers: &HeaderMap) -> Response {
        let builder = Response::default();
        if !self.path.exists() {
            return builder.status(StatusCode::NOT_FOUND);
        } else {
            match self.build().await {
                Ok(file) => file.send(req_headers).await,
                Err(_) => builder.status(StatusCode::INTERNAL_SERVER_ERROR),
            }
        }
    }

    /// Build a new [`NamedFile`].
    pub async fn build(self) -> Result<NamedFile> {
        let StaticFile {
            path,
            content_type,
            content_encoding,
            buffer_size,
            disposition_type,
            attached_name,
            flags,
        } = self;

        let file = File::open(&path).await?;
        let content_type = content_type.unwrap_or_else(|| {
            let ct = mime_guess::from_path(&path).first_or_octet_stream();
            let ftype = ct.type_();
            let stype = ct.subtype();
            if (ftype == mime::TEXT || stype == mime::JSON || stype == mime::JAVASCRIPT)
                && ct.get_param(mime::CHARSET).is_none()
            {
                //TODO: auto detect charset
                format!("{}; charset=utf-8", ct)
                    .parse::<mime::Mime>()
                    .unwrap_or(ct)
            } else {
                ct
            }
        });
        let metadata = file.metadata().await?;
        let modified = metadata.modified().ok();
        let content_encoding = match content_encoding {
            Some(content_encoding) => Some(
                content_encoding
                    .parse::<HeaderValue>()
                    .map_err(|e| Error::Other(e.to_string()))?,
            ),
            None => None,
        };

        let mut content_disposition = None;
        if attached_name.is_some() || disposition_type.is_some() {
            content_disposition = Some(build_content_disposition(
                &path,
                &content_type,
                disposition_type.as_deref(),
                attached_name.as_deref(),
            )?);
        }
        Ok(NamedFile {
            path,
            file,
            content_type,
            content_disposition,
            metadata,
            modified,
            content_encoding,
            buffer_size: buffer_size.unwrap_or(CHUNK_SIZE),
            flags,
        })
    }
}

#[async_trait]
impl Handler for StaticFile {
    #[inline]
    async fn handle(&self, req: &mut Request, _next: &Next) -> Result<Response> {
        let builder = Response::default();
        match self.clone().build().await {
            Ok(file) => Ok(file.send(req.headers()).await),
            Err(_) => Ok(builder.status(StatusCode::NOT_FOUND)),
        }
    }
}

impl NamedFile {
    /// Create new [`NamedFileBuilder`].
    #[inline]
    pub fn builder(path: impl Into<PathBuf>) -> StaticFile {
        StaticFile {
            path: path.into(),
            attached_name: None,
            disposition_type: None,
            content_type: None,
            content_encoding: None,
            buffer_size: None,
            flags: BitFlags::default(),
        }
    }

    /// Attempts to open a file in read-only mode.
    #[inline]
    pub async fn open(path: impl Into<PathBuf>) -> Result<NamedFile> {
        Self::builder(path).build().await
    }

    /// Attempts to send a file. If file not exists, not found error will occur.
    pub async fn send_file(path: impl Into<PathBuf>, req_headers: &HeaderMap) -> Response {
        let path = path.into();
        let builder = Response::default();
        if !path.exists() {
            return builder.status(StatusCode::NOT_FOUND);
        } else {
            match Self::builder(path).build().await {
                Ok(file) => file.send(req_headers).await,
                Err(_) => builder.status(StatusCode::INTERNAL_SERVER_ERROR),
            }
        }
    }

    /// Returns reference to the underlying `File` object.
    #[inline]
    pub fn file(&self) -> &File {
        &self.file
    }

    /// Retrieve the path of this file.
    #[inline]
    pub fn path(&self) -> &Path {
        self.path.as_path()
    }

    /// Get content type value.
    #[inline]
    pub fn content_type(&self) -> &mime::Mime {
        &self.content_type
    }
    /// Sets the MIME Content-Type for serving this file. By default
    /// the Content-Type is inferred from the filename extension.
    #[inline]
    pub fn set_content_type(&mut self, content_type: mime::Mime) {
        self.content_type = content_type;
    }

    /// Get Content-Disposition value.
    #[inline]
    pub fn content_disposition(&self) -> Option<&HeaderValue> {
        self.content_disposition.as_ref()
    }
    /// Sets the `Content-Disposition` for serving this file. This allows
    /// changing the inline/attachment disposition as well as the filename
    /// sent to the peer.
    ///
    /// By default the disposition is `inline` for text,
    /// image, and video content types, and `attachment` otherwise, and
    /// the filename is taken from the path provided in the `open` method
    /// after converting it to UTF-8 using
    /// [to_string_lossy](https://doc.rust-lang.org/std/ffi/struct.OsStr.html#method.to_string_lossy).
    #[inline]
    pub fn set_content_disposition(&mut self, content_disposition: HeaderValue) {
        self.content_disposition = Some(content_disposition);
        self.flags.insert(Flag::ContentDisposition);
    }

    /// Disable `Content-Disposition` header.
    ///
    /// By default Content-Disposition` header is enabled.
    #[inline]
    pub fn disable_content_disposition(&mut self) {
        self.flags.remove(Flag::ContentDisposition);
    }

    /// Get content encoding value reference.
    #[inline]
    pub fn content_encoding(&self) -> Option<&HeaderValue> {
        self.content_encoding.as_ref()
    }
    /// Sets content encoding for serving this file
    #[inline]
    pub fn set_content_encoding(&mut self, content_encoding: HeaderValue) {
        self.content_encoding = Some(content_encoding);
    }

    /// Get ETag value.
    pub fn etag(&self) -> Option<ETag> {
        // This etag format is similar to Apache's.
        self.modified.as_ref().and_then(|mtime| {
            let ino = {
                #[cfg(unix)]
                {
                    self.metadata.ino()
                }
                #[cfg(not(unix))]
                {
                    0
                }
            };

            let dur = mtime
                .duration_since(UNIX_EPOCH)
                .expect("modification time must be after epoch");
            let etag_str = format!(
                "\"{:x}-{:x}-{:x}-{:x}\"",
                ino,
                self.metadata.len(),
                dur.as_secs(),
                dur.subsec_nanos()
            );
            match etag_str.parse::<ETag>() {
                Ok(etag) => Some(etag),
                Err(e) => {
                    // tracing::error!(error = ?e, etag = %etag_str, "set file's etag failed");
                    println!("error = {},etag = {} set file's etag failed", e, etag_str);
                    None
                }
            }
        })
    }
    ///Specifies whether to use ETag or not.
    ///
    ///Default is true.
    #[inline]
    pub fn use_etag(&mut self, value: bool) {
        if value {
            self.flags.insert(Flag::Etag);
        } else {
            self.flags.remove(Flag::Etag);
        }
    }

    /// GEt last_modified value.
    #[inline]
    pub fn last_modified(&self) -> Option<SystemTime> {
        self.modified
    }
    ///Specifies whether to use Last-Modified or not.
    ///
    ///Default is true.
    #[inline]
    pub fn use_last_modified(&mut self, value: bool) {
        if value {
            self.flags.insert(Flag::LastModified);
        } else {
            self.flags.remove(Flag::LastModified);
        }
    }
    ///Consume self and send content to [`Response`].
    pub async fn send(mut self, req_headers: &HeaderMap) -> Response {
        let etag = if self.flags.contains(Flag::Etag) {
            self.etag()
        } else {
            None
        };
        let last_modified = if self.flags.contains(Flag::LastModified) {
            self.last_modified()
        } else {
            None
        };

        // check preconditions
        let precondition_failed = if !any_match(etag.as_ref(), req_headers) {
            true
        } else if let (Some(ref last_modified), Some(since)) =
            (last_modified, req_headers.typed_get::<IfUnmodifiedSince>())
        {
            !since.precondition_passes(*last_modified)
        } else {
            false
        };

        // check last modified
        let not_modified = if !none_match(etag.as_ref(), req_headers) {
            true
        } else if req_headers.contains_key("if-none-match") {
            false
        } else if let (Some(ref last_modified), Some(since)) =
            (last_modified, req_headers.typed_get::<IfModifiedSince>())
        {
            !since.is_modified(*last_modified)
        } else {
            false
        };

        let mut builder = Response::default();

        if self.flags.contains(Flag::ContentDisposition) {
            if let Some(content_disposition) = self.content_disposition.take() {
                builder
                    .headers_mut()
                    .insert("content-disposition", content_disposition);
            } else if !builder.headers().contains_key("content-disposition") {
                match build_content_disposition(&self.path, &self.content_type, None, None) {
                    Ok(content_disposition) => {
                        builder
                            .headers_mut()
                            .insert("content-disposition", content_disposition);
                    }
                    Err(e) => {
                        // tracing::error!(error = ?e, "build file's content disposition failed");
                        println!("error = {}, build file's content disposition failed", e)
                    }
                }
            }
        }

        builder
            .headers_mut()
            .typed_insert(ContentType::from(self.content_type.clone()));
        if let Some(lm) = last_modified {
            builder.headers_mut().typed_insert(LastModified::from(lm));
        }
        if let Some(etag) = self.etag() {
            builder.headers_mut().typed_insert(etag);
        }
        builder.headers_mut().typed_insert(AcceptRanges::bytes());

        let mut length = self.metadata.len();
        if let Some(content_encoding) = &self.content_encoding {
            builder
                .headers_mut()
                .insert("content-encoding", content_encoding.clone());
        }
        let mut offset = 0;
        if let Some(range) = req_headers.get("range") {
            if let Ok(range) = range.to_str() {
                if let Ok(range) = HttpRange::parse(range, length) {
                    length = range[0].length;
                    offset = range[0].start;
                } else {
                    builder
                        .headers_mut()
                        .typed_insert(ContentRange::unsatisfied_bytes(length));
                    return builder.status(StatusCode::RANGE_NOT_SATISFIABLE);
                };
            } else {
                return builder.status(StatusCode::BAD_REQUEST);
            };
        }

        if precondition_failed {
            return builder.status(StatusCode::PRECONDITION_FAILED);
        } else if not_modified {
            return builder.status(StatusCode::NOT_MODIFIED);
        }

        if offset != 0 || length != self.metadata.len() {
            let mut builder = builder.status(StatusCode::PARTIAL_CONTENT);
            match ContentRange::bytes(offset..offset + length - 1, self.metadata.len()) {
                Ok(content_range) => {
                    builder.headers_mut().typed_insert(content_range);
                }
                Err(e) => {
                    // tracing::error!(error = ?e, "set file's content ranage failed");
                    println!("error = {}, set file's content ranage failed", e);
                }
            }

            let reader = FileChunk {
                offset,
                chunk_size: cmp::min(length, self.metadata.len()),
                read_size: 0,
                state: ChunkedState::File(Some(self.file.into_std().await)),
                buffer_size: self.buffer_size,
            };

            builder
                .headers_mut()
                .typed_insert(ContentLength(reader.chunk_size));
            return builder.stream(reader);
        } else {
            let mut builder = builder.status(StatusCode::OK);
            let reader = FileChunk {
                offset,
                state: ChunkedState::File(Some(self.file.into_std().await)),
                chunk_size: length,
                read_size: 0,
                buffer_size: self.buffer_size,
            };
            builder
                .headers_mut()
                .typed_insert(ContentLength(length - offset));
            return builder.stream(reader);
        }
    }
}

impl Deref for NamedFile {
    type Target = File;

    fn deref(&self) -> &File {
        &self.file
    }
}

impl DerefMut for NamedFile {
    fn deref_mut(&mut self) -> &mut File {
        &mut self.file
    }
}

/// Returns true if `req_headers` has no `If-Match` header or one which matches `etag`.
fn any_match(etag: Option<&ETag>, req_headers: &HeaderMap) -> bool {
    match req_headers.typed_get::<IfMatch>() {
        None => true,
        Some(if_match) => {
            if if_match == IfMatch::any() {
                true
            } else if let Some(etag) = etag {
                if_match.precondition_passes(etag)
            } else {
                false
            }
        }
    }
}

/// Returns true if `req_headers` doesn't have an `If-None-Match` header matching `req`.
fn none_match(etag: Option<&ETag>, req_headers: &HeaderMap) -> bool {
    match req_headers.typed_get::<IfNoneMatch>() {
        None => true,
        Some(if_none_match) => {
            if if_none_match == IfNoneMatch::any() {
                false
            } else if let Some(etag) = etag {
                if_none_match.precondition_passes(etag)
            } else {
                true
            }
        }
    }
}

fn build_content_disposition(
    file_path: impl AsRef<Path>,
    content_type: &mime::Mime,
    disposition_type: Option<&str>,
    attached_name: Option<&str>,
) -> Result<HeaderValue> {
    let disposition_type = disposition_type.unwrap_or_else(|| {
        if attached_name.is_some() {
            "attachment"
        } else {
            match (content_type.type_(), content_type.subtype()) {
                (mime::IMAGE | mime::TEXT | mime::VIDEO | mime::AUDIO, _)
                | (_, mime::JAVASCRIPT | mime::JSON) => "inline",
                _ => "attachment",
            }
        }
    });
    let content_disposition = if disposition_type == "attachment" {
        let attached_name = match attached_name {
            Some(attached_name) => Cow::Borrowed(attached_name),
            None => file_path
                .as_ref()
                .file_name()
                .map(|file_name| file_name.to_string_lossy().to_string())
                .unwrap_or_else(|| "file".into())
                .into(),
        };
        format!("attachment; filename={}", attached_name)
            .parse::<HeaderValue>()
            .map_err(|e| Error::Other(e.to_string()))?
    } else {
        disposition_type
            .parse::<HeaderValue>()
            .map_err(|e| Error::Other(e.to_string()))?
    };
    Ok(content_disposition)
}

/// HTTP Range header representation.
#[derive(Clone, Debug, Copy)]
pub struct HttpRange {
    /// Start position.
    pub start: u64,
    /// Total length.
    pub length: u64,
}

static PREFIX: &str = "bytes=";
const PREFIX_LEN: usize = 6;

impl HttpRange {
    /// Parses Range HTTP header string as per RFC 2616.
    ///
    /// `header` is HTTP Range header (e.g. `bytes=bytes=0-9`).
    /// `size` is full size of response (file).
    pub fn parse(header: &str, size: u64) -> Result<Vec<HttpRange>, Error> {
        if header.is_empty() {
            return Ok(Vec::new());
        }
        if !header.starts_with(PREFIX) {
            return Err(Error::Other(String::from("InvalidRange")));
        }

        let size_sig = size as i64;
        let mut no_overlap = false;

        let all_ranges: Vec<Option<HttpRange>> = header[PREFIX_LEN..]
            .split(',')
            .map(|x| x.trim())
            .filter(|x| !x.is_empty())
            .map(|ra| {
                let mut start_end_iter = ra.split('-');

                let start_str = start_end_iter.next().ok_or(())?.trim();
                let end_str = start_end_iter.next().ok_or(())?.trim();

                if start_str.is_empty() {
                    // If no start is specified, end specifies the
                    // range start relative to the end of the file.
                    let mut length: i64 = end_str.parse().map_err(|_| ())?;

                    if length > size_sig {
                        length = size_sig;
                    }

                    Ok(Some(HttpRange {
                        start: (size_sig - length) as u64,
                        length: length as u64,
                    }))
                } else {
                    let start: i64 = start_str.parse().map_err(|_| ())?;

                    if start < 0 {
                        return Err(());
                    }
                    if start >= size_sig {
                        no_overlap = true;
                        return Ok(None);
                    }

                    let length = if end_str.is_empty() {
                        // If no end is specified, range extends to end of the file.
                        size_sig - start
                    } else {
                        let mut end: i64 = end_str.parse().map_err(|_| ())?;

                        if start > end {
                            return Err(());
                        }

                        if end >= size_sig {
                            end = size_sig - 1;
                        }

                        end - start + 1
                    };

                    Ok(Some(HttpRange {
                        start: start as u64,
                        length: length as u64,
                    }))
                }
            })
            .collect::<Result<_, _>>()
            .map_err(|_| Error::Other(String::from("InvalidRange")))?;

        let ranges: Vec<HttpRange> = all_ranges.into_iter().flatten().collect();

        if no_overlap && ranges.is_empty() {
            return Err(Error::Other(String::from("InvalidRange")));
        }

        Ok(ranges)
    }
}

pub(crate) enum ChunkedState<T> {
    File(Option<T>),
    Future(tokio::task::JoinHandle<Result<(T, Bytes), IoError>>),
}

pub struct FileChunk<T> {
    chunk_size: u64,
    read_size: u64,
    buffer_size: u64,
    offset: u64,
    state: ChunkedState<T>,
}

impl<T> Stream for FileChunk<T>
where
    T: Read + Seek + Unpin + Send + 'static,
{
    type Item = Result<Bytes, IoError>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
        if self.chunk_size == self.read_size {
            return Poll::Ready(None);
        }

        match self.state {
            ChunkedState::File(ref mut file) => {
                let mut file = file
                    .take()
                    .expect("ChunkedReadFile polled after completion");
                let max_bytes = cmp::min(
                    self.chunk_size.saturating_sub(self.read_size),
                    self.buffer_size,
                ) as usize;
                let offset = self.offset;
                let fut = tokio::task::spawn_blocking(move || {
                    let mut buf = Vec::with_capacity(max_bytes);
                    file.seek(std::io::SeekFrom::Start(offset))?;
                    let bytes = file.by_ref().take(max_bytes as u64).read_to_end(&mut buf)?;
                    if bytes == 0 {
                        return Err(ErrorKind::UnexpectedEof.into());
                    }
                    Ok((file, Bytes::from(buf)))
                });

                self.state = ChunkedState::Future(fut);
                self.poll_next(cx)
            }
            ChunkedState::Future(ref mut fut) => {
                let (file, bytes) = ready!(Pin::new(fut).poll(cx))
                    .map_err(|_| IoError::new(ErrorKind::Other, "BlockingErr"))??;
                self.state = ChunkedState::File(Some(file));

                self.offset += bytes.len() as u64;
                self.read_size += bytes.len() as u64;

                Poll::Ready(Some(Ok(bytes)))
            }
        }
    }
}