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, trace, warn};
use rand::prelude::*;
use serde::Serialize;
use serde_json::json;
use std::collections::HashMap;
use std::net::IpAddr;
use std::time::SystemTime;
use stop_token::{StopSource, StopToken};
use uuid::Uuid;

//@todo round 2, I think I need to break out of the concept of worker / miner being the same. We
//need to create a miner structg that contains all of the data shit.
#[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 session_id: u32,
    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 agent: 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,
    //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 I'm starting to think we pull this out into our own file to make it a lot easier to do
    //things.
    //impl functions directly onto agentworker rather than copy paste in here
    pub worker_list: Arc<Mutex<HashMap<u16, AgentWorkerInfo>>>,
}

#[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 miner_account: Option<i32>,
    pub id: Uuid,
}

#[derive(Debug, Clone)]
pub struct AgentWorkerInfo {
    pub client: Option<String>,
    pub name: Option<String>,
    pub sid: Option<Buffer>,
    //@todo these all seem like an extra waste of energy.
    pub account_id: i32,
    pub miner_account: Option<i32>,
    pub id: Uuid,
    pub stats: Arc<Mutex<MinerStats>>,
    pub job_stats: Arc<Mutex<JobStats>>,
}

impl<State: Clone + Send + Sync + 'static> Connection<State> {
    pub fn new(
        session_id: u32,
        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();

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

        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 {
            session_id,
            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)),
            agent: Arc::new(Mutex::new(false)),
            //@todo this should be passed in.
            difficulty: Arc::new(Mutex::new(initial_difficulty)),
            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,
            worker_list: Arc::new(Mutex::new(HashMap::new())),
        }
    }

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

    //@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)?;

        trace!("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<(serde_json::Value, serde_json::Value)> {
        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((json!(false), serde_json::Value::Null)),
        };

        Ok((values["result"].clone(), values["error"].clone()))
    }

    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.floor());

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

    pub async fn register_worker(
        &self,
        session_id: u16,
        client_agent: &str,
        worker_name: &str,
        worker_id: Uuid,
    ) {
        let worker = AgentWorkerInfo {
            client: Some(client_agent.to_owned()),
            name: Some(worker_name.to_owned()),
            //@todo I suppose this is technically session ID
            sid: Some(Buffer::from(session_id.to_le_bytes().to_vec())),
            //@todo we need to be careful of lockups here.
            account_id: self.get_account_id().await,
            miner_account: self.get_miner_account().await,
            id: worker_id,
            stats: Arc::new(Mutex::new(MinerStats {
                accepted_shares: 0,
                rejected_shares: 0,
                last_active: Utc::now().naive_utc(),
            })),
            job_stats: Arc::new(Mutex::new(JobStats {
                last_timestamp: Utc::now().naive_utc().timestamp(),
                last_share_timestamp: 0,
                last_retarget: Utc::now().naive_utc().timestamp() - self.options.retarget_time / 2,
                times: Vec::new(),
                current_difficulty: self.difficulty.lock().await.clone(),
                job_difficulty: 0.0,
            })),
        };

        self.worker_list.lock().await.insert(session_id, worker);
        //@todo one large thing we are missing here, and it's probs just for the above library but
        //EVENTS so new event for new worker.
    }

    pub async fn unregister_worker(&self, session_id: u16) {
        self.worker_list.lock().await.remove(&session_id);
    }

    //@todo I think wrapping this into it's own type would be nice just called WorkerList
    pub async fn get_worker_list(&self) -> HashMap<u16, AgentWorkerInfo> {
        self.worker_list.lock().await.clone()
    }

    pub async fn get_worker_by_session_id(&self, session_id: u16) -> Option<AgentWorkerInfo> {
        self.worker_list
            .lock()
            .await
            .get(&session_id)
            .map(|w| w.clone())
    }

    pub async fn update_worker_by_session_id(&self, session_id: u16, worker: AgentWorkerInfo) {
        self.worker_list.lock().await.insert(session_id, worker);
    }

    // ===== 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_miner_account(&self, id: Option<i32>) {
        self.worker_info.lock().await.miner_account = id;
    }

    pub async fn get_miner_account(&self) -> Option<i32> {
        self.worker_info.lock().await.miner_account
    }

    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 fn get_session_id(&self) -> u32 {
        self.session_id
    }

    // 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.stats.lock().await.last_active = Utc::now().naive_utc();
        self.consider_ban().await;
        if *self.var_diff.lock().await {
            self.retarget().await;
        }
    }

    pub async fn valid_share_by_session_id(&self, session_id: u16) {
        match self.get_worker_by_session_id(session_id).await {
            Some(worker) => {
                worker.stats.lock().await.accepted_shares += 1;
                worker.stats.lock().await.last_active = Utc::now().naive_utc();
                //@todo we need to fix these to be worker specific.
                self.consider_ban().await;
                if *self.var_diff.lock().await {
                    self.retarget().await;
                }
            }
            None => {}
        }
    }

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