ruststream-nats 0.7.1

//! The subject a service subscribes to: [`CoreSubject`] and [`CoreWildcard`] over Core NATS,
//! [`JetStreamSubject`] through a `JetStream` pull consumer.
//!
//! They are the crate's [`SubscriptionSource`]s: the value a `#[subscriber(..)]` handler carries,
//! and the value the runtime resolves once against the connected broker at startup.
//!
//! One type per delivery model, because the two models share no settings. A queue group is Core
//! NATS only and a durable name describes a `JetStream` consumer, so each is a method on one type
//! and absent from the other. Naming the wrong one is a compile error rather than a service that
//! starts and then refuses its own subscription.
//!
//! Core NATS then splits once more, over the one property a retry depends on: one subject is a
//! destination a copy of a delivery can be published to, a pattern is not. [`CoreSubject`] carries
//! the first and answers the framework with its own subject; [`CoreWildcard`] carries the second
//! and leaves the mount site to name where a copy goes.

use std::future::{Future, ready};
use std::num::NonZeroU64;
use std::time::Duration;

pub use async_nats::jetstream::consumer::DeliverPolicy;
#[cfg(feature = "asyncapi")]
use ruststream::asyncapi::{Binding, Bindings};
use ruststream::runtime::IntoSource;
use ruststream::{
    AddressedCopies, NamedCopies, RedeliveryAddress, RedeliveryAddressed, SubscriptionSource,
};

#[cfg(feature = "asyncapi")]
use serde::Serialize;

use self::sealed::Sealed;
use crate::{ConnectedNatsBroker, error::NatsError, subscriber::NatsSubscriber};

/// Ack window a `JetStream` consumer gets when the descriptor names none.
const DEFAULT_ACK_WAIT: Duration = Duration::from_secs(30);
/// In-flight unacked deliveries a `JetStream` consumer allows when the descriptor names no cap.
const DEFAULT_MAX_ACK_PENDING: i64 = 1024;
/// Fetch window a `JetStream` pull consumer uses when the descriptor names none.
const DEFAULT_PULL_EXPIRES: Duration = Duration::from_secs(5);

/// A [`Duration`] that is known not to be zero.
///
/// [`JetStreamSubject::pull_expires`] and [`JetStreamSubject::ack_wait`] take one. A fetch that
/// expires the instant it is issued turns the batch loop into a hot spin, so the zero is kept out
/// of the type instead of being caught when the subscription opens.
///
/// # Examples
///
/// ```
/// # mod demo {
/// use ruststream_nats::prelude::*;
/// use serde::Deserialize;
///
/// #[derive(Deserialize)]
/// struct Order {
///     amount: u64,
/// }
///
/// #[subscriber(
///     JetStreamSubject::new("orders.*", "ORDERS")
///         .durable("reconciler")
///         .ack_wait(NonZeroDuration::from_secs(nonzero!(60)))
///         .pull_expires(NonZeroDuration::from_millis(nonzero!(300)))
/// )]
/// async fn reconcile(orders: &[Order]) -> HandlerOutcome {
///     let total: u64 = orders.iter().map(|order| order.amount).sum();
///     tracing::info!(count = orders.len(), total, "reconciled");
///     HandlerOutcome::ack()
/// }
///
/// #[ruststream::app]
/// fn app() -> impl App {
///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
///         NatsBroker::new("nats://localhost:4222"),
///         |b| {
///             b.include(reconcile.batch(nonzero!(50)));
///         },
///     )
/// }
/// # }
/// # fn main() {}
/// ```
///
/// A literal is checked while the crate compiles, so a zero window does not build:
///
/// ```compile_fail
/// use ruststream::nonzero;
/// use ruststream_nats::NonZeroDuration;
///
/// let window = NonZeroDuration::from_millis(nonzero!(0));
/// ```
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
#[must_use]
pub struct NonZeroDuration(Duration);

impl NonZeroDuration {
    /// Wraps `value`, or returns `None` when it is zero.
    ///
    /// This is the entry for a duration that arrives at run time, from configuration or from a
    /// calculation. A literal reads better through [`from_secs`](Self::from_secs) or
    /// [`from_millis`](Self::from_millis).
    #[must_use]
    pub const fn new(value: Duration) -> Option<Self> {
        if value.is_zero() {
            None
        } else {
            Some(Self(value))
        }
    }

    /// Builds a duration of `secs` seconds.
    pub const fn from_secs(secs: NonZeroU64) -> Self {
        Self(Duration::from_secs(secs.get()))
    }

    /// Builds a duration of `millis` milliseconds.
    pub const fn from_millis(millis: NonZeroU64) -> Self {
        Self(Duration::from_millis(millis.get()))
    }

    /// The duration itself.
    #[must_use]
    pub const fn get(self) -> Duration {
        self.0
    }
}

impl From<NonZeroDuration> for Duration {
    fn from(value: NonZeroDuration) -> Self {
        value.get()
    }
}

/// What a descriptor asks the connected broker to open, with every default already resolved.
///
/// Reachable through the sealed [`NatsSubscription`], so it is `pub` for the visibility checker
/// only: `subject` is a private module and nothing re-exports this, so no crate outside
/// can name it.
#[doc(hidden)]
#[derive(Debug)]
pub enum SubscriptionPlan<'a> {
    Core {
        queue_group: Option<&'a str>,
    },
    JetStream {
        stream: &'a str,
        durable: Option<&'a str>,
        filter_subject: &'a str,
        ack_wait: Duration,
        max_ack_pending: i64,
        deliver_policy: DeliverPolicy,
        pull_expires: Duration,
    },
}

mod sealed {
    use super::SubscriptionPlan;

    pub trait Sealed {
        fn plan(&self) -> SubscriptionPlan<'_>;
    }
}

/// A NATS subscription descriptor: [`CoreSubject`], [`CoreWildcard`] or [`JetStreamSubject`].
///
/// Sealed, because the two delivery models NATS has are the two this crate ships.
///
/// The three differ in where a delayed copy of a delivery goes. A [`CoreSubject`] is its own
/// destination, so the framework publishes the copy back to it. A [`CoreWildcard`] names many
/// subjects and none in particular, so its mount site names one with `out_retry(..).to(..)`. A
/// [`JetStreamSubject`] needs no copy: the server holds the message for the delay.
///
/// # Examples
///
/// ```
/// # mod demo {
/// use std::time::Duration;
///
/// use ruststream_nats::prelude::*;
/// use serde::Deserialize;
///
/// #[derive(Deserialize)]
/// struct Payment {
///     id: u64,
///     settled: bool,
/// }
///
/// #[subscriber("payments.settled")]
/// async fn reconcile(payment: &Payment) -> HandlerOutcome {
///     if payment.settled {
///         return HandlerOutcome::ack();
///     }
///     HandlerOutcome::retry_after(Duration::from_secs(30))
/// }
///
/// #[subscriber(CoreWildcard::new("payments.*"))]
/// async fn audit(payment: &Payment) -> HandlerOutcome {
///     tracing::warn!(payment.id, "audit pending");
///     HandlerOutcome::retry_after(Duration::from_secs(30))
/// }
///
/// #[subscriber(JetStreamSubject::new("payments.stored", "PAYMENTS").durable("reconciler"))]
/// async fn store(payment: &Payment) -> HandlerOutcome {
///     tracing::warn!(payment.id, "store pending");
///     HandlerOutcome::retry_after(Duration::from_secs(30))
/// }
///
/// #[ruststream::app]
/// fn app() -> impl App {
///     RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(
///         NatsBroker::new("nats://localhost:4222"),
///         |b| {
///             // One subject is its own destination.
///             b.include(reconcile)
///                 .max_attempts(nonzero!(5u32))
///                 .dead_letter("payments.dead");
///
///             // A pattern is not, so the mount site names where a copy lands.
///             b.include(audit)
///                 .max_attempts(nonzero!(5u32))
///                 .dead_letter("dead.payments")
///                 .out_retry(Publish)
///                 .to("payments.settled");
///
///             // The server holds the message, so nothing is published at all.
///             b.include(store)
///                 .max_attempts(nonzero!(5u32))
///                 .dead_letter("payments.dead");
///         },
///     )
/// }
/// # }
/// # fn main() {}
/// ```
#[diagnostic::on_unimplemented(
    message = "`{Self}` does not describe a NATS subscription",
    label = "not a subscription descriptor",
    note = "use `CoreSubject::new(subject)` for one Core NATS subject, \
            `CoreWildcard::new(pattern)` for a Core NATS pattern, or \
            `JetStreamSubject::new(subject, stream)` for a JetStream pull consumer"
)]
pub trait NatsSubscription: Sealed {
    /// The subject pattern this subscription receives messages on.
    fn subject(&self) -> &str;

    /// The one thing about a subscription the types cannot settle: the subject is a string the
    /// caller supplies, often from configuration, and a descriptor constructor is the expression
    /// a `#[subscriber(..)]` attribute writes, so it cannot hand back a `Result`. Every
    /// `subscribe` implementation calls this before it does any work.
    ///
    /// # Errors
    ///
    /// Returns [`NatsError::InvalidOptions`] when the subject is empty.
    fn ensure_subject(&self) -> Result<(), NatsError> {
        if self.subject().is_empty() {
            return Err(NatsError::InvalidOptions(
                "subject must be non-empty".into(),
            ));
        }
        Ok(())
    }
}

/// Whether `subject` carries a NATS wildcard: `*` for one token, `>` for the rest of them.
///
/// A wildcard is matched on delivery. A publish takes it as a literal token (a pedantic connection
/// refuses it), so a copy published to a pattern would not reach the subject its delivery came
/// from, and this crate treats a pattern as a subscription and never as a destination.
fn has_wildcard(subject: &str) -> bool {
    subject.split('.').any(|token| token == "*" || token == ">")
}

/// The subject as a publish destination.
///
/// # Errors
///
/// Returns [`NatsError::InvalidOptions`] when the subject is empty or carries a wildcard.
pub(crate) fn publish_destination(subject: &str) -> Result<RedeliveryAddress, NatsError> {
    if subject.is_empty() {
        return Err(NatsError::InvalidOptions(
            "subject must be non-empty".into(),
        ));
    }
    if has_wildcard(subject) {
        return Err(NatsError::InvalidOptions(format!(
            "`{subject}` is a subscription pattern, not a subject a delayed copy can be \
             published to. Subscribe to a pattern with `CoreWildcard`, which leaves the mount \
             site to name where a retry copy goes"
        )));
    }
    Ok(RedeliveryAddress::new(subject.to_owned()))
}

/// One subject subscribed to over Core NATS.
///
/// Core NATS delivers to whoever is subscribed at that moment and stores nothing, so the only
/// setting it has is the queue group that load-balances a subject across several subscribers.
/// Reading a stream instead is [`JetStreamSubject`]; reading many subjects at once is
/// [`CoreWildcard`].
///
/// The subject is one destination, so a delayed redelivery needs nothing from the mount site: the
/// framework publishes its copy back to this very subject.
///
/// # Examples
///
/// ```
/// # mod demo {
/// use ruststream_nats::prelude::*;
/// use serde::Deserialize;
///
/// #[derive(Deserialize)]
/// struct Order {
///     id: u64,
/// }
///
/// // Every replica joins `workers`, so each order reaches one of them.
/// #[subscriber(CoreSubject::new("orders.created").queue_group("workers"))]
/// async fn fulfil(order: &Order) -> HandlerOutcome {
///     tracing::info!(order.id, "fulfilled");
///     HandlerOutcome::ack()
/// }
///
/// #[ruststream::app]
/// fn app() -> impl App {
///     RustStream::new(AppInfo::new("fulfillment", "0.1.0")).with_broker(
///         NatsBroker::new("nats://localhost:4222"),
///         |b| {
///             b.include(fulfil);
///         },
///     )
/// }
/// # }
/// # fn main() {}
/// ```
///
/// A `JetStream` setting is not a method here, so asking for one does not compile:
///
/// ```compile_fail
/// use ruststream_nats::CoreSubject;
///
/// // `durable` names a JetStream consumer, and Core NATS has none.
/// let bad = CoreSubject::new("orders.created").durable("worker-1");
/// ```
#[derive(Debug, Clone, PartialEq, Eq)]
#[must_use]
pub struct CoreSubject {
    subject: String,
    queue_group: Option<String>,
}

impl CoreSubject {
    /// Subscribes to `subject` over Core NATS.
    pub fn new(subject: impl Into<String>) -> Self {
        Self {
            subject: subject.into(),
            queue_group: None,
        }
    }

    /// Load-balances the subject across a queue group: the server hands each message to one
    /// member instead of all of them.
    pub fn queue_group(mut self, name: impl Into<String>) -> Self {
        self.queue_group = Some(name.into());
        self
    }

    /// The queue group this subscription joins, if any.
    #[cfg(feature = "asyncapi")]
    pub(crate) fn group(&self) -> Option<&str> {
        self.queue_group.as_deref()
    }
}

impl Sealed for CoreSubject {
    fn plan(&self) -> SubscriptionPlan<'_> {
        SubscriptionPlan::Core {
            queue_group: self.queue_group.as_deref(),
        }
    }
}

/// A pattern subscribed to over Core NATS: `orders.*`, `orders.>`, `>`.
///
/// The same subscription as [`CoreSubject`], on many subjects instead of one. That is the whole
/// difference between the two types, and it is the difference that matters to a retry: a pattern
/// names many subjects and none of them in particular, so this subscription cannot say where a
/// copy of a delivery would reach it again. The mount site says it instead, with
/// `.out_retry(Publish).to("orders.created")` or with a transform that reads the delivery and
/// names its own subject; a registration that says neither refuses to start.
///
/// # Examples
///
/// ```
/// # mod demo {
/// use std::time::Duration;
///
/// use ruststream_nats::prelude::*;
/// use serde::Deserialize;
///
/// #[derive(Deserialize)]
/// struct Order {
///     id: u64,
///     paid: bool,
/// }
///
/// #[subscriber(CoreWildcard::new("orders.*").queue_group("auditors"))]
/// async fn audit(order: &Order) -> HandlerOutcome {
///     if order.paid {
///         return HandlerOutcome::ack();
///     }
///     HandlerOutcome::retry_after(Duration::from_secs(30))
/// }
///
/// #[ruststream::app]
/// fn app() -> impl App {
///     RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(
///         NatsBroker::new("nats://localhost:4222"),
///         |b| {
///             // A pattern is no destination, so the mount site names where a delayed copy goes.
///             b.include(audit)
///                 .max_attempts(nonzero!(5u32))
///                 .dead_letter("orders.dead")
///                 .out_retry(Publish)
///                 .to("orders.audit");
///         },
///     )
/// }
/// # }
/// # fn main() {}
/// ```
#[derive(Debug, Clone, PartialEq, Eq)]
#[must_use]
pub struct CoreWildcard {
    subject: String,
    queue_group: Option<String>,
}

impl CoreWildcard {
    /// Subscribes to the pattern `subject` over Core NATS.
    pub fn new(subject: impl Into<String>) -> Self {
        Self {
            subject: subject.into(),
            queue_group: None,
        }
    }

    /// Load-balances the pattern across a queue group, exactly as
    /// [`CoreSubject::queue_group`] does for one subject.
    pub fn queue_group(mut self, name: impl Into<String>) -> Self {
        self.queue_group = Some(name.into());
        self
    }

    /// The queue group this subscription joins, if any.
    #[cfg(feature = "asyncapi")]
    pub(crate) fn group(&self) -> Option<&str> {
        self.queue_group.as_deref()
    }
}

impl Sealed for CoreWildcard {
    fn plan(&self) -> SubscriptionPlan<'_> {
        SubscriptionPlan::Core {
            queue_group: self.queue_group.as_deref(),
        }
    }
}

impl NatsSubscription for CoreWildcard {
    fn subject(&self) -> &str {
        &self.subject
    }
}

impl NatsSubscription for CoreSubject {
    fn subject(&self) -> &str {
        &self.subject
    }

    /// A wildcard belongs to [`CoreWildcard`], so it is refused here rather than opened as a
    /// subscription this type promises to address.
    ///
    /// The subject is a string the caller supplies, often from configuration, so the two forms
    /// cannot be told apart while the crate compiles. The subscription refuses to open instead,
    /// which is a service that does not start rather than one that starts and loses a delayed
    /// message.
    fn ensure_subject(&self) -> Result<(), NatsError> {
        publish_destination(self.subject()).map(|_| ())
    }
}

/// A subject read through a `JetStream` pull consumer on one stream.
///
/// The consumer is created when the subscription opens. Name it with [`durable`](Self::durable)
/// and the server keeps its position across restarts; leave the name out and the consumer is
/// ephemeral, discarded when the subscriber drops.
///
/// # Examples
///
/// ```
/// # mod demo {
/// use ruststream_nats::prelude::*;
/// use serde::Deserialize;
///
/// #[derive(Deserialize)]
/// struct Order {
///     id: u64,
/// }
///
/// // The durable keeps its position across restarts, and every replica shares it.
/// #[subscriber(
///     JetStreamSubject::new("orders.*", "ORDERS")
///         .durable("billing")
///         .ack_wait(NonZeroDuration::from_secs(nonzero!(30)))
/// )]
/// async fn bill(order: &Order) -> HandlerOutcome {
///     tracing::info!(order.id, "billed");
///     HandlerOutcome::ack()
/// }
///
/// #[ruststream::app]
/// fn app() -> impl App {
///     RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(
///         NatsBroker::new("nats://localhost:4222"),
///         |b| {
///             b.include(bill);
///         },
///     )
/// }
/// # }
/// # fn main() {}
/// ```
///
/// A queue group is Core NATS only, so it is not a method here:
///
/// ```compile_fail
/// use ruststream_nats::JetStreamSubject;
///
/// let bad = JetStreamSubject::new("orders.*", "ORDERS").queue_group("workers");
/// ```
#[derive(Debug, Clone)]
#[must_use]
pub struct JetStreamSubject {
    subject: String,
    stream: String,
    durable: Option<String>,
    filter_subject: Option<String>,
    ack_wait: Option<NonZeroDuration>,
    max_ack_pending: Option<i64>,
    deliver_policy: Option<DeliverPolicy>,
    pull_expires: Option<NonZeroDuration>,
}

impl JetStreamSubject {
    /// Reads `subject` from the stream named `stream`.
    ///
    /// `subject` is the pattern the subscription reports and the default consumer filter;
    /// `stream` is the `JetStream` stream that stores it.
    pub fn new(subject: impl Into<String>, stream: impl Into<String>) -> Self {
        Self {
            subject: subject.into(),
            stream: stream.into(),
            durable: None,
            filter_subject: None,
            ack_wait: None,
            max_ack_pending: None,
            deliver_policy: None,
            pull_expires: None,
        }
    }

    /// Names the consumer, so the server keeps its position across restarts.
    pub fn durable(mut self, name: impl Into<String>) -> Self {
        self.durable = Some(name.into());
        self
    }

    /// Narrows which subjects of the stream this consumer reads. Defaults to the subject the
    /// constructor named.
    pub fn filter_subject(mut self, subject: impl Into<String>) -> Self {
        self.filter_subject = Some(subject.into());
        self
    }

    /// Per-message acknowledgement timeout. After this window an unacked delivery is redelivered.
    /// Defaults to 30 seconds.
    pub const fn ack_wait(mut self, ack_wait: NonZeroDuration) -> Self {
        self.ack_wait = Some(ack_wait);
        self
    }

    /// Soft cap on in-flight unacked deliveries. Defaults to 1024.
    pub const fn max_ack_pending(mut self, max: i64) -> Self {
        self.max_ack_pending = Some(max);
        self
    }

    /// Where a newly created consumer starts reading. Defaults to `DeliverPolicy::All`.
    pub const fn deliver_policy(mut self, policy: DeliverPolicy) -> Self {
        self.deliver_policy = Some(policy);
        self
    }

    /// How long one fetch waits before delivering a partial (or retrying an empty) batch.
    /// Defaults to 5 seconds. Has no effect on the per-message
    /// [`Subscriber::stream`](ruststream::Subscriber::stream) path.
    ///
    /// How many messages a batch carries is not set here: the batch size travels with the
    /// registration (`b.include(handler.batch(nonzero!(6)))`) and reaches the fetch as the
    /// argument of [`BatchSubscriber::batches`](ruststream::BatchSubscriber::batches).
    pub const fn pull_expires(mut self, expires: NonZeroDuration) -> Self {
        self.pull_expires = Some(expires);
        self
    }

    /// The subjects the consumer actually reads: the filter where one is set, the subject
    /// otherwise.
    pub(crate) fn consumer_filter(&self) -> &str {
        self.filter_subject.as_deref().unwrap_or(&self.subject)
    }

    /// What this consumer is, in `JetStream`'s own words.
    ///
    /// The `nats` binding has one field and it belongs to a Core NATS queue group, so everything
    /// a stream consumer is configured with goes under an extension of this crate's own. Every
    /// value is the descriptor's; nothing here waits for a connection.
    #[cfg(feature = "asyncapi")]
    fn consumer_bindings(&self) -> Bindings {
        let body = JetStreamChannel {
            stream: &self.stream,
            durable: self.durable.as_deref(),
            filter_subject: self.consumer_filter(),
            ack_wait_seconds: self
                .ack_wait
                .map_or(DEFAULT_ACK_WAIT, NonZeroDuration::get)
                .as_secs_f64(),
            max_ack_pending: self.max_ack_pending.unwrap_or(DEFAULT_MAX_ACK_PENDING),
            deliver_policy: deliver_policy_name(self.deliver_policy.unwrap_or(DeliverPolicy::All)),
        };
        Binding::extension(JETSTREAM_EXTENSION, &body)
            .map(|binding| Bindings::new().with(binding))
            .unwrap_or_default()
    }
}

/// The extension key the `JetStream` consumer settings travel under.
///
/// The specification's `nats` binding declares its channel object empty, and the protocol keys are
/// a closed list, so a stream, a durable name and an ack window have no lawful binding field. An
/// `x-` extension sits at the same level and carries no `bindingVersion`.
#[cfg(feature = "asyncapi")]
pub(crate) const JETSTREAM_EXTENSION: &str = "x-ruststream-jetstream";

/// The whole `nats` binding, version 0.1.0: the queue group of a Core NATS subscription, on the
/// `receive` operation. A subscription outside a group contributes nothing rather than an empty
/// object.
#[cfg(feature = "asyncapi")]
fn core_operation_bindings(queue_group: Option<&str>) -> Bindings {
    queue_group
        .map(|queue| NatsOperation { queue })
        .and_then(|body| Binding::new("nats", "0.1.0", &body).ok())
        .map(|binding| Bindings::new().with(binding))
        .unwrap_or_default()
}

/// The name `JetStream` gives a delivery policy on the wire.
#[cfg(feature = "asyncapi")]
fn deliver_policy_name(policy: DeliverPolicy) -> &'static str {
    match policy {
        DeliverPolicy::All => "all",
        DeliverPolicy::Last => "last",
        DeliverPolicy::New => "new",
        DeliverPolicy::ByStartSequence { .. } => "by_start_sequence",
        DeliverPolicy::ByStartTime { .. } => "by_start_time",
        DeliverPolicy::LastPerSubject => "last_per_subject",
    }
}

/// The queue group, which is the whole `nats` binding.
#[cfg(feature = "asyncapi")]
#[derive(Serialize)]
struct NatsOperation<'a> {
    queue: &'a str,
}

/// What a `JetStream` consumer is configured with, under this crate's extension key.
#[cfg(feature = "asyncapi")]
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct JetStreamChannel<'a> {
    stream: &'a str,
    #[serde(skip_serializing_if = "Option::is_none")]
    durable: Option<&'a str>,
    filter_subject: &'a str,
    ack_wait_seconds: f64,
    max_ack_pending: i64,
    deliver_policy: &'static str,
}

impl Sealed for JetStreamSubject {
    fn plan(&self) -> SubscriptionPlan<'_> {
        SubscriptionPlan::JetStream {
            stream: &self.stream,
            durable: self.durable.as_deref(),
            filter_subject: self.filter_subject.as_deref().unwrap_or(&self.subject),
            ack_wait: self.ack_wait.map_or(DEFAULT_ACK_WAIT, NonZeroDuration::get),
            max_ack_pending: self.max_ack_pending.unwrap_or(DEFAULT_MAX_ACK_PENDING),
            deliver_policy: self.deliver_policy.unwrap_or(DeliverPolicy::All),
            pull_expires: self
                .pull_expires
                .map_or(DEFAULT_PULL_EXPIRES, NonZeroDuration::get),
        }
    }
}

impl NatsSubscription for JetStreamSubject {
    fn subject(&self) -> &str {
        &self.subject
    }
}

/// A descriptor is already a source, so the macro-free constructor takes it as it stands:
/// `subscriber(JetStreamSubject::new("orders.*", "ORDERS"), body)` names the same subscription
/// the `#[subscriber(..)]` attribute does.
///
/// # Examples
///
/// ```
/// # mod demo {
/// use ruststream::schemars::JsonSchema;
/// use ruststream_nats::prelude::*;
/// use serde::Deserialize;
///
/// #[derive(Deserialize, JsonSchema)]
/// # #[schemars(crate = "ruststream::schemars")]
/// struct Order {
///     id: u64,
/// }
///
/// struct Audit;
///
/// impl Handle<Order> for Audit {
///     async fn handle(
///         &self,
///         order: &Order,
///         _outs: &(),
///         _ctx: &mut Context<'_>,
///     ) -> Result<(), HandlerOutcome> {
///         tracing::info!(order.id, "audited");
///         Ok(())
///     }
/// }
///
/// #[ruststream::app]
/// fn app() -> impl App {
///     RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(
///         NatsBroker::new("nats://localhost:4222"),
///         |b| {
///             let audited = CoreSubject::new("orders.created").queue_group("auditors");
///             b.include(subscriber(audited, Audit).build());
///         },
///     )
/// }
/// # }
/// # fn main() {}
/// ```
impl IntoSource for CoreSubject {
    type Source = Self;

    fn into_source(self) -> Self {
        self
    }
}

impl IntoSource for JetStreamSubject {
    type Source = Self;

    fn into_source(self) -> Self {
        self
    }
}

impl IntoSource for CoreWildcard {
    type Source = Self;

    fn into_source(self) -> Self {
        self
    }
}

impl SubscriptionSource<ConnectedNatsBroker> for CoreSubject {
    type Subscriber = NatsSubscriber;
    // One subject, published to and subscribed to under one name, so the framework's deferred
    // copy needs nothing from the mount site.
    type Copies = AddressedCopies;

    fn name(&self) -> &str {
        NatsSubscription::subject(self)
    }

    async fn subscribe(
        self,
        connected: &ConnectedNatsBroker,
    ) -> Result<Self::Subscriber, NatsError> {
        connected.subscribe_with(self).await
    }

    #[cfg(feature = "asyncapi")]
    fn operation_bindings(&self) -> Bindings {
        core_operation_bindings(self.group())
    }
}

/// The subject itself, so the answer needs no connection and the future is ready.
impl RedeliveryAddressed<ConnectedNatsBroker> for CoreSubject {
    fn redelivery_address(
        &self,
        _connected: &ConnectedNatsBroker,
    ) -> impl Future<Output = Result<RedeliveryAddress, NatsError>> + Send {
        ready(publish_destination(NatsSubscription::subject(self)))
    }
}

impl SubscriptionSource<ConnectedNatsBroker> for CoreWildcard {
    type Subscriber = NatsSubscriber;
    // A pattern reads many subjects and names none of them.
    type Copies = NamedCopies;

    fn name(&self) -> &str {
        NatsSubscription::subject(self)
    }

    async fn subscribe(
        self,
        connected: &ConnectedNatsBroker,
    ) -> Result<Self::Subscriber, NatsError> {
        connected.subscribe_with(self).await
    }

    #[cfg(feature = "asyncapi")]
    fn operation_bindings(&self) -> Bindings {
        core_operation_bindings(self.group())
    }
}

impl SubscriptionSource<ConnectedNatsBroker> for JetStreamSubject {
    type Subscriber = NatsSubscriber;
    // The consumer reads its filter subject, and a publish there reaches it.
    type Copies = AddressedCopies;

    fn name(&self) -> &str {
        NatsSubscription::subject(self)
    }

    async fn subscribe(
        self,
        connected: &ConnectedNatsBroker,
    ) -> Result<Self::Subscriber, NatsError> {
        connected.subscribe_with(self).await
    }

    #[cfg(feature = "asyncapi")]
    fn channel_bindings(&self) -> Bindings {
        self.consumer_bindings()
    }
}

/// The consumer's filter subject, which defaults to the subject the descriptor names.
///
/// A `JetStream` consumer holds a delivery itself and takes the delay in its negative
/// acknowledgement, so the framework publishes no copy here and this address is what the
/// conformance suite checks rather than what a running service uses. A consumer filtering a
/// pattern reports that pattern, which is not a publish destination; nothing publishes to it, and
/// a `.to(name)` at the mount site overrides it.
impl RedeliveryAddressed<ConnectedNatsBroker> for JetStreamSubject {
    fn redelivery_address(
        &self,
        _connected: &ConnectedNatsBroker,
    ) -> impl Future<Output = Result<RedeliveryAddress, NatsError>> + Send {
        ready(Ok(RedeliveryAddress::new(
            self.consumer_filter().to_owned(),
        )))
    }
}

#[cfg(test)]
mod tests {
    use ruststream::nonzero;

    use super::*;

    #[test]
    fn a_plain_subscription_is_core_without_a_queue_group() {
        let opts = CoreSubject::new("orders.created");
        opts.ensure_subject().expect("core ok");
        assert!(matches!(
            opts.plan(),
            SubscriptionPlan::Core { queue_group: None }
        ));
    }

    #[test]
    fn a_queue_group_reaches_the_broker() {
        let opts = CoreSubject::new("orders.created").queue_group("workers");
        assert!(matches!(
            opts.plan(),
            SubscriptionPlan::Core {
                queue_group: Some("workers")
            }
        ));
    }

    #[test]
    fn jetstream_defaults_are_resolved_in_the_plan() {
        let consumer = JetStreamSubject::new("orders.*", "ORDERS");
        let SubscriptionPlan::JetStream {
            stream,
            durable,
            filter_subject,
            ack_wait,
            max_ack_pending,
            pull_expires,
            ..
        } = consumer.plan()
        else {
            panic!("a jetstream descriptor must plan a jetstream subscription");
        };
        assert_eq!(stream, "ORDERS");
        assert_eq!(durable, None);
        // The filter defaults to the subject, which only the descriptor knows.
        assert_eq!(filter_subject, "orders.*");
        assert_eq!(ack_wait, DEFAULT_ACK_WAIT);
        assert_eq!(max_ack_pending, DEFAULT_MAX_ACK_PENDING);
        assert_eq!(pull_expires, DEFAULT_PULL_EXPIRES);
    }

    // What a reported address promises is that a publish there arrives, so a pattern the server
    // would refuse a publish to has no business being one. `CoreSubject` carries the subject that
    // is a destination, and refuses to open on anything else.
    #[test]
    fn only_a_concrete_subject_is_a_redelivery_address() {
        assert_eq!(
            publish_destination("orders.created").expect("a literal subject is a destination"),
            RedeliveryAddress::new("orders.created")
        );
        for pattern in ["orders.*", "orders.>", ">", ""] {
            assert!(
                CoreSubject::new(pattern).ensure_subject().is_err(),
                "`{pattern}` is not a destination, so CoreSubject must refuse it"
            );
            assert!(publish_destination(pattern).is_err());
        }
    }

    // A pattern is the other type's business, and it opens on exactly what Core NATS accepts.
    #[test]
    fn a_pattern_opens_as_a_wildcard_subscription() {
        for pattern in ["orders.*", "orders.>", ">", "orders.created"] {
            CoreWildcard::new(pattern)
                .ensure_subject()
                .expect("a pattern is what this descriptor is for");
        }
        assert!(CoreWildcard::new("").ensure_subject().is_err());
    }

    // A JetStream consumer reads its filter, so a narrower filter is the address, not the subject
    // the descriptor was constructed with.
    #[test]
    fn a_jetstream_consumer_reports_the_subject_it_actually_reads() {
        assert_eq!(
            JetStreamSubject::new("orders.*", "ORDERS")
                .filter_subject("orders.created")
                .consumer_filter(),
            "orders.created"
        );
        assert_eq!(
            JetStreamSubject::new("orders.*", "ORDERS").consumer_filter(),
            "orders.*"
        );
    }

    #[test]
    fn jetstream_settings_reach_the_plan() {
        let consumer = JetStreamSubject::new("orders.*", "ORDERS")
            .durable("worker")
            .filter_subject("orders.created")
            .ack_wait(NonZeroDuration::from_secs(nonzero!(5)))
            .max_ack_pending(64)
            .deliver_policy(DeliverPolicy::New)
            .pull_expires(NonZeroDuration::from_millis(nonzero!(250)));
        let SubscriptionPlan::JetStream {
            durable,
            filter_subject,
            ack_wait,
            max_ack_pending,
            pull_expires,
            ..
        } = consumer.plan()
        else {
            panic!("a jetstream descriptor must plan a jetstream subscription");
        };
        assert_eq!(durable, Some("worker"));
        assert_eq!(filter_subject, "orders.created");
        assert_eq!(ack_wait, Duration::from_secs(5));
        assert_eq!(max_ack_pending, 64);
        assert_eq!(pull_expires, Duration::from_millis(250));
    }

    #[test]
    fn an_empty_subject_is_rejected_on_both_descriptors() {
        let core = CoreSubject::new("").ensure_subject().unwrap_err();
        assert!(matches!(core, NatsError::InvalidOptions(msg) if msg.contains("subject")));
        let js = JetStreamSubject::new("", "ORDERS")
            .ensure_subject()
            .unwrap_err();
        assert!(matches!(js, NatsError::InvalidOptions(msg) if msg.contains("subject")));
    }

    #[test]
    fn a_zero_duration_has_no_non_zero_form() {
        assert!(NonZeroDuration::new(Duration::ZERO).is_none());
        assert_eq!(
            NonZeroDuration::new(Duration::from_millis(1)).map(NonZeroDuration::get),
            Some(Duration::from_millis(1))
        );
    }
}