/*
 * Copyright 2020 Damian Peckett <damian@pecke.tt>.
 * Copyright 2013-2018 Docker, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

//! A modern async/await Docker client written in Rust. Ported directly from the reference
//! Golang implementation.
//!
//! # Examples
//!
//! ```
//! use docker_client_async::{opts, LocalDockerEngineClient};
//! use docker_client_async::error::Error;
//!
//! #[tokio::main]
//! async fn main() -> Result<(), Error> {
//!   let cli = LocalDockerEngineClient::new_client_with_opts(Some(vec![Box::new(opts::from_env)]))?;
//!   for container in cli.container_list(None).await? {
//!     println!("{} {}", container.id, container.image);
//!   }
//!   Ok(())
//! }
//! ```

use crate::error::*;
use crate::opts::DockerEngineClientOption;
use bytes::Bytes;
use futures::ready;
use futures::task::{Context, Poll};
use hyper::body::HttpBody;
use hyper::client::connect::Connect;
use hyper::{Body, Client, Response, Uri};
use hyperlocal::UnixConnector;
use lazy_static::lazy_static;
use percent_encoding::{utf8_percent_encode, NON_ALPHANUMERIC};
use regex::Regex;
use snafu::ResultExt;
use std::collections::HashMap;
use std::io;
use std::sync::Mutex;
use std::time::Duration;
use tokio::io::AsyncRead;
use tokio::macros::support::Pin;
use tokio::time::timeout;

pub mod container;
pub mod error;
pub mod network;
pub mod opts;
pub mod types;
pub mod version;
pub mod volume;

lazy_static! {
    static ref HEADER_REGEXP: Regex = Regex::new(r"\ADocker/.+\s\((.+)\)\z").unwrap();
}

/// A DockerEngineClient for communicating over a local unix domain socket.
pub type LocalDockerEngineClient = DockerEngineClient<UnixConnector>;

/// DockerEngineClient is the API client that performs all operations
/// against a docker server.
#[derive(Clone, Debug)]
pub struct DockerEngineClient<C: Connect + Clone + Send + Sync + 'static> {
    /// scheme sets the scheme for the client.
    pub(crate) scheme: Option<String>,
    /// host holds the server address to connect to.
    pub(crate) host: Option<String>,
    /// proto holds the client protocol i.e. unix.
    pub(crate) proto: Option<String>,
    /// addr holds the client address.
    pub(crate) addr: Option<String>,
    /// base_path holds the path to prepend to the requests.
    pub(crate) base_path: Option<String>,
    /// timeout holds the http client request timeout.
    pub(crate) timeout: Duration,
    // client used to send and receive http requests.
    pub(crate) client: Option<Client<C, Body>>,
    // version of the server to talk to.
    pub(crate) version: String,
    /// custom http headers configured by users.
    pub(crate) custom_http_headers: Option<HashMap<String, String>>,
    /// manualOverride is set to true when the version was set by users.
    pub(crate) manual_override: bool,
    /// negotiateVersion indicates if the client should automatically negotiate
    /// the API version to use when making requests. API version negotiation is
    /// performed on the first request, after which negotiated is set to "true"
    /// so that subsequent requests do not re-negotiate.
    pub(crate) negotiate_version: bool,
    /// negotiated indicates that API version negotiation took place.
    pub(crate) negotiated: bool,
}

impl<C: Connect + Clone + Send + Sync + 'static> Default for DockerEngineClient<C> {
    fn default() -> Self {
        Self {
            scheme: Some("http".into()),
            host: Some("unix:///var/run/docker.sock".into()),
            proto: Some("unix".into()),
            addr: Some("/var/run/docker.sock".into()),
            base_path: None,
            timeout: Duration::from_millis(5_000),
            client: None,
            version: "1.40".into(),
            custom_http_headers: None,
            manual_override: false,
            negotiate_version: false,
            negotiated: false,
        }
    }
}

impl<C: Connect + Clone + Send + Sync + 'static> DockerEngineClient<C> {
    pub fn new_client_with_opts(
        options: Option<Vec<DockerEngineClientOption<C>>>,
    ) -> Result<DockerEngineClient<C>, Error> {
        let mut client: DockerEngineClient<C> = Default::default();
        if let Some(client_options) = options {
            for client_option in &client_options {
                client_option(&mut client)?;
            }
        }
        Ok(client)
    }

    /// Construct a hyper Uri for a given request.
    fn request_uri(
        &self,
        path: &str,
        query_params: Option<HashMap<String, String>>,
    ) -> Result<Uri, Error> {
        let encoded_query_params = query_params
            .map(|query_params| {
                let mut encoded_query_params = String::from("?");
                for (key, value) in query_params {
                    encoded_query_params.push_str(&key);
                    encoded_query_params.push('=');
                    encoded_query_params
                        .push_str(&utf8_percent_encode(&value, NON_ALPHANUMERIC).to_string());
                    encoded_query_params.push('&');
                }
                encoded_query_params.trim_end_matches('&').to_string()
            })
            .unwrap_or_else(String::new);

        let path_and_query_params = format!(
            "{}{}{}",
            self.base_path
                .as_ref()
                .map(String::from)
                .unwrap_or_else(|| format!("/v{}", self.version)),
            path,
            encoded_query_params
        );

        Ok(match self.proto.as_ref().unwrap().as_str() {
            "tcp" => {
                let scheme = self
                    .scheme
                    .as_ref()
                    .map(String::from)
                    .unwrap_or_else(|| "http".to_string());
                let address = self
                    .addr
                    .as_ref()
                    .map(String::from)
                    .unwrap_or_else(|| "localhost".to_string());
                format!("{}://{}{}", scheme, address, path_and_query_params)
                    .parse()
                    .unwrap()
            }
            "unix" => {
                hyperlocal::Uri::new(self.addr.as_ref().unwrap(), &path_and_query_params).into()
            }
            _ => unimplemented!(),
        })
    }
}

/// Read all http response chunks and concatenate into a string.
pub(crate) async fn read_response_body(
    mut response: Response<Body>,
    read_timeout: Duration,
) -> Result<String, Error> {
    let mut response_body = String::new();
    while let Some(chunk) = timeout(read_timeout, response.body_mut().data())
        .await
        .context(HttpClientTimeoutError {})?
    {
        response_body.push_str(&String::from_utf8_lossy(
            &chunk.context(HttpClientError {})?,
        ));
    }
    Ok(response_body)
}

/// Read all http response chunks and concatenate into a raw vector.
pub(crate) async fn read_response_body_raw(
    mut response: Response<Body>,
    read_timeout: Duration,
) -> Result<Vec<u8>, Error> {
    let mut response_body = Vec::new();
    while let Some(chunk) = timeout(read_timeout, response.body_mut().data())
        .await
        .context(HttpClientTimeoutError {})?
    {
        response_body.append(&mut chunk.context(HttpClientError {})?.to_vec());
    }
    Ok(response_body)
}

/// parse_host_url parses a url string, validates the string is a host url, and
/// returns the parsed URL.
pub(crate) fn parse_host_url(host: &str) -> Result<Uri, Error> {
    host.parse::<Uri>().context(HttpUriError {})
}

/// get_docker_os returns the operating system based on the server header from the daemon.
pub(crate) fn get_docker_os(server_header: &str) -> String {
    let captures = HEADER_REGEXP.captures(server_header).unwrap();
    captures
        .get(1)
        .map(|m| String::from(m.as_str()))
        .unwrap_or_else(String::new)
}

/// get_filters_query returns a url query with "filters" query term, based on the
/// filters provided.
pub(crate) fn get_filters_query(
    filters: types::filters::Args,
) -> Result<HashMap<String, String>, Error> {
    let mut query_params: HashMap<String, String> = HashMap::new();
    if !filters.fields.is_empty() {
        query_params.insert(
            "filters".into(),
            serde_json::to_string(&filters.fields).context(JsonSerializationError {})?,
        );
    }
    Ok(query_params)
}

pub(crate) struct AsyncHttpBodyReader {
    body: Mutex<Pin<Box<dyn HttpBody<Data = Bytes, Error = hyper::Error>>>>,
}

impl AsyncHttpBodyReader {
    pub(crate) fn new(response: Response<Body>) -> Self {
        Self {
            body: Mutex::new(Box::pin(response.into_body())),
        }
    }
}

impl AsyncRead for AsyncHttpBodyReader {
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut [u8],
    ) -> Poll<Result<usize, std::io::Error>> {
        let mut body_reader = self.body.lock().unwrap();
        if let Some(data) = ready!(body_reader.as_mut().poll_data(cx)) {
            match data {
                Ok(data) => Poll::Ready(io::Read::read(
                    &mut Vec::from(data.as_ref()).as_mut_slice().as_ref(),
                    buf,
                )),
                Err(err) => Poll::Ready(Err(io::Error::new(
                    io::ErrorKind::Other,
                    format!("{}", err),
                ))),
            }
        } else {
            Poll::Ready(Ok(0))
        }
    }
}

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

    #[test]
    pub fn test_get_docker_os() {
        let os = get_docker_os("Docker/19.03.5 (linux)");
        assert_eq!(&os, "linux");
    }
}