use std::{
    hash::{Hash, Hasher},
    str::FromStr,
};

pub use safer_ffi::option::TaggedOption;
use safer_ffi::{
    derive_ReprC,
    dyn_traits::VirtualPtr,
    layout::{CLayoutOf, ReprC},
};

#[cfg(doc)]
use crate::bus::{IntoChannel, IntoPayload};
use crate::peer_pubkey::PeerPubkey;
pub type ConcurrentCallback<T> = safer_ffi::closure::ArcDynFn1<(), T>;
pub type Callback<T> = safer_ffi::closure::BoxDynFnMut1<(), T>;
pub type Continuation<'a, T> = safer_ffi::closure::BoxDynFnMut1<(), T>;

/// The Ditto Bus API uses [`Bytes`](safer_ffi::bytes::Bytes) to transfer data between
pub type Payload = safer_ffi::bytes::Bytes<'static>;

/// An inbound message.
///
/// Inbound messages currently only contain the message's payload, but metadata may be added future
/// updates.
#[derive_ReprC]
#[repr(C)]
#[derive(Clone)]
pub struct Inbound {
    pub payload: Payload,
}

/// # Ditto Data Streams
/// Ditto Data Streams allow you to send datagrams across a Ditto mesh.
///
/// This structure is the core accessor to the Ditto Data Streams, which may remind you of some TCP
/// APIs you've seen before (except this one can work without an IP connection too!).
///
/// While exploring the API from [`Bus::bind_topic`], [`Bus::connect`] and the examples their docs
/// contain is probably the easiest way to approach it, there are a few subjects that are global to
/// the API: giving them a look could be helpful.
///
/// The entire API is built around the builder pattern: a rather convenient way to explore it is to
/// simply let your code completion guide you around it.
///
/// ## Topics
/// Much like TCP and UDP have ports, Ditto Data Streams have `topic`s: instead of integers, you can
/// name your ports after anything you like!
///
/// The equivalent of binding a socket is to use [`Bus::bind_topic`], to which you'll provide a
/// topic (or name).
///
/// Any attempt to [`Bus::connect`] to your peer will have to specify on which topic the connection
/// should be made, the connection appearing as a new [`Stream`] for your [`Acceptor`].
///
/// ## [`Payload`] and [`IntoPayload`]
/// To carry data around, the Ditto Data Streams API uses [`Payload`]s, also known as
/// [`safer_ffi::bytes::Bytes`]: much like the wider known [`bytes::Bytes`](https://docs.rs/bytes/latest/bytes/struct.Bytes.html), it's a cheaply cloneable and sliceable chunk of contiguous memory... But it does have a few low-level tricks up its sleeve to be even more efficient.
///
/// And while it's easy enough to construct it from common owned slice types, [`IntoPayload`] is a
/// very friendly option to make serialization less verbose, yet explicit and flexible.
///
/// ## [`IntoChannel`]: We promise, it's not as scary as it looks
/// In this API, anytime we think you may want the choice between closures and channels to handle
/// certain events you could subscribe to, we've used the following signature pattern:
/// ```
/// # mod dittolive_ditto {
/// #   pub mod experimental {
/// #       pub use dittolive_ditto_base::*;
/// #   }
/// # }
/// # use dittolive_ditto::experimental::bus::{Channel, IntoChannel};
/// # struct Event; struct Subscriber<T>(T);
/// fn subscribe<OverloadId, Handler>(handler: Handler) -> Subscriber<Handler::Receiver>
/// where
///     Handler: IntoChannel<OverloadId, Event>,
///     Handler::Sender: Channel<OverloadId, Event>,
/// # {unimplemented!()}
/// ```
///
/// While this looks imposing, it's actually a very magic formula: it allows you to write little to
/// no boiler plate in 99% of cases, while giving you the flexibility to use any handler you please
/// and allowing us to find every trick possible to minimize the overhead of calling your handler.
///
/// In most cases, all you need to do to use such an API is to call it with
/// - a closure: `let handle = subscribe(|e: Event| println!("{e}"));` (you may want to keep
///   `handle` around, as dropping it will cancel that subscription).
/// - any channel implementation you like: `let handle = subscribe(std::sync::mpsc::channel())`
///   (here, `handle` will deref to the receiver half of the channel, letting you call
///   `handle.recv()` whenever you like).
///
/// `OverloadId` is actually a charm that tricks the compiler into letting you bypass the orphan
/// rule<sup>[(1)](https://doc.rust-lang.org/book/ch10-02-traits.html#:~:text=orphan%20rule)[(2)](https://github.com/Ixrec/rust-orphan-rules)</sup>, and is described in more detail in [`IntoChannel`]'s documentation, which also contains an
/// example of how to use that trick to add support for [`flume`](https://crates.io/crates/flume)'s excellent channel implementation
/// to Ditto's Data Streams API.
#[derive_ReprC]
#[repr(C)]
#[derive(Clone)]
pub struct Bus {
    pub inner: VirtualPtr<dyn IBus + Send + Sync>,
}

/// The Bus itself. It allows listening for and opening connections.
#[derive_ReprC(dyn, Clone)]
pub trait IBus: 'static {
    /// Binds a callback to a new topic.
    extern "C" fn bind_topic(
        &self,
        topic: MaybeTopic<'_>,
        reliability: Reliability,
        on_accept: ConcurrentCallback<StreamCandidate>,
    ) -> BindResult;

    /// Attempts to establish a connection, yielding a connection result once done.
    extern "C" fn connect(
        &self,
        to: PeerPubkey,
        topic: MaybeTopic<'_>,
        reliability: Reliability,
        request_compression: bool,
        then: Continuation<'static, ConnectionResult>,
    );
    /// Returns a hash of the bus's type ID, allowing internal code to identify the bus type and
    /// downcast it if necessary.
    ///
    /// There's never any point in using it outside of Ditto's codebase.
    extern "C" fn type_id(&self) -> u64 {
        type_id_hash::<Self>()
    }
}

pub fn type_id_hash<T: 'static + ?Sized>() -> u64 {
    let id = std::any::TypeId::of::<T>();
    let mut hasher = std::hash::DefaultHasher::new();
    id.hash(&mut hasher);
    hasher.finish()
}

#[derive_ReprC]
#[repr(C)]
#[derive(Clone)]
pub struct Acceptor {
    pub inner: VirtualPtr<dyn IAcceptor + Send + Sync>,
}
impl core::fmt::Debug for Acceptor {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Acceptor")
            .field("topic", &self.inner.topic())
            .field("reliability", &self.inner.reliability())
            .finish()
    }
}

#[derive_ReprC(dyn, Clone)]
pub trait IAcceptor {
    fn topic(&self) -> Topic;
    fn reliability(&self) -> Reliability;
}

/// The result of attempting to bind a topic.
#[repr(C, u8)]
#[derive(Debug, Clone)]
pub enum BindResult {
    /// Upon success, the acceptor is yielded.
    Ok(Acceptor) = 1,

    /// Upon failure, the reason for that failure is yielded.
    Err(BindError) = 0,
}

/// The reasons for a bind attempt to fail.
#[derive_ReprC]
#[derive(thiserror::Error, Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord)]
#[repr(u8)]
pub enum BindError {
    /// The topic you attempted to bind was already bound.
    #[error("The topic you attempted to bind was already bound.")]
    TopicAlreadyBound,
    /// The provided topic was invalid.
    #[error("The provided topic was invalid.")]
    InvalidTopic,
    /// There was an internal error.
    #[error("There was an internal error.")]
    InternalError,
}

mod bind_result {
    use super::*;

    /// A safer-ffi friendly equivalent layout to [`ConnectionResult`].
    #[derive_ReprC]
    #[repr(C)]
    pub struct BindResult_Layout {
        /// `true` if accepted, `false` if rejected.
        pub ok: bool,

        /// Initialized if `accepted` is `true`
        /// if not, must be a valid [`BindError`] instead.
        pub acceptor: Acceptor,
    }
    unsafe impl ReprC for BindResult {
        type CLayout = CLayoutOf<BindResult_Layout>;

        // ALLOW: the appropriate type for the transmute source is private
        #[allow(clippy::missing_transmute_annotations)]
        fn is_valid(it: &'_ Self::CLayout) -> bool {
            BindResult_Layout::is_valid(it)
                || unsafe {
                    bool::is_valid(&it.ok)
                        && !core::mem::transmute::<_, bool>(it.ok)
                        && BindError::is_valid(core::mem::transmute::<
                            &Acceptor_Layout,
                            &BindError_Layout,
                        >(&it.acceptor))
                }
        }
    }
    #[test]
    fn bind_result_layout() {
        for expected in [
            BindResult::Err(BindError::InvalidTopic),
            BindResult::Err(BindError::TopicAlreadyBound),
        ] {
            assert!(BindResult::is_valid(&unsafe {
                core::mem::transmute::<BindResult, BindResult_Layout_Layout>(expected)
            }))
        }
    }
}

/// Requested reliability level for a message to be transmitted to another peer.
#[derive_ReprC]
#[repr(u8)]
#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Hash)]
pub enum Reliability {
    /// No guarantees of successful delivery, ordering, or once-only delivery
    Unreliable,

    /// Every message will be delivered in order or else the connection fails
    Reliable,
}

impl std::fmt::Display for Reliability {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let s = match self {
            Reliability::Unreliable => "unreliable",
            Reliability::Reliable => "reliable",
        };
        f.write_str(s)
    }
}

/// 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.
#[derive_ReprC]
#[derive(Clone)]
#[repr(C)]
#[must_use = "Until it is `open`ed, a `StreamCandidate` won't allow the connection to be used."]
pub struct StreamCandidate {
    pub inner: VirtualPtr<dyn IStreamCandidate + Send + Sync>,
}
/// A ready-to-open stream.
///
/// To finish opening the stream, [`IStreamCandidate::open`] must be called, providing the callback
/// to be called upon receiving packets.
///
/// If the candidate is destroyed before caling `open`, the connection will be closed, leading to a
/// [`ConnectionError::HandshakeFailed`].
///
/// If the candidate is never destroyed, the handshake will eventually [time
/// out](ConnectionError::HandshakeTimedOut).
#[derive_ReprC(dyn, Clone)]
pub trait IStreamCandidate: 'static {
    /// The peer with which the stream is/will be connected.
    fn peer_pubkey(&self) -> PeerPubkey;

    /// The topic of the stream.
    ///
    /// The return value _may_ alias `self`. You can ensure it's fully owned by calling
    /// [`Bytes::upgrade`](safer_ffi::bytes::Bytes::upgrade) which will copy only if necessary.
    fn topic(&self) -> Topic;

    /// Confirms the stream's opening, setting its optional callback.
    ///
    /// Failing to call this before the candidate is destroyed will close the connection.
    fn open(&self, on_recv: TaggedOption<Callback<Inbound>>) -> TaggedOption<Stream>;

    /// Returns a hash of the candidate's type ID, allowing internal code to identify the candidate
    /// type and downcast it if necessary.
    ///
    /// There's never any point in using it outside of Ditto's codebase.
    extern "C" fn type_id(&self) -> u64 {
        type_id_hash::<Self>()
    }
}

/// An open, bidirectional stream.
#[derive_ReprC]
#[repr(C)]
#[derive(Clone)]
#[must_use = "Dropping a stream closes the connection."]
pub struct Stream {
    pub inner: VirtualPtr<dyn IStream + Send + Sync>,
}
#[derive_ReprC(dyn, Clone)]
/// An open, bidirectional stream.
pub trait IStream: 'static {
    /// The peer with which the stream is/will be connected.
    extern "C" fn peer_pubkey(&self) -> PeerPubkey;

    /// The topic of the stream.
    ///
    /// The return value _may_ alias `self`. You can ensure it's fully owned by calling
    /// [`Bytes::upgrade`](safer_ffi::bytes::Bytes::upgrade) which will copy only if necessary.
    extern "C" fn topic(&self) -> Topic;

    /// Send a payload.
    ///
    /// The returned SendHandle can be used to cancel the send (provided it hasn't already been
    /// delivered). `finished` is called once the send is considered complete
    /// (immediately for unreliable streams, upon ACK otherwise). If the send failed
    /// (only allowed for cancelations), the payload
    extern "C" fn send(&self, payload: Payload) -> SendHandle;

    /// Returns the stream's closer if the stream was indeed closed.
    extern "C" fn current_status(&self) -> StreamStatus;

    /// Adds a `continuation` on the stream's closure.
    ///
    /// All continuations added to the stream's closure will be called once it is closed.
    ///
    /// The `continuation` is guaranteed to be called if the stream was already closed.
    extern "C" fn add_on_close(&self, continuation: Continuation<'static, StreamStatus>);

    /// Returns the maximum size of a payload that can be sent on this stream.
    ///
    /// This may be 0 if the stream has been closed.
    ///
    /// Note that a stream may receive messages larger than this size.
    extern "C" fn max_send_size(&self) -> usize;

    /// Returns a hash of the candidate's type ID, allowing internal code to identify the candidate
    /// type and downcast it if necessary.
    ///
    /// There's never any point in using it outside of Ditto's codebase.
    extern "C" fn type_id(&self) -> u64 {
        type_id_hash::<Self>()
    }
}

/// A handle on an send operation, allowing to await its completion or attempt to cancel it.
///
/// If dropped, the handle MUST neither `cancel` nor await the operation, but simply detach.
#[derive_ReprC]
#[repr(transparent)]
#[derive(Clone)]
pub struct SendHandle(pub VirtualPtr<dyn ISendHandle + Send + Sync>);

/// A handle on an send operation, allowing to await its completion or attempt to cancel it.
///
/// If dropped, the handle MUST neither `cancel` nor await the operation, but simply detach.
#[derive_ReprC(dyn, Clone)]
pub trait ISendHandle {
    /// Attempts to cancel the send operation, returning the payload of the message once the
    /// cancellation succeeds.
    ///
    /// This cancellation may fail to cancel the operation, even if the message is still in a
    /// send queue.
    ///
    /// If a continuation has been passed to `Self::then`, and hasn't been called yet, it _must_ be
    /// called with the current result of `Self::poll`.
    extern "C" fn cancel(&self) -> CancellationResult;

    /// Returns the current status of the operation.
    ///
    /// Note that consistency isn't guaranteed: calling `poll` after `then` may yield a
    /// different result than that yielded to the continuation. In such a case, the
    /// continuation's argument is considered "canon".
    extern "C" fn poll(&self) -> SendStatus;

    /// Calls the `callback` when the send status changes.
    ///
    /// Previously set callbacks MUST:
    /// - Never be called upon a change in the [`SendStatus`] if `keep_previous == false` once this
    ///   method returns.
    /// - Also be called upon a change in the [`SendStatus`] if `keep_previous == true`, without any
    ///   call order being specified.
    extern "C" fn set_on_change(&self, callback: Callback<SendStatus>, keep_previous: bool);
}

/// Identifies whether the stream was closed locally or remotely.
#[derive_ReprC]
#[repr(u8)]
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum StreamStatus {
    /// The stream is open
    Open,

    /// The stream was closed by the remote peer.
    ClosedByRemote,

    /// The stream was closed by the locally.
    ClosedByLocal,
}

/// The result of attempting to connect.
#[repr(C, u8)]
#[derive(Debug)]
pub enum ConnectionResult {
    /// Upon success, the stream is yielded.
    Accepted(StreamCandidate) = 1,

    /// Upon failure, the reason for that failure is yielded.
    Rejected(ConnectionError) = 0,
}

impl ConnectionResult {
    /// Returns the [`ConnectionResult`] as a [`Result`].
    pub fn accepted(self) -> Result<StreamCandidate, ConnectionError> {
        match self {
            ConnectionResult::Accepted(stream) => Ok(stream),
            ConnectionResult::Rejected(err) => Err(err),
        }
    }
}

mod connection_result {
    use super::*;

    /// A safer-ffi friendly equivalent layout to [`ConnectionResult`].
    #[derive_ReprC]
    #[repr(C)]
    pub struct ConnectionResult_Layout {
        /// `true` if accepted, `false` if rejected.
        pub accepted: bool,

        /// Initialized if `accepted` is `true`
        /// if not, must be a valid [`ConnectionError`] instead
        pub stream: StreamCandidate,
    }
    unsafe impl ReprC for ConnectionResult {
        type CLayout = CLayoutOf<ConnectionResult_Layout>;

        #[expect(clippy::missing_transmute_annotations)]
        fn is_valid(it: &'_ Self::CLayout) -> bool {
            ConnectionResult_Layout::is_valid(it)
                || unsafe {
                    bool::is_valid(&it.accepted)
                        && !core::mem::transmute::<_, bool>(it.accepted)
                        && ConnectionError::is_valid(core::mem::transmute::<
                            &StreamCandidate_Layout,
                            &ConnectionError_Layout,
                        >(&it.stream))
                }
        }
    }
    #[test]
    fn connection_result_layout() {
        for expected in [
            ConnectionResult::Rejected(ConnectionError::PeerNotFound),
            ConnectionResult::Rejected(ConnectionError::ConnectionRejected),
            ConnectionResult::Rejected(ConnectionError::HandshakeFailed),
            ConnectionResult::Rejected(ConnectionError::HandshakeTimedOut),
        ] {
            assert!(ConnectionResult::is_valid(&unsafe {
                #[expect(clippy::missing_transmute_annotations)]
                core::mem::transmute(expected)
            }))
        }
    }
}

mod cancelaltion_result {
    use super::*;

    /// A safer-ffi friendly equivalent layout to [`CancellationResult`].
    #[derive_ReprC]
    #[repr(C)]
    pub struct CancellationResult_Layout {
        /// `true` if cancelling succeeded, `false` otherwise.
        pub success: bool,

        /// Initialized iff `accepted` is `true`,
        /// if not, must be a valid [`CancellationError`] instead
        pub error_reason: CancellationError,
    }

    #[expect(clippy::missing_transmute_annotations)]
    unsafe impl ReprC for CancellationResult {
        type CLayout = CLayoutOf<CancellationResult_Layout>;

        fn is_valid(it: &'_ Self::CLayout) -> bool {
            CancellationResult_Layout::is_valid(it)
                || unsafe {
                    bool::is_valid(&it.success) && core::mem::transmute::<_, bool>(it.success)
                }
        }
    }

    #[test]
    fn cancellation_result_layout() {
        for expected in [
            CancellationResult::Ok,
            CancellationResult::Err(CancellationError::CancellationFailed),
        ] {
            assert!(CancellationResult::is_valid(&unsafe {
                #[expect(clippy::missing_transmute_annotations)]
                core::mem::transmute(expected)
            }))
        }
        assert!(CancellationResult::is_valid(&unsafe {
            core::mem::transmute::<[bool; 2], CancellationResult_Layout_Layout>([false, false])
        }));

        assert!(CancellationResult::is_valid(&unsafe {
            core::mem::transmute::<[bool; 2], CancellationResult_Layout_Layout>([true, false])
        }));

        assert!(!CancellationResult::is_valid(&unsafe {
            core::mem::transmute::<[u8; 2], CancellationResult_Layout_Layout>([2, 0])
        }));

        assert!(!CancellationResult::is_valid(&unsafe {
            core::mem::transmute::<[u8; 2], CancellationResult_Layout_Layout>([2, 1])
        }));

        assert!(!CancellationResult::is_valid(&unsafe {
            core::mem::transmute::<[u8; 2], CancellationResult_Layout_Layout>([0, 1])
        }));
    }
}

/// The reason for a connection attempt's failure.
#[derive_ReprC]
#[repr(u8)]
#[derive(thiserror::Error, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, Clone, Copy)]
pub enum ConnectionError {
    /// The peer you attempted to target was deemed invalid.
    #[error("The peer you attempted to target was deemed invalid.")]
    InvalidTarget,

    /// The peer you attempted to connect to was not found or failed to respond.
    #[error("The peer you attempted to connect to was not found or failed to respond in time.")]
    PeerNotFound,

    /// The peer you attempted to connect to was found, but refused your topic.
    #[error("The peer you attempted to connect to was found, but refused your topic.")]
    ConnectionRejected,

    /// The topic you passed was invalid: it must fit the following regex: /^[a-zA-Z0-9_ ]{0,15}$/
    #[error(
        "The topic you passed was invalid: it must fit the following regex: /^[a-zA-Z0-9_ \
         ]{{0,15}}$/"
    )]
    InvalidTopic,

    /// The Stream connection handshake didn't finish in time.
    #[error("The Stream connection handshake didn't finish in time.")]
    HandshakeTimedOut,

    /// The Stream connection handshake failed.
    #[error("The Stream connection handshake failed.")]
    HandshakeFailed,

    /// There was an internal error.
    #[error("There was an internal error.")]
    InternalError,
}

/// The reason for a connection attempt's failure.
#[derive_ReprC]
#[repr(u8)]
#[derive(thiserror::Error, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, Clone, Copy)]
pub enum CancellationError {
    /// Cancelling the send operation failed.
    #[error("Cancelling the send operation failed.")]
    CancellationFailed,
}

/// An FFI-safe `Result<Payload, CancellationError>`
#[repr(C, u8)]
pub enum CancellationResult {
    /// Cancelling the send operation was successful.
    ///
    /// The send operation's payload is returned if possible.
    Ok = 1,

    /// Cancelling the send operation was unsuccessful.
    Err(CancellationError) = 0,
}

impl From<Result<(), CancellationError>> for CancellationResult {
    fn from(value: Result<(), CancellationError>) -> Self {
        match value {
            Ok(()) => Self::Ok,
            Err(error) => Self::Err(error),
        }
    }
}

impl From<CancellationResult> for Result<(), CancellationError> {
    fn from(value: CancellationResult) -> Self {
        match value {
            CancellationResult::Ok => Ok(()),
            CancellationResult::Err(error) => Err(error),
        }
    }
}

/// Indicates the status of a given send operation.
///
/// A [`SendHandle`]'s state may become [`Unknown`](SendStatus::Unknown) at any time,
/// but may never return to another state after that.
///
/// Transition details:
/// - `Pending -> Sent`: Trivial, each state is technically allowed to represent the other.
/// - `Pending -> Failed`: This likely indicates that the stream has been closed before the message
///   could be sent. This may be due to a network issue or simply your peer closing the stream on
///   their end.
/// - `Pending` -> `Cancelled`: This indicates that the send operation was cancelled, although it
///   does not guarantee that the peer will never receive the cancelled message.
#[derive_ReprC]
#[derive(Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
#[repr(u8)]
pub enum SendStatus {
    /// The send status couldn't be recovered.
    ///
    /// Transition to this state may happen from any state (including "final" states),
    /// as the status may stop being tracked.
    Unknown = 0,

    /// The send operation is still pending.
    ///
    /// Note this doesn't guarantee that the message _hasn't_ been sent or acknowledged yet,
    /// just that the handle wasn't able to confirm that either of these states has been
    /// reached.
    Pending = 1,

    /// The send operation has been executed, but no acknowledgement has been received by the
    /// handle.
    Sent = 2,

    /// The send was identified as having failed.
    ///
    /// This typically would happen if the target disconnected before reaching one of the other end
    /// states.
    Failed = 3,

    /// The send was cancelled.
    ///
    /// Note that this does not guarantee that the message won't be delivered, even if it hadn't
    /// been sent yet.
    Cancelled = 4,
}

impl ISendHandle for SendStatus {
    extern "C" fn cancel(&self) -> CancellationResult {
        CancellationResult::Err(CancellationError::CancellationFailed)
    }

    extern "C" fn poll(&self) -> SendStatus {
        *self
    }

    extern "C" fn set_on_change(
        &self,
        mut continuation: Callback<SendStatus>,
        _keep_previous: bool,
    ) {
        #[allow(deprecated)]
        continuation.call(*self)
    }
}

impl core::fmt::Debug for SendStatus {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let s = match self {
            SendStatus::Unknown => "SendStatus::Unknown",
            SendStatus::Pending => "SendStatus::Pending",
            SendStatus::Sent => "SendStatus::Sent",
            SendStatus::Failed => "SendStatus::Failed",
            SendStatus::Cancelled => "SendStatus::Cancelled",
        };
        f.write_str(s)
    }
}

impl core::fmt::Display for SendStatus {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let s = match self {
            SendStatus::Unknown => "Unknown",
            SendStatus::Pending => "Pending",
            SendStatus::Sent => "Sent",
            SendStatus::Failed => "Failed",
            SendStatus::Cancelled => "Cancelled",
        };
        f.write_str(s)
    }
}

pub type MaybeTopic<'a> = safer_ffi::string::str_ref<'a>;

pub use topic::Topic;
/// This module is purely intended to silence a lint that gets emitted for code generated by
/// safer-ffi.
#[allow(clippy::size_of_in_element_count)]
mod topic {
    use super::*;
    /// A generalization of TCP/UDP's concept of "port", topics identify a Stream's purpose.
    ///
    /// Topics MUST be ASCII strings, restricted to the following regex: `^[a-zA-Z0-9_ ]{1,15}$`.
    ///
    /// ```
    /// # use std::{convert::TryFrom, str::FromStr};
    /// # use dittolive_ditto_base::bus::Topic;
    /// assert!(Topic::new(b"").is_none());
    /// Topic::try_from("a").unwrap();
    /// let _: Topic = "a".parse().unwrap();
    /// Topic::try_from(b"b").unwrap();
    /// Topic::try_from("topic is too long").unwrap_err();
    /// Topic::try_from("invalid.").unwrap_err();
    /// let s = "Hi there";
    /// assert_eq!(Topic::try_from(s).unwrap().as_str(), s);
    /// ```
    #[derive_ReprC]
    #[derive(Clone, PartialEq, Eq, Hash, PartialOrd, Ord)]
    #[repr(C)]
    pub struct Topic {
        pub(crate) len: u8,
        pub(crate) buffer: [u8; 15],
    }
}

impl Topic {
    /// The maximum length of a topic (inclusive).
    pub const CHAR_LIMIT: usize = 15;

    /// Converts a slice into a topic, returning `None` if the slice isn't a valid topic.
    pub const fn new(topic: &[u8]) -> Option<Self> {
        if !Self::is_valid(topic) {
            return None;
        }
        // UNWRAP: `is_valid` ensures the slice is inlinable for all architectures.
        const _: () = {
            if Topic::CHAR_LIMIT > safer_ffi::bytes::Bytes::MAX_INLINE_SIZE {
                panic!("The topic's length exceeds the maximum inline size of Bytes.")
            }
        };
        let mut buffer = [0; 15];
        let len = topic.len();
        let mut i = 0;
        while i < len {
            buffer[i] = topic[i];
            i += 1;
        }
        Some(Self {
            len: len as u8,
            buffer,
        })
    }

    const fn as_slice(&self) -> &[u8] {
        self.buffer.split_at(self.len as usize).0
    }

    pub const fn as_str(&self) -> &str {
        // SAFETY: `Topic` is always a valid ASCII string.
        unsafe { core::str::from_utf8_unchecked(self.as_slice()) }
    }

    /// Checks if a slice is a valid topic.
    pub const fn is_valid(slice: &[u8]) -> bool {
        if slice.len() > Self::CHAR_LIMIT || slice.is_empty() {
            return false;
        }
        let mut i = 0;
        while i < slice.len() {
            if !matches!(slice[i], b'a'..=b'z' | b'A'..=b'Z' | b'0'..=b'9' | b'_' | b' ') {
                return false;
            }
            i += 1;
        }
        true
    }
}

impl FromStr for Topic {
    type Err = ();
    fn from_str(s: &str) -> Result<Self, Self::Err> {
        Self::new(s.as_bytes()).ok_or(())
    }
}

impl TryFrom<&[u8]> for Topic {
    type Error = ();
    fn try_from(t: &[u8]) -> Result<Self, Self::Error> {
        Self::new(t).ok_or(())
    }
}

impl<const N: usize> TryFrom<[u8; N]> for Topic {
    type Error = ();
    fn try_from(t: [u8; N]) -> Result<Self, Self::Error> {
        Self::new(&t[..]).ok_or(())
    }
}

impl<const N: usize> TryFrom<&[u8; N]> for Topic {
    type Error = ();
    fn try_from(t: &[u8; N]) -> Result<Self, Self::Error> {
        Self::new(&t[..]).ok_or(())
    }
}

impl TryFrom<&str> for Topic {
    type Error = ();
    fn try_from(t: &str) -> Result<Self, Self::Error> {
        Self::new(t.as_ref()).ok_or(())
    }
}

impl core::ops::Deref for Topic {
    type Target = str;
    fn deref(&self) -> &Self::Target {
        self.as_str()
    }
}

impl core::fmt::Debug for Topic {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        f.write_str(self)
    }
}
