ruststream 0.7.5

//! The one mount chain: the attachments a form needs before it becomes a route.
//!
//! Every publish policy a registration names is attached with one verb,
//! [`out`](RouterWith::out): [`Reply`] names the policy the value a `reply("dest")` handler
//! returns leaves through, a slot marker names one [`Out`](crate::runtime::Out) slot's, and
//! [`Retry`] names the one a deferred `retry_after` copy leaves through.
//! [`out_reply`](RouterWith::out_reply) and [`out_retry`](RouterWith::out_retry) are the same
//! call with the marker spelled out.
//!
//! The steps after a call ride the position it named - [`codec`](RouterWith::codec),
//! [`transform`](RouterWith::transform),
//! [`map_publisher`](crate::runtime::MapPublisher::map_publisher) on either kind,
//! [`batch_transform`](RouterWith::batch_transform) and
//! [`transactional`](RouterWith::transactional) on the reply alone - so the chain reads the same
//! whichever position it is filling.
//!
//! A [`Router`](super::Router) is a consuming builder, so the chain commits through an explicit
//! terminal, [`build`](RouterWith::build): `Drop` cannot return the router the registration grew
//! into. A chain left unfinished never becomes a router, so a forgotten `.build()` is a compile
//! error at the next use. A [`BrokerScope`](crate::runtime::BrokerScope) drives the same chain
//! through a guard that commits when the statement ends.

use std::borrow::Cow;
use std::fmt;
use std::marker::PhantomData;
use std::num::NonZeroU32;

use crate::runtime::retry::{
    CapOpen, DeadLetterOpen, DeclareCap, DeclareDeadLetter, DestinationLast, DestinationOpen, Retry,
};
use crate::runtime::slot::{
    AdmitsAt, BatchTransformLast, CodecLast, MapPolicyLast, NamedStep, NoOutBound, OutPosition,
    Reply, ReplyLast, ReplyStep, TransactionalLast, TransformLast,
};

use super::builder::RouterBroker;
use super::routes::RetryDestinationsDeclared;

/// One commit strategy of a mount chain's attachment, keyed by its `Mount` token and the chain
/// it grew on. Machinery; never named directly.
#[doc(hidden)]
#[diagnostic::on_unimplemented(
    message = "this registration is not ready to mount",
    label = "the attachment `{Self}` has an unfilled position",
    note = "bind every Out slot the handler declares with .out(marker, policy) before .build(); \
            an unbound one shows as `MissingSlot<..>` naming its marker"
)]
pub trait RouterCommit<Mount, R, Def>: Sized {
    /// The router the committed registration grows into.
    type Out;

    fn commit(self, def: Def, router: R) -> Self::Out;
}

/// A mount chain over one registration's attachment.
///
/// The attachment is a pair: the reply position (the broker's default until `.out(Reply, ..)`
/// names a policy, or [`NoReply`](crate::runtime::NoReply) on a handler that declares no reply)
/// and the slot tuple (one position per [`Out`](crate::runtime::Out) marker the handler declares,
/// each unbound until its `.out(marker, ..)`). `Last` is the position the chain named most
/// recently, which is what the steps after it ride; it starts as
/// [`NoOutBound`](crate::runtime::NoOutBound), where those steps do not exist.
#[must_use = "a router registration is only added once .build() commits it"]
pub struct RouterWith<Mount, R, Def, Attach, Last = NoOutBound> {
    def: Def,
    attach: Attach,
    router: R,
    _mount: PhantomData<fn() -> Mount>,
    _last: PhantomData<fn() -> Last>,
}

impl<Mount, R, Def, Attach, Last> RouterWith<Mount, R, Def, Attach, Last> {
    pub(crate) fn new(def: Def, attach: Attach, router: R) -> Self {
        Self {
            def,
            attach,
            router,
            _mount: PhantomData,
            _last: PhantomData,
        }
    }

    /// Names the publish policy of one position: [`Reply`] for the value a `reply("dest")`
    /// handler returns, an [`Out`](crate::runtime::Out) slot's own marker for a slot, [`Retry`]
    /// for the deferred `retry_after` copy.
    ///
    /// `policy` is one of the broker prelude's (`Publish`, `TransactionalPublish`, ...), or a
    /// [`Bound`](crate::runtime::Bound) token wrapping one for a cross-broker target; the runtime
    /// pairs it after the brokers connect. Calls bind by marker, so their order does not matter,
    /// and each position takes exactly one: binding one twice does not compile. Omitting
    /// `.out(Reply, ..)` leaves the reply on the broker's own
    /// [`DefaultPublish`](crate::DefaultPublish) policy; omitting a slot's call does not compile;
    /// omitting `.out(Retry, ..)` leaves a `retry_after` copy on that same default policy.
    ///
    /// The steps after the call fill the rest of that position's wiring, whichever position it
    /// is: the deferred-retry copy is a slot publish like any other, so it takes the slot steps
    /// too.
    ///
    /// [`out_reply`](Self::out_reply) and [`out_retry`](Self::out_retry) are this call with the
    /// marker spelled out.
    // The unit marker drives inference, so it travels by value to keep the call site
    // `.out(Reply, ..)`; the return type names the chain with the bound position.
    #[allow(clippy::needless_pass_by_value, clippy::type_complexity)]
    pub fn out<M, Policy, Index>(
        self,
        marker: M,
        policy: Policy,
    ) -> RouterWith<
        Mount,
        R,
        Def,
        <M as OutPosition<Mount, R::Broker, Attach, Policy, Index>>::Out,
        Index,
    >
    where
        R: RouterBroker,
        M: OutPosition<Mount, R::Broker, Attach, Policy, Index>,
    {
        // The marker is inference input only; its value carries no data.
        let _ = marker;
        RouterWith::new(self.def, M::bind(self.attach, policy), self.router)
    }

    /// Names the publish policy the handler's reply leaves through: [`out`](Self::out) with the
    /// [`Reply`] marker, and the same chain afterwards.
    ///
    /// # Examples
    ///
    /// The reply type fixes its destination, and the mount site names the policy the reply leaves through:
    ///
    /// ```
    /// # #[cfg(all(feature = "macros", feature = "memory", feature = "json"))]
    /// # mod demo {
    /// use ruststream::memory::prelude::*;
    /// use serde::{Deserialize, Serialize};
    ///
    /// #[derive(Deserialize)]
    /// struct Request {
    ///     id: u64,
    /// }
    ///
    /// /// The reply type fixes its destination, so the clause names none.
    /// #[derive(Serialize, Outgoing)]
    /// #[outgoing(name = "receipts")]
    /// struct Receipt {
    ///     id: u64,
    /// }
    ///
    /// #[subscriber("requests", reply)]
    /// async fn issue(req: &Request) -> Result<Receipt, HandlerOutcome> {
    ///     if req.id == 0 {
    ///         return Err(HandlerOutcome::drop());
    ///     }
    ///     Ok(Receipt { id: req.id })
    /// }
    ///
    /// fn app() -> RustStream {
    ///     RustStream::new(AppInfo::new("receipts", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
    ///         b.include(issue).out_reply(Publish);
    ///     })
    /// }
    /// # }
    /// # fn main() {}
    /// ```
    #[allow(clippy::type_complexity)] // the chain's own state; an alias would hide the position
    pub fn out_reply<Policy>(
        self,
        policy: Policy,
    ) -> RouterWith<
        Mount,
        R,
        Def,
        <Reply as OutPosition<Mount, R::Broker, Attach, Policy, ReplyLast>>::Out,
        ReplyLast,
    >
    where
        R: RouterBroker,
        Reply: OutPosition<Mount, R::Broker, Attach, Policy, ReplyLast>,
    {
        self.out(Reply, policy)
    }

    /// Replaces the publish policy a `retry_after` copy leaves through: [`out`](Self::out) with
    /// the [`Retry`] marker, and the same chain afterwards.
    ///
    /// The position is filled already, from the broker's own
    /// [`DefaultPublish`](crate::DefaultPublish) policy, so this call is a customisation and not a
    /// requirement. It comes after [`max_attempts`](Self::max_attempts) and
    /// [`dead_letter`](Self::dead_letter). The position is an `Out` slot like any other, so the
    /// steps after it are the slot steps: [`codec`](Self::codec), [`transform`](Self::transform)
    /// and a broker's own [`map_publisher`](MapPublisher::map_publisher) settings.
    #[allow(clippy::type_complexity)] // the chain's own state; an alias would hide the position
    pub fn out_retry<Policy, Index>(
        self,
        policy: Policy,
    ) -> RouterWith<
        Mount,
        R,
        Def,
        <Retry as OutPosition<Mount, R::Broker, Attach, Policy, Index>>::Out,
        Index,
    >
    where
        R: RouterBroker,
        Retry: OutPosition<Mount, R::Broker, Attach, Policy, Index>,
    {
        self.out(Retry, policy)
    }

    /// Caps the deliveries one message of this registration gets, counting the first.
    ///
    /// A delivery that has used them up is not retried: it goes to the destination
    /// [`dead_letter`](Self::dead_letter) names, and where the registration names none it is
    /// rejected, which keeps the broker's own dead-letter policy in play where one is configured.
    /// The count is the broker's own where the transport keeps one
    /// ([`IncomingMessage::redelivery_count`](crate::IncomingMessage::redelivery_count)) and the
    /// framework's `x-ruststream-retry-count` header otherwise, never both.
    ///
    /// Declared once per registration, before `out_retry(policy)` names the publisher the copies
    /// leave through. Where the broker applies a delivery limit itself, the subscription
    /// descriptor takes the declaration into its own topology and this process publishes nothing.
    ///
    /// A broker with native delayed redelivery gets a `retry_after` delay only while the delivery
    /// is below the cap: the count is read first, and a delivery at the cap goes to the
    /// dead-letter destination or is rejected rather than coming back. The same one count is read
    /// there, so the cap holds where the broker holds the message back itself and where its crate
    /// honours the delay by publishing a copy that carries the framework's header.
    ///
    /// An immediate [`retry`](crate::runtime::HandlerOutcome::retry) obeys the cap too. On a
    /// transport that counts its own redeliveries it stays the broker's requeue; on one that
    /// counts none the runtime republishes the delivery at once so the count travels with it, so
    /// under a cap an immediate retry is no longer a broker requeue.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # #[cfg(all(feature = "memory", feature = "macros", feature = "json"))]
    /// # fn build() {
    /// use ruststream::memory::MemoryBroker;
    /// use ruststream::runtime::{HandlerOutcome, Router, RouterDef};
    /// use ruststream::{nonzero, subscriber};
    /// # #[derive(serde::Deserialize)]
    /// # struct Order { id: u64 }
    ///
    /// #[subscriber("orders")]
    /// async fn reconcile(order: &Order) -> HandlerOutcome {
    ///     let _ = order.id;
    ///     HandlerOutcome::retry()
    /// }
    ///
    /// fn routes() -> impl RouterDef<MemoryBroker> {
    ///     Router::<MemoryBroker>::new()
    ///         .include(reconcile)
    ///         .max_attempts(nonzero!(5u32))
    ///         .dead_letter("orders.dead")
    ///         .build()
    /// }
    /// # }
    /// ```
    #[allow(clippy::type_complexity)] // the chain's own state; an alias would hide the position
    pub fn max_attempts(
        self,
        attempts: NonZeroU32,
    ) -> RouterWith<Mount, R, Def, <Attach as DeclareCap>::Out, Last>
    where
        Attach: DeclareCap<Step: CapOpen>,
    {
        RouterWith::new(self.def, self.attach.declare_cap(attempts), self.router)
    }

    /// Names where a delivery of this registration goes once its attempts run out.
    ///
    /// A plain destination, the way a reply's is: a name template does not resolve here, because
    /// nothing binds its placeholders. The delivery is republished there as it arrived - payload
    /// and headers, with the retry-count header incremented - so a dead-letter consumer reads the
    /// message the handler could not process. The generated `AsyncAPI` document reports it as a
    /// channel this registration publishes to.
    ///
    /// Declared on its own, without [`max_attempts`](Self::max_attempts), it takes over every
    /// copy: a handler that asks for a retry has its delivery sent there instead of back to the
    /// subscription.
    ///
    /// Declared once per registration, before `out_retry(policy)`. Where the broker applies a
    /// dead-letter destination itself, the subscription descriptor takes the declaration into its
    /// own topology - and only when a cap is declared beside it, because a native dead-letter
    /// policy needs both.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # #[cfg(all(feature = "memory", feature = "macros", feature = "json"))]
    /// # fn build() {
    /// use ruststream::memory::MemoryBroker;
    /// use ruststream::runtime::{HandlerOutcome, Router, RouterDef};
    /// use ruststream::subscriber;
    /// # #[derive(serde::Deserialize)]
    /// # struct Order { id: u64 }
    ///
    /// #[subscriber("orders")]
    /// async fn reconcile(order: &Order) -> HandlerOutcome {
    ///     let _ = order.id;
    ///     HandlerOutcome::ack()
    /// }
    ///
    /// fn routes() -> impl RouterDef<MemoryBroker> {
    ///     Router::<MemoryBroker>::new()
    ///         .include(reconcile)
    ///         .dead_letter("orders.dead")
    ///         .build()
    /// }
    /// # }
    /// ```
    #[allow(clippy::type_complexity)] // the chain's own state; an alias would hide the position
    pub fn dead_letter(
        self,
        destination: impl Into<Cow<'static, str>>,
    ) -> RouterWith<Mount, R, Def, <Attach as DeclareDeadLetter>::Out, Last>
    where
        Attach: DeclareDeadLetter<Step: DeadLetterOpen>,
    {
        RouterWith::new(
            self.def,
            self.attach.declare_dead_letter(destination.into()),
            self.router,
        )
    }

    /// Names where the copies of the position named last go: a plain destination, as a reply's
    /// is, and a name template does not resolve here because nothing binds its placeholders.
    ///
    /// Only the deferred-retry position takes one, and only right after `out_retry(policy)`
    /// named it, so a reader sees what the destination belongs to. A descriptor that addresses
    /// its own subscription already answers, and this overrides that answer; a descriptor that
    /// addresses nothing - a wildcard subject, an MQTT filter, a pattern, a list of topics - has
    /// no answer, so naming the publisher and then the destination is what such a registration
    /// owes. The other way is a transform that names the destination per delivery, and the two
    /// are mutually exclusive, as a declared reply and a naming transform are.
    ///
    /// A `Router` chain refuses a registration that names neither at `.build()`. A scope's
    /// registration commits when the statement ends, so there the refusal is the subscription's
    /// at startup.
    ///
    /// A name given here is a channel the registration sends to, and the generated `AsyncAPI`
    /// document reports it with a `send` operation. The address a descriptor takes its own
    /// copies back at is not reported: it is the subscription's own channel, which the document
    /// already carries.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # #[cfg(all(feature = "memory", feature = "macros", feature = "json"))]
    /// # fn build() {
    /// use ruststream::memory::{MemoryBroker, MemoryPublish};
    /// use ruststream::runtime::{HandlerOutcome, Router, RouterDef};
    /// use ruststream::subscriber;
    /// # #[derive(serde::Deserialize)]
    /// # struct Order { id: u64 }
    ///
    /// #[subscriber("orders")]
    /// async fn reconcile(order: &Order) -> HandlerOutcome {
    ///     let _ = order.id;
    ///     HandlerOutcome::retry()
    /// }
    ///
    /// fn routes() -> impl RouterDef<MemoryBroker> {
    ///     Router::<MemoryBroker>::new()
    ///         .include(reconcile)
    ///         .out_retry(MemoryPublish)
    ///         .to("orders.retry")
    ///         .build()
    /// }
    /// # }
    /// ```
    #[allow(clippy::type_complexity)] // the chain's own state; an alias would hide the position
    pub fn to(
        self,
        destination: impl Into<Cow<'static, str>>,
    ) -> RouterWith<Mount, R, Def, <Attach as DestinationLast<Last>>::Out, Last>
    where
        Attach: DestinationLast<Last, Step: DestinationOpen>,
    {
        RouterWith::new(
            self.def,
            self.attach.destination_last(destination.into()),
            self.router,
        )
    }

    /// Encodes what leaves the position named last with `codec` instead of the registration
    /// surface's own.
    ///
    /// Named once per position: the codec slot the call fills is filled, so a second one does not
    /// compile. A byte-for-byte ([`Serialized`](crate::runtime::Serialized)) reply carries its own
    /// bytes and has no codec position at all.
    #[allow(clippy::type_complexity)] // the chain's own state; an alias would hide the position
    pub fn codec<Cd>(
        self,
        codec: Cd,
    ) -> RouterWith<Mount, R, Def, <Attach as CodecLast<Cd, Last>>::Out, Last>
    where
        Attach: CodecLast<Cd, Last, Step: NamedStep>,
    {
        RouterWith::new(self.def, self.attach.codec_last(codec), self.router)
    }

    /// Composes a static [`PublishTransform`](crate::runtime::PublishTransform) onto everything
    /// that leaves the position named last, of that position's own
    /// [`ContextKind`](crate::runtime::ContextKind): a reply hands the transform the delivery it
    /// answers, a slot hands it the slot's name.
    ///
    /// The step repeats and they run in the order the chain writes them, so a chain names as many
    /// as it likes per position:
    /// `.out(Reply, Publish).transform(StampSource).out(Audit, Publish).transform(Envelope)`.
    /// Before any `.out(..)` the step has no position to ride, and the call fails naming the fix.
    ///
    /// A transform declaring
    /// [`Destination = Names`](crate::runtime::PublishTransform::Destination) needs the position to
    /// offer that right, and the call fails here when it does not: the reply type already declares
    /// its channel, the replies are a batch's, the right has been taken by an earlier transform, or
    /// the slot's `#[publishes(..)]` dictionary contains a type that fixes its own channel.
    #[allow(clippy::type_complexity)] // the chain's own state; an alias would hide the position
    pub fn transform<N>(
        self,
        transform: N,
    ) -> RouterWith<Mount, R, Def, <Attach as TransformLast<N, Last>>::Out, Last>
    where
        R: RouterBroker,
        Attach: TransformLast<N, Last, Step: NamedStep> + AdmitsAt<N, Last, Mount, Def, R::Broker>,
    {
        RouterWith::new(self.def, self.attach.transform_last(transform), self.router)
    }

    /// Composes a [`BatchPublishTransform`](crate::runtime::BatchPublishTransform) onto every
    /// reply of a batch (`&[T]` plus `reply(..)`), after the per-message stack. Wrap a
    /// per-message transform with [`for_batch`](crate::runtime::for_batch) to reuse it here.
    ///
    /// Reply-only: a slot publish is one message with no batch to run a batch transform over.
    #[allow(clippy::type_complexity)] // the chain's own state; an alias would hide the position
    pub fn batch_transform<N>(
        self,
        transform: N,
    ) -> RouterWith<Mount, R, Def, <Attach as BatchTransformLast<N, Last>>::Out, Last>
    where
        Attach: BatchTransformLast<N, Last, Step: ReplyStep>,
    {
        RouterWith::new(
            self.def,
            self.attach.batch_transform_last(transform),
            self.router,
        )
    }

    /// Publishes a batch's replies inside one broker transaction: they all become visible
    /// atomically on commit, or none of them do.
    ///
    /// The policy's live publisher has to be a
    /// [`TransactionalPublisher`](crate::TransactionalPublisher), which the pairing checks against
    /// the chain's own broker; a one-message reply has no batch to make atomic, so the wiring only
    /// mounts on the batch forms. Reply-only, for the same reason
    /// [`batch_transform`](Self::batch_transform) is.
    #[allow(clippy::type_complexity)] // the chain's own state; an alias would hide the position
    pub fn transactional(
        self,
    ) -> RouterWith<Mount, R, Def, <Attach as TransactionalLast<Last>>::Out, Last>
    where
        Attach: TransactionalLast<Last, Step: ReplyStep>,
    {
        RouterWith::new(self.def, self.attach.transactional_last(), self.router)
    }

    /// Adds the registration to the router, with whatever the chain attached - the broker's own
    /// [`DefaultPublish`](crate::DefaultPublish) policy for a reply no `.out(Reply, ..)` named.
    ///
    /// This is also where a registration that bound the deferred-retry position over a
    /// subscription which addresses nothing, and then named no destination for the copies, is
    /// refused.
    #[allow(clippy::type_complexity)] // the commit's own output; an alias would hide the router
    pub fn build(self) -> <Attach as RouterCommit<Mount, R, Def>>::Out
    where
        Attach: RouterCommit<Mount, R, Def, Out: RetryDestinationsDeclared>,
    {
        self.commit_chain()
    }

    /// Commits the chain without asking whether every registration named a destination.
    ///
    /// A scope's guard commits when the statement ends, so the chain there passes through no
    /// state that could carry the question: `.to(name)` is written after `out_retry(policy)`, and
    /// by then the guard would already have had to be committable. That surface takes the startup
    /// refusal instead, and this is the commit it uses.
    pub(crate) fn commit_chain(self) -> <Attach as RouterCommit<Mount, R, Def>>::Out
    where
        Attach: RouterCommit<Mount, R, Def>,
    {
        self.attach.commit(self.def, self.router)
    }
}

impl<Mount, R, Def, Attach, Last> fmt::Debug for RouterWith<Mount, R, Def, Attach, Last> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("RouterWith").finish_non_exhaustive()
    }
}

/// Replaces the publish policy of the position a mount chain named last, keeping every step the
/// chain already filled: the hook a broker crate layers its own publisher settings on.
///
/// The publish-side mirror of
/// [`map_source`](crate::runtime::SubscriberSettings::map_source). Core cannot know that a publish
/// has an exchange, a partition key or a confirm mode, so the broker declares a trait over the
/// mount chain bound to its own policy type and implements each method as one `map_publisher`
/// call. The bound means those methods simply do not exist on a chain that named another broker's
/// policy - or none at all.
///
/// A mount site then reads `b.include(confirm).out(Reply, StreamPublish::default()).stream("ORDERS")`,
/// and the same one impl serves the reply position and every slot, on a router and on a broker
/// scope alike.
///
/// # Examples
///
/// ```
/// use ruststream::runtime::MapPublisher;
///
/// /// A broker's publish policy: the mode, plus the options that shape it. Freely
/// /// constructible, because it holds no connection.
/// #[derive(Debug, Clone, Default)]
/// pub struct StreamPublish {
///     stream: Option<String>,
///     expect_last_sequence: Option<u64>,
/// }
///
/// /// What the broker crate ships next to it: the publisher's own settings, reachable wherever
/// /// a mount chain named that policy.
/// pub trait StreamPublishSettings: Sized {
///     /// Publishes into `stream` rather than letting the server pick by subject.
///     fn stream(self, stream: impl Into<String>) -> Self;
///
///     /// Rejects the publish unless the stream's last sequence is `seq`.
///     fn expect_last_sequence(self, seq: u64) -> Self;
/// }
///
/// // Each method is one `map_publisher` call: take the policy the chain carries, set the
/// // option on it, hand it back. Everything else the chain named stays where it was.
/// impl<T: MapPublisher<Policy = StreamPublish>> StreamPublishSettings for T {
///     fn stream(self, stream: impl Into<String>) -> Self {
///         let stream = stream.into();
///         self.map_publisher(move |mut policy| {
///             policy.stream = Some(stream);
///             policy
///         })
///     }
///
///     fn expect_last_sequence(self, seq: u64) -> Self {
///         self.map_publisher(move |mut policy| {
///             policy.expect_last_sequence = Some(seq);
///             policy
///         })
///     }
/// }
/// ```
pub trait MapPublisher: Sized {
    /// The policy the position a mount chain named last carries.
    type Policy;

    /// Replaces it with what the broker's own settings made of it.
    ///
    /// The replacement is the same policy type: a broker's publisher settings are that policy's
    /// own fields (an exchange, a partition key, a confirm mode), while a different policy type
    /// is a different publish mode and belongs in the `.out(marker, policy)` call itself.
    // No `#[must_use]`, unlike what the lint asks for: the steps beside it on the chain carry
    // none either, because a scope registration commits when its guard drops at the end of the
    // statement. The attribute would fire on
    // `b.include(confirm).out(Reply, Publish).stream("ORDERS");`, which is correct as written,
    // and a broker's settings trait built on this hook would have to carry the same false
    // positive at every one of its own methods. What genuinely must be used - a router chain, and
    // a scope chain with unbound slots - says so on its own type instead.
    #[allow(clippy::return_self_not_must_use)]
    fn map_publisher(self, f: impl FnOnce(Self::Policy) -> Self::Policy) -> Self;
}

impl<Mount, R, Def, Attach, Last> MapPublisher for RouterWith<Mount, R, Def, Attach, Last>
where
    Attach: MapPolicyLast<Last, Step: NamedStep>,
{
    type Policy = Attach::Policy;

    fn map_publisher(self, f: impl FnOnce(Self::Policy) -> Self::Policy) -> Self {
        Self::new(self.def, self.attach.map_policy_last(f), self.router)
    }
}

use super::mount::DefaultReply;
use crate::runtime::slot::NoReply;

/// The attachment a reply-only form starts with: the broker's default policy, no slots.
pub(crate) type ReplyOnly = (DefaultReply, ());

/// The attachment a slot-only form starts with: no reply position, one unbound slot per marker.
pub(crate) type SlotsOnly<Slots> = (NoReply, Slots);

/// The attachment a reply-and-slots form starts with.
pub(crate) type ReplyAndSlots<Slots> = (DefaultReply, Slots);

/// The chain a reply-only form hands back.
pub type RouterPublishing<Mount, R, Def> = RouterWith<Mount, R, Def, ReplyOnly>;

/// The chain a slot-carrying form hands back.
pub type RouterOut<Mount, R, Def, Slots> = RouterWith<Mount, R, Def, SlotsOnly<Slots>>;

/// The chain a reply-and-slots form hands back.
pub type RouterPublishingOut<Mount, R, Def, Slots> =
    RouterWith<Mount, R, Def, ReplyAndSlots<Slots>>;