use async_std::net::{SocketAddr, TcpStream};
use async_std::sync::{Arc, Mutex, RwLock};
use chrono::{Duration, NaiveDateTime, Utc};
use futures::io::BufReader;
use futures::io::{AsyncBufReadExt, AsyncWriteExt};
use futures::io::{ReadHalf, WriteHalf};
use log::{info, warn};
use serde::Serialize;
use std::time::SystemTime;
use stratum_types::params::{Params, Results, SetDiff};
use stratum_types::traits::{
    AuthManager, Authorize, AuthorizeResult, BlockValidator, DataProvider, Notify, StratumManager,
    SubscribeResult,
};
use stratum_types::traits::{PoolParams, StratumParams};
use stratum_types::Result;
use stratum_types::{
    Error, MinerAuth, MinerInfo, MinerJobStats, Request, Response, StratumError, StratumMethod,
    StratumPacket, ID,
};
use uuid::Uuid;

#[derive(PartialEq, Debug)]
pub enum State {
    Connected,
    Disconnect,
}

//@todo Review which of these we need Mutexs/Arcs/etc. Might be over indulging here.
#[derive(Debug)]
pub struct Connection<SM: StratumManager> {
    pub id: ID,
    pub write_half: Arc<Mutex<WriteHalf<TcpStream>>>,
    pub read_half: Arc<Mutex<BufReader<ReadHalf<TcpStream>>>>,
    pub authorized: Arc<Mutex<bool>>,
    pub data_provider: Arc<SM::DataProvider>,
    pub block_validator: Arc<SM::BlockValidator>,
    pub auth_manager: Arc<SM::AuthManager>,
    pub session_start: SystemTime,
    pub state: Mutex<State>,
    pub subscribed: Arc<Mutex<bool>>,
    pub subscriber_id: Arc<Mutex<String>>,
    pub miner_info: Arc<RwLock<MinerInfo>>,
    //Possibly pull these out into their own var.
    //Makes it easier to operate on them.
    pub difficulty: Arc<Mutex<f64>>,
    pub submissions: Arc<Mutex<u64>>,
    pub last_retarget: Arc<Mutex<SystemTime>>,
    pub next_difficulty: Arc<Mutex<f64>>,
    pub job_stats: Arc<Mutex<JobStats>>,
    pub options: Arc<MinerOptions>,
    pub stats: Arc<Mutex<MinerStats>>,
    pub needs_ban: Arc<Mutex<bool>>,
    pub var_diff: bool,
    pub ban_stats: Arc<Mutex<BanStats>>,
    pub classic: Arc<Mutex<bool>>,
    pub last_message_id: Arc<Mutex<ID>>,
}

//@todo probably move these over to types.
#[derive(Debug, Default)]
pub struct JobStats {
    last_share_timestamp: i64,
    last_retarget: i64,
    times: Vec<i64>,
    current_difficulty: f64,
    job_difficulty: f64,
}

//@todo this should probably come from builder pattern
#[derive(Debug, Default)]
pub struct MinerOptions {
    retarget_time: i64,
    target_time: f64,
    min_diff: f64,
    max_delta: f64,
    variance_percent: u32,
    share_time_min: f64,
    share_time_max: f64,
}

#[derive(Debug, Clone)]
pub struct MinerStats {
    accepted_shares: u64,
    rejected_shares: u64,
    last_active: NaiveDateTime,
}

#[derive(Debug)]
pub struct BanStats {
    accepted_shares: u64,
    rejected_shares: u64,
    last_active: NaiveDateTime,
}

impl<SM> Connection<SM>
where
    SM: StratumManager,
{
    pub fn new(
        addr: SocketAddr,
        rh: BufReader<ReadHalf<TcpStream>>,
        wh: WriteHalf<TcpStream>,
        data_provider: Arc<SM::DataProvider>,
        auth_manager: Arc<SM::AuthManager>,
        block_validator: Arc<SM::BlockValidator>,
        var_diff: bool,
        initial_difficulty: f64,
    ) -> Self {
        //@todo could store this as a UUID type as well.
        let id = Uuid::new_v4().to_string();

        info!("Accepting new miner. ID: {}", &id);

        let info = MinerInfo {
            ip: addr.ip(),
            auth: None,
            sid: None,
            job_stats: None,
            worker_name: None,
        };

        let options = MinerOptions {
            retarget_time: 120,
            target_time: 6.0,
            min_diff: 0.0001,
            max_delta: 1.0, //@todo make this adjustable, not sure if this is solid or not.
            //@todo probably don't store, get from above and then calcualte the others.
            variance_percent: 30,
            share_time_min: 4.2,
            share_time_max: 7.8,
        };

        //Make this an impl on stats same for each above. That way we can just do minerstats::new(current_time)
        let stats = MinerStats {
            accepted_shares: 0,
            rejected_shares: 0,
            last_active: Utc::now().naive_utc(),
        };

        let ban_stats = BanStats {
            accepted_shares: 0,
            rejected_shares: 0,
            last_active: Utc::now().naive_utc(),
        };

        Connection {
            id: ID::Str(id),
            write_half: Arc::new(Mutex::new(wh)),
            read_half: Arc::new(Mutex::new(rh)),
            authorized: Arc::new(Mutex::new(false)),
            data_provider,
            auth_manager,
            block_validator,
            session_start: SystemTime::now(),
            state: Mutex::new(State::Connected),
            subscribed: Arc::new(Mutex::new(false)),
            //@todo this should be passed in.
            difficulty: Arc::new(Mutex::new(initial_difficulty)),
            subscriber_id: Arc::new(Mutex::new(String::new())),
            miner_info: Arc::new(RwLock::new(info)),
            submissions: Arc::new(Mutex::new(0)),
            last_retarget: Arc::new(Mutex::new(SystemTime::now())),
            next_difficulty: Arc::new(Mutex::new(0.0)),
            job_stats: Arc::new(Mutex::new(Default::default())),
            options: Arc::new(options),
            stats: Arc::new(Mutex::new(stats)),
            var_diff,
            needs_ban: Arc::new(Mutex::new(false)),
            ban_stats: Arc::new(Mutex::new(ban_stats)),
            classic: Arc::new(Mutex::new(false)),
            last_message_id: Arc::new(Mutex::new(ID::Str(String::from("")))),
        }
    }

    pub async fn start(&self) -> Result<()> {
        loop {
            if *self.state.lock().await == State::Disconnect {
                break;
            }

            let msg: StratumPacket<SM::PoolParams, SM::StratumParams> = self.next_message().await?;

            match msg {
                StratumPacket::Request(req) => self.handle_requests(req).await?,
                StratumPacket::Response(res) => self.handle_responses(res).await?,
            };
        }

        Ok(())
    }

    pub async fn handle_requests(
        &self,
        req: Request<SM::PoolParams, SM::StratumParams>,
    ) -> Result<()> {
        if let Params::Authorize(auth) = &req.params {
            if req.method == StratumMethod::Authorize {
                return Ok(self.handle_authorize(auth, false).await?);
            } else if req.method == StratumMethod::ClassicAuthorize {
                *self.classic.lock().await = true;
                *self.last_message_id.lock().await = req.id;
                return Ok(self.handle_authorize(auth, true).await?);
            }
        } else if let Params::Subscribe(sub) = &req.params {
            if req.method == StratumMethod::Subscribe {
                return Ok(self.handle_subscribe(sub).await?);
            } else if req.method == StratumMethod::ClassicSubscribe {
                *self.classic.lock().await = true;
                *self.last_message_id.lock().await = req.id;
                return Ok(self.handle_subscribe(sub).await?);
            }
        }

        if *self.classic.lock().await {
            *self.last_message_id.lock().await = req.id;
        }

        let authorized = self.authorized.lock().await;
        if !*authorized {
            //This would be a huge ban
            // return Err(Error::;
            //@todo needs to return an error here.
            return Ok(());
        }

        info!("Got new param message");

        dbg!(&req.method);

        match &req.params {
            Params::Submit(value) => self.handle_submit(value).await?,
            Params::Subscribe(value) => self.handle_subscribe(value).await?,
            Params::Unknown(value) => self.handle_unknown(value).await?,
            Params::Authorize(_) => {}
            _ => {}
        };
        Ok(())
    }

    pub async fn handle_responses(&self, _res: Response<SM::PoolParams>) -> Result<()> {
        // if *self.classic.lock().await {
        //     *self.last_message_id.lock().await = res.id;
        // }
        //@todo maybe throw this into a function - "checkAuth" then error is easier to throw
        let authorized = self.authorized.lock().await;
        if !*authorized {
            //This would be a huge ban
            // return Err(Error::;
            //@todo needs to return an error here.
            return Ok(());
        }

        Ok(())
    }

    pub async fn next_message(&self) -> Result<StratumPacket<SM::PoolParams, SM::StratumParams>> {
        //I don't actually think this has to loop here.
        loop {
            let mut stream = self.read_half.lock().await;

            let mut buf = String::new();
            let num_bytes = stream.read_line(&mut buf).await?;

            if num_bytes == 0 {
                self.shutdown().await?;
                return Err(Error::StreamClosed);
            }

            if !buf.is_empty() {
                //@smells
                buf = buf.trim().to_owned();
                //@todo make this a debug.
                info!("Received Message: {}", &buf);
                let msg: StratumPacket<SM::PoolParams, SM::StratumParams> =
                    serde_json::from_str(&buf)?;

                // info!("Received Message with id: {}", msg.id);
                info!("Processing message");

                return Ok(msg);
            };
        }
    }

    async fn send<T: Serialize>(&self, message: T) -> Result<()> {
        let msg = serde_json::to_vec(&message)?;
        let msg_string = serde_json::to_string(&message)?;

        info!("Sending message: {}", msg_string);

        let mut stream = self.write_half.lock().await;

        stream.write_all(&msg).await?;
        stream.write_all(b"\n").await?;

        Ok(())
    }

    //Helper function to make sending requests more simple.
    async fn send_request(
        &self,
        method: StratumMethod,
        params: Params<SM::PoolParams, SM::StratumParams>,
    ) -> Result<()> {
        let id = if *self.classic.lock().await {
            if method == StratumMethod::ClassicNotify {
                ID::Null(serde_json::Value::Null)
            } else {
                self.last_message_id.lock().await.clone()
            }
        } else {
            self.id.clone()
        };

        let request = Request { id, method, params };

        Ok(self.send(request).await?)
    }

    async fn send_response(
        &self,
        method: StratumMethod,
        result: Results<SM::PoolParams>,
    ) -> Result<()> {
        let id = if *self.classic.lock().await {
            self.last_message_id.lock().await.clone()
        } else {
            self.id.clone()
        };

        let response = Response {
            id,
            method,
            result: Some(result),
            error: None,
        };

        Ok(self.send(response).await?)
    }

    async fn send_error(&self, method: StratumMethod, error: StratumError) -> Result<()> {
        let response: Response<SM::PoolParams> = Response {
            id: self.id.clone(),
            method,
            result: None,
            error: Some(error),
        };

        Ok(self.send(response).await?)
    }

    pub async fn shutdown(&self) -> Result<()> {
        *self.state.lock().await = State::Disconnect;

        //Only returning a result here because we might want to add more functionality in the
        //future.
        //Here is where we actually will write upstream to KILL the connection if we are using that
        //proxy.
        Ok(())
    }

    // ===== Handler Functions ===== //
    pub async fn handle_authorize(
        &self,
        auth: &<SM::PoolParams as PoolParams>::Authorize,
        classic: bool,
    ) -> Result<()> {
        let mut method = StratumMethod::Authorize;

        if classic {
            method = StratumMethod::ClassicAuthorize;
        }

        self.miner_info.write().await.worker_name = auth.worker_name();

        let result = self
            .auth_manager
            .authorize(self.miner_info.read().await.clone(), &auth, classic)
            .await;

        match result {
            Ok(auth_result) => {
                // if !classic {
                self.send_response(method, Results::AuthorizeResult(auth_result.clone()))
                    .await?;
                // }

                if auth_result.authorized() {
                    info!("Authorized Miner: {}", &self.id);
                    *self.authorized.lock().await = true;
                    let miner_auth = MinerAuth {
                        username: auth.username(),
                        id: auth_result.id(),
                        client: auth.client(),
                    };

                    self.miner_info.write().await.auth = Some(miner_auth);
                // self.miner_info.write().await.worker_name = auth.worker_name();
                } else {
                    self.shutdown().await?;
                }
            }
            Err(e) => {
                self.send_error(method, e).await?;
            }
        }

        Ok(())
    }

    //@todo can make these this same as below.
    pub async fn send_initial_work(&self) -> Result<()> {
        let mut method = StratumMethod::Notify;

        if *self.classic.lock().await {
            method = StratumMethod::ClassicNotify;
        }

        let job = self
            .data_provider
            .get_job(self.classic.lock().await.clone())
            .await;

        let mut job_stats = self.job_stats.lock().await;
        job_stats.last_share_timestamp = Utc::now().timestamp();
        job_stats.last_retarget = Utc::now().timestamp();
        job_stats.times = Vec::new();
        job_stats.job_difficulty = job.get_difficulty();
        job_stats.current_difficulty = *self.difficulty.lock().await;

        let info_job_stats = MinerJobStats {
            expected_difficulty: job_stats.current_difficulty,
        };

        self.miner_info.write().await.job_stats = Some(info_job_stats);

        self.send_request(method, Params::Notify(job)).await?;

        Ok(())
    }

    //@todo add a rebroadcast work, where we do NOT reset all the stats here
    pub async fn send_work(&self) -> Result<()> {
        //Should be a function called check_auth @todo.
        //If the miner is not authorized, don't send work. Found this bug while testing w/
        //Vaultminer.
        if !*self.authorized.lock().await {
            return Ok(());
        }

        //Don't send work until the miner has actually subscribed
        if !*self.subscribed.lock().await {
            return Ok(());
        }

        let mut method = StratumMethod::Notify;

        if *self.classic.lock().await {
            method = StratumMethod::ClassicNotify;
        }

        let job = self
            .data_provider
            .get_job(self.classic.lock().await.clone())
            .await;
        let job_difficulty = job.get_difficulty();

        let ban_stats = self.ban_stats.lock().await;

        let last_active_ago = Utc::now().naive_utc() - ban_stats.last_active;

        //If the miner has not been active (sending shares) for 5 minutes, we disconnect this dude.
        //@todo before live, check this guy. Also should come from options.
        if last_active_ago > Duration::seconds(600) {
            warn!(
                "Miner not active since {}. Disconnecting",
                ban_stats.last_active
            );
            self.disconnect().await;
            return Ok(());
        }

        self.send_request(method, Params::Notify(job)).await?;

        let mut job_stats = self.job_stats.lock().await;
        job_stats.last_share_timestamp = Utc::now().timestamp();
        job_stats.last_retarget = Utc::now().timestamp();
        job_stats.times = Vec::new();
        job_stats.job_difficulty = job_difficulty;
        job_stats.current_difficulty = *self.difficulty.lock().await;

        let info_job_stats = MinerJobStats {
            expected_difficulty: job_stats.current_difficulty,
        };

        self.miner_info.write().await.job_stats = Some(info_job_stats);

        Ok(())
    }

    pub async fn handle_submit(
        &self,
        share: &<SM::StratumParams as StratumParams>::Submit,
    ) -> Result<()> {
        info!("Received share from miner: {}", &self.id);

        let mut method = StratumMethod::Submit;

        if *self.classic.lock().await {
            method = StratumMethod::ClassicSubmit;
        }

        let result = self
            .block_validator
            .validate_share(
                self.miner_info.read().await.clone(),
                share.clone(),
                self.classic.lock().await.clone(),
            )
            .await;

        match result {
            Ok(valid) => {
                info!("Accepted share from miner: {}", &self.id);

                self.send_response(method, Results::SubmitResult(valid))
                    .await?;

                let mut ban_stats = self.ban_stats.lock().await;

                ban_stats.accepted_shares += 1;
                ban_stats.last_active = Utc::now().naive_utc();

                if self.var_diff {
                    //@todo come back to if we want this or not.
                    let mut job_stats = self.job_stats.lock().await;
                    //self.stats.valid_shares += 1;
                    //@todo make sure this is initialized in notify.
                    let now = Utc::now().timestamp();
                    let duration_since_last_share = now - job_stats.last_share_timestamp;

                    job_stats.times.push(duration_since_last_share);
                    job_stats.last_share_timestamp = now;

                    let time_total: i64 = job_stats.times.iter().sum();
                    let avg = time_total as f64 / job_stats.times.len() as f64;

                    if now - job_stats.last_retarget >= self.options.retarget_time
                        || avg < self.options.share_time_min && avg > self.options.share_time_max
                    {
                        self.retarget(avg, &mut job_stats).await?
                    }
                }
            }
            Err(e) => {
                info!("Rejecting share from miner: {}. Reason: {}", &self.id, e);
                self.send_error(method, e).await?;

                let mut ban_stats = self.ban_stats.lock().await;

                ban_stats.rejected_shares += 1;
                ban_stats.last_active = Utc::now().naive_utc();

                //Banning logic.
                let total_shares = ban_stats.accepted_shares + ban_stats.rejected_shares;
                //Come from options @todo.
                let check_threshold = 500;
                let invalid_percent = 50.0;
                if total_shares >= check_threshold {
                    let percent_bad =
                        (ban_stats.rejected_shares as f64 / total_shares as f64) * 100.0;
                    if percent_bad > invalid_percent {
                        warn!(
                            "{} out of the last {} shares were rejected. Banning miner: {}",
                            ban_stats.rejected_shares, total_shares, self.id
                        );
                        self.ban().await;
                        self.disconnect().await;
                    } else {
                        ban_stats.accepted_shares = 0;
                        ban_stats.rejected_shares = 0;
                    }
                }
            }
        }

        Ok(())
    }

    pub async fn handle_subscribe(
        &self,
        subscribe: &<SM::PoolParams as PoolParams>::Subscribe,
    ) -> Result<()> {
        let result = self
            .auth_manager
            .subscribe(
                self.miner_info.read().await.clone(),
                subscribe,
                self.classic.lock().await.clone(),
            )
            .await;

        info!("Handling subscribe message");

        match result {
            Ok(sub_info) => {
                //@todo remove this I think. since we have it in miner_info.
                *self.subscriber_id.lock().await = sub_info.id();

                *self.subscribed.lock().await = true;

                self.miner_info.write().await.sid = Some(sub_info.id());

                //2. Send the miner Subsription ID aka SubscribeResult response. save that the miner is
                //   subcribed
                if *self.classic.lock().await {
                    //Classic requires a bit of a hack.
                    let mut stream = self.write_half.lock().await;
                    let msg = format!(
                        "{{\"id\":\"{1}\",\"result\":[[[\"mining.set_difficulty\",\"{0}\"],[\"mining.notify\",\"{0}\"]],\"{0}\",24],\"error\":null}}",
                        sub_info.id(), *self.last_message_id.lock().await
                    );
                    info!("Sending message: {}", msg.clone());
                    stream.write_all(&msg.into_bytes()).await?;
                    stream.write_all(b"\n").await?;
                    drop(stream);
                } else {
                    self.send_response(
                        StratumMethod::Subscribe,
                        Results::SubscribeResult(sub_info.clone()),
                    )
                    .await?;
                }

                info!("Subscribed Miner: {} with SID: {}", &self.id, sub_info.id());

                //3. Send the miner Set Difficulty and save it internally
                self.set_difficulty(*self.difficulty.lock().await).await?;

                //4. Send the miner the current work
                self.send_initial_work().await?;
            }
            Err(e) => {
                self.send_error(StratumMethod::Subscribe, e).await?;
            }
        }

        Ok(())
    }

    //This function sets the difficulty for this connection, and sends a set difficulty message to
    //the client.
    async fn set_difficulty(&self, difficulty: f64) -> Result<()> {
        let method = StratumMethod::SetDifficulty;

        if *self.classic.lock().await {
            // method = StratumMethod::ClassicSetDifficulty;
            let mut stream = self.write_half.lock().await;
            let msg = format!(
                "{{\"id\":null,\"method\":\"mining.set_difficulty\",\"params\":[{}]}}",
                difficulty.floor() as u64
            );
            info!("Sending message: {}", msg.clone());
            stream.write_all(&msg.into_bytes()).await?;
            stream.write_all(b"\n").await?;
            drop(stream);
        } else {
            self.send_request(method, Params::SetDifficulty(SetDiff { difficulty }))
                .await?;
        }

        Ok(())
    }

    //Can probably not send avg here either.
    async fn retarget(&self, avg: f64, stats: &mut JobStats) -> Result<()> {
        // let mut stats = self.job_stats.lock().await;

        let mut new_difficulty = stats.current_difficulty * (self.options.target_time / avg);

        let delta = (new_difficulty - stats.current_difficulty).abs();

        if delta > self.options.max_delta {
            if new_difficulty > stats.current_difficulty {
                //@smells come back here later.
                new_difficulty = new_difficulty - (delta - self.options.max_delta);
            } else if new_difficulty < stats.current_difficulty {
                new_difficulty = new_difficulty + (delta - self.options.max_delta);
            }
        }

        if new_difficulty < self.options.min_diff {
            new_difficulty = self.options.min_diff;
        } else if new_difficulty > stats.job_difficulty {
            new_difficulty = stats.job_difficulty;
        }

        if new_difficulty < stats.current_difficulty || new_difficulty > stats.current_difficulty {
            stats.last_retarget = Utc::now().timestamp();

            //Clear some of the stats.
            stats.times = Vec::new();
            stats.current_difficulty = new_difficulty;
            let job_stats = MinerJobStats {
                expected_difficulty: new_difficulty,
            };
            self.miner_info.write().await.job_stats = Some(job_stats);

            self.set_difficulty(stats.current_difficulty).await?;
        }

        Ok(())
    }

    //Unimplemented - Probably just log the value, and see what's going on.
    //Make a handle_unknown function that exists in stratum manager - then the pool can decide what
    //to do.
    pub async fn handle_unknown(&self, _msg: &serde_json::Value) -> Result<()> {
        Ok(())
    }

    pub async fn disconnect(&self) {
        *self.state.lock().await = State::Disconnect;
    }

    pub async fn ban(&self) {
        *self.needs_ban.lock().await = true;
    }

    pub async fn needs_ban(&self) -> bool {
        *self.needs_ban.lock().await
    }

    pub async fn get_stats(&self) -> MinerStats {
        self.stats.lock().await.clone()
    }
}
