use super::*;

/// Used by the `Bus` API to turn handlers into a pair of a `Sender` which is called when a new
/// event must be handled, and a [`IntoChannel::Receiver`] which is often wrapped in the
/// subscription handle to that event, encouraging you to keep said handle available.
///
/// The most common implementations for this pattern require `Sender` to be a closure:
/// - Closures expose `()` as their receiver and themselves as the `Sender`, letting you use
///   closures in any API that takes an [`impl IntoChannel`](IntoChannel).
/// - Channel pairs, such as the return value of [`std::sync::mpsc::channel`], are split into a
///   closure that sends its input onto the send end of the channel, and the receive end of the
///   channel is returned as the [`IntoChannel::Receiver`]. This commonly manifests in letting you
///   use channel constructors such as `tokio::sync::mpsc::unbounded_channel()` as the argument for
///   methods that take an [`impl IntoChannel`](IntoChannel).
///
/// # What is this `OverloadId`?
/// The `OverloadId` generic parameter has no semantic role, and exists solely to allow some
/// type magic:
/// - It lets you bypass the [orphan rule](https://doc.rust-lang.org/book/ch10-02-traits.html#:~:text=more%20specifically%20the-,orphan%20rule,-%2C%20so%20named%20because)
///   by using a type defined by your own crate as the `OverloadId`. You can use this to add support
///   to your favorite mpsc channel implementation if it's not covered by our own, for example.
/// - It allows to write multiple generic implementations for `IntoChannel<_, Sender>` that Rust
///   thinks conflict even though they don't in practice. While code that uses that trait now needs
///   to be generic over `OverloadId`, type inference will resolve it on its own provided there are
///   no conflicting implementations. Your `OverloadId` should still be public and understandable so
///   that picking the appropriate overload can be done easily should multiple overloads match the
///   requirement.
///
/// # Implementing [`IntoChannel`]
/// Here's how you _too_ can implement [`IntoChannel`], letting you use whatever flavor of
/// channel you like, regardless of whether or not we support it. Feel free to straight up
/// copy this implementation if you want `flume` support:
/// ```
/// # mod dittolive_ditto {
/// #   pub mod experimental {
/// #       pub use dittolive_ditto_base::*;
/// #   }
/// # }
/// use dittolive_ditto::experimental::bus::{
///     Channel, ConcurrentChannel, IntoChannel, OneShotChannel,
/// };
///
/// /// This allows us to bypass the orphan rule
/// pub struct MyOverloadId;
///
/// impl<T: core::marker::Send + 'static> IntoChannel<MyOverloadId, T>
///     for (flume::Sender<T>, flume::Receiver<T>)
/// {
///     type Sender = flume::Sender<T>;
///     type Receiver = flume::Receiver<T>;
///     fn into_channel(self) -> (flume::Sender<T>, Self::Receiver) {
///         self
///     }
/// }
/// impl<T: core::marker::Send + 'static> ConcurrentChannel<MyOverloadId, T> for flume::Sender<T> {
///     fn send_infallibly(&self, value: T) {
///         core::mem::drop(self.send(value))
///     }
/// }
/// impl<T: core::marker::Send + 'static> Channel<MyOverloadId, T> for flume::Sender<T> {
///     fn send_infallibly_mut(&mut self, value: T) {
///         core::mem::drop(self.send(value))
///     }
/// }
/// impl<T: core::marker::Send + 'static> OneShotChannel<MyOverloadId, T> for flume::Sender<T> {
///     fn send_infallibly_once(mut self, value: T) {
///         core::mem::drop(self.send(value))
///     }
/// }
/// ```
///
/// Notice how this doctest compiles despite owning neither the `IntoChannel`, `Channel` or
/// `OneShotChannel` traits, and not owning the types in `flume` either? This is thanks to the
/// fact that `MyOverloadId`, which it does own, is the first generic argument to each of
/// these traits.
///
/// But thanks to type inference, you'll never actually have to type `MyOverloadId` in any code
/// that passes `flume::unbounded()` to a function such as [`ConnectionBuilder::finish`]
/// (unless another crate defines the same implementations with `TheirOverloadId` in your
/// dependency tree).
pub trait IntoChannel<OverloadId, T> {
    /// The receiver side of the pair.
    type Receiver;
    /// The sender side of the pair.
    type Sender: OneShotChannel<OverloadId, T>;
    /// Splits the pair into its sender and receiver parts.
    fn into_channel(self) -> (Self::Sender, Self::Receiver);

    /// Map inputs through a function before passing it through the channel, allowing for
    /// example to add context to a payload before sending it through a global channel:
    ///
    /// ```
    /// # mod dittolive_ditto {
    /// #   pub mod experimental {
    /// #       pub use dittolive_ditto_base::*;
    /// #   }
    /// #   pub mod prelude {
    /// #       pub struct Ditto;
    /// #       impl Ditto {
    /// #           pub fn bus(&self) -> super::experimental::bus::Bus { unreachable!() }
    /// #       }
    /// #   }
    /// # }
    /// # use dittolive_ditto::prelude::Ditto;
    /// # use dittolive_ditto::experimental::{bus::{Acceptor, Bus, IntoChannel, Payload}, 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 bus = ditto.bus();
    /// let mut acceptor: Acceptor<_> = bus.bind_topic("logger").finish(channel()).expect("Topic should be unique");
    /// let (tx, mut rx) = channel();
    /// tokio::task::spawn(async move {
    ///     while let Some((peer, packet)) = rx.recv().await {
    ///         dbg!((peer, packet));
    ///     }
    /// });
    /// while let Some(stream) = acceptor.recv().await {
    ///     let peer = stream.peer_pubkey();
    ///     let add_context = move |payload| (peer.clone(), payload);
    ///     let stream = stream.open(tx.clone().map_input(add_context));
    ///     // Because a `StreamSubscriber` will unsubscribe its stream's callback as soon as it is dropped,
    ///     // keeping it around is generally important.
    ///     // Note that this is also the case with `Acceptor`, but by wrapping it around the channel's receiver,
    ///     // we've given you an explicit reason to keep it around.
    ///     stream.detach()
    /// }
    /// # }
    /// ```
    fn map_input<U, F: FnOnce(U) -> T + Send + Sync + 'static>(
        self,
        convert: F,
    ) -> Map<U, T, F, OverloadId, Self>
    where
        Self: Sized,
        OverloadId: 'static,
        T: 'static,
        U: 'static,
        Map<U, T, F, OverloadId, Self>: IntoChannel<ChannelPair, U>,
    {
        Map {
            convert,
            handler: self,
            marker: Default::default(),
        }
    }
}

/// Used when `Self` already implements [`OneShotChannel`] (and possibly [`Channel`]), returning
/// `()` as the [`IntoChannel::Receiver`].
pub struct NoReceiver;

/// Converts a `(tx, rx)` pair into a closure that pushes on said channel, and uses `rx` as the
/// [`IntoChannel::Receiver`].
pub struct ChannelPair;

/// A channel that may only be used once.
///
/// For example, [`ConnectionBuilder::finish`] may only yield a single connection result.
pub trait OneShotChannel<OverloadId, T>: Send + Sync + 'static {
    /// Send the value over the channel.
    fn send_infallibly_once(self, value: T);
}

/// A channel that may be used multiple times.
///
/// This is used when multiple values may be yielded by a given operation, but that the processing
/// of individual values isn't expected to be a long-lived task, such as the [`Inbound`] messages
/// yielded by an [open stream](StreamCandidate::open).
///
/// While this property _can_ be used to implement backpressure, this should be avoided, as doing so
/// may stall the executor.
pub trait Channel<OverloadId, T>: OneShotChannel<OverloadId, T> {
    /// Send the value over the channel.
    fn send_infallibly_mut(&mut self, value: T);
}

/// A channel that may be used multiple times, and concurrently.
///
/// This is used when multiple values may be yielded by a given operation, and that it's expected
/// that the yielded values may often be the object of a long-lived task.
///
/// This property being required by an API is a strong hint that the API will call the handler from
/// a dedicated task that allows blocking without disrupting the underlying executor. This does mean
/// that backpressure _cannot_ be applied by stalling in `send_infallibly`.
///
/// For example, this allows treating any closure passed to [`AcceptorBuilder::finish`] (or its
/// [`finish_with`](AcceptorBuilder::finish_with) counterpart) as if it was passed to
/// [`std::thread::spawn`].
pub trait ConcurrentChannel<OverloadId, T>: Channel<OverloadId, T> {
    /// Send the value over the channel.
    fn send_infallibly(&self, value: T);
}

impl<Sender: OneShotChannel<NoReceiver, T>, T> IntoChannel<NoReceiver, T> for Sender {
    type Receiver = ();
    type Sender = Self;
    fn into_channel(self) -> (Sender, Self::Receiver) {
        (self, ())
    }
}

impl<T: Send + 'static> IntoChannel<ChannelPair, T>
    for (
        tokio::sync::oneshot::Sender<T>,
        tokio::sync::oneshot::Receiver<T>,
    )
{
    type Receiver = tokio::sync::oneshot::Receiver<T>;
    type Sender = tokio::sync::oneshot::Sender<T>;
    fn into_channel(
        self,
    ) -> (
        tokio::sync::oneshot::Sender<T>,
        tokio::sync::oneshot::Receiver<T>,
    ) {
        self
    }
}

impl<T: Send + 'static> IntoChannel<ChannelPair, T>
    for (std::sync::mpsc::Sender<T>, std::sync::mpsc::Receiver<T>)
{
    type Receiver = std::sync::mpsc::Receiver<T>;
    type Sender = std::sync::mpsc::Sender<T>;
    fn into_channel(self) -> (std::sync::mpsc::Sender<T>, Self::Receiver) {
        self
    }
}

impl<T: Send + 'static> IntoChannel<ChannelPair, T>
    for (std::sync::mpsc::SyncSender<T>, std::sync::mpsc::Receiver<T>)
{
    type Receiver = std::sync::mpsc::Receiver<T>;
    type Sender = std::sync::mpsc::SyncSender<T>;
    fn into_channel(self) -> (std::sync::mpsc::SyncSender<T>, Self::Receiver) {
        self
    }
}

impl<T: Send + 'static> IntoChannel<ChannelPair, T>
    for (
        tokio::sync::mpsc::UnboundedSender<T>,
        tokio::sync::mpsc::UnboundedReceiver<T>,
    )
{
    type Receiver = tokio::sync::mpsc::UnboundedReceiver<T>;
    type Sender = tokio::sync::mpsc::UnboundedSender<T>;
    fn into_channel(self) -> (tokio::sync::mpsc::UnboundedSender<T>, Self::Receiver) {
        self
    }
}

impl<T, F: Fn(T) + Send + Sync + 'static> ConcurrentChannel<NoReceiver, T> for F {
    fn send_infallibly(&self, value: T) {
        self(value);
    }
}

impl<T, F: FnMut(T) + Send + Sync + 'static> Channel<NoReceiver, T> for F {
    fn send_infallibly_mut(&mut self, value: T) {
        self(value);
    }
}

impl<T: Send + 'static> ConcurrentChannel<ChannelPair, T> for std::sync::mpsc::Sender<T> {
    fn send_infallibly(&self, value: T) {
        core::mem::drop(self.send(value))
    }
}
impl<T: Send + 'static> Channel<ChannelPair, T> for std::sync::mpsc::Sender<T> {
    fn send_infallibly_mut(&mut self, value: T) {
        core::mem::drop(self.send(value))
    }
}
impl<T: Send + 'static> ConcurrentChannel<NoReceiver, T> for std::sync::mpsc::Sender<T> {
    fn send_infallibly(&self, value: T) {
        core::mem::drop(self.send(value))
    }
}
impl<T: Send + 'static> Channel<NoReceiver, T> for std::sync::mpsc::Sender<T> {
    fn send_infallibly_mut(&mut self, value: T) {
        core::mem::drop(self.send(value))
    }
}

impl<T: Send + 'static> ConcurrentChannel<ChannelPair, T> for std::sync::mpsc::SyncSender<T> {
    fn send_infallibly(&self, value: T) {
        core::mem::drop(self.send(value))
    }
}
impl<T: Send + 'static> Channel<ChannelPair, T> for std::sync::mpsc::SyncSender<T> {
    fn send_infallibly_mut(&mut self, value: T) {
        core::mem::drop(self.send(value))
    }
}
impl<T: Send + 'static> ConcurrentChannel<NoReceiver, T> for std::sync::mpsc::SyncSender<T> {
    fn send_infallibly(&self, value: T) {
        core::mem::drop(self.send(value))
    }
}
impl<T: Send + 'static> Channel<NoReceiver, T> for std::sync::mpsc::SyncSender<T> {
    fn send_infallibly_mut(&mut self, value: T) {
        core::mem::drop(self.send(value))
    }
}

impl<T: Send + 'static> ConcurrentChannel<ChannelPair, T>
    for tokio::sync::mpsc::UnboundedSender<T>
{
    fn send_infallibly(&self, value: T) {
        core::mem::drop(self.send(value))
    }
}
impl<T: Send + 'static> Channel<ChannelPair, T> for tokio::sync::mpsc::UnboundedSender<T> {
    fn send_infallibly_mut(&mut self, value: T) {
        core::mem::drop(self.send(value))
    }
}
impl<T: Send + 'static> ConcurrentChannel<NoReceiver, T> for tokio::sync::mpsc::UnboundedSender<T> {
    fn send_infallibly(&self, value: T) {
        core::mem::drop(self.send(value))
    }
}
impl<T: Send + 'static> Channel<NoReceiver, T> for tokio::sync::mpsc::UnboundedSender<T> {
    fn send_infallibly_mut(&mut self, value: T) {
        core::mem::drop(self.send(value))
    }
}

impl<T, F: FnOnce(T) + Send + Sync + 'static> OneShotChannel<NoReceiver, T> for F {
    fn send_infallibly_once(self, value: T) {
        self(value);
    }
}

impl<T: Send + 'static> OneShotChannel<ChannelPair, T> for std::sync::mpsc::Sender<T> {
    fn send_infallibly_once(self, value: T) {
        core::mem::drop(self.send(value))
    }
}
impl<T: Send + 'static> OneShotChannel<NoReceiver, T> for std::sync::mpsc::Sender<T> {
    fn send_infallibly_once(self, value: T) {
        core::mem::drop(self.send(value))
    }
}

impl<T: Send + 'static> OneShotChannel<ChannelPair, T> for std::sync::mpsc::SyncSender<T> {
    fn send_infallibly_once(self, value: T) {
        core::mem::drop(self.send(value))
    }
}
impl<T: Send + 'static> OneShotChannel<NoReceiver, T> for std::sync::mpsc::SyncSender<T> {
    fn send_infallibly_once(self, value: T) {
        core::mem::drop(self.send(value))
    }
}

impl<T: Send + 'static> OneShotChannel<ChannelPair, T> for tokio::sync::mpsc::UnboundedSender<T> {
    fn send_infallibly_once(self, value: T) {
        core::mem::drop(self.send(value))
    }
}
impl<T: Send + 'static> OneShotChannel<NoReceiver, T> for tokio::sync::mpsc::UnboundedSender<T> {
    fn send_infallibly_once(self, value: T) {
        core::mem::drop(self.send(value))
    }
}

impl<T: Send + 'static> OneShotChannel<ChannelPair, T> for tokio::sync::oneshot::Sender<T> {
    fn send_infallibly_once(self, value: T) {
        core::mem::drop(self.send(value))
    }
}
impl<T: Send + 'static> OneShotChannel<NoReceiver, T> for tokio::sync::oneshot::Sender<T> {
    fn send_infallibly_once(self, value: T) {
        core::mem::drop(self.send(value))
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn impls_os_channel<OverloadId, T: IntoChannel<OverloadId, u8>>(_: T)
    where
        T::Sender: OneShotChannel<OverloadId, u8>,
    {
    }
    fn impls_channel<OverloadId, T: IntoChannel<OverloadId, u8>>(
        t: T,
        check: impl FnOnce(T::Receiver),
    ) where
        T::Sender: Channel<OverloadId, u8> + ConcurrentChannel<OverloadId, u8>,
    {
        let (mut tx, rx) = t.into_channel();
        tx.send_infallibly(1);
        tx.send_infallibly_mut(2);
        check(rx);
    }
    #[test]
    fn verify_channel_impls() {
        impls_channel(|_| (), |()| ());
        impls_channel(std::sync::mpsc::channel(), |rx| {
            assert_eq!(rx.recv().unwrap(), 1);
            assert_eq!(rx.recv().unwrap(), 2);
        });
        let (tx, rx) = std::sync::mpsc::channel();
        impls_channel(tx, |()| {
            assert_eq!(rx.recv().unwrap(), 1);
            assert_eq!(rx.recv().unwrap(), 2);
        });
        impls_channel(tokio::sync::mpsc::unbounded_channel(), |mut rx| {
            assert_eq!(rx.blocking_recv().unwrap(), 1);
            assert_eq!(rx.blocking_recv().unwrap(), 2);
        });
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        impls_channel(tx, |()| {
            assert_eq!(rx.blocking_recv().unwrap(), 1);
            assert_eq!(rx.blocking_recv().unwrap(), 2);
        });
        impls_os_channel(tokio::sync::oneshot::channel());
        impls_os_channel(tokio::sync::oneshot::channel().0);
        let map = |_| "hi";
        impls_channel(std::sync::mpsc::channel().map_input(map), |rx| {
            assert_eq!(rx.recv().unwrap(), "hi")
        });
    }
}
