ruststream-pulsar 0.7.1

//! [`PulsarPublisher`], its [`PulsarPublish`] policy, and the per-message settings a call site
//! adjusts ([`PulsarPublishOptions`], [`PulsarPublishSteps`]).

// Without the `testing` feature a transport 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::future::{Future, ready};
use std::sync::Arc;

use pulsar::routing_policy::RoutingPolicy;
use pulsar::{ProducerOptions, Pulsar, TokioExecutor};
#[cfg(feature = "asyncapi")]
use ruststream::asyncapi::Bindings;
use ruststream::runtime::{PublishBuilder, PublishSink};
use ruststream::{BytesMut, OutgoingMessage, PairError, PublishPolicy, Publisher, Take};
use tokio::sync::Mutex;

#[cfg(feature = "asyncapi")]
use crate::bindings::topic_channel;
use crate::broker::{ConnectedPulsarBroker, Core, CoreCell, Transport};
use crate::error::{PulsarError, box_err};
use crate::message::to_pulsar_message;
use crate::topic::PulsarTopic;

pub(crate) type PulsarProducer = pulsar::Producer<TokioExecutor>;

/// The producer options every producer this crate opens is built with: place a keyed message by
/// its key.
///
/// A partitioned topic is many topics behind one name, and the producer picks the partition per
/// message. The client's unset routing policy rotates through the partitions and never looks at
/// the key, which sends one key's messages to every partition in turn and leaves a `KeyShared`
/// subscription no per-key order to keep. Naming the policy is what makes the key place the
/// message: it hashes a key when the message carries one, and rotates when it does not. A topic
/// with no partitions has one producer and reads nothing here.
fn keyed_routing() -> ProducerOptions {
    ProducerOptions {
        routing_policy: Some(RoutingPolicy::RoundRobin),
        ..ProducerOptions::default()
    }
}

/// What one Pulsar publish differs from the next in.
///
/// Every field is optional: a message carries what its call site named with a step of
/// [`PulsarPublishSteps`], and nothing else. The [`PulsarPublish`] policy fixes no key of its
/// own, so a publish that names no step goes unkeyed.
///
/// # Examples
///
/// A test reads them back off the slot the handler published through:
///
/// ```
/// # #[cfg(feature = "testing")]
/// # mod demo {
/// use std::error::Error;
///
/// use ruststream::testing::TestApp;
/// use ruststream_pulsar::prelude::*;
/// use serde::{Deserialize, Serialize};
///
/// #[derive(Deserialize, Outgoing, Serialize)]
/// struct Order {
///     id: u64,
/// }
///
/// #[derive(Outgoing, Serialize)]
/// struct Receipt {
///     id: u64,
/// }
///
/// #[derive(OutSlot)]
/// #[publishes(Receipt)]
/// struct Ledger;
///
/// #[subscriber(PulsarSubscription::new("orders", "workers"))]
/// async fn record(
///     order: &Order,
///     Out(ledger): Out<impl Publisher<Options = PulsarPublishOptions>, Ledger>,
/// ) -> HandlerOutcome {
///     let sent = ledger
///         .message(&Receipt { id: order.id })
///         .to("receipts")
///         .partition_key(format!("user-{}", order.id))
///         .publish()
///         .await;
///     if sent.is_err() {
///         return HandlerOutcome::retry();
///     }
///     HandlerOutcome::ack()
/// }
///
/// pub fn app() -> RustStream {
///     RustStream::new(AppInfo::new("orders", "0.1.0"))
///         .with_broker(PulsarBroker::new("pulsar://localhost:6650"), |b| {
///             b.include(record).out(Ledger, Publish).build();
///         })
/// }
///
/// pub async fn a_receipt_is_keyed_by_its_user() -> Result<(), Box<dyn Error>> {
///     let tb = TestApp::start(app()).await?;
///     tb.broker::<PulsarBroker>()
///         .message(&Order { id: 7 })
///         .to("orders")
///         .publish()
///         .await?;
///
///     tb.out::<Ledger>()
///         .assert_called_once()
///         .with_options(&PulsarPublishOptions {
///             partition_key: Some("user-7".to_owned()),
///         });
///     tb.shutdown().await?;
///     Ok(())
/// }
/// # }
/// # #[cfg(feature = "testing")]
/// # #[tokio::main]
/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
/// #     demo::a_receipt_is_keyed_by_its_user().await
/// # }
/// # #[cfg(not(feature = "testing"))]
/// # fn main() {}
/// ```
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct PulsarPublishOptions {
    /// The message's partition key: keyed routing places the message by it and `KeyShared`
    /// subscriptions order by it.
    ///
    /// `None` leaves the key to the `partition-key` header, the portable spelling of the same
    /// value; with neither, the message is unkeyed.
    pub partition_key: Option<String>,
}

/// Publishes messages to Pulsar topics, one producer per topic, created lazily and shared
/// through the broker core (so `shutdown` can close them).
///
/// The partition key travels per message: [`PulsarPublishSteps::partition_key`] names it at the
/// call site, and a `partition-key` header names the same key for a caller that writes its
/// headers itself. Awaits the broker's send receipt, so `Ok` means the
/// broker stored the message. A payload over the broker's
/// [`max_message_size`](crate::PulsarBroker::max_message_size) fails with
/// [`PulsarError::Publish`] without being sent. Buildable before `connect` and usable until
/// `shutdown`; afterwards every publish reports [`PulsarError::NotConnected`].
#[derive(Clone)]
pub struct PulsarPublisher {
    cell: CoreCell,
}

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

impl PulsarPublisher {
    pub(crate) fn new(cell: CoreCell) -> Self {
        Self { cell }
    }

    fn core(&self) -> Result<&Core, PulsarError> {
        let core = self.cell.get().ok_or(PulsarError::NotConnected)?;
        core.ensure_open()?;
        Ok(core)
    }

    /// The per-topic producer, created on first use and cached on the core.
    // The map guard intentionally spans the build so two callers cannot race a double
    // producer for the same topic.
    #[allow(clippy::significant_drop_tightening)]
    async fn producer_for(
        &self,
        core: &Core,
        client: &Pulsar<TokioExecutor>,
        topic: &str,
    ) -> Result<Arc<Mutex<PulsarProducer>>, PulsarError> {
        let full = PulsarTopic::parse(topic)?.as_str().to_owned();
        let mut producers = core.producers.lock().await;
        if let Some(producer) = producers.get(&full) {
            return Ok(Arc::clone(producer));
        }
        // Built on the runtime the broker connected on: the client starts the connection a
        // producer needs on the runtime that asks, and the first publish may come from a runtime
        // that stops right after (a handler on a dedicated thread).
        let (client, named) = (client.clone(), full.clone());
        let producer = core
            .runtime
            .spawn(async move {
                Box::pin(
                    client
                        .producer()
                        .with_topic(named)
                        .with_options(keyed_routing())
                        .build(),
                )
                .await
            })
            .await
            .map_err(|e| PulsarError::Publish {
                topic: topic.to_owned(),
                source: box_err(e),
            })?
            .map_err(|e| PulsarError::Publish {
                topic: topic.to_owned(),
                source: box_err(e),
            })?;
        let producer = Arc::new(Mutex::new(producer));
        producers.insert(full, Arc::clone(&producer));
        Ok(producer)
    }
}

impl Publisher for PulsarPublisher {
    // The client's message owns its payload as a `Vec<u8>`, so this transport keeps the buffer
    // the framework wrote.
    type Payload = Take;
    type Error = PulsarError;
    /// The partition key, the one value Pulsar lets one publish differ from the next in on this
    /// crate's surface.
    type Options = PulsarPublishOptions;

    async fn publish(
        &self,
        msg: OutgoingMessage<'_, BytesMut>,
        options: Option<&Self::Options>,
    ) -> Result<(), Self::Error> {
        let core = self.core()?;
        let size = msg.payload().len();
        if size > core.max_message_size {
            return Err(PulsarError::Publish {
                topic: msg.name().to_owned(),
                source: Box::from(format!(
                    "the payload is {size} bytes, over the {} the broker's max_message_size \
                     allows",
                    core.max_message_size
                )),
            });
        }
        let key = options.and_then(|options| options.partition_key.as_deref());
        let client = match &core.transport {
            Transport::Client(client) => client,
            #[cfg(feature = "testing")]
            Transport::InProcess(bus) => return bus.publish(msg, key),
        };
        // The destination outlives the message, so the client's message can consume it.
        let topic = msg.name();
        let producer = Box::pin(self.producer_for(core, client, topic)).await?;
        let message = to_pulsar_message(msg, key)?;
        let receipt = {
            let mut producer = producer.lock().await;
            Box::pin(producer.send_non_blocking(message))
                .await
                .map_err(|e| PulsarError::Publish {
                    topic: topic.to_owned(),
                    source: box_err(e),
                })?
        };
        receipt.await.map(|_| ()).map_err(|e| PulsarError::Publish {
            topic: topic.to_owned(),
            source: box_err(e),
        })
    }
}

/// Pulsar's per-message steps on the framework's publish builder.
///
/// A step sets one field of [`PulsarPublishOptions`] for the publish being assembled and returns
/// the builder, so the message still leaves through the entry the mount site named, with that
/// entry's codec, transforms and slot attribution. The bound is on the publisher's settings type,
/// so these steps appear on a builder over a Pulsar publisher and over no other broker's.
///
/// The trait is in the crate [prelude](crate::prelude). A handler body that names a step imports
/// that glob and bounds its slot `Out<impl Publisher<Options = PulsarPublishOptions>, Marker>`;
/// every other body imports the framework prelude alone.
///
/// # Examples
///
/// ```
/// # mod demo {
/// use ruststream_pulsar::prelude::*;
/// use serde::{Deserialize, Serialize};
///
/// #[derive(Deserialize)]
/// struct Payment {
///     account: String,
///     cents: u64,
/// }
///
/// #[derive(Outgoing, Serialize)]
/// #[outgoing(name = "ledger")]
/// struct Entry {
///     account: String,
///     cents: u64,
/// }
///
/// #[derive(OutSlot)]
/// #[publishes(Entry)]
/// struct Ledger;
///
/// /// Keys every entry by its account, so a `KeyShared` reader of the ledger sees one account's
/// /// entries in order.
/// #[subscriber(PulsarSubscription::new("payments", "bookkeeping"))]
/// async fn book(
///     payment: &Payment,
///     Out(ledger): Out<impl Publisher<Options = PulsarPublishOptions>, Ledger>,
/// ) -> HandlerOutcome {
///     let entry = Entry {
///         account: payment.account.clone(),
///         cents: payment.cents,
///     };
///     match ledger.message(&entry).partition_key(&payment.account).publish().await {
///         Ok(()) => HandlerOutcome::ack(),
///         Err(_) => HandlerOutcome::retry(),
///     }
/// }
/// #
/// # #[ruststream::app]
/// # fn app() -> impl App {
/// #     RustStream::new(AppInfo::new("bookkeeping", "0.1.0")).with_broker(
/// #         PulsarBroker::new("pulsar://localhost:6650"),
/// #         |b| {
/// #             b.include(book).out(Ledger, Publish).build();
/// #         },
/// #     )
/// # }
/// # }
/// # fn main() {}
/// ```
pub trait PulsarPublishSteps {
    /// Sends this one message under `key` as its partition key, which keyed routing places the
    /// message by and `KeyShared` subscriptions order by.
    ///
    /// The key wins over a `partition-key` header the call site wrote itself; without either the
    /// message is unkeyed.
    ///
    /// # Examples
    ///
    /// ```
    /// # mod demo {
    /// use ruststream_pulsar::prelude::*;
    /// use serde::{Deserialize, Serialize};
    ///
    /// #[derive(Deserialize, Outgoing, Serialize)]
    /// struct Click {
    ///     session: String,
    ///     page: String,
    /// }
    ///
    /// #[derive(OutSlot)]
    /// #[publishes(Click)]
    /// struct Sessions;
    ///
    /// /// Re-keys raw clicks by session, so one session's clicks land on one partition.
    /// #[subscriber(PulsarSubscription::new("clicks", "sessionizer"))]
    /// async fn sessionize(
    ///     click: &Click,
    ///     Out(sessions): Out<impl Publisher<Options = PulsarPublishOptions>, Sessions>,
    /// ) -> HandlerOutcome {
    ///     let sent = sessions
    ///         .message(click)
    ///         .to("clicks-by-session")
    ///         .partition_key(&click.session)
    ///         .publish()
    ///         .await;
    ///     if sent.is_err() {
    ///         return HandlerOutcome::retry();
    ///     }
    ///     HandlerOutcome::ack()
    /// }
    /// #
    /// # #[ruststream::app]
    /// # fn app() -> impl App {
    /// #     RustStream::new(AppInfo::new("sessions", "0.1.0")).with_broker(
    /// #         PulsarBroker::new("pulsar://localhost:6650"),
    /// #         |b| {
    /// #             b.include(sessionize).out(Sessions, Publish).build();
    /// #         },
    /// #     )
    /// # }
    /// # }
    /// # fn main() {}
    /// ```
    #[must_use]
    fn partition_key(self, key: impl Into<String>) -> Self;
}

impl<Sink, Body, Enc, Hdrs, Dest> PulsarPublishSteps for PublishBuilder<Sink, Body, Enc, Hdrs, Dest>
where
    Sink: PublishSink<Options = PulsarPublishOptions>,
{
    fn partition_key(mut self, key: impl Into<String>) -> Self {
        self.options_mut()
            .get_or_insert_with(PulsarPublishOptions::default)
            .partition_key = Some(key.into());
        self
    }
}

/// The publish policy for [`PulsarPublisher`]: pure declaration, constructible anywhere,
/// paired with the connected broker by the runtime after `connect`.
///
/// In the generated `AsyncAPI` document it describes the channel it publishes to: the
/// specification's `pulsar` channel binding is a namespace and a persistence, and both live in
/// the destination's topic name, which the runtime hands the policy. Its operation and message
/// objects are empty, and a partition key belongs to one message rather than to the policy.
/// Replies go to the destination the reply type or the mount site declares, the channel the
/// document reports for them.
///
/// # Examples
///
/// A routes file that does not import this crate's prelude names the policy by this name:
///
/// ```
/// # mod demo {
/// use ruststream::prelude::*;
/// use ruststream_pulsar::{PulsarBroker, PulsarPublish, PulsarSubscription};
/// use serde::{Deserialize, Serialize};
///
/// #[derive(Deserialize)]
/// struct Order {
///     id: u64,
/// }
///
/// #[derive(Outgoing, Serialize)]
/// #[outgoing(name = "receipts")]
/// struct Receipt {
///     id: u64,
/// }
///
/// #[subscriber(PulsarSubscription::new("orders", "workers"), reply)]
/// async fn confirm(order: &Order) -> Receipt {
///     Receipt { id: order.id }
/// }
///
/// #[ruststream::app]
/// fn app() -> impl App {
///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
///         PulsarBroker::new("pulsar://localhost:6650"),
///         |b| {
///             b.include(confirm).out_reply(PulsarPublish);
///         },
///     )
/// }
/// # }
/// # fn main() {}
/// ```
#[derive(Debug, Clone, Copy, Default)]
#[must_use]
pub struct PulsarPublish;

#[cfg(feature = "asyncapi")]
impl PulsarPublish {
    /// The `pulsar` channel binding of the destination the runtime resolved for this position:
    /// the reply type's own name or the `reply("dest")` clause, a slot entry's name, the
    /// `dead_letter("dlq")` declaration.
    ///
    /// A destination Pulsar would refuse as a topic name reports nothing rather than a guess,
    /// the way an unserializable binding body does: a document describes the deployment or it
    /// stays silent about it.
    fn describe_channel(channel: &str) -> Bindings {
        PulsarTopic::parse(channel).map_or_else(|_| Bindings::new(), |topic| topic_channel(&topic))
    }
}

impl PublishPolicy<ConnectedPulsarBroker> for PulsarPublish {
    type Live = PulsarPublisher;

    fn pair(
        self,
        connected: &ConnectedPulsarBroker,
    ) -> impl Future<Output = Result<Self::Live, PairError>> {
        ready(Ok(connected.publisher()))
    }

    #[cfg(feature = "asyncapi")]
    fn channel_bindings(&self, channel: &str) -> Bindings {
        Self::describe_channel(channel)
    }
}

#[cfg(all(test, feature = "testing"))]
mod tests {
    use ruststream::runtime::PublishExt;
    use ruststream::testing::{InProcess, TestableBroker};
    use ruststream::{HeaderMap, Outgoing, Serialized};
    use serde::Serialize;

    use super::PulsarPublishSteps;
    use crate::{ConnectedPulsarBroker, PARTITION_KEY_HEADER, PulsarBroker};

    /// The payload these tests carry: what they assert on is the key the publish carried, so the
    /// body is deliberately opaque bytes rather than a model.
    #[derive(Outgoing, Serialized)]
    struct Record(&'static [u8]);

    #[derive(Serialize)]
    struct OrderMeta {
        tenant: &'static str,
    }

    #[derive(Outgoing, Serialize)]
    #[outgoing(name = "orders.done", headers = OrderMeta)]
    struct OrderDone {
        id: u64,
    }

    async fn connected() -> ConnectedPulsarBroker {
        PulsarBroker::new("pulsar://localhost:6650")
            .connect_in_process()
            .await
            .expect("the broker connects in process")
    }

    /// The step wins over the portable spelling of the same value, so a call site that sets both
    /// gets the one it wrote last in the chain rather than a silent merge.
    #[tokio::test]
    async fn the_step_wins_over_a_call_site_header() {
        let broker = connected().await;
        let mut headers = HeaderMap::new();
        headers.insert(PARTITION_KEY_HEADER, "user-7");
        broker
            .publisher()
            .message(&Record(b"{}"))
            .with_headers(headers)
            .to("orders")
            .partition_key("user-42")
            .publish()
            .await
            .expect("publish succeeds");

        let sent = broker.published("orders");
        assert_eq!(
            sent[0].headers().get(PARTITION_KEY_HEADER),
            Some(b"user-42".as_slice())
        );
    }

    /// The case a header cannot reach: the publish's headers position is taken by a declared
    /// contract, and the key is a setting rather than one more entry in that map.
    #[tokio::test]
    async fn the_step_composes_with_a_declared_header_contract() {
        let broker = connected().await;
        broker
            .publisher()
            .message(&OrderDone { id: 1 })
            .with_headers(&OrderMeta { tenant: "acme" })
            .partition_key("user-42")
            .publish()
            .await
            .expect("publish succeeds");

        let sent = broker.published("orders.done");
        assert_eq!(sent[0].headers().get_str("tenant"), Some("acme"));
        assert_eq!(
            sent[0].headers().get(PARTITION_KEY_HEADER),
            Some(b"user-42".as_slice())
        );
    }
}