#![allow(clippy::uninlined_format_args)]
#![deny(unused_qualifications)]
//#![deny(unused_crate_dependencies)]
#![deny(unused_extern_crates)]
#![deny(warnings)]

use std::io::Cursor;
use std::path::Path;
use std::sync::Arc;

use openraft::Config;
use openraft::TokioRuntime;
//use tokio::net::TcpListener;
use tokio::net::lookup_host;
use tokio::net::TcpSocket;
use futures::TryStreamExt;
use tokio::task;
use tokio_stream::wrappers::TcpListenerStream;
use tokio_rustls::TlsAcceptor;
use tokio::signal;

use crate::app::App;
use crate::network::api;
use crate::network::management;
use crate::network::Network;
use crate::store::new_storage;
use crate::store::Request;
use crate::store::Response;

pub mod app;
pub mod client;
pub mod network;
pub mod sqlite_store;

pub mod cipher;
use cipher::EncryptData;
pub use cipher::NoEncrypt;

pub mod notifications;

pub mod key_providers;

pub mod convert;

pub use rxqlite_common::RSQliteClientTlsConfig;
use key_providers::KeyProvider;

use sqlite_store as store;
//use std::net::SocketAddr;
//use std::str::FromStr;
use warp::Filter;

use std::collections::{BTreeMap, BTreeSet};

use rustls::{server::WantsServerCert, ConfigBuilder, ServerConfig};

use serde::{Deserialize, Serialize};

#[cfg(feature = "sqlcipher")]
use crate::cipher::aes_gcm_siv::Aes256GcmSivEncryptor;
#[cfg(feature = "sqlcipher")]
use base64::{engine::general_purpose::URL_SAFE, Engine as _};
#[cfg(feature = "sqlcipher")]
use ring::digest;

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct RSQliteNodeTlsConfig {
    /*
    pub key_path: String,
    pub cert_path: String,
    */
    pub accept_invalid_certificates: bool,
}

/*
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::time::{timeout, Duration};
use futures::future::poll_fn;
*/

use toy_rpc_ha421;
/*
pub type NodeId = u64;

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq, Default)]
pub struct Node {
    pub rpc_addr: String,
    pub api_addr: String,
}
*/

pub use rxqlite_lite_common::{NodeId,Node};


pub type SnapshotData = Cursor<Vec<u8>>;

openraft::declare_raft_types!(
    pub TypeConfig:
        D = Request,
        R = Response,
        NodeId = NodeId,
        Node = Node,
        Entry = openraft::Entry<TypeConfig>,
        SnapshotData = SnapshotData,
        AsyncRuntime = TokioRuntime
);

pub mod typ {
    use openraft::error::Infallible;

    use crate::Node;
    use crate::NodeId;
    use crate::TypeConfig;

    pub type Entry = openraft::Entry<TypeConfig>;

    pub type LeaderId = openraft::LeaderId<NodeId>;
    pub type LogId = openraft::log_id::LogId<NodeId>;
    pub type Vote = openraft::Vote<NodeId>;
    pub type Membership = openraft::Membership<NodeId, Node>;
    pub type StoredMembership = openraft::StoredMembership<NodeId, Node>;
    
    pub type RaftError<E = Infallible> = openraft::error::RaftError<NodeId, E>;
    pub type RPCError<E = Infallible> = openraft::error::RPCError<NodeId, Node, RaftError<E>>;

    pub type ClientWriteError = openraft::error::ClientWriteError<NodeId, Node>;
    pub type CheckIsLeaderError = openraft::error::CheckIsLeaderError<NodeId, Node>;
    pub type ForwardToLeader = openraft::error::ForwardToLeader<NodeId, Node>;
    pub type InitializeError = openraft::error::InitializeError<NodeId, Node>;

    pub type ClientWriteResponse = openraft::raft::ClientWriteResponse<TypeConfig>;

    pub type RaftMetrics = openraft::metrics::RaftMetrics<NodeId, Node>;

    pub type RaftServerMetrics = openraft::metrics::RaftServerMetrics<NodeId, Node>;
}

pub type RXQliteRaft = openraft::Raft<TypeConfig>;

fn with_app(
    app: Arc<App>,
) -> impl Filter<Extract = (Arc<App>,), Error = std::convert::Infallible> + Clone {
    warp::any().map(move || app.clone())
}
/*
async fn poll_future<T>(future: Pin<&mut T>, duration: Duration) -> Poll<T::Output>
where
    T: Future,
{
    let res = timeout(duration, poll_fn(|cx| Future::poll(future, cx))).await;
    match res {
        Ok(poll) => poll,
        Err(_) => Poll::Pending,
    }
}
*/
/*
#[derive(Serialize, Deserialize, Clone)]
pub struct InstanceParams {
    pub http_addr: String,
    pub rpc_addr: String,
    pub notifications_addr: String,
    tls_config: Option<RSQliteNodeTlsConfig>,
}
*/

pub struct InstanceParams {
    pub node_id: NodeId,
    pub http_addr: String,
    pub rpc_addr: String,
    pub notifications_addr: String,
    pub tls_config: Option<RSQliteNodeTlsConfig>,
    pub key_provider: Option<Box<dyn KeyProvider>>,
    pub no_database_encryption: bool,
    pub is_test_node: bool,
}


async fn do_init_rxqlite<P>(
    base_dir: P,
    instance_params: &mut InstanceParams,
) -> anyhow::Result<
  (
    Arc<App>, 
    tokio::task::JoinHandle<anyhow::Result<()>>,
    tokio::task::JoinHandle<()>,
    tokio::task::JoinHandle<()>,
  )>
where
    P: AsRef<Path>,
{
    let (
      _key,
      encrypt_data,
      private_key,
      certificates,
      ): (
        Option<String>, 
        Option<Arc<Box<dyn EncryptData>>>,
        Option<rustls::pki_types::PrivatePkcs8KeyDer<'static>>,
        Option<Vec<rustls::pki_types::CertificateDer<'static>>>,
      ) = {
        #[cfg(feature = "sqlcipher")]
        {
            /*if let Some(tls_config) = instance_params.tls_config.as_ref() {
              */
                if let Some(key_provider) = instance_params.key_provider.take() {
                  let certificates = key_provider.get_certificates().clone();
                  
                  let private_key = key_provider.get_private_key().clone_key();
                  
                  
                  let hashed_key = digest::digest(&digest::SHA256, private_key.secret_pkcs8_der());

                  let hashed_key = URL_SAFE.encode(hashed_key.as_ref());
                  
                  let encrypt_data = Aes256GcmSivEncryptor::new(&private_key);
                  (
                    Some(hashed_key), 
                    Some(Arc::new(Box::new(encrypt_data))),
                    Some(private_key),
                    Some(certificates),
                  )
                /*
                } else {
                  (None, None, None, None,)
                }
                */
            } else {
                (None, None, None, None,)
            }
        }
        #[cfg(not(feature = "sqlcipher"))]
        {
            (None, None, None, None,)
        }
    };
    let rocksdb_dir = base_dir.as_ref().join("rocksdb");
    let sqlite_path = base_dir.as_ref().join("sqlite.db");

    // Create a configuration for the raft instance.
    // under stress, the cluster might timeout, so multiply by 10 timeouts
    let config = Config {
        heartbeat_interval: 2500,
        election_timeout_min: 2990,
        election_timeout_max: 3000,
        ..Default::default()
    };

    let config = Arc::new(config.validate().unwrap());

    let (log_store, state_machine_store) = new_storage(
        &rocksdb_dir,
        &sqlite_path,
        #[cfg(feature = "sqlcipher")]
        _key,
        encrypt_data,
    )
    .await?;

    let sqlite_and_path = state_machine_store.data.sqlite_and_path.clone();

    // Create the network layer that will connect and communicate the raft instances and
    // will be used in conjunction with the store created above.
    let network = Network {
        tls_config: instance_params.tls_config.clone(),
    };

    // Create a local raft instance.
    let raft = openraft::Raft::new(
        instance_params.node_id,
        config.clone(),
        network,
        log_store,
        state_machine_store,
    )
    .await
    .unwrap();

    let app = Arc::new(App {
        id: instance_params.node_id,
        api_addr: instance_params.http_addr.clone(),
        rpc_addr: instance_params.rpc_addr.clone(),
        raft,
        //key_values: kvs,
        sqlite_and_path,
        config,
    });
    let echo_service = Arc::new(network::raft::Raft::new(app.clone()));

    let mut server_builder = toy_rpc_ha421::Server::builder();
    let (toy_rpc_handle,
        notifications_handle
      ) = if instance_params.tls_config.is_some() {
        let config_builder : tokio_rustls::rustls::ConfigBuilder<_,_> = tokio_rustls::rustls::ServerConfig::builder()
        .with_no_client_auth();
        let notification_config =
            config_builder.with_single_cert(certificates.as_ref().unwrap().clone(), private_key.as_ref().unwrap().clone_key().into()).unwrap();
        let notifications_addr = instance_params.notifications_addr.clone();
        let is_test_node= instance_params.is_test_node;
        let notifications_handle = task::spawn(async move {
            if let Err(err) = notifications::start_notification_server_tls(notifications_addr, notification_config,is_test_node)
                .await {
              tracing::error!("Notification server error: {}",err);
            }
        });

        let config_builder: ConfigBuilder<ServerConfig, WantsServerCert> = ServerConfig::builder()
            .with_no_client_auth();

        let config = config_builder.with_single_cert(certificates.as_ref().unwrap().clone(), private_key.as_ref().unwrap().clone_key().into())?;

        server_builder = server_builder.register(echo_service);
        let server = server_builder.build();
        
        let socket = TcpSocket::new_v4()?;
        if instance_params.is_test_node {
          socket.set_reuseaddr(true)?;
          //socket.set_reuseport(true)?;
          //#[cfg(target_os = "linux")]
        }
        let mut rpc_addr = lookup_host(&instance_params.rpc_addr).await?;
        socket.bind(rpc_addr.next().unwrap())?;
        
        
        let listener = socket.listen(1024)?;
        
        let toy_rpc_handle = task::spawn(async move {
            
            if let Err(err) = server
              .accept_with_tls_config(listener, config)
              .await {
                tracing::error!("Toy rpc error: {}",err);
            }
        });
        (toy_rpc_handle,notifications_handle)
    } else {
        let notifications_addr = instance_params.notifications_addr.clone();
        let is_test_node = instance_params.is_test_node;
        let notifications_handle = task::spawn(async move {
            if let Err(err) = notifications::start_notification_server(notifications_addr,is_test_node)
              .await {
              tracing::error!("Notification server error: {}",err);
            }
        });

        server_builder = server_builder.register(echo_service);
        let server = server_builder.build();
        
        let socket = TcpSocket::new_v4()?;
        if instance_params.is_test_node {
          socket.set_reuseaddr(true)?;
          //#[cfg(target_os = "linux")]
          //socket.set_reuseport(true)?;
        }
        let mut rpc_addr = lookup_host(&instance_params.rpc_addr).await?;
        socket.bind(rpc_addr.next().unwrap())?;
        
        
        let listener = socket.listen(1024)?;
        
        let toy_rpc_handle = task::spawn(async move {
          
          if let Err(err) = server
            .accept_websocket(listener)
            .await {
              tracing::error!("Toy rpc error: {}",err);
            }
        });
        (toy_rpc_handle,notifications_handle)
    };

    let execute_consistent_query = warp::post()
        .and(warp::path!("api" / "sql-consistent"))
        .and(warp::body::json())
        .and(with_app(app.clone()))
        .and_then(|arg0: Message, arg1: Arc<App>| api::sql_consistent(arg0, arg1));
    let lite_execute_consistent_query = warp::post()
        .and(warp::path!("lite-api" / "sql-consistent"))
        .and(warp::body::json())
        .and(with_app(app.clone()))
        .and_then(|arg0: Message, arg1: Arc<App>| api::lite_sql_consistent(arg0, arg1));
        
    let execute_query = warp::post()
        .and(warp::path!("api" / "sql"))
        .and(warp::body::json())
        .and(with_app(app.clone()))
        .and_then(|arg0: Message, arg1: Arc<App>| api::sql(arg0, arg1));
    
    let lite_execute_query = warp::post()
        .and(warp::path!("lite-api" / "sql"))
        .and(warp::body::json())
        .and(with_app(app.clone()))
        .and_then(|arg0: Message, arg1: Arc<App>| api::lite_sql(arg0, arg1));
        
    let management_add_learner = warp::post()
        .and(warp::path!("cluster" / "add-learner"))
        .and(warp::body::json())
        .and(with_app(app.clone()))
        .and_then(management::add_learner);

    let management_change_membership = warp::post()
        .and(warp::path!("cluster" / "change-membership"))
        .and(warp::body::json())
        .and(with_app(app.clone()))
        .and_then(management::change_membership);

    let management_metrics = warp::get()
        .and(warp::path!("cluster" / "metrics"))
        .and(with_app(app.clone()))
        .and_then(management::metrics);
    
    let lite_management_metrics = warp::get()
        .and(warp::path!("lite-cluster" / "metrics"))
        .and(with_app(app.clone()))
        .and_then(management::lite_metrics);
        
    let management_snapshot = warp::post()
        .and(warp::path!("cluster" / "snapshot"))
        .and(warp::body::json())
        .and(with_app(app.clone()))
        .and_then(management::snapshot);

    let routes = execute_query
        .or(lite_execute_query)
        .or(execute_consistent_query)
        .or(lite_execute_consistent_query)
        .or(management_add_learner)
        .or(management_change_membership)
        .or(management_metrics)
        .or(lite_management_metrics)
        .or(management_snapshot);

    //let instance_params_ = instance_params.clone();
    
    let socket = TcpSocket::new_v4()?;
    if instance_params.is_test_node {
      socket.set_reuseaddr(true)?;
      //#[cfg(target_os = "linux")]
      //socket.set_reuseport(true)?;
    }
    let mut http_addr = lookup_host(&instance_params.http_addr).await?;
    socket.bind(http_addr.next().unwrap())?;
    
    let listener = socket.listen(1024)?;
    let server_config = if instance_params.tls_config.is_some() {
      let config_builder: tokio_rustls::rustls::ConfigBuilder<_,_> = tokio_rustls::rustls::ServerConfig::builder()
      .with_no_client_auth();

      let config = config_builder.with_single_cert(certificates.unwrap(), private_key.  unwrap().into())?;
      Some(config)
    } else {
      None
    };
    let http_handle = tokio::spawn(async move {
      if let Some(server_config)= server_config {
        let acceptor = TlsAcceptor::from(Arc::new(server_config));
      
        let incoming_stream = TcpListenerStream::new(listener)
          .map_ok(move |stream| {
              let acceptor = acceptor.clone();
              async move {
                  let tls_stream = acceptor.accept(stream).await?;
                  Ok::<_, std::io::Error>(tls_stream)
              }
          })
          .try_buffer_unordered(100);
        warp::serve(routes)
          .run_incoming(incoming_stream)
          .await;
      } else {
        let incoming_stream = TcpListenerStream::new(listener);
        warp::serve(routes)
            .run_incoming(incoming_stream)
            .await;
      }
      Ok::<(),anyhow::Error>(())
    });

    Ok((app, http_handle,toy_rpc_handle,notifications_handle))
}

pub async fn init_rxqlite<P>(
    base_dir: P,
    mut instance_params: InstanceParams,
    leader: bool,
    members: Vec<(NodeId, String, String)>,
) -> anyhow::Result<()>
where
    P: AsRef<Path>,
{
    /*
    if http_addr.is_none() {
        return Err(anyhow::anyhow!(
            "http_addr must be sepcified on server initialization"
        ));
    }
    if rpc_addr.is_none() {
        return Err(anyhow::anyhow!(
            "rpc_addr must be sepcified on server initialization"
        ));
    }
    if notifications_addr.is_none() {
        return Err(anyhow::anyhow!(
            "notifications_addr must be sepcified on server initialization"
        ));
    }
    */
    //std::fs::create_dir_all(&base_dir)?;
    /*
    let http_addr = instance_params.http_addr.unwrap();
    let rpc_addr = rpc_addr.unwrap();
    let notifications_addr = notifications_addr.unwrap();
    
    let instance_params = InstanceParams {
        http_addr: http_addr.clone(),
        rpc_addr: rpc_addr.clone(),
        notifications_addr: notifications_addr.clone(),
        tls_config,
    };

    let instance_params_json = serde_json::to_string(&instance_params)?;

    tokio::fs::write(
        base_dir.as_ref().join("instance_params.json"),
        instance_params_json.as_bytes(),
    )
    .await?;
    */
    let (app, http_handle,toy_rpc_handle,notifications_handle) 
      = do_init_rxqlite(base_dir, &mut instance_params).await?;

    if leader {
        let mut nodes = BTreeMap::new();
        let node = Node {
            api_addr: instance_params.http_addr.clone(),
            rpc_addr: instance_params.rpc_addr.clone(),
        };
        nodes.insert(app.id, node);
        app.raft.initialize(nodes).await?;

        if members.len() > 0 {
            let mut member_ship: BTreeSet<NodeId> =
                members.iter().map(|(node_id, _, _)| *node_id).collect();

            for (node_id_, api_addr, rpc_addr) in members.into_iter() {
                let node = Node {
                    rpc_addr,
                    api_addr,
                };
                tracing::debug!(
                    "{}({}):adding learner : {}/{}",
                    file!(),
                    line!(),
                    node_id_,
                    node
                );
                app.raft.add_learner(node_id_, node, true).await?;
                tracing::debug!("{}({}):learner added: {}", file!(), line!(), node_id_);
            }
            /*
            loop {
              tracing::info!("{}({})",file!(),line!());
              let mut cluster_initialized = true;
              let membership_config = {
                let metrics = app.raft.metrics().borrow().clone();
                metrics.membership_config.clone()
              };
              let mut node_iter=membership_config.nodes();
              for node_id in member_ship.iter() {
                if node_iter.position(|(nid,_)| nid==node_id).is_none() {
                  cluster_initialized=false;
                  break;
                }
              }
              if cluster_initialized {
                break;
              }
              tokio::time::sleep(std::time::Duration::from_millis(250)).await;
              //poll_future(Pin::new(&mut handle), std::time::Duration::from_millis(250)).await;
            }
            */
            tracing::debug!("{}({}):changing membership", file!(), line!());
            member_ship.insert(app.id);
            app.raft.change_membership(member_ship, false).await?;

            tracing::debug!("{}({}):membership changed", file!(), line!());
        }
    }

    tokio::select! {
      res = http_handle => {
        if let Err(err) = res {
          tracing::error!("http server error: {}",err);
        }
      }
      _ = toy_rpc_handle => {
      }
      _ = notifications_handle => {
      }
      _ = signal::ctrl_c() => {
      }
    }
    app.raft.shutdown().await?;
    Ok(())
}

pub async fn start_rxqlite<P>(
    base_dir: P,
    mut instance_params: InstanceParams,
) -> anyhow::Result<()>
where
    P: AsRef<Path>,
{
    /*
    let tls_instance_params_json =
        tokio::fs::read_to_string(base_dir.as_ref().join("instance_params.json")).await?;
    let instance_params: InstanceParams = serde_json::from_str(&tls_instance_params_json)?;
      */
    let (app, http_handle,toy_rpc_handle,notifications_handle) 
      = do_init_rxqlite(base_dir, &mut instance_params).await?;

    tokio::select! {
      res = http_handle => {
        if let Err(err) = res {
          tracing::error!("http server error: {}",err);
        }
      }
      _ = toy_rpc_handle => {
      }
      _ = notifications_handle => {
      }
      _ = signal::ctrl_c() => {
      }
    }
    app.raft.shutdown().await?;
    
    Ok(())
}

pub use rxqlite_common::{Message, MessageResponse, Value};
/*
#[derive(Default, Debug, Clone)]
pub struct ConnectOptions {
    pub leader_id: NodeId,
    pub leader_host: String,
    pub leader_port: u16,
    pub tls_config: Option<RSQliteClientTlsConfig>,
}
*/

//pub use rxqlite_lite_common::ConnectOptions;

pub type RXQLiteError = anyhow::Error;
/*
impl ConnectOptions {
    pub async fn connect(&self) -> Result<client::RXQLiteClient, RXQLiteError> {
        Ok(client::RXQLiteClient::with_options(self))
    }
}
*/
pub use client::RXQLiteClient as Connection;

pub use rxqlite_common::FromValueRef;

#[cfg(test)]
mod tests;

#[cfg(feature = "test-dependency")]
pub mod tests;
