ruststream-rumqttc 0.7.1

//! The broker ladder: [`MqttBroker`] -> [`ConnectedMqttBroker`].
//!
//! Construction is synchronous and I/O-free; `connect` spawns the connection task and waits
//! for the broker's first `CONNACK`, and the connected form holds the live client directly.
//! One shared cell remains so publishers can be handed out while the application is still
//! being assembled, before `connect` runs.

// Without the `testing` feature a link has one variant, so a `match` on it has a single arm; the
// matches stay so that the in-process arm has its place when the feature is on.
#![cfg_attr(
    not(feature = "testing"),
    allow(clippy::infallible_destructuring_match)
)]

use std::borrow::Cow;
use std::collections::VecDeque;
use std::num::{NonZeroU16, NonZeroU32};
use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::time::Duration;

use percent_encoding::percent_decode_str;
use rumqttc::Transport;
#[cfg(feature = "testing")]
use rumqttc::v5::mqttbytes::matches;
use rumqttc::v5::mqttbytes::v5::{LastWill, Publish};
use rumqttc::v5::{AsyncClient, MqttOptions};
#[cfg(feature = "testing")]
use ruststream::testing::{Coordinator, InProcess, TestableBroker};
use ruststream::{
    AckError, AddressedCopies, Broker, ConnectedBroker, DefaultPublish, DescribeServer, ServerSpec,
    Subscribe,
};
#[cfg(feature = "testing")]
use ruststream::{OutgoingMessage, RawMessage};
use tokio::sync::{OnceCell, mpsc, oneshot};

#[cfg(feature = "asyncapi")]
use crate::asyncapi::{self, MqttLastWill, MqttServer};
use crate::conn::{Conn, Shared, run};
use crate::error::MqttError;
use crate::filter::{MqttFilter, MqttTopic, Qos};
#[cfg(feature = "testing")]
use crate::in_process::Bus;
use crate::publisher::{MqttPublish, MqttPublisher};
use crate::subscriber::{MqttSubscriber, OnDrop};

/// What a connected broker and every handle derived from it speak over: the live client, or,
/// under the `testing` feature, the in-process transport the test harness connected instead.
///
/// Without the feature there is one variant, so the type is the client handle itself and every
/// `match` on it is irrefutable: a production build carries no second transport and no branch to
/// it.
#[derive(Clone)]
pub(crate) enum Link {
    Wire(AsyncClient),
    #[cfg(feature = "testing")]
    InProcess(Arc<Bus>),
}

// The zero-cost promise of the in-process mode, held by the compiler: a build without it gives
// the link exactly the size of the client it wraps, and every handle holding a link keeps its
// size with it.
/// How long shutdown waits for the connection task to write what was queued before it.
const SHUTDOWN_FLUSH: Duration = Duration::from_secs(5);

#[cfg(not(feature = "testing"))]
const _: () = assert!(size_of::<Link>() == size_of::<AsyncClient>());
#[cfg(not(feature = "testing"))]
const _: () = assert!(size_of::<Core>() == size_of::<(AsyncClient, Arc<Shared>)>());
#[cfg(not(feature = "testing"))]
const _: () =
    assert!(size_of::<Option<(Link, Publish)>>() == size_of::<Option<(AsyncClient, Publish)>>());

impl std::fmt::Debug for Link {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Link").finish_non_exhaustive()
    }
}

impl Link {
    /// Acknowledges the delivery `publish` to the server.
    pub(crate) async fn ack(&self, publish: &Publish) -> Result<(), AckError> {
        let acknowledged = match self {
            Self::Wire(client) => client.ack(publish).await.is_ok(),
            #[cfg(feature = "testing")]
            Self::InProcess(bus) => bus.is_open(),
        };
        if acknowledged {
            Ok(())
        } else {
            Err(AckError::Broker(Box::from(
                "the mqtt connection task has shut down",
            )))
        }
    }

    /// Closes the session with a clean `DISCONNECT`, and returns once the connection task has
    /// flushed it.
    async fn disconnect(&self, shared: &Shared) {
        match self {
            Self::Wire(client) => {
                let _ = client.disconnect().await;
                // An acknowledgement reports success once it is queued, so the session is closed
                // only when the task has written the queue up to the DISCONNECT, or has lost the
                // connection. The bound covers a broker that stops reading.
                if tokio::time::timeout(SHUTDOWN_FLUSH, shared.stopped())
                    .await
                    .is_err()
                {
                    tracing::warn!(
                        timeout = ?SHUTDOWN_FLUSH,
                        "mqtt shutdown stopped waiting for the connection task to flush its queue"
                    );
                }
            }
            #[cfg(feature = "testing")]
            Self::InProcess(bus) => bus.close(),
        }
    }
}

/// The connection state shared by the connected form and every handle derived from it.
pub(crate) struct Core {
    pub(crate) link: Link,
    pub(crate) shared: Arc<Shared>,
}

impl std::fmt::Debug for Core {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Core").finish_non_exhaustive()
    }
}

pub(crate) type CoreCell = Arc<OnceCell<Core>>;

// The client's incoming default is 10 KiB, far below real payloads.
const DEFAULT_MAX_PACKET_SIZE: NonZeroU32 = NonZeroU32::new(1024 * 1024).unwrap();
const DEFAULT_RECEIVE_MAXIMUM: NonZeroU16 = NonZeroU16::new(1000).unwrap();

/// An MQTT 5 broker for the `RustStream` messaging framework.
///
/// `new` is synchronous and records only configuration; the runtime connects once at startup
/// via the consuming [`Broker::connect`], which waits for the broker's `CONNACK`. That is
/// what lets a service compose with the synchronous `#[ruststream::app]` builder.
///
/// # Examples
///
/// ```
/// # mod demo {
/// use std::time::Duration;
///
/// use ruststream_rumqttc::prelude::*;
/// use serde::Deserialize;
///
/// #[derive(Deserialize)]
/// struct Order {
///     id: u64,
/// }
///
/// #[subscriber(MqttTopic::new("orders/created"))]
/// async fn confirm(order: &Order) -> HandlerOutcome {
///     tracing::info!(order.id, "order confirmed");
///     HandlerOutcome::ack()
/// }
///
/// #[ruststream::app]
/// fn app() -> impl App {
///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
///         MqttBroker::new("mqtt://localhost:1883", "orders-svc")
///             .credentials("user", "pass")
///             .keep_alive(Duration::from_secs(30))
///             .clean_start(false)
///             .session_expiry(Duration::from_hours(1)),
///         |b| {
///             b.include(confirm);
///         },
///     )
/// }
/// # }
/// # fn main() {}
/// ```
///
/// With the `testing` feature the same broker has an in-process mode, so a test hands the app
/// `main` runs to `TestApp::start` unchanged and addresses the broker by this type. The
/// in-process server reads this broker's settings and models the one this connection talks to:
/// wildcards, share groups, retained messages and the quality of service each subscription reads
/// at. `TestApp::start_live` runs the same test body against a running server, for a persistent
/// session, the last will and the receive-maximum window.
///
/// ```
/// # #[cfg(feature = "testing")]
/// # mod demo {
/// use std::error::Error;
///
/// use ruststream::testing::TestApp;
/// use ruststream_rumqttc::prelude::*;
/// use serde::{Deserialize, Serialize};
///
/// #[derive(Debug, Deserialize, Serialize, Outgoing)]
/// struct Telemetry {
///     device: String,
///     temperature: f64,
/// }
///
/// #[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
/// #[outgoing(name = "alerts")]
/// struct Alert {
///     device: String,
/// }
///
/// #[subscriber(MqttFilter::new("devices/+/telemetry").qos(Qos::AtLeastOnce).shared("workers"))]
/// async fn handle(telemetry: &Telemetry, Out(alerts): Out<impl Publisher>) -> HandlerOutcome {
///     if telemetry.temperature <= 30.0 {
///         return HandlerOutcome::ack();
///     }
///     let alert = Alert { device: telemetry.device.clone() };
///     if alerts.message(&alert).publish().await.is_err() {
///         return HandlerOutcome::retry();
///     }
///     HandlerOutcome::ack()
/// }
///
/// /// The app `main` runs, and the one the tests hand the harness.
/// pub fn app() -> impl App {
///     RustStream::new(AppInfo::new("telemetry", "0.1.0")).with_broker(
///         MqttBroker::new("mqtt://localhost:1883", "telemetry-svc"),
///         |b| {
///             b.include(handle)
///                 .out(DefaultSlot, Publish::default().qos(Qos::AtLeastOnce))
///                 .out_retry(Publish::default())
///                 .to("devices/retry/telemetry")
///                 .build();
///         },
///     )
/// }
///
/// pub async fn a_hot_reading_raises_an_alert() -> Result<(), Box<dyn Error>> {
///     let tb = TestApp::start(app()).await?;
///
///     tb.broker::<MqttBroker>()
///         .message(&Telemetry { device: "dev42".to_owned(), temperature: 31.5 })
///         .to("devices/dev42/telemetry")
///         .publish()
///         .await?;
///
///     tb.broker::<MqttBroker>()
///         .published::<Alert>("alerts")
///         .assert_called_once()
///         .with(&Alert { device: "dev42".to_owned() });
///     tb.shutdown().await?;
///     Ok(())
/// }
/// # }
/// # #[cfg(feature = "testing")]
/// # fn main() -> Result<(), Box<dyn std::error::Error>> {
/// #     tokio::runtime::Builder::new_multi_thread()
/// #         .enable_all()
/// #         .build()?
/// #         .block_on(demo::a_hot_reading_raises_an_alert())
/// # }
/// # #[cfg(not(feature = "testing"))]
/// # fn main() {}
/// ```
#[derive(Debug, Clone)]
#[must_use]
pub struct MqttBroker {
    url: String,
    client_id: String,
    credentials: Option<(String, String)>,
    keep_alive: Option<Duration>,
    clean_start: Option<bool>,
    session_expiry: Option<u32>,
    max_packet_size: NonZeroU32,
    receive_maximum: NonZeroU16,
    last_will: Option<(String, Vec<u8>, Qos, bool)>,
    tls_ca: Option<Vec<u8>>,
    tls_client_auth: Option<(Vec<u8>, Vec<u8>)>,
    cell: CoreCell,
}

impl MqttBroker {
    /// Records the broker URL (`mqtt://host:port` or `mqtts://host:port`) and the client id.
    /// No I/O.
    ///
    /// A user name and a password in the URL (`mqtt://user:password@host:port`) log the
    /// connection in, percent-decoded: `p%40ss` is the password `p@ss`. The URL logs in only
    /// with both a non-empty user name and a password, and
    /// [`credentials`](Self::credentials) wins over it. A login that is not UTF-8 once decoded
    /// is refused when the broker connects, before any I/O.
    pub fn new(url: impl Into<String>, client_id: impl Into<String>) -> Self {
        Self {
            url: url.into(),
            client_id: client_id.into(),
            credentials: None,
            keep_alive: None,
            clean_start: None,
            session_expiry: None,
            max_packet_size: DEFAULT_MAX_PACKET_SIZE,
            receive_maximum: DEFAULT_RECEIVE_MAXIMUM,
            last_will: None,
            tls_ca: None,
            tls_client_auth: None,
            cell: Arc::new(OnceCell::new()),
        }
    }

    /// The user name and password the connection logs in with. They win over a login the URL
    /// carries.
    pub fn credentials(mut self, username: impl Into<String>, password: impl Into<String>) -> Self {
        self.credentials = Some((username.into(), password.into()));
        self
    }

    /// Sets the keep-alive interval: the longest silence the connection allows before the server
    /// ends it. The `rumqttc` client's minimum is 5 seconds, and a shorter interval fails
    /// [`connect`](Broker::connect) with [`MqttError::Invalid`] before any I/O.
    pub fn keep_alive(mut self, keep_alive: Duration) -> Self {
        self.keep_alive = Some(keep_alive);
        self
    }

    /// Starts a fresh session (`true`) or resumes a persistent one (`false`). Resuming pairs
    /// with [`session_expiry`](Self::session_expiry), and is what redelivers unacknowledged
    /// messages.
    pub fn clean_start(mut self, clean_start: bool) -> Self {
        self.clean_start = Some(clean_start);
        self
    }

    /// How long the broker keeps the session (and its subscriptions and unacked messages)
    /// after a disconnect.
    pub fn session_expiry(mut self, expiry: Duration) -> Self {
        self.session_expiry = Some(u32::try_from(expiry.as_secs()).unwrap_or(u32::MAX));
        self
    }

    /// The maximum incoming packet size. Defaults to 1 MiB (the client's own default is a
    /// 10 KiB cap that kills the connection on larger payloads). The server discards a message
    /// larger than this instead of sending it.
    ///
    /// # Examples
    ///
    /// ```
    /// # mod demo {
    /// use std::num::NonZeroU32;
    ///
    /// use ruststream_rumqttc::prelude::*;
    /// use serde::Deserialize;
    ///
    /// // A telemetry frame never exceeds 4 KiB; a larger packet is not worth receiving.
    /// const MAX_FRAME: NonZeroU32 = NonZeroU32::new(4 * 1024).unwrap();
    ///
    /// #[derive(Deserialize)]
    /// struct Frame {
    ///     sensor: String,
    /// }
    ///
    /// #[subscriber(MqttTopic::new("telemetry/frames"))]
    /// async fn record(frame: &Frame) -> HandlerOutcome {
    ///     tracing::info!(sensor = %frame.sensor, "frame recorded");
    ///     HandlerOutcome::ack()
    /// }
    ///
    /// #[ruststream::app]
    /// fn app() -> impl App {
    ///     RustStream::new(AppInfo::new("telemetry", "0.1.0")).with_broker(
    ///         MqttBroker::new("mqtt://localhost:1883", "telemetry-svc").max_packet_size(MAX_FRAME),
    ///         |b| {
    ///             b.include(record);
    ///         },
    ///     )
    /// }
    /// # }
    /// # fn main() {}
    /// ```
    ///
    /// A zero limit is a protocol error the server refuses the connection with, so it does not
    /// compile:
    ///
    /// ```compile_fail,E0308
    /// use ruststream_rumqttc::MqttBroker;
    ///
    /// let broker = MqttBroker::new("mqtt://localhost:1883", "telemetry-svc").max_packet_size(0);
    /// ```
    pub fn max_packet_size(mut self, bytes: NonZeroU32) -> Self {
        self.max_packet_size = bytes;
        self
    }

    /// Flow control: how many unacknowledged `QoS` 1/2 deliveries the broker may have in
    /// flight, which is also what bounds an unread subscriber queue. Defaults to 1000.
    ///
    /// # Examples
    ///
    /// ```
    /// # mod demo {
    /// use std::num::NonZeroU16;
    ///
    /// use ruststream_rumqttc::prelude::*;
    /// use serde::Deserialize;
    ///
    /// // Each render takes seconds, so the broker holds the rest of the backlog.
    /// const IN_FLIGHT: NonZeroU16 = NonZeroU16::new(8).unwrap();
    ///
    /// #[derive(Deserialize)]
    /// struct Render {
    ///     scene: String,
    /// }
    ///
    /// #[subscriber(MqttTopic::new("renders/requested").qos(Qos::AtLeastOnce))]
    /// async fn render(job: &Render) -> HandlerOutcome {
    ///     tracing::info!(scene = %job.scene, "scene rendered");
    ///     HandlerOutcome::ack()
    /// }
    ///
    /// #[ruststream::app]
    /// fn app() -> impl App {
    ///     RustStream::new(AppInfo::new("renderer", "0.1.0")).with_broker(
    ///         MqttBroker::new("mqtt://localhost:1883", "renderer-svc").receive_maximum(IN_FLIGHT),
    ///         |b| {
    ///             b.include(render);
    ///         },
    ///     )
    /// }
    /// # }
    /// # fn main() {}
    /// ```
    ///
    /// A zero maximum is a protocol error the server refuses the connection with, so it does not
    /// compile:
    ///
    /// ```compile_fail,E0308
    /// use ruststream_rumqttc::MqttBroker;
    ///
    /// let broker = MqttBroker::new("mqtt://localhost:1883", "renderer-svc").receive_maximum(0);
    /// ```
    pub fn receive_maximum(mut self, maximum: NonZeroU16) -> Self {
        self.receive_maximum = maximum;
        self
    }

    /// The last-will message the broker publishes if this session dies unexpectedly.
    pub fn last_will(
        mut self,
        topic: impl Into<String>,
        payload: impl Into<Vec<u8>>,
        qos: Qos,
        retain: bool,
    ) -> Self {
        self.last_will = Some((topic.into(), payload.into(), qos, retain));
        self
    }

    /// Trusts `ca` (PEM) for the TLS connection; selects TLS regardless of the URL scheme.
    ///
    /// The client verifies the server against this certificate authority and no other store, so
    /// it is what a TLS connection needs rather than an option on one: a `mqtts://` URL without
    /// it is refused when the broker connects, naming this setting.
    pub fn tls_ca(mut self, ca: impl Into<Vec<u8>>) -> Self {
        self.tls_ca = Some(ca.into());
        self
    }

    /// Authenticates with a TLS client certificate (both PEM), as managed MQTT services
    /// commonly require. Asks for TLS on its own, and [`tls_ca`](Self::tls_ca) is what verifies
    /// the server, so a connection that sets this and not that one is refused.
    pub fn tls_client_auth(mut self, cert: impl Into<Vec<u8>>, key: impl Into<Vec<u8>>) -> Self {
        self.tls_client_auth = Some((cert.into(), key.into()));
        self
    }

    /// A publisher sharing this broker's connection cell; buildable before `connect`.
    #[must_use]
    pub fn publisher(&self) -> MqttPublisher {
        MqttPublisher::new(Arc::clone(&self.cell), Qos::default(), false)
    }

    /// Whether the URL's scheme selects TLS.
    fn tls_from_scheme(&self) -> bool {
        self.url.starts_with("mqtts://")
    }

    /// The URL's authority: the host and port, without the scheme, the userinfo or anything after
    /// the host.
    ///
    /// An MQTT URL routinely carries `user:password@`, which logs the connection in and must not
    /// reach the generated document. The framework owns that trimming, so the rule is the same one
    /// every broker crate applies.
    fn authority(&self) -> String {
        ServerSpec::host_from_url(&self.url)
    }

    /// The login the connection opens with: the one set with [`credentials`](Self::credentials),
    /// or else the user name and password the URL carries, percent-decoded, or else none.
    ///
    /// A URL logs in only with both a non-empty user name and a password, the way the other
    /// broker crates read one.
    fn login(&self) -> Result<Option<(String, String)>, MqttError> {
        if let Some(explicit) = &self.credentials {
            return Ok(Some(explicit.clone()));
        }
        let after_scheme = self
            .url
            .split_once("://")
            .map_or(self.url.as_str(), |(_, rest)| rest);
        // The same authority `ServerSpec::host_from_url` reads the host from: it ends at the
        // first '/', '?' or '#', and inside it the last '@' ends the userinfo.
        let authority = after_scheme
            .split_once(['/', '?', '#'])
            .map_or(after_scheme, |(authority, _)| authority);
        let Some((userinfo, _)) = authority.rsplit_once('@') else {
            return Ok(None);
        };
        let Some((username, password)) = userinfo.split_once(':') else {
            return Ok(None);
        };
        if username.is_empty() {
            return Ok(None);
        }
        let decode = |part: &str| {
            percent_decode_str(part)
                .decode_utf8()
                .map(Cow::into_owned)
                .map_err(|_| {
                    MqttError::Invalid(
                        "the URL's user name or password is not UTF-8 once percent-decoded"
                            .to_owned(),
                    )
                })
        };
        Ok(Some((decode(username)?, decode(password)?)))
    }

    /// The host and port a client connects to, and what the generated document reports.
    fn endpoint(&self) -> Result<(String, u16), MqttError> {
        let authority = self.authority();
        let (host, port) = match authority.rsplit_once(':') {
            Some((host, port)) => (
                host.to_owned(),
                port.parse::<u16>()
                    .map_err(|_| MqttError::Invalid(format!("'{port}' is not a valid port")))?,
            ),
            None => (
                authority.clone(),
                if self.tls_from_scheme() { 8883 } else { 1883 },
            ),
        };
        if host.is_empty() {
            return Err(MqttError::Invalid("host must be non-empty".into()));
        }
        Ok((host, port))
    }

    fn options(&self) -> Result<MqttOptions, MqttError> {
        let tls_from_scheme = self.tls_from_scheme();
        let (host, port) = self.endpoint()?;
        if let Some(keep_alive) = self.keep_alive
            && keep_alive < Duration::from_secs(5)
        {
            return Err(MqttError::Invalid(
                "keep_alive must be at least 5 seconds, the rumqttc client's minimum".into(),
            ));
        }

        let mut options = MqttOptions::new(self.client_id.clone(), host, port);
        if let Some(keep_alive) = self.keep_alive {
            options.set_keep_alive(keep_alive);
        }
        if let Some(clean_start) = self.clean_start {
            options.set_clean_start(clean_start);
        }
        if let Some((username, password)) = self.login()? {
            options.set_credentials(username, password);
        }
        if let Some(expiry) = self.session_expiry {
            options.set_session_expiry_interval(Some(expiry));
        }
        options.set_max_packet_size(Some(self.max_packet_size.get()));
        options.set_receive_maximum(Some(self.receive_maximum.get()));
        options.set_manual_acks(true);
        if let Some((topic, payload, qos, retain)) = &self.last_will {
            options.set_last_will(LastWill::new(
                topic.clone(),
                payload.clone(),
                qos.to_client(),
                *retain,
                None,
            ));
        }
        if tls_from_scheme || self.tls_ca.is_some() || self.tls_client_auth.is_some() {
            // The client verifies the server against the certificates handed to it here and
            // against no other store, so an empty one trusts nothing: every handshake would fail,
            // and the connection task would read that as transient and retry until `connect`
            // times out. Refusing here is the difference between a message naming the missing
            // setting and half a minute of silence.
            let Some(ca) = self.tls_ca.clone() else {
                return Err(MqttError::Invalid(
                    "a TLS connection needs the certificate authority to trust: call tls_ca(pem)"
                        .to_owned(),
                ));
            };
            options.set_transport(Transport::tls(ca, self.tls_client_auth.clone(), None));
        }
        Ok(options)
    }
}

impl Broker for MqttBroker {
    type Error = MqttError;
    type Connected = ConnectedMqttBroker;

    async fn connect(self) -> Result<Self::Connected, Self::Error> {
        let core = self
            .cell
            .get_or_try_init(async || {
                let options = self.options()?;
                let (client, eventloop) = AsyncClient::new(options, 64);
                let shared = Arc::new(Shared::new());
                let (connack_tx, connack_rx) = oneshot::channel();
                tokio::spawn(run(Conn {
                    client: client.clone(),
                    eventloop,
                    shared: Arc::clone(&shared),
                    first_connack: Some(connack_tx),
                    parked: VecDeque::new(),
                }));
                // The task retries transient failures itself; the first CONNACK (or a fatal
                // refusal) decides whether connect succeeds.
                match tokio::time::timeout(Duration::from_secs(30), connack_rx).await {
                    Ok(Ok(Ok(()))) => {}
                    Ok(Ok(Err(err))) => return Err(err),
                    Ok(Err(_)) => {
                        return Err(MqttError::Connect(Box::from(
                            "the connection task exited before the first CONNACK",
                        )));
                    }
                    Err(_) => {
                        shared.closed.store(true, Ordering::Release);
                        // A task still dialing would otherwise connect later and run unowned;
                        // the queued DISCONNECT is what stops a task that got through.
                        let _ = client.try_disconnect();
                        // Everything the task retried is gone by now, so the wait is all there
                        // would be to report: a wrong port, an unreachable host and a handshake
                        // the broker refused after the fact would all read the same.
                        let reason = shared.last_error().map_or_else(
                            || "timed out waiting for the broker's CONNACK".to_owned(),
                            |last| {
                                format!(
                                    "timed out waiting for the broker's CONNACK; \
                                     last connection error: {last}"
                                )
                            },
                        );
                        return Err(MqttError::Connect(Box::from(reason)));
                    }
                }
                Ok::<_, MqttError>(Core {
                    link: Link::Wire(client),
                    shared,
                })
            })
            .await?;
        Ok(ConnectedMqttBroker {
            link: core.link.clone(),
            shared: Arc::clone(&core.shared),
            cell: self.cell,
        })
    }
}

/// The in-process mode: the connected form a test runs the production app against, carrying the
/// in-process transport in place of the client and every setting of this broker.
///
/// The options are built as `connect` builds them, so a configuration a service could not connect
/// with is not one a test can connect with either.
#[cfg(feature = "testing")]
impl InProcess for MqttBroker {
    async fn connect_in_process(self) -> Result<Self::Connected, Self::Error> {
        self.options()?;
        let max_packet_size = self.max_packet_size.get();
        let core = self
            .cell
            .get_or_try_init(async || {
                Ok::<_, MqttError>(Core {
                    link: Link::InProcess(Bus::new(max_packet_size)),
                    shared: Arc::new(Shared::new()),
                })
            })
            .await?;
        Ok(ConnectedMqttBroker {
            link: core.link.clone(),
            shared: Arc::clone(&core.shared),
            cell: self.cell,
        })
    }
}

#[cfg(feature = "testing")]
ruststream::register_testable_broker!(MqttBroker);

impl MqttBroker {
    /// The session this broker opens, as the `mqtt` server binding describes it.
    ///
    /// Credentials are absent by construction: the binding carries the client identity and the
    /// session settings, never the user name or the password. The last will contributes its
    /// coordinates and not its payload, which is content rather than a coordinate.
    #[cfg(feature = "asyncapi")]
    fn server_binding(&self) -> ruststream::asyncapi::Bindings {
        asyncapi::server(&MqttServer {
            client_id: self.client_id.clone(),
            clean_session: self.clean_start,
            last_will: self
                .last_will
                .as_ref()
                .map(|(topic, _payload, qos, retain)| MqttLastWill {
                    topic: topic.clone(),
                    qos: qos.level(),
                    retain: *retain,
                }),
            keep_alive: self.keep_alive.map(|interval| interval.as_secs()),
            session_expiry_interval: self.session_expiry,
            maximum_packet_size: self.max_packet_size.get(),
        })
    }
}

impl DescribeServer for MqttBroker {
    /// Reports the host and port a client connects to, the protocol version it speaks, and the
    /// session settings the `mqtt` binding has room for. A URL's credentials stay out of the
    /// generated document, which teams publish and share.
    ///
    /// The port is stated even where the URL leaves it out, because the protocol's default is
    /// what a reader of the document would otherwise have to know.
    fn describe_server(&self) -> ServerSpec {
        // A URL `connect` will reject still must not hold up the document, so the fallback keeps
        // the framework's stripped authority and drops the unusable port.
        let spec = self.endpoint().map_or_else(
            |_| ServerSpec::from_url(&self.url, "mqtt"),
            |(host, port)| ServerSpec::new(format!("{host}:{port}"), "mqtt"),
        );
        // The crate speaks MQTT 5 and nothing else: user properties and shared subscriptions
        // exist only there.
        let spec = spec.protocol_version("5");
        #[cfg(feature = "asyncapi")]
        let spec = spec.bindings(self.server_binding());
        spec
    }
}

/// The typed witness that `connect` succeeded: holds the live client directly, or the in-process
/// transport when the test harness connected it.
pub struct ConnectedMqttBroker {
    link: Link,
    shared: Arc<Shared>,
    // Keeps the cell of publishers handed out before connect alive and filled.
    cell: CoreCell,
}

impl std::fmt::Debug for ConnectedMqttBroker {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ConnectedMqttBroker")
            .finish_non_exhaustive()
    }
}

impl ConnectedMqttBroker {
    /// A publisher from the connected form with the default policy (`QoS` 1, not retained).
    #[must_use]
    pub fn publisher(&self) -> MqttPublisher {
        MqttPublisher::new(Arc::clone(&self.cell), Qos::default(), false)
    }

    pub(crate) fn publisher_with(&self, policy: MqttPublish) -> MqttPublisher {
        policy.into_publisher(Arc::clone(&self.cell))
    }

    /// Opens the subscription described by `topic` and waits for the broker's `SUBACK`.
    ///
    /// Dropping the subscriber unsubscribes the filter once no other subscriber on this
    /// connection shares it.
    ///
    /// # Errors
    ///
    /// Returns [`MqttError`] when the value is not a topic, the broker rejects it, or the broker
    /// is shut down.
    pub async fn subscribe_topic(&self, topic: MqttTopic) -> Result<MqttSubscriber, MqttError> {
        self.open_topic(topic, OnDrop::Unsubscribe).await
    }

    /// Opens the subscription described by `filter` and waits for the broker's `SUBACK`.
    ///
    /// Dropping the subscriber unsubscribes the filter once no other subscriber on this
    /// connection shares it.
    ///
    /// # Errors
    ///
    /// Returns [`MqttError`] when the descriptor is invalid, the broker rejects the filter,
    /// or the broker is shut down.
    pub async fn subscribe_filter(&self, filter: MqttFilter) -> Result<MqttSubscriber, MqttError> {
        self.open_filter(filter, OnDrop::Unsubscribe).await
    }

    /// Opens the subscription described by `topic`, whose drop does `on_drop` to its filter.
    pub(crate) async fn open_topic(
        &self,
        topic: MqttTopic,
        on_drop: OnDrop,
    ) -> Result<MqttSubscriber, MqttError> {
        topic.validate()?;
        self.open(topic.into_filter(), on_drop).await
    }

    /// Opens the subscription described by `filter`, whose drop does `on_drop` to its filter.
    pub(crate) async fn open_filter(
        &self,
        filter: MqttFilter,
        on_drop: OnDrop,
    ) -> Result<MqttSubscriber, MqttError> {
        filter.validate()?;
        self.open(filter, on_drop).await
    }

    /// The wire half both descriptors share, with validation already done.
    async fn open(&self, topic: MqttFilter, on_drop: OnDrop) -> Result<MqttSubscriber, MqttError> {
        self.shared.ensure_open()?;

        let wire_filter = topic.wire_filter();
        let qos = topic.qos_value().to_client();
        let (tx, rx) = mpsc::unbounded_channel();
        let id = self
            .shared
            .open(&self.link, &wire_filter, topic.filter(), qos, tx)
            .await?;

        Ok(MqttSubscriber::new(
            topic.filter().to_owned(),
            id,
            Arc::clone(&self.shared),
            self.link.clone(),
            on_drop,
            rx,
        ))
    }
}

impl ConnectedBroker for ConnectedMqttBroker {
    type Error = MqttError;
    type Closed = ();

    async fn shutdown(self) -> Result<(), Self::Error> {
        self.shared.closed.store(true, Ordering::Release);
        let held = self.shared.held();
        if held > 0 {
            // Unacknowledged, so the broker redelivers them when the session resumes; they are
            // lost only if the session is not persistent.
            tracing::warn!(
                held,
                "mqtt shutdown with deliveries no subscription ever matched"
            );
        }
        // A clean DISCONNECT lets the broker publish no last will and expire the session per
        // policy; the connection task exits once it has flushed it, after every acknowledgement
        // queued before it.
        self.link.disconnect(&self.shared).await;
        Ok(())
    }
}

impl Subscribe for ConnectedMqttBroker {
    type Subscriber = MqttSubscriber;
    /// A name written at the mount site is a topic, which is also the topic a publisher names, so
    /// a deferred retry reaches the subscription that took it. A filter with `+` or `#` is
    /// subscribe-only, and subscribing to one is what [`MqttFilter`] is for.
    type Copies = AddressedCopies;

    /// Opens the subscription of a `#[subscriber("topic")]` handler, which stays in the session
    /// when the service stops: its drop leaves the filter on the server.
    async fn subscribe(&self, name: &str) -> Result<Self::Subscriber, Self::Error> {
        self.open_topic(MqttTopic::new(name), OnDrop::Keep).await
    }
}

impl DefaultPublish for ConnectedMqttBroker {
    type Policy = MqttPublish;
}

/// The harness's view of the in-process transport: what it injects, what it reads back, the
/// coordinator it counts the in-flight deliveries with, and which subscriptions a publish reaches.
///
/// # Panics
///
/// `inject` and `published` panic on a broker connected with `connect`: the harness drives only
/// the transport `connect_in_process` produced, and a live connection has no log to read and no
/// synchronous way to take a message.
#[cfg(feature = "testing")]
impl TestableBroker for ConnectedMqttBroker {
    fn install_coordinator(&self, coordinator: Coordinator) {
        if let Link::InProcess(bus) = &self.link {
            bus.install(coordinator);
        }
    }

    fn inject(&self, message: OutgoingMessage<'_>) {
        let bus = self.bus("inject");
        // An external producer is another client of the server, so its publish goes through the
        // server's routing like any other, at the default quality of service and not retained.
        if let Err(err) = bus.inject(&self.shared, &message) {
            panic!(
                "the injected message to {:?} is not one a server takes: {err}",
                message.name()
            );
        }
    }

    fn published(&self, name: &str) -> Vec<RawMessage> {
        self.bus("published").published(name)
    }

    /// Every subscription whose topic filter matches the destination, by the match the
    /// connection demultiplexes deliveries with; the subscriptions this connection opened on one
    /// wire filter (the members of one share group, or one filter subscribed twice) are one
    /// subscription to the server and take one delivery between them in turn, so the answer names
    /// the one whose turn it is.
    fn routes(&self, destination: &str, subscriptions: &[&str]) -> Vec<usize> {
        let mut routed = Vec::new();
        let mut answered: Vec<&str> = Vec::new();
        for filter in subscriptions {
            if !matches(destination, filter) || answered.contains(filter) {
                continue;
            }
            answered.push(filter);
            // The subscriptions the app opened under this name, in the order it opened them.
            let positions = subscriptions
                .iter()
                .enumerate()
                .filter(|(_, name)| *name == filter)
                .map(|(position, _)| position);
            let takers = self.shared.next_takers(filter);
            if takers.is_empty() {
                // A filter this connection has not opened answers by its name alone.
                routed.extend(positions);
            } else {
                let positions: Vec<usize> = positions.collect();
                routed.extend(takers.iter().filter_map(|rank| positions.get(*rank)));
            }
        }
        routed.sort_unstable();
        routed
    }
}

#[cfg(feature = "testing")]
impl ConnectedMqttBroker {
    /// The in-process transport, which is all the harness drives.
    fn bus(&self, what: &str) -> &Arc<Bus> {
        match &self.link {
            Link::InProcess(bus) => bus,
            Link::Wire(_) => panic!(
                "TestableBroker::{what} reached a broker connected with `connect`; the harness \
                 drives the connection `connect_in_process` produces"
            ),
        }
    }
}

#[cfg(test)]
mod tests {
    use std::hash::{BuildHasher, Hasher, RandomState};

    use super::*;

    fn broker(url: &str) -> MqttBroker {
        MqttBroker::new(url, "describe")
    }

    #[test]
    fn the_endpoint_is_the_host_and_port_whatever_else_the_url_carries() {
        for (url, expected) in [
            ("mqtt://broker.example.com", ("broker.example.com", 1883)),
            ("mqtts://broker.example.com", ("broker.example.com", 8883)),
            (
                "mqtt://broker.example.com:1884",
                ("broker.example.com", 1884),
            ),
            (
                "mqtt://alice:s3cret@broker.example.com:1884",
                ("broker.example.com", 1884),
            ),
            (
                "mqtts://alice:s3cret@broker.example.com",
                ("broker.example.com", 8883),
            ),
            // A password may contain the separator, so the split takes the last one.
            (
                "mqtt://alice:p@ss@broker.example.com:1884",
                ("broker.example.com", 1884),
            ),
            (
                "mqtt://broker.example.com:1884/ns",
                ("broker.example.com", 1884),
            ),
            ("mqtt://broker.example.com/ns", ("broker.example.com", 1883)),
            // The authority ends before the path, so an `@` after it is not a separator.
            (
                "mqtt://broker.example.com/a@b",
                ("broker.example.com", 1883),
            ),
        ] {
            let broker = broker(url);
            let (host, port) = broker.endpoint().expect("the url parses");
            assert_eq!((host.as_str(), port), expected, "parsing {url}");
        }
    }

    /// TLS with nothing to verify the server against is a connection that can never succeed, so
    /// the setting that is missing is named instead. Each of the three ways to ask for TLS
    /// answers the same.
    #[tokio::test]
    async fn tls_without_a_certificate_authority_is_refused_before_any_io() {
        let scheme = MqttBroker::new("mqtts://127.0.0.1:8883", "tls-scheme");
        let client_auth =
            broker("mqtt://127.0.0.1:1").tls_client_auth(b"cert".to_vec(), b"key".to_vec());

        for (broker, asked) in [
            (scheme, "the URL scheme"),
            (client_auth, "a client certificate"),
        ] {
            let error = broker
                .connect()
                .await
                .expect_err("nothing can verify the server");
            assert!(
                matches!(&error, MqttError::Invalid(reason) if reason.contains("tls_ca")),
                "{asked} asks for TLS, and the refusal names what is missing: {error}"
            );
        }
    }

    /// The minimum is the `rumqttc` client's, which asserts on a smaller value, so a value below
    /// it is refused where the service can still see it: before any I/O, naming the number and
    /// whose minimum it is, instead of a panic inside the client.
    #[tokio::test]
    async fn a_keep_alive_below_the_client_minimum_is_refused_before_any_io() {
        let error = broker("mqtt://127.0.0.1:1")
            .keep_alive(Duration::from_secs(1))
            .connect()
            .await
            .expect_err("the client's minimum is five seconds");

        assert!(
            matches!(&error, MqttError::Invalid(reason)
                if reason.contains("5 seconds") && reason.contains("rumqttc client's minimum")),
            "the refusal names the minimum and whose it is: {error}"
        );
    }

    /// The login the connection opens with, as the client will send it.
    fn login(broker: &MqttBroker) -> Option<(String, String)> {
        broker
            .options()
            .expect("the options build")
            .credentials()
            .map(|login| (login.username, login.password))
    }

    #[test]
    fn the_url_user_name_and_password_log_in_percent_decoded() {
        let chosen = login(&broker("mqtt://us%40er:p%3Ass@broker.example.com:1884"));

        assert_eq!(chosen, Some(("us@er".to_owned(), "p:ss".to_owned())));
    }

    /// A password made up at run time, so no secret is written into the source.
    fn made_up_password() -> String {
        RandomState::new().build_hasher().finish().to_string()
    }

    #[test]
    fn explicit_credentials_win_over_the_url() {
        let password = made_up_password();
        let chosen = login(
            &broker("mqtt://other:wrong@broker.example.com:1884")
                .credentials("svc", password.as_str()),
        );

        assert_eq!(chosen, Some(("svc".to_owned(), password)));
    }

    #[test]
    fn a_url_without_a_user_name_and_a_password_logs_in_with_nothing() {
        for url in [
            "mqtt://broker.example.com:1884",
            "mqtt://svc@broker.example.com:1884",
            "mqtt://:secret@broker.example.com:1884",
        ] {
            let chosen = login(&broker(url));
            assert!(chosen.is_none(), "{url}: got {chosen:?}");
        }
    }

    #[test]
    fn a_url_login_that_is_not_utf8_once_decoded_is_refused_before_any_io() {
        for url in [
            "mqtt://%FF:secret@broker.example.com:1884",
            "mqtt://svc:%C3@broker.example.com:1884",
        ] {
            let refused = broker(url)
                .options()
                .expect_err("an MQTT user name and password are UTF-8");
            assert!(
                matches!(&refused, MqttError::Invalid(reason) if reason.contains("UTF-8")),
                "{url}: got {refused}"
            );
        }
    }

    #[test]
    fn a_url_with_credentials_describes_a_server_without_them() {
        let spec = broker("mqtt://alice:s3cret@broker.example.com:1884").describe_server();
        let host = spec.host.expect("a networked broker states its host");

        assert_eq!(host, "broker.example.com:1884");
        assert!(
            !host.contains('@'),
            "the userinfo separator is gone: {host}"
        );
        assert!(!host.contains("alice"), "the user name is gone: {host}");
        assert!(!host.contains("s3cret"), "the password is gone: {host}");

        let spec = broker("mqtt://us%40er:p%3Ass@broker.example.com:1884").describe_server();
        let host = spec.host.expect("a networked broker states its host");
        assert_eq!(
            host, "broker.example.com:1884",
            "percent-encoded userinfo is gone too"
        );
    }
}