use super::*;

/// Before a [`Stream`] can be fully open, its receive callback must be set through
/// [`StreamCandidate::open`] or [`StreamCandidate::open_write_only`].
///
/// If a [`StreamCandidate`] is dropped before it is fully open, the underlying [`Stream`] will
/// be closed, notifying the remote peer.
///
/// Moreover, an [`Acceptor`] may wait for its yielded [`StreamCandidate`] to be opened before
/// notifying the connecting peer of its acceptance, preventing the opposing [`ConnectionFuture`]
/// from resolving.
///
/// Note that [`ConnectionBuilder`](ConnectionBuilder::on_receive_factory) and
/// [`AcceptorBuilder`](AcceptorBuilder::on_receive_factory), which are the only sources of
/// [`StreamCandidate`]s, both have a `on_receive_factory` method: this method allows you to set a
/// common policy for [`StreamCandidate::open`]ing the [`Stream`], which is generally more
/// convenient and may spare you some head-scratching with regard to the above note on
/// [`ConnectionFuture`]'s resolution.
pub struct StreamCandidate<Arguments> {
    pub(crate) inner: datastreams_ffi::StreamCandidate,
    pub(crate) arguments: Arguments,
}

impl<Arguments: IArguments> core::fmt::Debug for StreamCandidate<Arguments> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("StreamCandidate")
            .field("peer_pubkey", &self.peer_pubkey())
            .field("topic", &self.topic())
            .field("arguments", &self.arguments())
            .finish()
    }
}

impl StreamCandidate<NoArguments> {
    /// Sets the arguments to be sent when accepting the stream.
    ///
    /// This is only available when receiving a [`StreamCandidate`] through an [`Acceptor`], as
    /// clients will instead set arguments through [`ConnectionBuilder::with_arguments`].
    ///
    /// Just like when set with [`ConnectionBuilder::with_arguments`], the arguments set here will
    /// be visible to the remote peer through [`StreamCandidate::arguments`] as soon as the
    /// candidate is raised; and will be transferred reliably regardless of the opened stream's
    /// reliability.
    ///
    /// # Example
    /// ```
    /// # mod dittolive_ditto {
    /// #   pub mod preview {
    /// #       pub use dittolive_ditto_base::*;
    /// #   }
    /// # }
    /// # use dittolive_ditto::preview::{datastreams::{Endpoint, Payload, Stream}, peer_pubkey::PeerPubkey};
    /// # use safer_ffi::bytes::Bytes;
    /// # use std::fs::File;
    /// # async fn usage(
    /// #     endpoint: Endpoint,
    /// #     send_file: impl Fn(File, Stream<()>, bool) + Send + Sync + 'static,
    /// #     resolve_file: impl Fn(&str) -> Result<File, ()> + Send + Sync + 'static
    /// # ) {
    /// #[derive(serde::Deserialize)]
    /// struct FileTransferArguments {
    ///     id: String,
    ///     compression: Vec<String>
    /// }
    /// let _acceptor = endpoint.bind_topic("file_transfer")
    ///     .finish_with(move |candidate| {
    ///     let Some(args) = candidate.arguments() else { return; /* No arguments provided */ };
    ///     let Ok(FileTransferArguments { id, compression }) =
    ///         serde_json::from_slice(&args) else {
    ///            return; // Invalid arguments
    ///        };
    ///     let Ok(file) = resolve_file(&id) else {return; /* File not found */};
    ///     let use_lz4 = compression.iter().any(|c| c=="lz4");
    ///     let stream = candidate.with_arguments(use_lz4.then(|| Payload::from("lz4")))
    ///                           .open_write_only();
    ///     send_file(file, stream, use_lz4);
    /// });
    /// # }
    /// ```
    pub fn with_arguments<Arguments: IArguments>(
        self,
        arguments: Arguments,
    ) -> StreamCandidate<Arguments> {
        StreamCandidate {
            inner: self.inner,
            arguments,
        }
    }
}

impl<Arguments: IArguments> StreamCandidate<Arguments> {
    /// The public key of the peer this stream is connected to.
    pub fn peer_pubkey(&self) -> PeerPubkey {
        self.inner.inner.peer_pubkey()
    }

    /// The [`Topic`] of the stream.
    pub fn topic(&self) -> Topic {
        self.inner.inner.topic()
    }

    /// Returns the arguments set by the remote peer when initiating the stream, if any.
    ///
    /// Arguments may have been set either through [`ConnectionBuilder::with_arguments`] for the
    /// client, or through [`StreamCandidate::with_arguments`] for the server.
    ///
    /// You can find examples of usage for [`StreamCandidate::arguments`] in the documentation of
    /// both [`ConnectionBuilder::with_arguments`] and [`StreamCandidate::with_arguments`].
    pub fn arguments(&self) -> Option<Payload> {
        self.inner.inner.arguments()
    }

    /// Opens the stream with no intent of receiving data: all received messages on the stream
    /// will be dropped immediately.
    pub fn open_write_only(self) -> Stream<()> {
        let Self { inner, arguments } = self;
        inner.open_write_only(arguments.into_arguments())
    }
    /// Opens the stream, setting `handler` as its receiver.
    ///
    /// Note that once a stream is open, the callback can never be removed: it will be called
    /// until all instances descending from the returned [`Stream`] have been destroyed.
    pub fn open<OverloadId, Handler>(self, handler: Handler) -> Stream<Handler::Receiver>
    where
        Handler: IntoChannel<OverloadId, Inbound>,
        Handler::Sender: Channel<OverloadId, Inbound>,
    {
        let Self { inner, arguments } = self;
        inner.open(handler, arguments.into_arguments())
    }

    /// Returns a reference to the underlying [`datastreams_ffi::StreamCandidate`].
    ///
    /// There's generally never a reason to use this other than for Ditto's internal codebase,
    /// you _will_ shoot your foot off if you use both StreamCandidate types in your code.
    pub fn inner(&self) -> &datastreams_ffi::StreamCandidate {
        &self.inner
    }

    /// Disallows setting arguments when accepting this stream.
    ///
    /// This is convenient when you want to handle new stream candidates the same way regardless of
    /// whether you are the client or the server for that stream.
    pub fn disallow_arguments(self) -> StreamCandidate<ArgumentsDisallowed> {
        StreamCandidate {
            inner: self.inner,
            arguments: ArgumentsDisallowed,
        }
    }
}
