// License: see LICENSE file at root directory of `master` branch

//! # Server

use std::{
    collections::HashMap,
    io::{self, BufReader, Error, ErrorKind, Read, Write},
    mem,
    net::{SocketAddr, TcpListener, TcpStream},
    sync::{Arc, Mutex},
    thread::{self, JoinHandle},
    time::Duration,
};

use blackhole::{Blackhole, Job};
use zeros::Zeros;

use namaste::Uid;

type ZeroMap = Arc<Mutex<HashMap<Zeros, ClientInfo>>>;

/// # Default read/write timeout
pub const DEFAULT_RW_TIMEOUT: Duration = Duration::from_secs(3);

const JOB_QUEUE_LIMIT: usize = 65_535;

/// # Client info
#[derive(Clone)]
struct ClientInfo {

    /// # Port
    port: u16,

    /// # Handshake UID
    handshake_uid: Uid,

}

/// # Server
pub struct Server {

    /// # Server handler
    handler: JoinHandle<()>,

}

impl Server {

    /// # Starts server
    pub fn start(port: Option<u16>, rw_timeout: Duration, active_job_limit: usize) -> io::Result<Self> {
        let server = TcpListener::bind(SocketAddr::from((namaste::DEFAULT_IP, port.unwrap_or(namaste::server::DEFAULT_PORT))))?;
        Ok(Self {
            handler: spawn_server_handler(server, rw_timeout, active_job_limit)?,
        })
    }

    /// # Joins
    ///
    /// Calls [`JoinHandle::join()`][JoinHandle::join()] on internal thread.
    ///
    /// ## Panics
    ///
    /// [`JoinHandle::join()`][JoinHandle::join()] might panic.
    ///
    /// [JoinHandle::join()]: https://doc.rust-lang.org/std/thread/struct.JoinHandle.html#method.join
    pub fn join(self) -> io::Result<()> {
        self.handler.join().map_err(|e| Error::new(ErrorKind::Other, format!("{:?}", e)))
    }

}

/// # Spawns server handler
fn spawn_server_handler(server: TcpListener, rw_timeout: Duration, active_job_limit: usize) -> io::Result<JoinHandle<()>> {
    let zero_map = Arc::new(Mutex::new(HashMap::with_capacity(JOB_QUEUE_LIMIT)));
    let blackhole = Blackhole::make(active_job_limit.into(), JOB_QUEUE_LIMIT)?;
    Ok(thread::spawn(move || {
        for stream in server.incoming() {
            let stream = match stream {
                Ok(stream) => stream,
                Err(err) => {
                    eprintln!("Failed accepting new client: {}", err);
                    continue;
                },
            };

            if let Err(err) = stream.set_read_timeout(Some(rw_timeout)) {
                eprintln!("[{:?}] Failed setting read timeout: {}", &stream, err);
            }
            if let Err(err) = stream.set_write_timeout(Some(rw_timeout)) {
                eprintln!("[{:?}] Failed setting write timeout: {}", &stream, err);
            }

            match blackhole.throw(ClientHandler {
                client: stream,
                zero_map: zero_map.clone(),
                rw_timeout,
            }) {
                Ok(job) => if let Some(ClientHandler { client, .. }) = job {
                    eprintln!("Job queue is full, dropping new client: {:?}", client);
                },
                Err(err) => {
                    eprintln!("Failed handling new client because of internal error: {:?}.\nStopping server...", err);
                    break;
                },
            };
        }

        if let Err(err) = blackhole.escape_on_idle() {
            eprintln!("Failed stopping server: {}", err);
        }
    }))
}

/// # Client handler
struct ClientHandler {

    /// # Client
    client: TcpStream,

    /// # Zero map
    zero_map: ZeroMap,

    /// # Read/write timeout
    rw_timeout: Duration,

}

impl ClientHandler {

    /// # Handles BOOK command
    fn handle_book_cmd(&mut self, uid: Uid, handshake_uid: Uid, local_server_port: u16) {
        let new_client_info = ClientInfo {
            port: local_server_port,
            handshake_uid,
        };

        let zeros = Zeros::from(uid);
        let current_client_info = match self.zero_map.lock() {
            Ok(zero_map) => zero_map.get(&zeros).map(|i| i.clone()),
            Err(err) => {
                eprintln!("Zero map is poisoned: {}", err);
                if let Err(err) = write_response(&mut self.client, false) {
                    eprintln!("[{:?}] Failed writing response: {}", self.client, err);
                }
                return;
            },
        };
        let response = match current_client_info {
            Some(current_client_info) => match self.verify_client_info(&uid, &current_client_info) {
                true => false,
                false => self.verify_client_info(&uid, &new_client_info) && self.insert_new_client_info(zeros, new_client_info),
            },
            None => self.verify_client_info(&uid, &new_client_info) && self.insert_new_client_info(zeros, new_client_info),
        };
        if let Err(err) = write_response(&mut self.client, response) {
            eprintln!("[{:?}] Failed writing response: {}", self.client, err);
            match self.zero_map.lock() {
                Ok(mut zero_map) => drop(zero_map.remove(&Zeros::from(uid))),
                Err(err) => eprintln!("Zero map is poisoned: {}", err),
            };
        }
    }

    /// # Handles CHECK-OUT command
    fn handle_check_out_cmd(&mut self, uid: Uid, handshake_uid: Uid) {
        let response = match self.zero_map.lock() {
            Ok(mut zero_map) => {
                let zeros = Zeros::from(uid);
                match zero_map.get(&zeros).map(|info| namaste::cmp_uids(&info.handshake_uid, &handshake_uid)).unwrap_or(false) {
                    true => zero_map.remove(&zeros).is_some(),
                    false => false,
                }
            },
            Err(err) => {
                eprintln!("Zero map is poisoned: {}", err);
                false
            },
        };

        if let Err(err) = write_response(&mut self.client, response) {
            eprintln!("[{:?}] Failed writing response: {}", self.client, err);
        }
    }

    /// # Verifies client info
    fn verify_client_info(&mut self, uid: &Uid, client_info: &ClientInfo) -> bool {
        let mut stream = match TcpStream::connect_timeout(&SocketAddr::from((namaste::DEFAULT_IP, client_info.port)), self.rw_timeout) {
            Ok(stream) => stream,
            Err(_) => return false,
        };

        if let Err(err) = stream.set_read_timeout(Some(self.rw_timeout)) {
            eprintln!("[{:?}] Failed setting read timeout: {}", &stream, err);
        }
        if let Err(err) = stream.set_write_timeout(Some(self.rw_timeout)) {
            eprintln!("[{:?}] Failed setting write timeout: {}", &stream, err);
        }

        if stream.write_all(&client_info.handshake_uid).is_err() {
            return false;
        }
        match namaste::read_uid(&mut stream) {
            Ok(msg) => namaste::cmp_uids(uid, &msg),
            Err(_) => false,
        }
    }

    /// # Inserts new client info
    fn insert_new_client_info(&self, zeros: Zeros, client_info: ClientInfo) -> bool {
        match self.zero_map.lock() {
            Ok(mut zero_map) => {
                zero_map.insert(zeros, client_info);
                true
            },
            Err(err) => {
                eprintln!("Zero map is poisoned: {}", err);
                false
            },
        }
    }

}

impl Job for ClientHandler {

    fn run(&mut self) -> Option<Box<Job>> {
        match read_command(&mut self.client) {
            Ok(Cmd::Book { uid, handshake_uid, local_server_port }) => self.handle_book_cmd(uid, handshake_uid, local_server_port),
            Ok(Cmd::CheckOut { uid, handshake_uid }) => self.handle_check_out_cmd(uid, handshake_uid),
            Err(err) => eprintln!("[{:?}] Failed reading command: {}", self.client, err),
        };
        None
    }

}

/// # Command
enum Cmd {

    /// # Book
    Book {

        /// # UID
        uid: Uid,

        /// # Handshake UID
        handshake_uid: Uid,

        /// # Local server port
        local_server_port: u16,

    },

    /// # CheckOut
    CheckOut {

        /// # UID
        uid: Uid,

        /// # Handshake UID
        handshake_uid: Uid,

    },

}

/// # Reads command
fn read_command(stream: &mut TcpStream) -> io::Result<Cmd> {
    const UID_LEN: usize = mem::size_of::<Uid>();

    let mut reader = BufReader::new(stream);

    let is_book_cmd = {
        let mut buf = [0];
        reader.read_exact(&mut buf)?;
        match buf[0] {
            namaste::server::CMD_BOOK => true,
            namaste::server::CMD_CHECK_OUT => false,
            other => return Err(Error::new(ErrorKind::InvalidData, format!("Invalid command: {:?}", other))),
        }
    };

    let mut uid = [0; UID_LEN];
    reader.read_exact(&mut uid)?;

    let mut handshake_uid = [0; UID_LEN];
    reader.read_exact(&mut handshake_uid)?;

    match is_book_cmd {
        true => {
            let mut buf = [0; mem::size_of::<u16>()];
            reader.read_exact(&mut buf).map(|()| Cmd::Book {
                uid,
                handshake_uid,
                local_server_port: u16::from_be_bytes(buf),
            })
        },
        false => Ok(Cmd::CheckOut {
            uid,
            handshake_uid,
        })
    }
}

/// # Writes response
fn write_response(stream: &mut TcpStream, result: bool) -> io::Result<()> {
    stream.write_all(&[result as u8])?;
    Ok(())
}
