use crate::server::VarDiffConfig;
use crate::{Result, ID};
use async_std::net::SocketAddr;
use async_std::sync::{Arc, Mutex, RwLock};
use chrono::{NaiveDateTime, Utc};
use encodings::{FromHex, ToHex};
use extended_primitives::Buffer;
use futures::channel::mpsc::{UnboundedReceiver, UnboundedSender};
use futures::SinkExt;
use futures::StreamExt;
use log::{debug, info, warn};
use rand::prelude::*;
use serde::Serialize;
use std::net::IpAddr;
use std::time::SystemTime;
use stop_token::{StopSource, StopToken};
use uuid::Uuid;

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

//@todo Review which of these we need Mutexs/Arcs/etc. Might be over indulging here.
//@todo change the ID of this miner to the ID provided by the pool after subscribing/authing is
//done. Can just be a check in each one of those handles if the other one has already been
//completed.
#[derive(Debug)]
pub struct Connection<State> {
    // pub write_half: Arc<Mutex<WriteHalf<TcpStream>>>,
    // pub read_half: Arc<Mutex<BufReader<ReadHalf<TcpStream>>>>,
    pub sender: Arc<Mutex<UnboundedSender<String>>>,
    pub upstream_sender: Arc<Mutex<UnboundedSender<String>>>,
    pub upstream_receiver:
        Arc<Mutex<UnboundedReceiver<serde_json::map::Map<String, serde_json::Value>>>>,
    pub authorized: Arc<Mutex<bool>>,
    pub subscribed: Arc<Mutex<bool>>,
    pub session_start: SystemTime,
    pub connection_state: Mutex<ConnectionState>,
    pub difficulty: Arc<Mutex<f64>>,
    pub stats: Arc<Mutex<MinerStats>>,
    pub needs_ban: Arc<Mutex<bool>>,
    pub next_difficulty: Arc<Mutex<Option<f64>>>,

    //@todo review the rest of these variables here.
    pub id: Uuid,
    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 job_stats: Arc<Mutex<JobStats>>,
    pub options: Arc<MinerOptions>,
    pub var_diff: Arc<Mutex<bool>>,
    pub last_message_id: Arc<Mutex<ID>>,
    pub worker_info: Arc<Mutex<WorkerInfo>>,
    pub state: Arc<Mutex<State>>,
    pub stop_source: Arc<Mutex<Option<StopSource>>>,
    pub stop_token: StopToken,
}

//@todo review this. I'm not sure if this is the best strategy here, but I'd like for the ability
//to pass some information to the implementer of traits. I.E. For logging in, the implementer
//probably wants the IP of the connection that is attempting to log in, so that they can rate limit
//or ban. Further, for share submitting and other features, the implementer probably wants the same
//ability. Rather than pass the entire stream into the trait functions, I figured we should build a
//struct that holds some information about the miner that can be edited and passed into those
//functions.
#[derive(Clone, Debug)]
pub struct MinerInfo {
    pub ip: IpAddr,
    pub auth: Option<MinerAuth>,
    pub id: Option<Uuid>,
    pub sid: Option<String>,
    pub job_stats: Option<MinerJobStats>,
    pub name: Option<String>,
}

#[derive(Clone, Debug)]
pub struct MinerAuth {
    pub id: String,
    pub username: String,
    pub client: String,
}

#[derive(Clone, Debug)]
pub struct MinerJobStats {
    pub expected_difficulty: f64,
}

//@todo probably move these over to types.
//@todo maybe rename this as vardiff stats.
#[derive(Debug)]
pub struct JobStats {
    last_timestamp: i64,
    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_diff: f64,
    max_delta: f64,
    variance_percent: f64,
    // 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, Clone, Default)]
pub struct WorkerInfo {
    pub client: Option<String>,
    pub name: Option<String>,
    pub sid: Option<Buffer>,
    pub account_id: i32,
    pub id: Uuid,
}

impl<State: Clone + Send + Sync + 'static> Connection<State> {
    pub fn new(
        addr: SocketAddr,
        sender: UnboundedSender<String>,
        upstream_sender: UnboundedSender<String>,
        upstream_receiver: UnboundedReceiver<serde_json::map::Map<String, serde_json::Value>>,
        initial_difficulty: f64,
        var_diff_config: VarDiffConfig,
        state: State,
        //@todo we should probably kill this, but for now it has to live here to make things
        //easier.
    ) -> Self {
        //@todo could store this as a UUID type as well.
        let id = Uuid::new_v4();

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

        let info = MinerInfo {
            ip: addr.ip(),
            auth: None,
            id: None,
            sid: None,
            job_stats: None,
            //@todo delete this as it's in WorkerInfo.
            name: None,
        };

        let options = MinerOptions {
            retarget_time: var_diff_config.retarget_time,
            target_time: var_diff_config.target_time,
            //@todo these values make no sense so let's trim them a bit.
            min_diff: var_diff_config.minimum_difficulty,
            max_diff: var_diff_config.maximum_difficulty,
            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: var_diff_config.variance_percent,
            // 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 job_stats = JobStats {
            last_share_timestamp: 0,
            last_timestamp: Utc::now().naive_utc().timestamp(),
            //@todo our firs retarget occurs in half the time to speed up the process if the
            //Difficulty starts too low.
            last_retarget: Utc::now().naive_utc().timestamp() - options.retarget_time / 2,
            times: Vec::new(),
            current_difficulty: initial_difficulty,
            job_difficulty: 0.0,
        };

        let stop_source = StopSource::new();
        let stop_token = stop_source.stop_token();

        //@todo clean up the comments here.
        Connection {
            id,
            // write_half: Arc::new(Mutex::new(wh)),
            // read_half: Arc::new(Mutex::new(rh)),
            sender: Arc::new(Mutex::new(sender)),
            upstream_sender: Arc::new(Mutex::new(upstream_sender)),
            upstream_receiver: Arc::new(Mutex::new(upstream_receiver)),
            authorized: Arc::new(Mutex::new(false)),
            session_start: SystemTime::now(),
            connection_state: Mutex::new(ConnectionState::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)),
            next_difficulty: Arc::new(Mutex::new(None)),
            job_stats: Arc::new(Mutex::new(job_stats)),
            options: Arc::new(options),
            stats: Arc::new(Mutex::new(stats)),
            var_diff: Arc::new(Mutex::new(var_diff_config.var_diff)),
            needs_ban: Arc::new(Mutex::new(false)),
            last_message_id: Arc::new(Mutex::new(ID::Str(String::from("")))),
            worker_info: Arc::new(Mutex::new(Default::default())),
            state: Arc::new(Mutex::new(state)),
            stop_source: Arc::new(Mutex::new(Some(stop_source))),
            stop_token,
        }
    }

    pub async fn info(&self) -> MinerInfo {
        self.miner_info.read().await.clone()
    }

    pub async fn is_disconnected(&self) -> bool {
        *self.connection_state.lock().await == ConnectionState::Disconnect
    }

    //pub async fn next_message(
    //    &self,
    //) -> Result<(String, serde_json::map::Map<String, serde_json::Value>)> {
    //    //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();
    //            debug!("Received Message: {}", &buf);
    //            let msg: Map<String, Value> = serde_json::from_str(&buf)?;

    //            let method = if msg.contains_key("method") {
    //                match msg.get("method") {
    //                    Some(method) => method.as_str(),
    //                    //@todo need better stratum erroring here.
    //                    None => return Err(Error::MethodDoesntExist),
    //                }
    //            } else if msg.contains_key("messsage") {
    //                match msg.get("message") {
    //                    Some(method) => method.as_str(),
    //                    //@todo need better stratum erroring here.
    //                    None => return Err(Error::MethodDoesntExist),
    //                }
    //            } else {
    //                return Err(Error::MethodDoesntExist);
    //            };

    //            if let Some(method_string) = method {
    //                //Mark the sender as active as we received a message.
    //                //We only mark them as active if the message/method was valid
    //                self.stats.lock().await.last_active = Utc::now().naive_utc();

    //                return Ok((method_string.to_owned(), msg));
    //            } else {
    //                //@todo improper format
    //                return Err(Error::MethodDoesntExist);
    //            }
    //        };
    //    }
    //}

    //@todo we have disabled last_active for now... Need to reimplement this desperately.
    pub async fn send<T: Serialize>(&self, message: T) -> Result<()> {
        //let last_active = self.stats.lock().await.last_active;

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

        ////@todo rewrite this comment
        ////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.
        ////@todo make the last_active thing a config.
        //if last_active_ago > Duration::seconds(600) {
        //    warn!(
        //        "Miner: {} not active since {}. Disconnecting",
        //        self.id, last_active
        //    );
        //    self.ban().await;
        //    return Ok(());
        //}

        let msg_string = serde_json::to_string(&message)?;

        debug!("Sending message: {}", msg_string.clone());

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

        //@todo this feels inefficient, maybe we do send bytes here.
        sender.send(msg_string).await?;
        // stream.write_all(b"\n").await?;

        Ok(())
    }

    pub async fn upstream_send<T: Serialize>(&self, message: T) -> Result<()> {
        //let last_active = self.stats.lock().await.last_active;

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

        ////@todo rewrite this comment
        ////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.
        ////@todo make the last_active thing a config.
        //if last_active_ago > Duration::seconds(600) {
        //    warn!(
        //        "Miner: {} not active since {}. Disconnecting",
        //        self.id, last_active
        //    );
        //    self.ban().await;
        //    return Ok(());
        //}

        let msg_string = serde_json::to_string(&message)?;

        debug!("Sending message: {}", msg_string.clone());

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

        //@todo this feels inefficient, maybe we do send bytes here.
        upstream_sender.send(msg_string).await?;
        // stream.write_all(b"\n").await?;

        Ok(())
    }

    pub async fn upstream_result(&self) -> Result<bool> {
        let mut upstream_receiver = self.upstream_receiver.lock().await;

        let values = match upstream_receiver.next().await {
            Some(values) => values,
            //@todo return error here.
            None => return Ok(false),
        };

        let result = values["result"].as_bool().unwrap();

        Ok(result)
    }

    pub async fn shutdown(&self) -> Result<()> {
        //@todo this should also have an internal stop_token, and each one of these should get
        //killed?
        *self.connection_state.lock().await = ConnectionState::Disconnect;

        //This will kill everything that uses are own stop_token.
        //@todo might be able to get rid of connection state with this.
        *self.stop_source.lock().await = None;

        //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(())
    }

    // "varDiff": {
    //             "minDiff": 8, //Minimum difficulty
    //             "maxDiff": 512, //Network difficulty will be used if it is lower than this
    //             "targetTime": 15, //Try to get 1 share per this many seconds
    //             "retargetTime": 90, //Check to see if we should retarget every this many seconds
    //             "variancePercent": 30 //Allow time to very this % from target without retargeting
    //         }

    //Can probably not send avg here either.
    //@todo does this need to return a result? Ideally not, but if we send difficulty, then maybe.
    //@todo see if we can solve a lot of these recasting issues.
    async fn retarget(&self) {
        let now = Utc::now().naive_utc().timestamp();

        let mut job_stats = self.job_stats.lock().await;
        //@todo check if this makes sense.
        //@note: The difficulty may have changed throughout the last time we ran this function, so
        //we want to save any updated difficulty here. We also don't want to keep a lock on
        //self.difficulty the entire function, as it's likely used elsewhere so this is a good way
        //to do it.
        //
        //One more note, we should turn current_difficulty into a RwLock so that we can just grab
        //the read portion here and not have to worry about the write portion.
        job_stats.current_difficulty = *self.difficulty.lock().await;

        let since_last = now - job_stats.last_timestamp;

        job_stats.times.push(since_last);
        job_stats.last_timestamp = now;

        if now - job_stats.last_retarget < self.options.retarget_time {
            return;
        }

        let variance = self.options.target_time * (self.options.variance_percent as f64 / 100.0);
        let time_min = self.options.target_time - variance;
        let time_max = self.options.target_time + variance;
        job_stats.last_retarget = now;

        let avg: i64 = job_stats.times.iter().sum::<i64>() / job_stats.times.len() as i64;
        let mut d_dif = self.options.target_time / avg as f64;

        if avg as f64 > time_max && job_stats.current_difficulty > self.options.min_diff {
            //@note can speed up vardiff with this optional mode. Think about implementing.
            // if self.options.x2_mode {
            //     d_dif = 0.5;
            // }

            if d_dif * job_stats.current_difficulty < self.options.min_diff {
                d_dif = self.options.min_diff / job_stats.current_difficulty;
            }
        } else if (avg as f64) < time_min {
            //@note can speed up vardiff with this optional mode. Think about implementing.
            // if self.options.x2_mode {
            //     d_dif = 2.0;
            // }

            if d_dif * job_stats.current_difficulty > self.options.max_diff {
                d_dif = self.options.max_diff / job_stats.current_difficulty;
            }
        } else {
            return;
        }

        let new_diff = job_stats.current_difficulty * d_dif;

        *self.next_difficulty.lock().await = Some(new_diff);

        job_stats.times.clear();

        // 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?;
        //}
    }

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

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

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

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

    pub async fn ip(&self) -> IpAddr {
        self.info().await.ip
    }

    pub fn id(&self) -> Uuid {
        self.id
    }

    pub async fn set_sid(&self, sid: Option<String>, sid_size: usize) -> String {
        let sid = match sid {
            Some(sid) => Buffer::from_hex(sid).unwrap(),
            None => {
                let mut id = vec![0; sid_size];
                let mut rng = rand::thread_rng();

                rng.fill_bytes(&mut id);

                Buffer::from(id)
            }
        };

        self.worker_info.lock().await.sid = Some(sid.clone());

        sid.to_hex()
    }

    // ===== Worker Helper functions ===== //
    pub async fn set_var_diff(&self, var_diff: bool) {
        *self.var_diff.lock().await = var_diff;
    }

    pub async fn set_worker_name(&self, name: Option<String>) {
        self.worker_info.lock().await.name = name;
    }

    pub async fn set_account_id(&self, id: i32) {
        self.worker_info.lock().await.account_id = id;
    }

    pub async fn get_account_id(&self) -> i32 {
        self.worker_info.lock().await.account_id
    }

    pub async fn set_client(&self, client: &str) {
        self.worker_info.lock().await.client = Some(client.to_owned());
    }

    pub async fn get_client(&self) -> Option<String> {
        self.worker_info.lock().await.client.clone()
    }

    pub async fn set_worker_id(&self, id: Uuid) {
        self.worker_info.lock().await.id = id;
    }

    pub async fn get_sid(&self) -> Option<Buffer> {
        self.worker_info.lock().await.sid.clone()
    }

    // pub async fn set_sid(&self, sid: &str) {
    //     self.worker_info.lock().await.sid = Some(sid.to_owned());
    // }

    pub async fn get_worker(&self) -> WorkerInfo {
        self.worker_info.lock().await.clone()
    }

    //@todo make this a read/write
    pub async fn authorized(&self) -> bool {
        *self.authorized.lock().await
    }

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

    //@todo make this a read/write
    pub async fn subscribed(&self) -> bool {
        *self.subscribed.lock().await
    }

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

    pub async fn set_difficulty(&self, difficulty: f64) {
        *self.difficulty.lock().await = difficulty;
    }

    pub async fn get_difficulty(&self) -> f64 {
        *self.difficulty.lock().await
    }

    pub async fn get_state(&self) -> State {
        self.state.lock().await.clone()
    }

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

    pub async fn valid_share(&self) {
        self.stats.lock().await.accepted_shares += 1;
        self.consider_ban().await;
        if *self.var_diff.lock().await {
            self.retarget().await;
        }
    }

    //@todo I don't know if we actually need this function.
    // pub async fn get_next_difficulty(&self) -> Option<f64> {
    //     self.next_difficulty.lock().await.clone()
    // }

    pub async fn update_difficulty(&self) -> Option<f64> {
        let next_difficulty = *self.next_difficulty.lock().await;

        if let Some(next_difficulty) = next_difficulty {
            *self.difficulty.lock().await = next_difficulty;

            *self.next_difficulty.lock().await = None;

            Some(next_difficulty)
        } else {
            None
        }
    }

    pub async fn invalid_share(&self) {
        self.stats.lock().await.rejected_shares += 1;
        self.consider_ban().await;
        //@todo see below
        //I don't think we want to retarget on invalid shares, but let's double check later.
        // self.retarget().await;
    }

    //Consider making this pub(crate)? Although, I think it could be useful for other things.
    pub fn get_stop_token(&self) -> StopToken {
        self.stop_token.clone()
    }

    pub async fn consider_ban(&self) {
        let accepted = self.stats.lock().await.accepted_shares;
        let rejected = self.stats.lock().await.rejected_shares;

        let total = accepted + rejected;

        //@todo come from options.
        let check_threshold = 500;
        let invalid_percent = 50.0;

        if total >= check_threshold {
            let percent_bad: f64 = (rejected as f64 / total as f64) * 100.0;

            if percent_bad < invalid_percent {
                //@todo make this possible. Reset stats to 0.
                // self.stats.lock().await = MinerStats::default();
            } else {
                warn!(
                    "Miner: {} banned. {} out of the last {} shares were invalid",
                    self.id, rejected, total
                );
                self.ban().await;
            }
        }
    }
}