

use crate::yig_io_worker::YigIoWorkerFactory;
use common::runtime::Executor;
use common::error::Errno;
use io_engine::{backend_storage::{BackendStore, BackendStoreFactory}, 
types::{MsgFileOpResp, MsgFileReadOp, MsgFileStatOp, MsgFileWriteOp}};
use io_engine::types::{MsgFileOp, MsgFileStatResult};
use io_engine::io_thread_pool::IoThreadPool;
use std::collections::HashMap;
use crossbeam_channel::bounded;
use crossbeam_channel::Sender;
use std::sync::Arc;
use log::{error, warn};


// make sure that YigBackend is thread safe.
pub struct YigBackend{
    bucket: String,
    yig_pool: IoThreadPool,
}

impl BackendStore for YigBackend{
    fn get(&self, name: String)->Result<Vec<u8>, Errno>{
        let thr = self.yig_pool.get_thread_roundrobin();
        let (tx, rx) = bounded::<MsgFileOpResp>(1);
        let msg_read = MsgFileReadOp{
            id0: 0,
            id1: 0,
            name: name.clone(),
            dir: self.bucket.clone(),
            offset: 0,
            size: 0,
            data_sender: tx,
        };
        let ret = thr.do_io(MsgFileOp::OpRead(msg_read));
        if !ret.is_success(){
            error!("YigBackend::get: failed to send OpRead for bucket: {}, object: {}, err: {:?}",
            self.bucket, name, ret);
            return Err(ret);
        }
        let ret = rx.recv();
        match ret {
            Ok(ret) => {
                match ret {
                    MsgFileOpResp::OpRespRead(ret) => {
                        if ret.err.is_success() {
                            if let Some(data) = ret.data {
                                return Ok(data);
                            }
                            warn!("YigBackend::get: got empty data from {}/{}", self.bucket, name);
                            return Ok(Vec::new());
                        }
                        error!("YigBackend::get: failed to read bucket: {}, object: {}, err: {:?}",
                    self.bucket, name, ret.err);
                        return Err(ret.err);
                    }
                    _ => {
                        error!("YigBackend::get: got invalid read resp for bucket: {}, object: {}",
                        self.bucket, name);
                        return Err(Errno::Eintr);
                    }
                }
            }
            Err(err) => {
                error!("YigBackend::get: failed to recv read resp for bucket: {}, object: {}, err: {}",
            self.bucket, name, err);
                return Err(Errno::Eintr);
            }
        }
    }

    fn get_async(&self, name: String, resp_sender: Sender<MsgFileOpResp>) -> Errno{
        let thr = self.yig_pool.get_thread_roundrobin();
        let msg_read = MsgFileReadOp{
            id0: 0,
            id1: 0,
            name: name.clone(),
            dir: self.bucket.clone(),
            offset: 0,
            size: 0,
            data_sender: resp_sender,
        };
        let ret = thr.do_io(MsgFileOp::OpRead(msg_read));
        if !ret.is_success(){
            error!("YigBackend::get_async: failed to send OpRead for bucket: {}, object: {}, err: {:?}",
            self.bucket, name, ret);
            return ret;
        }
        return Errno::Esucc;
    }

    fn put(&self, name: String, data: &[u8])->Errno{
        let thr = self.yig_pool.get_thread_hashring(name.clone());
        let (tx, rx) = bounded::<MsgFileOpResp>(1);
        let msg = MsgFileWriteOp{
            id0: 0,
            id1: 0,
            name: name.clone(),
            dir: self.bucket.clone(),
            max_size: 0, // currently, not used.
            offset: 0,
            data: data.to_vec(),
            resp_sender: tx,
        };
        let ret = thr.do_io(MsgFileOp::OpWrite(msg));
        if !ret.is_success(){
            error!("YigBackend::put: failed to send OpWrite for {}/{}, data_size: {}, err: {:?}",
            self.bucket, name, data.len(), ret);
            return ret;
        }
        let ret = rx.recv();
        match ret {
            Ok(ret) => {
                match ret {
                    MsgFileOpResp::OpRespWrite(ret) => {
                        if !ret.err.is_success(){
                            error!("YigBackend::put: failed to put {}/{}, err: {:?}", self.bucket, name, ret.err);
                        }
                        return ret.err;
                    }
                    _ => {
                        error!("YigBackend::put: got invalid write resp for {}/{}, size: {}",
                        self.bucket, name, data.len());
                        return Errno::Eintr;
                    }
                }
            }
            Err(err) => {
                error!("YigBackend::write: failed to recv write result for {}/{}, size: {}, err: {:?}",
            self.bucket, name, data.len(), err);
                return Errno::Eintr;
            }
        }
    }

    fn put_async(&self, name: String, data: &[u8], resp_sender: Sender<MsgFileOpResp>)->Errno{
        let thr = self.yig_pool.get_thread_roundrobin();
        let msg = MsgFileWriteOp{
            id0: 0,
            id1: 0,
            name: name.clone(),
            dir: self.bucket.clone(),
            max_size: 0, // currently, not used.
            offset: 0,
            data: data.to_vec(),
            resp_sender: resp_sender,
        };
        let ret = thr.do_io(MsgFileOp::OpWrite(msg));
        if !ret.is_success(){
            error!("YigBackend::put_async: failed to send OpWrite for {}/{}, size: {}, err: {:?}",
            self.bucket, name, data.len(), ret);
            return ret;
        }
        return Errno::Esucc;
    }

    fn stat(&self, name: String) -> MsgFileStatResult{
        let mut result = MsgFileStatResult{
            id0: 0,
            id1: 0,
            name: name.clone(),
            size: 0,
            err: Errno::Eintr,
        };
        let (tx, rx) = bounded::<MsgFileStatResult>(1);
        let thr = self.yig_pool.get_thread_roundrobin();
        let msg = MsgFileStatOp{
            id0: 0,
            id1: 0,
            dir: self.bucket.clone(),
            name: name.clone(),
            result_tx: tx,
        };
        let ret = thr.do_io(MsgFileOp::OpStat(msg));
        if !ret.is_success(){
            error!("YigBackend::stat: failed to perform stat for {}/{}, err: {:?}", self.bucket, name, ret);
            result.err = ret;
            return result;
        }
        let ret = rx.recv();
        match ret{
            Ok(ret) => {
                result = ret;
            }
            Err(err) => {
                error!("YigBackend::stat: failed to receive the result from stat {}/{}, err: {}",
                self.bucket, name, err);
                result.err = Errno::Eintr;
            }
        }

        return result;
    }

    fn stat_async(&self, name: String, resp_sender: Sender<MsgFileOpResp>) -> Errno{
        let result: MsgFileStatResult;
        let (tx, rx) = bounded::<MsgFileStatResult>(1);
        let thr = self.yig_pool.get_thread_roundrobin();
        let msg = MsgFileStatOp{
            id0: 0,
            id1: 0,
            dir: self.bucket.clone(),
            name: name.clone(),
            result_tx: tx,
        };
        let ret = thr.do_io(MsgFileOp::OpStat(msg));
        if !ret.is_success(){
            error!("YigBackend::stat_async: failed to perform stat for {}/{}, err: {:?}", self.bucket, name, ret);
            return ret;
        }
        let ret = rx.recv();
        match ret{
            Ok(ret) => {
                result = ret;
            }
            Err(err) => {
                error!("YigBackend::stat_async: failed to receive the result from stat {}/{}, err: {}",
                self.bucket, name, err);
                return Errno::Eintr;
            }
        }
        
        let ret = resp_sender.send(MsgFileOpResp::OpRespStat(result));
        match ret{
            Ok(_) => {
                return Errno::Esucc;
            }
            Err(err) => {
                error!("YigBackend::stat_async: failed to send response for stat {}/{}, err: {}",
                self.bucket, name, err);
                return Errno::Eintr;
            }
        }
    }
}

impl Drop for YigBackend{
    fn drop(&mut self){
        self.yig_pool.stop();
    }
}

impl YigBackend {
    pub fn new(region: &String, endpoint: &String, ak: &String, sk: &String, bucket: &String, num: u32, exec: &Executor) -> Self {
        YigBackend{
            bucket: bucket.clone(),
            yig_pool: IoThreadPool::new(num, 
                &format!("yig_io_thread_"), 
                exec, 
  &YigIoWorkerFactory::new(region, endpoint, ak, sk)),
        }
    }
}

pub struct YigBackendFactory{
    exec: Executor,
}

impl BackendStoreFactory for YigBackendFactory{
    fn new_backend_store(&self, cfg: &HashMap<String, String>) -> Result<Arc<dyn BackendStore>, Errno>{
        let region: String;
        let endpoint: String;
        let ak: String;
        let sk: String;
        let bucket: String;
        let thread_num: u32;

        if let Some(r) = cfg.get("region"){
            region = r.clone();
        } else {
            error!("new_backend_store: missing region setting");
            return Err(Errno::Eintr);
        }

        if let Some(e) = cfg.get("endpoint"){
            endpoint = e.clone();
        } else {
            error!("new_backend_store: missing endpoint setting");
            return Err(Errno::Eintr);
        }

        if let Some(a) = cfg.get("ak") {
            ak = a.clone();
        } else {
            error!("new_backend_store: missing ak");
            return Err(Errno::Eintr);
        }

        if let Some(s) = cfg.get("sk") {
            sk = s.clone();
        } else {
            error!("new_backend_store: missing sk");
            return Err(Errno::Eintr);
        }

        if let Some(n) = cfg.get("thread_num") {
            thread_num = n.parse::<u32>().unwrap();
        } else {
            error!("new_backend_store: missing thread_num");
            return Err(Errno::Eintr);
        }

        if let Some(b) = cfg.get("bucket") {
            bucket = b.clone();
        } else {
            error!("new_backend_store: missing bucket");
            return Err(Errno::Eintr);
        }

        Ok(Arc::new(YigBackend::new(&region, &endpoint, &ak, &sk, &bucket, thread_num, &self.exec)))
    }
}

impl YigBackendFactory {
    pub fn new(exec: &Executor) -> Box<dyn BackendStoreFactory> {
        Box::new(YigBackendFactory{
            exec: exec.clone(),
        })
    }
}