use std::any::TypeId;

use std::collections::HashMap;

use std::fmt::Debug;

use std::ops::Deref;

use std::sync::atomic::{AtomicU32, Ordering};

use std::sync::Arc;

use std::time::Duration;

use async_trait::async_trait;

use bonsaidb_core::admin::{Admin, ADMIN_DATABASE_NAME};

use bonsaidb_core::api::{self, Api, ApiName};

use bonsaidb_core::arc_bytes::serde::Bytes;

use bonsaidb_core::arc_bytes::OwnedBytes;

use bonsaidb_core::connection::{

    AsyncStorageConnection, Database, HasSession, IdentityReference, Session,

};

use bonsaidb_core::networking::{

    AlterUserPermissionGroupMembership, AlterUserRoleMembership, AssumeIdentity, CreateDatabase,

    CreateUser, DeleteDatabase, DeleteUser, ListAvailableSchemas, ListDatabases, LogOutSession,

    MessageReceived, Payload, UnregisterSubscriber, CURRENT_PROTOCOL_VERSION,

};

use bonsaidb_core::permissions::Permissions;

use bonsaidb_core::schema::{Nameable, Schema, SchemaName, SchemaSummary, Schematic};

use bonsaidb_utils::fast_async_lock;

use flume::Sender;

use futures::future::BoxFuture;

use futures::{Future, FutureExt};

use parking_lot::Mutex;

#[cfg(not(target_arch = "wasm32"))]

use tokio::runtime::Handle;

use url::Url;

pub use self::remote_database::{AsyncRemoteDatabase, AsyncRemoteSubscriber};

#[cfg(not(target_arch = "wasm32"))]

pub use self::sync::{BlockingClient, BlockingRemoteDatabase, BlockingRemoteSubscriber};

use crate::builder::Async;

use crate::error::Error;

use crate::{ApiError, Builder};

#[cfg(not(target_arch = "wasm32"))]

mod quic_worker;

mod remote_database;

#[cfg(not(target_arch = "wasm32"))]

mod sync;

#[cfg(all(feature = "websockets", not(target_arch = "wasm32")))]

mod tungstenite_worker;

#[cfg(all(feature = "websockets", target_arch = "wasm32"))]

mod wasm_websocket_worker;

pub struct SubscriberMap(Arc<Mutex<HashMap<u64, flume::Sender<Message>>>>);

impl SubscriberMap {

}

impl Deref for SubscriberMap {

    type Target = Mutex<HashMap<u64, flume::Sender<Message>>>;

}

use bonsaidb_core::circulate::Message;

#[cfg(all(feature = "websockets", not(target_arch = "wasm32")))]

pub type WebSocketError = tokio_tungstenite::tungstenite::Error;

#[cfg(all(feature = "websockets", target_arch = "wasm32"))]

pub type WebSocketError = wasm_websocket_worker::WebSocketError;

/// Client for connecting to a BonsaiDb server.

///

/// ## How this type automatically reconnects

///

/// This type is designed to automatically reconnect if the underlying network

/// connection has been lost. When a disconnect happens, the error that caused

/// the disconnection will be returned to at least one requestor. If multiple

/// pending requests are outstanding, all remaining pending requests will have

/// an [`Error::Disconnected`](bonsaidb_core::networking::Error::Disconnected)

/// returned. This allows the application to detect when a networking issue has

/// arisen.

///

/// If the disconnect happens while the client is completely idle, the next

/// request will report the disconnection error. The subsequent request will

/// cause the client to begin reconnecting again.

///

/// When unauthenticated, this reconnection behavior is mostly transparent --

/// disconnection errors can be shown to the user, and service will be restored

/// automatically. However, when dealing with authentication, the client does

/// not store credentials to be able to send them again when reconnecting. This

/// means that the existing client handles will lose their authentication when

/// the network connection is broken. The current authentication status can be

/// checked using [`HasSession::session()`].

///

/// ## Connecting via QUIC

///

/// The URL scheme to connect via QUIC is `bonsaidb`. If no port is specified,

/// port 5645 is assumed.

///

/// ### With a valid TLS certificate

///

/// ```rust

/// # use bonsaidb_client::{AsyncClient, fabruic::Certificate, url::Url};

/// # async fn test_fn() -> anyhow::Result<()> {

/// let client = AsyncClient::build(Url::parse("bonsaidb://my-server.com")?).build()?;

/// # Ok(())

/// # }

/// ```

///

/// ### With a Self-Signed Pinned Certificate

///

/// When using `install_self_signed_certificate()`, clients will need the

/// contents of the `pinned-certificate.der` file within the database. It can be

/// specified when building the client:

///

/// ```rust

/// # use bonsaidb_client::{AsyncClient, fabruic::Certificate, url::Url};

/// # async fn test_fn() -> anyhow::Result<()> {

/// let certificate =

///     Certificate::from_der(std::fs::read("mydb.bonsaidb/pinned-certificate.der")?)?;

/// let client = AsyncClient::build(Url::parse("bonsaidb://localhost")?)

///     .with_certificate(certificate)

///     .build()?;

/// # Ok(())

/// # }

/// ```

///

/// ## Connecting via WebSockets

///

/// WebSockets are built atop the HTTP protocol. There are two URL schemes for

/// WebSockets:

///

/// - `ws`: Insecure WebSockets. Port 80 is assumed if no port is specified.

/// - `wss`: Secure WebSockets. Port 443 is assumed if no port is specified.

///

/// ### Without TLS

///

/// ```rust

/// # use bonsaidb_client::{AsyncClient, fabruic::Certificate, url::Url};

/// # async fn test_fn() -> anyhow::Result<()> {

/// let client = AsyncClient::build(Url::parse("ws://localhost")?).build()?;

/// # Ok(())

/// # }

/// ```

///

/// ### With TLS

///

/// ```rust

/// # use bonsaidb_client::{AsyncClient, fabruic::Certificate, url::Url};

/// # async fn test_fn() -> anyhow::Result<()> {

/// let client = AsyncClient::build(Url::parse("wss://my-server.com")?).build()?;

/// # Ok(())

/// # }

/// ```

///

/// ## Using a `Api`

///

/// Our user guide has a [section on creating and

/// using](https://dev.bonsaidb.io/release/guide/about/access-models/custom-api-server.html)

/// an [`Api`](api::Api).

///

/// ```rust

/// # use bonsaidb_client::{AsyncClient, fabruic::Certificate, url::Url};

/// // `bonsaidb_core` is re-exported to `bonsaidb::core` or `bonsaidb_client::core`.

/// use bonsaidb_core::api::{Api, ApiName, Infallible};

/// use bonsaidb_core::schema::Qualified;

/// use serde::{Deserialize, Serialize};

///

/// #[derive(Serialize, Deserialize, Debug)]

/// pub struct Ping;

///

/// #[derive(Serialize, Deserialize, Clone, Debug)]

/// pub struct Pong;

///

/// impl Api for Ping {

///     type Error = Infallible;

///     type Response = Pong;

///

///     fn name() -> ApiName {

///         ApiName::private("ping")

///     }

/// }

///

/// # async fn test_fn() -> anyhow::Result<()> {

/// let client = AsyncClient::build(Url::parse("bonsaidb://localhost")?).build()?;

/// let Pong = client.send_api_request(&Ping).await?;

/// # Ok(())

/// # }

/// ```

///

/// ### Receiving out-of-band messages from the server

///

/// If the server sends a message that isn't in response to a request, the

/// client will invoke it's [api callback](Builder::with_api_callback):

///

/// ```rust

/// # use bonsaidb_client::{AsyncClient, ApiCallback, fabruic::Certificate, url::Url};

/// # // `bonsaidb_core` is re-exported to `bonsaidb::core` or `bonsaidb_client::core`.

/// # use bonsaidb_core::{api::{Api, Infallible, ApiName}, schema::{Qualified}};

/// # use serde::{Serialize, Deserialize};

/// # #[derive(Serialize, Deserialize, Debug)]

/// # pub struct Ping;

/// # #[derive(Serialize, Deserialize, Clone, Debug)]

/// # pub struct Pong;

/// # impl Api for Ping {

/// #     type Response = Pong;

/// #     type Error = Infallible;

/// #

/// #     fn name() -> ApiName {

/// #         ApiName::private("ping")

/// #     }

/// # }

/// # async fn test_fn() -> anyhow::Result<()> {

/// let client = AsyncClient::build(Url::parse("bonsaidb://localhost")?)

///     .with_api_callback(ApiCallback::<Ping>::new(|result: Pong| async move {

///         println!("Received out-of-band Pong");

///     }))

///     .build()?;

/// # Ok(())

/// # }

/// ```

pub struct AsyncClient {

    pub(crate) data: Arc<Data>,

    session: ClientSession,

    request_timeout: Duration,

}

impl Drop for AsyncClient {

}

impl PartialEq for AsyncClient {

}

pub struct Data {

    request_sender: Sender<PendingRequest>,

    effective_permissions: Mutex<Option<Permissions>>,

    schemas: Mutex<HashMap<TypeId, Arc<Schematic>>>,

    connection_counter: Arc<AtomicU32>,

    request_id: AtomicU32,

    subscribers: SubscriberMap,

}

impl AsyncClient {

    /// Returns a builder for a new client connecting to `url`.

    /// Initialize a client connecting to `url`. This client can be shared by

    /// cloning it. All requests are done asynchronously over the same

    /// connection.

    ///

    /// If the client has an error connecting, the first request made will

    /// present that error. If the client disconnects while processing requests,

    /// all requests being processed will exit and return

    /// [`Error::Disconnected`](bonsaidb_core::networking::Error::Disconnected).

    /// The client will automatically try reconnecting.

    ///

    /// The goal of this design of this reconnection strategy is to make it

    /// easier to build resilliant apps. By allowing existing Client instances

    /// to recover and reconnect, each component of the apps built can adopt a

    /// "retry-to-recover" design, or "abort-and-fail" depending on how critical

    /// the database is to operation.

    /// Initialize a client connecting to `url` with `certificate` being used to

    /// validate and encrypt the connection. This client can be shared by

    /// cloning it. All requests are done asynchronously over the same

    /// connection.

    ///

    /// If the client has an error connecting, the first request made will

    /// present that error. If the client disconnects while processing requests,

    /// all requests being processed will exit and return

    /// [`Error::Disconnected`]. The client will automatically try reconnecting.

    ///

    /// The goal of this design of this reconnection strategy is to make it

    /// easier to build resilliant apps. By allowing existing Client instances

    /// to recover and reconnect, each component of the apps built can adopt a

    /// "retry-to-recover" design, or "abort-and-fail" depending on how critical

    /// the database is to operation.

            #[cfg(feature = "websockets")]

        }

    #[cfg(not(target_arch = "wasm32"))]

    #[cfg(all(feature = "websockets", not(target_arch = "wasm32")))]

    #[cfg(all(feature = "websockets", target_arch = "wasm32"))]

    fn new_websocket_client(

        server: ConnectionInfo,

        protocol_version: &'static str,

        custom_apis: HashMap<ApiName, Option<Arc<dyn AnyApiCallback>>>,

    ) -> Self {

        let (request_sender, request_receiver) = flume::unbounded();

        let connection_counter = Arc::new(AtomicU32::default());

        wasm_websocket_worker::spawn_client(

            Arc::new(server.url),

            protocol_version,

            request_receiver,

            Arc::new(custom_apis),

            server.subscribers.clone(),

            connection_counter.clone(),

            None,

            server.connect_timeout,

        );

        #[cfg(feature = "test-util")]

        let background_task_running = Arc::new(AtomicBool::new(true));

        Self {

            data: Arc::new(Data {

                request_sender,

                schemas: Mutex::default(),

                request_id: AtomicU32::default(),

                connection_counter,

                effective_permissions: Mutex::default(),

                subscribers: server.subscribers,

                #[cfg(feature = "test-util")]

                background_task_running,

            }),

            session: ClientSession::default(),

            request_timeout: server.request_timeout,

        }

    }

        #[cfg(target_arch = "wasm32")]

        let result = {

            use wasm_bindgen::JsCast;

            let (timeout_sender, mut timeout_receiver) = futures::channel::oneshot::channel();

            // Install the timeout.

            {

                if let Some(window) = web_sys::window() {

                    let timeout = wasm_bindgen::closure::Closure::once_into_js(move || {

                        let _result = timeout_sender.send(());

                    });

                    let _: Result<_, _> = window

                        .set_timeout_with_callback_and_timeout_and_arguments_0(

                            timeout.as_ref().unchecked_ref(),

                            self.request_timeout

                                .as_millis()

                                .try_into()

                                .unwrap_or(i32::MAX),

                        );

                }

            }

            futures::select! {

                result = result_receiver.recv_async() => Ok(result),

                _ = timeout_receiver => Err(Error::Network(bonsaidb_core::networking::Error::RequestTimeout)),

            }

        };

        #[cfg(not(target_arch = "wasm32"))]

        }

    #[cfg(not(target_arch = "wasm32"))]

    /// Sends an api `request`.

    #[cfg(not(target_arch = "wasm32"))]

    /// Returns the current effective permissions for the client. Returns None

    /// if unauthenticated.

    #[must_use]

        );

    #[cfg(not(target_arch = "wasm32"))]

        } else {

        };

    /// Sets this instance's request timeout.

    ///

    /// Each client has its own timeout. When cloning a client, this timeout

    /// setting will be copied to the clone.

}

impl HasSession for AsyncClient {

}

#[async_trait]

impl AsyncStorageConnection for AsyncClient {

    type Authenticated = Self;

    type Database = AsyncRemoteDatabase;

        Ok(self

            .send_api_request(&DeleteUser {

            })

    #[cfg(feature = "password-hashing")]

        Ok(self

            .send_api_request(&bonsaidb_core::networking::SetUserPassword {

            })

    #[cfg(any(feature = "token-authentication", feature = "password-hashing"))]

        self.send_api_request(&AlterUserPermissionGroupMembership {

            should_be_member: true,

        })

        self.send_api_request(&AlterUserPermissionGroupMembership {

            should_be_member: false,

        })

        self.send_api_request(&AlterUserRoleMembership {

            should_be_member: true,

        })

        self.send_api_request(&AlterUserRoleMembership {

            should_be_member: false,

        })

}

type OutstandingRequestMap = HashMap<u32, PendingRequest>;

type OutstandingRequestMapHandle = Arc<async_lock::Mutex<OutstandingRequestMap>>;

type PendingRequestResponder = Sender<Result<Bytes, Error>>;

pub struct PendingRequest {

    request: Payload,

    responder: PendingRequestResponder,

}

    } else {

    }

trait ApiWrapper<Response>: Send + Sync {

    fn invoke(&self, response: Response) -> BoxFuture<'static, ()>;

}

/// A callback that is invoked when an [`Api::Response`](Api::Response)

/// value is received out-of-band (not in reply to a request).

pub struct ApiCallback<Api: api::Api> {

    generator: Box<dyn ApiWrapper<Api::Response>>,

}

/// The trait bounds required for the function wrapped in a [`ApiCallback`].

pub trait ApiCallbackFn<Request, F>: Fn(Request) -> F + Send + Sync + 'static {}

impl<T, Request, F> ApiCallbackFn<Request, F> for T where T: Fn(Request) -> F + Send + Sync + 'static

{}

struct ApiFutureBoxer<Response: Send + Sync, F: Future<Output = ()> + Send + Sync>(

    Box<dyn ApiCallbackFn<Response, F>>,

);

impl<Response: Send + Sync, F: Future<Output = ()> + Send + Sync + 'static> ApiWrapper<Response>

    for ApiFutureBoxer<Response, F>

{

}

impl<Api: api::Api> ApiCallback<Api> {

    /// Returns a new instance wrapping the provided function.

    /// Returns a new instance wrapping the provided function, passing a clone

    /// of `context` as the second parameter. This is just a convenience wrapper

    /// around `new()` that produces more readable code when needing to access

    /// external information inside of the callback.

}

#[async_trait]

pub trait AnyApiCallback: Send + Sync + 'static {

    /// An out-of-band `response` was received. This happens when the server

    /// sends a response that isn't in response to a request.

    async fn response_received(&self, response: Bytes);

}

#[async_trait]

impl<Api: api::Api> AnyApiCallback for ApiCallback<Api> {

            }

        }

}

pub struct ClientSession {

    session: Arc<Session>,

    connection_id: u32,

}

struct ConnectionInfo {

    pub url: Url,

    pub subscribers: SubscriberMap,

    #[cfg_attr(target_arch = "wasm32", allow(dead_code))]

    pub connect_timeout: Duration,

    pub request_timeout: Duration,

}