use fast_pool::plugin::{CheckMode, DurationManager};
use futures_core::future::BoxFuture;
use futures_core::stream::BoxStream;
use log::error;
use rbdc::db::{ConnectOptions, Driver};
use rbdc::db::{Connection, ExecResult, Row};
use rbdc::pool::ConnectionGuard;
use rbdc::pool::ConnectionManager;
use rbdc::pool::Pool;
use rbdc::Error;
use rbs::value::map::ValueMap;
use rbs::Value;
use std::ops::Deref;
use std::sync::Arc;
use std::time::Duration;

#[derive(Debug, Clone)]
pub struct FastPool {
    pub manager: Arc<ConnectionManager>,
    pub inner: fast_pool::Pool<DurationManager<ConnManagerProxy>>,
}

impl FastPool {
    pub fn new_url<D: Driver + 'static>(driver: D, url: &str) -> Result<Self, Error>
    where
        Self: Sized,
    {
        Self::new(ConnectionManager::new(driver, url)?)
    }

    pub fn new_option<D: Driver + 'static, Options: ConnectOptions>(
        driver: D,
        options: Options,
    ) -> Result<Self, Error>
    where
        Self: Sized,
    {
        Self::new(ConnectionManager::new_options(driver, options))
    }
}

#[derive(Debug)]
pub struct ConnManagerProxy {
    inner: ConnectionManager,
}

impl ConnManagerProxy {
    pub fn new(inner: ConnectionManager) -> Self {
        Self { inner }
    }
}

#[derive(Debug)]
pub struct ConnProxy {
    conn: Option<fast_pool::ConnectionGuard<DurationManager<ConnManagerProxy>>>,
}

impl From<ConnectionManager> for ConnManagerProxy {
    fn from(value: ConnectionManager) -> Self {
        ConnManagerProxy { inner: value }
    }
}

#[async_trait::async_trait]
impl Pool for FastPool {
    fn new(manager: ConnectionManager) -> Result<Self, Error>
    where
        Self: Sized,
    {
        Ok(Self {
            manager: manager.clone().into(),
            inner: fast_pool::Pool::new(DurationManager::new(
                ConnManagerProxy::new(manager),
                CheckMode::NoLimit,
            )),
        })
    }

    async fn get(&self) -> Result<Box<dyn Connection>, Error> {
        let v = self
            .inner
            .get_timeout(self.inner.get_timeout_wait())
            .await
            .map_err(|e| Error::from(e.to_string()))?;
        let proxy = ConnProxy { conn: Some(v) };
        Ok(Box::new(proxy))
    }

    async fn get_timeout(&self, mut d: Duration) -> Result<Box<dyn Connection>, Error> {
        if d.is_zero() {
            let state = self.inner.state();
            if state.in_use < state.max_open {
                d = Duration::from_secs(10);
            } else {
                return Err(Error::from("Time out in the connection pool"));
            }
        }
        let v = self
            .inner
            .get_timeout(Some(d))
            .await
            .map_err(|e| Error::from(e.to_string()))?;
        let proxy = ConnProxy { conn: Some(v) };
        Ok(Box::new(proxy))
    }

    async fn set_timeout(&self, timeout: Option<Duration>) {
        self.inner.set_timeout_wait(timeout);
    }

    async fn set_conn_max_lifetime(&self, max_lifetime: Option<Duration>) {
        let manager = self
            .inner
            .downcast_manager::<DurationManager<ConnManagerProxy>>();
        if let Some(manager) = manager {
            if let Some(max_lifetime) = max_lifetime {
                manager.mode.set_mode(CheckMode::MaxLifetime(max_lifetime));
            } else {
                manager.mode.set_mode(CheckMode::NoLimit);
            }
        } else {
            error!("FastPool method set_conn_max_lifetime need use DurationManager to init");
        }
    }

    async fn set_max_idle_conns(&self, n: u64) {
        self.inner.set_max_idle_conns(n);
    }

    async fn set_max_open_conns(&self, n: u64) {
        self.inner.set_max_open(n);
    }

    fn driver_type(&self) -> &str {
        self.manager.driver_type()
    }

    async fn state(&self) -> Value {
        let mut m = ValueMap::with_capacity(10);
        let state = self.inner.state();
        m.insert("max_open".to_string().into(), state.max_open.into());
        m.insert("connections".to_string().into(), state.connections.into());
        m.insert("in_use".to_string().into(), state.in_use.into());
        m.insert("idle".to_string().into(), state.idle.into());
        m.insert("waits".to_string().into(), state.waits.into());
        m.insert("connecting".to_string().into(), state.connecting.into());
        m.insert("checking".to_string().into(), state.checking.into());
        Value::Map(m)
    }

    fn driver(&self) -> &dyn Driver {
        self.manager.driver.deref()
    }
}

impl fast_pool::Manager for ConnManagerProxy {
    type Connection = ConnectionGuard;
    type Error = Error;

    async fn connect(&self) -> Result<Self::Connection, Self::Error> {
        self.inner.connect().await
    }

    async fn check(&self, conn: &mut Self::Connection) -> Result<(), Self::Error> {
        let r = self.inner.check(conn).await;
        match r {
            Ok(_) => Ok(()),
            Err(e) => {
                _ = conn.close().await;
                Err(e)
            }
        }
    }
}

impl Connection for ConnProxy {
    fn exec_rows(
        &mut self,
        sql: &str,
        params: Vec<Value>,
    ) -> BoxFuture<'_, Result<BoxStream<'_, Result<Box<dyn Row>, Error>>, Error>> {
        match &mut self.conn {
            Some(conn) => conn.exec_rows(sql, params),
            None => Box::pin(async { Err(Error::from("conn is drop")) }),
        }
    }

    fn exec_decode(
        &mut self,
        sql: &str,
        params: Vec<Value>,
    ) -> BoxFuture<'_, Result<Value, Error>> {
        match &mut self.conn {
            Some(conn) => conn.exec_decode(sql, params),
            None => Box::pin(async { Err(Error::from("conn is drop")) }),
        }
    }

    fn exec(&mut self, sql: &str, params: Vec<Value>) -> BoxFuture<'_, Result<ExecResult, Error>> {
        match &mut self.conn {
            Some(conn) => conn.exec(sql, params),
            None => Box::pin(async { Err(Error::from("conn is drop")) }),
        }
    }

    fn ping(&mut self) -> BoxFuture<'_, Result<(), Error>> {
        match &mut self.conn {
            Some(conn) => conn.ping(),
            None => Box::pin(async { Err(Error::from("conn is drop")) }),
        }
    }

    fn close(&mut self) -> BoxFuture<'_, Result<(), Error>> {
        match &mut self.conn {
            Some(conn) => conn.close(),
            None => Box::pin(async { Err(Error::from("conn is drop")) }),
        }
    }

    fn begin(&mut self) -> BoxFuture<'_, Result<(), Error>> {
        match &mut self.conn {
            Some(conn) => conn.begin(),
            None => Box::pin(async { Err(Error::from("conn is drop")) }),
        }
    }
    fn commit(&mut self) -> BoxFuture<'_, Result<(), Error>> {
        match &mut self.conn {
            Some(conn) => conn.commit(),
            None => Box::pin(async { Err(Error::from("conn is drop")) }),
        }
    }
    fn rollback(&mut self) -> BoxFuture<'_, Result<(), Error>> {
        match &mut self.conn {
            Some(conn) => conn.rollback(),
            None => Box::pin(async { Err(Error::from("conn is drop")) }),
        }
    }
}

#[cfg(test)]
mod test {
    use crate::FastPool;
    use futures_core::future::BoxFuture;
    use futures_core::stream::BoxStream;
    use rbdc::db::{ConnectOptions, Connection, Driver, ExecResult, Row};
    use rbdc::pool::ConnectionManager;
    use rbdc::pool::Pool;
    use rbdc::try_stream;
    use rbs::{Error, Value};
    use std::pin::Pin;

    #[derive(Debug)]
    pub struct Opt {}
    impl ConnectOptions for Opt {
        fn connect(&self) -> BoxFuture<'_, Result<Box<dyn Connection>, Error>> {
            Box::pin(async { Ok(Box::new(Conn {}) as Box<dyn Connection>) })
        }

        fn set_uri(&mut self, _uri: &str) -> Result<(), Error> {
            Ok(())
        }
    }

    #[derive(Debug)]
    pub struct Conn {}

    impl Connection for Conn {
        fn exec_rows(
            &mut self,
            _sql: &str,
            _params: Vec<Value>,
        ) -> BoxFuture<'_, Result<BoxStream<'_, Result<Box<dyn Row>, Error>>, Error>> {
            Box::pin(async move {
                let rows: Vec<Box<dyn Row>> = vec![];
                let stream = try_stream! {
                    for row in rows {
                        r#yield!(row);
                    }
                    Ok(())
                };
                let stream: Pin<
                    Box<dyn futures_core::Stream<Item = Result<Box<dyn Row>, Error>> + Send>,
                > = Box::pin(stream);
                Ok(stream)
            })
        }

        fn exec(
            &mut self,
            _sql: &str,
            _params: Vec<Value>,
        ) -> BoxFuture<'_, Result<ExecResult, Error>> {
            Box::pin(async { Ok(ExecResult::default()) })
        }

        fn ping(&mut self) -> BoxFuture<'_, Result<(), Error>> {
            Box::pin(async { Ok(()) })
        }

        fn close(&mut self) -> BoxFuture<'_, Result<(), Error>> {
            Box::pin(async { Ok(()) })
        }
    }

    #[derive(Debug)]
    pub struct D {}
    impl Driver for D {
        fn name(&self) -> &str {
            "d"
        }

        fn connect(&self, _url: &str) -> BoxFuture<'_, Result<Box<dyn Connection>, Error>> {
            Box::pin(async { Ok(Box::new(Conn {}) as Box<dyn Connection>) })
        }

        fn connect_opt<'a>(
            &'a self,
            _opt: &'a dyn ConnectOptions,
        ) -> BoxFuture<'a, Result<Box<dyn Connection>, Error>> {
            Box::pin(async { Ok(Box::new(Conn {}) as Box<dyn Connection>) })
        }

        fn default_option(&self) -> Box<dyn ConnectOptions> {
            Box::new(Opt {})
        }
    }

    #[test]
    fn test() {
        let pool = Box::new(FastPool::new(ConnectionManager::new(D {}, "").unwrap()));
        println!("ok={}", pool.is_ok());
    }
}
