#![warn(missing_docs, clippy::missing_safety_doc, clippy::missing_errors_doc)]

use std::{
    future::IntoFuture,
    num::NonZeroUsize,
    ops::{Deref, DerefMut},
    sync::Arc,
};

pub use datastreams_ffi::{ConnectionError, Endpoint, Payload, Reliability, SendStatus, Topic};
use safer_ffi::bytes::Bytes;

use crate::{
    datastreams_ffi::{
        self, Callback, IAcceptor, IEndpoint, ISendHandle, IStream, IStreamCandidate,
    },
    peer_pubkey::PeerPubkey,
};

impl Endpoint {
    /// Returns a builder for creating a new [`Acceptor`] for a given `topic` on this peer.
    ///
    /// An [`Acceptor`] is responsible for handling remotely initiated stream connections.
    ///
    /// You can only have 1 [`Acceptor`] per topic at any given time. The builder returns a
    /// handle to the [`Acceptor`], dropping the handle will remove the [`Acceptor`] from the
    /// endpoint.
    ///
    /// ```
    /// # mod dittolive_ditto {
    /// #   pub mod preview {
    /// #       pub use dittolive_ditto_base::*;
    /// #   }
    /// #   pub mod prelude {
    /// #       pub struct Ditto;
    /// #       impl Ditto {
    /// #           pub fn datastreams(&self) -> super::preview::datastreams::Endpoint { unreachable!() }
    /// #       }
    /// #   }
    /// # }
    /// # use dittolive_ditto::prelude::Ditto;
    /// # use dittolive_ditto::preview::{datastreams::{Acceptor, Endpoint, Reliability}, peer_pubkey::PeerPubkey};
    /// # use safer_ffi::bytes::Bytes;
    /// # async fn usage(ditto: &Ditto, target: PeerPubkey) {
    /// use tokio::sync::mpsc::unbounded_channel as channel; // (`std::sync::mpsc::channel` would also work)
    /// let endpoint = ditto.datastreams();
    /// let mut acceptor: Acceptor<_> = endpoint.bind_topic("ping")
    ///                                    .reliability(Reliability::Reliable)
    ///                                    .on_receive_factory(channel)
    ///                                    .finish(channel())
    ///                                   .expect("Topic should be unique");
    /// while let Some(mut stream) = acceptor.recv().await {
    ///     tokio::task::spawn(async move {
    ///         while let Some(packet) = stream.recv().await {
    ///             stream.message(packet).send();
    ///         }
    ///     });
    /// }
    /// # }
    /// ```
    pub fn bind_topic(
        &self,
        topic: impl TryInto<datastreams_ffi::Topic>,
    ) -> AcceptorBuilder<'_, ExactMatch> {
        AcceptorBuilder {
            endpoint: self,
            inner: ExactMatch(topic.try_into().ok()),
            reliability: datastreams_ffi::Reliability::Reliable,
            opener: (),
            marker: core::marker::PhantomData,
        }
    }

    /// Returns a builder for creating a new [`Stream`] to a remote peer.
    ///
    /// A `topic` is a unique identifier for the stream that must be agreed upon by both peers. It
    /// is possible to create multiple [`Stream`]s to the same peer with different topics.
    ///
    /// ```
    /// # mod dittolive_ditto {
    /// #   pub mod preview {
    /// #       pub use dittolive_ditto_base::*;
    /// #   }
    /// #   pub mod prelude {
    /// #       pub struct Ditto;
    /// #       impl Ditto {
    /// #           pub fn datastreams(&self) -> super::preview::datastreams::Endpoint { unreachable!() }
    /// #       }
    /// #   }
    /// # }
    /// # use dittolive_ditto::prelude::Ditto;
    /// # use dittolive_ditto::preview::{datastreams::{Endpoint, Reliability}, peer_pubkey::PeerPubkey};
    /// # use safer_ffi::bytes::Bytes;
    /// # async fn usage(ditto: &Ditto, target: PeerPubkey) {
    /// use tokio::sync::mpsc::unbounded_channel as channel; // (`std::sync::mpsc::channel` would also work)
    /// let endpoint = ditto.datastreams();
    /// let mut stream = endpoint.connect(target, "ping")
    ///                     .reliability(Reliability::Reliable)
    ///                     .on_receive_factory(channel)
    ///                     .finish_async().await.unwrap();
    /// for i in 0..10 {
    ///     let start = std::time::Instant::now();
    ///     stream.message("ping").send();
    ///     stream.recv().await;
    ///     println!("Ping {i} = {t:?}", t=start.elapsed());
    /// }
    /// # }
    /// ```
    pub fn connect(
        &self,
        peer: impl Into<PeerPubkey>,
        topic: impl TryInto<datastreams_ffi::Topic>,
    ) -> ConnectionBuilder<'_> {
        ConnectionBuilder {
            endpoint: self,
            peer: peer.into(),
            topic: topic.try_into().ok(),
            reliability: datastreams_ffi::Reliability::Reliable,
            enable_compression: false,
            timeout_ms: 0,
            on_receive: (),
            marker: core::marker::PhantomData,
            arguments: NoArguments,
        }
    }
}

pub use acceptor::*;
mod acceptor;

pub use connection_builder::*;
mod connection_builder;

pub use inbound::*;
mod inbound;

pub use stream_candidate::*;
mod stream_candidate;

pub use stream::*;
mod stream;

pub use message_builder::*;
mod message_builder;

pub use send_handle::*;
mod send_handle;

pub use helper_traits::*;
mod helper_traits;

pub use arguments::*;
mod arguments;

pub use datastreams_ffi::StreamStatus;
