ruststream 0.7.4

//! The publish builder: one call shape for every publish, with the positions a message type
//! leaves open.
//!
//! A publish is a value (or bytes), a destination, optionally typed headers, and - for a value -
//! a codec. The builder carries one position per piece and resolves each at the most specific
//! level that names it, so the surfaces differ in what they *declare*, never in which method to
//! call. Which positions exist at a call site is decided by the message type's
//! [`OutgoingDestination`] declaration, so an under-specified publish does not compile.

use std::borrow::Cow;
use std::error::Error as StdError;
use std::fmt;
use std::future::Future;
use std::marker::PhantomData;

use bytes::BytesMut;
use serde::Serialize;
use thiserror::Error;

use crate::codec::{Codec, CodecError};
use crate::{
    CallerName, FixedName, HeaderMap, HeadersContract, MessageHeaders, NameTemplate, NoHeaders,
    OutgoingDestination, OutgoingMessage, PayloadForm, SerializeHeadersError, WithHeaders,
};

use super::sink::{CallCodec, PublishCodec, PublishSink};

/// A publish under construction: the entry point's payload plus the positions still open.
///
/// Built by the `message(..)` entry point of every publish surface (an
/// [`Out`](crate::runtime::Out) slot, a transaction scope, any [`Publisher`](crate::Publisher)
/// through [`PublishExt`](super::PublishExt)), and finished by
/// [`publish`](Self::publish). The type parameters are the positions:
///
/// * `Sink` - where the bytes go, a [`PublishSink`].
/// * `Body` - [`MessageBody`], the value to encode. The runtime also fills this position with a
///   crate-internal byte body where it republishes a delivery it already holds as bytes.
/// * `Enc` - the codec position ([`PublishCodec`]); `()` on the byte body, which has none.
/// * `Hdrs` - [`HeadersUnset`], [`TypedHeaders`] or [`MapHeaders`].
/// * `Dest` - the destination: [`FixedName`] and [`SuppliedName`] are resolved, [`CallerName`]
///   and [`NameTemplate`] are still waiting for the call site.
///
/// Beside the positions it carries the broker's per-message settings
/// ([`PublishSink::Options`]), unset until a step of the broker's own extension trait fills one
/// in through [`options_mut`](Self::options_mut); what no step touches keeps what the publish
/// policy fixed.
///
/// You never name this type; the entry points return it and the compiler carries it.
#[must_use = "a publish builder does nothing until publish() is awaited"]
pub struct PublishBuilder<Sink: PublishSink, Body, Enc, Hdrs, Dest> {
    sink: Sink,
    body: Body,
    codec: Enc,
    headers: Hdrs,
    dest: Dest,
    options: Option<Sink::Options>,
}

impl<Sink: PublishSink, Body, Enc, Hdrs, Dest> fmt::Debug
    for PublishBuilder<Sink, Body, Enc, Hdrs, Dest>
{
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("PublishBuilder").finish_non_exhaustive()
    }
}

/// The payload of a typed publish: the value to encode with the resolved codec.
#[derive(Debug, Clone, Copy)]
pub struct MessageBody<'a, T>(&'a T);

/// The payload of a byte publish: the bytes to send as they are.
#[derive(Debug, Clone, Copy)]
pub struct RawBody<'a>(&'a [u8]);

/// The headers position before anything fills it: the message travels with no headers of its own.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)]
pub struct HeadersUnset;

/// The headers position filled with a typed contract value, serialized into the header map one
/// entry per field when the publish runs.
#[derive(Debug, Clone, Copy)]
pub struct TypedHeaders<'a, H: ?Sized>(&'a H);

/// The headers position filled with an already-built [`HeaderMap`], sent as it is.
#[derive(Debug, Clone, Default)]
pub struct MapHeaders(HeaderMap);

/// A destination named at the call site, the resolved form of [`CallerName`].
#[derive(Debug, Clone)]
pub struct SuppliedName<'a>(Cow<'a, str>);

/// A destination that the publish can be sent to: fixed by the declaration, or named at the call.
///
/// This is what separates a publish that is ready to run from one whose call site has not said
/// where the message goes. [`CallerName`] and [`NameTemplate`] deliberately do not implement it.
#[diagnostic::on_unimplemented(
    message = "this publish has no destination yet",
    note = "a message type declaring no name takes one at the call: `.to(\"orders.archived\")`; \
            one declaring a name template opens its placeholder setters with `.to()`, and the \
            publish appears once every placeholder is bound"
)]
pub trait ResolvedName {
    /// The destination this publish goes to, given the type's declared name (used by the
    /// fixed form, ignored by the supplied one).
    fn resolve<'n>(&'n self, declared: &'n str) -> &'n str;
}

impl ResolvedName for FixedName {
    fn resolve<'n>(&'n self, declared: &'n str) -> &'n str {
        declared
    }
}

impl ResolvedName for SuppliedName<'_> {
    fn resolve<'n>(&'n self, _declared: &'n str) -> &'n str {
        &self.0
    }
}

/// What a headers position contributes to the outgoing message.
///
/// Implemented by the three header states; the publish calls it once, just before the message
/// leaves, against the map the sink's base
/// ([`base_headers`](crate::Publisher::base_headers)) already filled - empty when there is no
/// base. A position writes over that map key by key, which is what gives the call site the last
/// word over the handle it published through.
pub trait PublishHeaders {
    /// Writes this position's headers over the outgoing message's map, one key at a time.
    ///
    /// # Errors
    ///
    /// Returns [`SerializeHeadersError`] when a typed contract value is not a flat struct or
    /// string-keyed map.
    fn write(self, headers: &mut HeaderMap) -> Result<(), SerializeHeadersError>;
}

impl PublishHeaders for HeadersUnset {
    fn write(self, _headers: &mut HeaderMap) -> Result<(), SerializeHeadersError> {
        Ok(())
    }
}

impl<H: Serialize + ?Sized> PublishHeaders for TypedHeaders<'_, H> {
    fn write(self, headers: &mut HeaderMap) -> Result<(), SerializeHeadersError> {
        headers.insert_typed(self.0)
    }
}

impl PublishHeaders for MapHeaders {
    fn write(self, headers: &mut HeaderMap) -> Result<(), SerializeHeadersError> {
        headers.overwrite_with(self.0);
        Ok(())
    }
}

/// A headers position that satisfies a message's declared header contract.
///
/// The contract comes from the message type ([`MessageHeaders`]): a type declaring
/// `#[outgoing(headers = Meta)]` publishes only with `with_headers(&meta)`, and a type declaring
/// none publishes without typed headers. The byte entry point has no contract, so every position
/// is admissible there.
#[diagnostic::on_unimplemented(
    message = "the headers of this publish do not match the message's `{C}` contract",
    note = "a message declaring `#[outgoing(headers = Meta)]` is published as \
            `.message(&value).with_headers(&meta)`; one declaring no contract takes no contract \
            value (an arbitrary map still travels through `.with_headers(headers)`)"
)]
pub trait SatisfiesContract<C: HeadersContract> {}

impl SatisfiesContract<NoHeaders> for HeadersUnset {}

// A contract-less message may still carry arbitrary transport headers; that map says nothing
// about the message's declared shape, so it does not stand in for a contract.
impl SatisfiesContract<NoHeaders> for MapHeaders {}

impl<H> SatisfiesContract<WithHeaders<H>> for TypedHeaders<'_, H> {}
/// The failure a value reports when its own encoder rejects it, erased so that one error type
/// serves every wire format.
///
/// It is what `#[derive(Serialized)]` names as [`Serialized::Error`] for a type that serializes
/// itself through `#[wire(encode = ..)]`. A hand-written impl names its own error instead -
/// [`Infallible`](core::convert::Infallible) where the bytes are already there.
#[derive(Debug, Error)]
#[error(transparent)]
pub struct SerializePayloadError(Box<dyn StdError + Send + Sync>);

impl SerializePayloadError {
    /// Erases one wire format's encode failure.
    #[must_use]
    pub fn new(source: impl StdError + Send + Sync + 'static) -> Self {
        Self(Box::new(source))
    }
}

/// What a `#[wire(encode = ..)]` function may return: nothing, or a result. Machinery behind the
/// derive, so that a format whose writer cannot fail is not made to pretend it can.
///
/// There is no counterpart for `decode`: `Result<Self, E>` and a bare `Self` overlap as soon as
/// the decoded type is itself a `Result`, so the decode half names a fallible function and says
/// so - which every self-describing format's reader is anyway.
#[doc(hidden)]
pub trait EncodeOutcome {
    /// Normalizes the outcome into the trait's error.
    ///
    /// # Errors
    ///
    /// Returns [`SerializePayloadError`] carrying the format's own error, when there was one.
    fn finish(self) -> Result<(), SerializePayloadError>;
}

impl EncodeOutcome for () {
    fn finish(self) -> Result<(), SerializePayloadError> {
        Ok(())
    }
}

impl<E: StdError + Send + Sync + 'static> EncodeOutcome for Result<(), E> {
    fn finish(self) -> Result<(), SerializePayloadError> {
        self.map_err(SerializePayloadError::new)
    }
}

/// A value that owns its byte layout: `Serialize` means the framework's codec does the work,
/// `Serialized` means the type produces the bytes itself.
///
/// A `Serialized` value leaves the service byte-for-byte with no codec anywhere on the path,
/// selected purely by the type, on every typed surface: the `message(&value)` entry of a
/// publish builder (an [`Out`](crate::runtime::Out) slot, a transaction scope, any publisher),
/// and the reply position (the same `.reply()` chain and `reply("dest")` clause an encoded
/// reply uses).
///
/// The two shapes it covers are one method apart. A byte bag lends what it holds; a generated
/// message (Protobuf, Cap'n Proto, a hand-rolled frame) holds fields, and writes them into the
/// buffer the publish path hands it - so nothing is encoded twice and nothing intermediate is
/// allocated.
///
/// # Implementing by hand
///
/// `#[derive(Serialized)]` (under the `macros` feature) covers a newtype or single-field struct
/// over a byte buffer, and, with `#[wire(encode = ..)]`, any type whose format names an encode
/// function. Any other shape is the bytes plus the wire spellings that route the type onto the
/// serialized wire: [`MessageWire`] for the typed publish entry, and
/// [`ReplyShape`](crate::runtime::ReplyShape) for the reply position.
///
/// ```
/// use std::convert::Infallible;
///
/// use ruststream::prelude::*;
///
/// struct Export(Vec<u8>);
///
/// impl Serialized for Export {
///     type Error = Infallible;
///
///     fn wire_bytes(&self, _buf: &mut BytesMut) -> Result<WireBytes<'_>, Infallible> {
///         Ok(WireBytes::Own(&self.0))
///     }
/// }
///
/// impl MessageWire for Export {
///     type Wire = SerializedWire;
/// }
///
/// impl ReplyShape for Export {
///     type Body = Self;
///     type Headers = ();
///     type Wire = SerializedReply;
/// }
///
/// # fn check() -> Result<(), Box<dyn std::error::Error>> {
/// let export = Export(vec![7, 9]);
/// let mut buf = BytesMut::new();
/// assert!(matches!(export.wire_bytes(&mut buf)?, WireBytes::Own(bytes) if bytes == [7, 9]));
/// # Ok(())
/// # }
/// # check().unwrap();
/// ```
///
/// A type that computes its bytes writes them into the buffer and lends that:
///
/// ```
/// use std::convert::Infallible;
///
/// use ruststream::prelude::*;
///
/// struct Tick(u32);
///
/// impl Serialized for Tick {
///     type Error = Infallible;
///
///     fn wire_bytes(&self, buf: &mut BytesMut) -> Result<WireBytes<'_>, Infallible> {
///         buf.extend_from_slice(&self.0.to_be_bytes());
///         Ok(WireBytes::InBuffer)
///     }
/// }
///
/// impl MessageWire for Tick {
///     type Wire = SerializedWire;
/// }
///
/// # fn check() -> Result<(), Box<dyn std::error::Error>> {
/// let mut buf = BytesMut::new();
/// assert!(matches!(Tick(1).wire_bytes(&mut buf)?, WireBytes::InBuffer));
/// assert_eq!(&buf[..], &[0, 0, 0, 1]);
/// # Ok(())
/// # }
/// # check().unwrap();
/// ```
pub trait Serialized {
    /// The failure of producing the bytes; [`Infallible`](core::convert::Infallible) for a value
    /// that already holds them.
    // Tighter than `Deserialized::Error`'s `Display`: a publish failure travels as the `#[source]`
    // of `PublishError`, where a construction failure is only reported to the decode policy.
    type Error: StdError + Send + Sync + 'static;

    /// The bytes the value publishes, exactly as they leave on the wire: the ones it already
    /// holds, answered as [`WireBytes::Own`], or the ones it writes into `buf`, answered as
    /// [`WireBytes::InBuffer`].
    ///
    /// One method rather than a lend-or-compute pair, so a value cannot answer the two
    /// differently, and a value that carries its bytes never touches `buf` - which is why the
    /// publish path can hand it a buffer that has not allocated. Which answer it gives is what
    /// the path above it needs to know: a transport that keeps the payload is handed the buffer
    /// itself where the value wrote into it, and copies only where the bytes are the value's own.
    ///
    /// # Errors
    ///
    /// Returns [`Self::Error`] when the value's own encoder rejects it. A value that already
    /// holds its bytes cannot fail, and says so with `Infallible`.
    fn wire_bytes(&self, buf: &mut BytesMut) -> Result<WireBytes<'_>, Self::Error>;
}

/// Where a [`Serialized`] value's bytes are: in the buffer it was handed, or in the value
/// itself.
///
/// It is the answer [`Serialized::wire_bytes`] gives, and the publish path needs it because the
/// two are not interchangeable underneath: a transport that keeps the payload takes the buffer
/// where the value wrote into it, and copies where the bytes belong to the value. A pointer
/// comparison would guess at the same fact; this states it.
///
/// # Examples
///
/// ```
/// use std::convert::Infallible;
///
/// use ruststream::prelude::*;
///
/// /// A frame that already holds its encoded bytes: it lends them, and the buffer it was handed
/// /// stays untouched.
/// struct Frame(Vec<u8>);
///
/// impl Serialized for Frame {
///     type Error = Infallible;
///
///     fn wire_bytes(&self, _buf: &mut BytesMut) -> Result<WireBytes<'_>, Infallible> {
///         Ok(WireBytes::Own(&self.0))
///     }
/// }
/// ```
#[derive(Debug, Clone, Copy)]
#[non_exhaustive]
pub enum WireBytes<'a> {
    /// The value wrote its bytes into the buffer it was handed.
    InBuffer,
    /// The bytes are the value's own, lent for the length of the publish.
    Own(&'a [u8]),
}

impl<'a> WireBytes<'a> {
    /// The bytes themselves, given the buffer the value was handed.
    ///
    /// What a caller that only reads them uses; the publish path answers the two arms
    /// differently, because only one of them can be taken.
    ///
    /// # Examples
    ///
    /// ```
    /// use ruststream::BytesMut;
    /// use ruststream::runtime::Serialized;
    ///
    /// /// The size a value has on the wire, for a check against the broker's payload limit before
    /// /// the publish is attempted.
    /// fn wire_len<T: Serialized>(value: &T) -> Result<usize, T::Error> {
    ///     let mut buf = BytesMut::new();
    ///     Ok(value.wire_bytes(&mut buf)?.of(&buf).len())
    /// }
    /// ```
    #[must_use]
    pub fn of(self, buf: &'a BytesMut) -> &'a [u8] {
        match self {
            Self::InBuffer => &buf[..],
            Self::Own(bytes) => bytes,
        }
    }
}

/// The encoded wire of a typed publish: the resolved codec serializes the value.
#[derive(Debug, Clone, Copy)]
pub struct EncodedWire;

/// The serialized wire of a typed publish: the value already carries its bytes
/// ([`Serialized`]), and they leave as they are, with no codec.
#[derive(Debug, Clone, Copy)]
pub struct SerializedWire;

/// The wire of a typed publish value: whether the resolved codec serializes it.
///
/// Implemented for every `serde::Serialize` value (the [`EncodedWire`]: the resolved codec
/// encodes it), and per-type for [`Serialized`] values (the [`SerializedWire`]: the bytes leave
/// as they are) - `#[derive(Serialized)]` writes that impl, or see [`Serialized`] for the
/// hand-written form. The projection is what lets one `message(&value)` entry serve both wires:
/// the value's type picks the lane, and the call site does not change.
#[diagnostic::on_unimplemented(
    message = "`{Self}` does not pick a publish wire",
    note = "a typed publish takes a `serde::Serialize` value (the resolved codec encodes it) or \
            a `#[derive(Serialized)]` type (its bytes leave as they are)"
)]
pub trait MessageWire {
    /// The value's wire: [`EncodedWire`] or [`SerializedWire`].
    type Wire;
}

// Without the opt-out, a type on neither lane is reported only through this impl's nested
// `Serialize` obligation - serde's missing-derive guidance, one lane. With it the failure names
// `MessageWire` (and, where rustc consults the trait directly, its both-lanes note); the
// method-call path still leads with serde's note, which stays the right fix for the common case.
#[diagnostic::do_not_recommend]
impl<T: Serialize> MessageWire for T {
    type Wire = EncodedWire;
}

/// What can go wrong producing a publish payload: the resolved codec rejected the value, or the
/// value's own encoder did. Machinery behind [`WirePayload`]; never named in user code.
#[doc(hidden)]
#[derive(Debug, Error)]
pub enum PayloadError {
    /// The encoded wire's codec failed.
    #[error(transparent)]
    Codec(#[from] CodecError),
    /// The serialized wire's own encoder failed.
    #[error(transparent)]
    Serialize(Box<dyn StdError + Send + Sync>),
}

/// How one wire produces the outgoing payload for a value. Machinery behind the publish
/// terminals, keyed by the value's [`MessageWire`] projection; never named in user code.
///
/// This is the point at which an encoded publish demands a codec: the encoded wire's impl needs
/// `Enc: PublishCodec`, the serialized wire's needs nothing of `Enc` at all.
#[diagnostic::on_unimplemented(
    message = "no codec is available to encode `{T}` for this publish",
    label = "this publish has no codec to encode with",
    note = "enable a codec feature on `ruststream` (`json`, `cbor` or `msgpack`), name one for \
            this publish with `.with_codec(JsonCodec)`, or give `{T}` its own bytes with \
            `#[derive(Serialized)]` so no codec is needed"
)]
#[doc(hidden)]
pub trait WirePayload<T, Enc> {
    /// Produces the bytes that leave, in the form the sink's publisher declared: the encoded
    /// wire runs the resolved codec, the serialized wire asks the value and never touches the
    /// codec. `buf` is the publish path's scratch, which a lending publisher's payload is
    /// written into and lent from.
    ///
    /// # Errors
    ///
    /// Returns [`PayloadError`] when the encoded wire's codec rejects the value, or the
    /// serialized wire's own encoder does.
    fn payload<'v, Form: PayloadForm>(
        value: &'v T,
        codec: &Enc,
        buf: &'v mut BytesMut,
    ) -> Result<Form::Form<'v>, PayloadError>;
}

impl<T: Serialize, Enc: PublishCodec> WirePayload<T, Enc> for EncodedWire {
    fn payload<'v, Form: PayloadForm>(
        value: &'v T,
        codec: &Enc,
        buf: &'v mut BytesMut,
    ) -> Result<Form::Form<'v>, PayloadError> {
        Ok(Form::encoded(codec.codec(), value, Form::encode_slot(buf))?)
    }
}

impl<T: Serialized, Enc> WirePayload<T, Enc> for SerializedWire {
    fn payload<'v, Form: PayloadForm>(
        value: &'v T,
        _codec: &Enc,
        buf: &'v mut BytesMut,
    ) -> Result<Form::Form<'v>, PayloadError> {
        Form::serialized(value, Form::encode_slot(buf))
            .map_err(|err| PayloadError::Serialize(Box::new(err)))
    }
}

/// The error of the publish builder: encoding the payload, serializing the typed headers, or the
/// sink itself.
#[derive(Debug, Error)]
#[non_exhaustive]
pub enum PublishError<E> {
    /// The codec failed to encode the message payload.
    #[error("encoding the message failed: {0}")]
    Encode(#[source] CodecError),
    /// The value's own encoder failed to produce its bytes.
    #[error("serializing the message's own bytes failed: {0}")]
    Serialize(#[source] Box<dyn StdError + Send + Sync>),
    /// The typed headers did not serialize into a header map.
    #[error("serializing the typed headers failed: {0}")]
    Headers(#[source] SerializeHeadersError),
    /// The sink (the broker publisher, or the transaction) rejected the message.
    #[error("publishing the message failed: {0}")]
    Publish(#[source] E),
}

impl<E> From<PayloadError> for PublishError<E> {
    fn from(err: PayloadError) -> Self {
        match err {
            PayloadError::Codec(err) => Self::Encode(err),
            PayloadError::Serialize(err) => Self::Serialize(err),
        }
    }
}

impl<Sink: PublishSink, Body, Enc, Hdrs, Dest> PublishBuilder<Sink, Body, Enc, Hdrs, Dest> {
    /// The whole-tuple constructor the entry points and the position transitions share.
    pub(crate) const fn new(
        sink: Sink,
        body: Body,
        codec: Enc,
        headers: Hdrs,
        dest: Dest,
        options: Option<Sink::Options>,
    ) -> Self {
        Self {
            sink,
            body,
            codec,
            headers,
            dest,
            options,
        }
    }

    /// The broker's per-message settings for this publish, still unset (`None`) until something
    /// fills them in.
    ///
    /// The one seam a broker extends the builder through. A broker ships an extension trait
    /// bounded on its own options type, so its steps appear on a builder over its own publisher
    /// and nowhere else; each step reaches in here, sets one field, and returns the builder. What
    /// no step touches stays unset, and the publish policy's own settings apply to it.
    ///
    /// # Examples
    ///
    /// ```
    /// use ruststream::runtime::{PublishBuilder, PublishSink};
    ///
    /// // The broker's settings: every field optional, so a call says only what it changes.
    /// #[derive(Clone, Default)]
    /// struct MqttOptions {
    ///     retain: Option<bool>,
    /// }
    ///
    /// // The broker's own step, in its prelude. The bound is what keeps it off every other
    /// // broker's builder.
    /// trait MqttSteps {
    ///     #[must_use]
    ///     fn retain(self, retain: bool) -> Self;
    /// }
    ///
    /// impl<Sink, Body, Enc, Hdrs, Dest> MqttSteps for PublishBuilder<Sink, Body, Enc, Hdrs, Dest>
    /// where
    ///     Sink: PublishSink<Options = MqttOptions>,
    /// {
    ///     fn retain(mut self, retain: bool) -> Self {
    ///         self.options_mut().get_or_insert_with(MqttOptions::default).retain = Some(retain);
    ///         self
    ///     }
    /// }
    /// ```
    pub const fn options_mut(&mut self) -> &mut Option<Sink::Options> {
        &mut self.options
    }
}

/// Starts a typed publish: `value` is encoded with `codec`, and the destination position is the
/// one the type declares.
pub(crate) fn message_of<Sink, T, Enc>(
    sink: Sink,
    value: &T,
    codec: Enc,
) -> PublishBuilder<Sink, MessageBody<'_, T>, Enc, HeadersUnset, T::Form>
where
    Sink: PublishSink,
    T: OutgoingDestination,
{
    PublishBuilder::new(
        sink,
        MessageBody(value),
        codec,
        HeadersUnset,
        <T::Form as Default>::default(),
        None,
    )
}

/// Starts a byte publish: the payload travels as it is, and the destination is named at the call.
///
/// Every runtime path carries a typed value or its own erased publisher, so this entry exists for
/// the suite that drives the builder's own steps over bytes.
#[cfg(test)]
pub(crate) fn raw_of<Sink, B>(
    sink: Sink,
    payload: &B,
) -> PublishBuilder<Sink, RawBody<'_>, (), HeadersUnset, CallerName>
where
    Sink: PublishSink,
    B: AsRef<[u8]> + ?Sized,
{
    PublishBuilder::new(
        sink,
        RawBody(payload.as_ref()),
        (),
        HeadersUnset,
        CallerName,
        None,
    )
}

impl<'a, Sink: PublishSink, T, Enc, Hdrs, Dest>
    PublishBuilder<Sink, MessageBody<'a, T>, Enc, Hdrs, Dest>
{
    /// Names the codec for this publish, the most specific level of the codec ladder: it wins
    /// over the surface's codec, which in turn won over the application's and the crate default.
    ///
    /// Only the encoded wire has this position: bytes are sent as they are, and a
    /// [`Serialized`] value's bytes leave as they are too - a codec named for one would be
    /// silently ignored, so naming it does not compile.
    // The wire bound rejects the misuse where it is written instead of letting the named codec
    // become a no-op on the serialized wire.
    pub fn with_codec<C: Codec>(
        self,
        codec: C,
    ) -> PublishBuilder<Sink, MessageBody<'a, T>, CallCodec<C>, Hdrs, Dest>
    where
        T: MessageWire<Wire = EncodedWire>,
    {
        PublishBuilder::new(
            self.sink,
            self.body,
            CallCodec(codec),
            self.headers,
            self.dest,
            self.options,
        )
    }
}

/// What may fill the headers position of a publish: a borrowed contract value, or an
/// already-built [`HeaderMap`].
///
/// The two differ in what they say about the message, not in how they are written: a borrowed
/// value is the message's declared contract, serialized one entry per field, while a map is the
/// transport level and stands for no contract at all (a message type declaring one still demands
/// its value). Which of the two a call site passed rides in the type, so
/// [`with_headers`](PublishBuilder::with_headers) is one method and the contract check stays a compile
/// error either way.
///
/// You never implement it: [`HeaderMap`] and `&H` of every serializable `H` already do.
#[diagnostic::on_unimplemented(
    message = "`{Self}` cannot fill the headers of a publish",
    note = "pass the message's declared contract by reference (`.with_headers(&meta)`), or an \
            already-built header map by value (`.with_headers(headers)`)"
)]
pub trait HeaderSource {
    /// The headers position this source resolves to: [`TypedHeaders`] or [`MapHeaders`].
    type Position;

    /// Moves the source into its position.
    fn into_position(self) -> Self::Position;
}

// A borrowed value is the contract form: it is serialized into the map when the publish runs, so
// the `Serialize` bound stays on `PublishHeaders` - an unserializable value must fail at the
// publish, with serde's own guidance, not at "no such method".
impl<'a, H: ?Sized> HeaderSource for &'a H {
    type Position = TypedHeaders<'a, H>;

    fn into_position(self) -> Self::Position {
        TypedHeaders(self)
    }
}

impl HeaderSource for HeaderMap {
    type Position = MapHeaders;

    fn into_position(self) -> Self::Position {
        MapHeaders(self)
    }
}

impl<Sink: PublishSink, Body, Enc, Dest> PublishBuilder<Sink, Body, Enc, HeadersUnset, Dest> {
    /// Supplies the headers of this publish: the message's declared contract by reference, or an
    /// already-built [`HeaderMap`] by value.
    ///
    /// A contract value is checked against the message's declaration, so a missing, extra or
    /// mismatched one is a compile error; a map carries whatever the caller put in it and stands
    /// for no contract, so a message type declaring one is not satisfied by a map. On the byte
    /// entry point either form is accepted: bytes declare no contract.
    ///
    /// The position is filled once. A borrow is cheap - nothing is serialized until the publish
    /// runs.
    ///
    /// # Examples
    ///
    /// A type that declares a header contract publishes only with it:
    ///
    /// ```
    /// # #[cfg(all(feature = "macros", feature = "memory", feature = "json"))]
    /// # mod demo {
    /// use ruststream::memory::prelude::*;
    /// use serde::{Deserialize, Serialize};
    ///
    /// #[derive(Serialize, Deserialize)]
    /// struct ChunkMeta {
    ///     task_id: u64,
    ///     chunk_no: u32,
    /// }
    ///
    /// #[derive(Serialize, Outgoing)]
    /// #[outgoing(name = "chunks.done", headers = ChunkMeta)]
    /// struct ChunkDone {
    ///     output_key: String,
    /// }
    ///
    /// #[derive(OutSlot)]
    /// #[publishes(ChunkDone)]
    /// struct Events;
    ///
    /// #[derive(Deserialized)]
    /// struct Chunk<'a>(&'a [u8]);
    ///
    /// #[subscriber("chunks.raw")]
    /// async fn convert(
    ///     chunk: &Chunk<'_>,
    ///     Headers(meta): Headers<ChunkMeta>,
    ///     Out(events): Out<impl Publisher, Events>,
    /// ) -> HandlerOutcome {
    ///     let done = ChunkDone {
    ///         output_key: format!("chunks/{}/{}.part", meta.task_id, meta.chunk_no),
    ///     };
    ///     if events.message(&done).with_headers(&meta).publish().await.is_err() {
    ///         return HandlerOutcome::retry();
    ///     }
    ///     let _ = chunk.0.len();
    ///     HandlerOutcome::ack()
    /// }
    /// # }
    /// # fn main() {}
    /// ```
    pub fn with_headers<S: HeaderSource>(
        self,
        headers: S,
    ) -> PublishBuilder<Sink, Body, Enc, S::Position, Dest> {
        PublishBuilder::new(
            self.sink,
            self.body,
            self.codec,
            headers.into_position(),
            self.dest,
            self.options,
        )
    }
}

impl<Sink: PublishSink, Body, Enc, Hdrs> PublishBuilder<Sink, Body, Enc, Hdrs, CallerName> {
    /// Names the destination of this publish.
    ///
    /// Present when the message type declares no name of its own, and on the byte entry point.
    /// Pass a `&str` (the borrowed, no-allocation case) or a `String` (a computed one).
    pub fn to<'n>(
        self,
        name: impl Into<Cow<'n, str>>,
    ) -> PublishBuilder<Sink, Body, Enc, Hdrs, SuppliedName<'n>> {
        PublishBuilder::new(
            self.sink,
            self.body,
            self.codec,
            self.headers,
            SuppliedName(name.into()),
            self.options,
        )
    }
}

impl<Sink, T, Enc, Hdrs> PublishBuilder<Sink, MessageBody<'_, T>, Enc, Hdrs, NameTemplate>
where
    Sink: PublishSink,
    T: TemplateAddress<Self>,
{
    /// Opens the placeholder setters of a templated destination: one per `{placeholder}` of the
    /// type's declared name, in any order, with the publish terminal appearing once the last one
    /// is bound.
    ///
    /// An unbound placeholder rides in the returned builder's type, so the compile error names
    /// the segment that is missing.
    pub fn to(self) -> T::Builder {
        T::begin(self)
    }
}

/// A destination template's binding builder: what `to()` opens for a type whose declared name
/// carries placeholders.
///
/// Implemented by `#[derive(Outgoing)]` on a type with a templated `name`; the associated
/// [`Builder`](Self::Builder) is the generated value carrying one setter per placeholder. You
/// never implement it by hand.
pub trait TemplateAddress<Cont> {
    /// The generated builder, with every placeholder still unbound.
    type Builder;

    /// Opens the builder over the publish it will finish.
    fn begin(cont: Cont) -> Self::Builder;
}

/// The unbound state of one address placeholder.
///
/// The segment marker rides in the type so the compile error of an unfinished address names the
/// placeholder that was forgotten, the way an unbound [`Out`](crate::runtime::Out) slot names
/// itself. `#[derive(Outgoing)]` generates one marker per placeholder of the declared name.
#[derive(Debug, Clone, Copy, Default)]
pub struct MissingSegment<S>(PhantomData<fn() -> S>);

impl<S> MissingSegment<S> {
    /// The unbound placeholder, as the generated builder starts it.
    #[must_use]
    pub const fn new() -> Self {
        Self(PhantomData)
    }
}

/// One address placeholder that carries a value, and so can be rendered into the destination.
///
/// The publish terminal of a generated address builder takes this witness once per placeholder,
/// so an address with a placeholder still in its [`MissingSegment`] state is rejected by a bound
/// the compiler can explain rather than by the terminal not existing. Implemented for the bound
/// representation only; you never implement it by hand.
#[diagnostic::on_unimplemented(
    message = "this address still has an unbound placeholder",
    note = "a name template opens one setter per placeholder, and the publish appears once \
            every one of them is bound"
)]
pub trait BoundSegment: fmt::Display {}

impl BoundSegment for String {}

/// A publish whose destination is rendered by a generated address builder.
///
/// The terminal a templated `to()` chain calls once every placeholder is bound; the rendered
/// address is the only thing the generated code adds to the publish the builder already carries.
pub trait PublishAt {
    /// The sink's error, as reported through [`PublishError`].
    type Error;

    /// Runs the publish against the rendered address.
    ///
    /// # Errors
    ///
    /// Returns [`PublishError`] when the payload cannot be encoded, the typed headers cannot be
    /// serialized, or the sink rejects the message.
    fn publish_at(
        self,
        name: &str,
    ) -> impl Future<Output = Result<(), PublishError<Self::Error>>> + Send;
}

/// Sends one message, applying the headers position, the broker's settings and the sink.
async fn deliver<'v, Sink: PublishSink, Hdrs: PublishHeaders>(
    mut sink: Sink,
    name: &'v str,
    payload: <Sink::Payload as PayloadForm>::Form<'v>,
    headers: Hdrs,
    options: Option<&Sink::Options>,
) -> Result<(), PublishError<Sink::Error>> {
    // The one place the outgoing map is built: the sink's base first, the publish's own headers
    // over it. A sink without a base yields the empty map this always started from, so nothing is
    // cloned and nothing is allocated on the path every publisher takes today.
    let mut map = sink.base_headers().cloned().unwrap_or_default();
    headers.write(&mut map).map_err(PublishError::Headers)?;
    let msg = OutgoingMessage::assembled(name, payload, map);
    sink.send(msg, options).await.map_err(PublishError::Publish)
}

/// Produces the value's wire payload and sends it, shared by the resolved and the rendered
/// terminals. The wire is the value's own ([`MessageWire`]): a `Serialize` value encodes with
/// the resolved codec, a [`Serialized`] one leaves byte-for-byte and the codec is never called.
async fn wire_and_send<Sink, T, Enc, Hdrs>(
    sink: Sink,
    codec: Enc,
    value: &T,
    headers: Hdrs,
    name: &str,
    options: Option<&Sink::Options>,
) -> Result<(), PublishError<Sink::Error>>
where
    Sink: PublishSink,
    T: MessageWire + Sync,
    T::Wire: WirePayload<T, Enc>,
    Hdrs: PublishHeaders,
{
    // The buffer a publish to a lending transport writes into, and the one a serialized value
    // writes into whichever form it leaves in. A handle used outside a dispatch loop has no
    // scratch to reuse, so this is its own; `BytesMut::new` does not allocate, so a value that
    // lends bytes it already holds pays nothing for it, and a taking transport is handed the
    // codec's own buffer and never touches this one.
    let mut buf = BytesMut::new();
    let payload =
        <T::Wire as WirePayload<T, Enc>>::payload::<Sink::Payload>(value, &codec, &mut buf)?;
    deliver(sink, name, payload, headers, options).await
}

// No `Enc: PublishCodec` here: what a publish needs of its codec position is the wire's business,
// and the serialized wire needs nothing. The encoded wire carries the bound through
// `WirePayload`, so a value that must be encoded still demands a resolved codec - and a value
// carrying its own bytes publishes in a build with no codec at all.
impl<Sink, T, Enc, Hdrs, Dest> PublishBuilder<Sink, MessageBody<'_, T>, Enc, Hdrs, Dest>
where
    Sink: PublishSink,
    T: OutgoingDestination + MessageHeaders + Sync,
    Hdrs: PublishHeaders,
{
    /// Publishes the value to the resolved destination, on the value's own wire
    /// ([`MessageWire`]): a `Serialize` value encodes with the resolved codec, a
    /// [`Serialized`] one leaves byte-for-byte.
    ///
    /// # Errors
    ///
    /// Returns [`PublishError`] when the codec rejects an encoded value, the typed headers do
    /// not serialize, or the sink rejects the message.
    ///
    /// # Cancel safety
    ///
    /// As cancel-safe as the sink's own send: dropping the future mid-flight may leave the
    /// message in an indeterminate state.
    // The position bounds sit on the method, not on the impl block: on the block an
    // under-specified publish reads as "method not found" and loses the guidance the bounds
    // carry, the wire projection's both-lanes note included.
    pub async fn publish(self) -> Result<(), PublishError<Sink::Error>>
    where
        T: MessageWire,
        T::Wire: WirePayload<T, Enc>,
        Hdrs: SatisfiesContract<T::Contract>,
        Dest: ResolvedName,
    {
        let Self {
            sink,
            body,
            codec,
            headers,
            dest,
            options,
        } = self;
        // The fixed form's name is borrowed straight from the declaration; the supplied form
        // ignores it and lends its own.
        let name = dest.resolve(T::DESTINATION);
        wire_and_send(sink, codec, body.0, headers, name, options.as_ref()).await
    }
}

impl<Sink, T, Enc, Hdrs> PublishAt
    for PublishBuilder<Sink, MessageBody<'_, T>, Enc, Hdrs, NameTemplate>
where
    Sink: PublishSink,
    T: OutgoingDestination + MessageHeaders + MessageWire + Sync,
    T::Wire: WirePayload<T, Enc>,
    Enc: Send,
    Hdrs: PublishHeaders + SatisfiesContract<T::Contract> + Send,
{
    type Error = Sink::Error;

    async fn publish_at(self, name: &str) -> Result<(), PublishError<Sink::Error>> {
        let Self {
            sink,
            body,
            codec,
            headers,
            options,
            ..
        } = self;
        wire_and_send(sink, codec, body.0, headers, name, options.as_ref()).await
    }
}

impl<Sink, Hdrs, Dest> PublishBuilder<Sink, RawBody<'_>, (), Hdrs, Dest>
where
    Sink: PublishSink,
    Hdrs: PublishHeaders,
{
    /// Publishes the bytes as they are, to the destination named at the call site.
    ///
    /// # Errors
    ///
    /// Returns [`PublishError`] when the typed headers do not serialize, or the sink rejects the
    /// message.
    ///
    /// # Cancel safety
    ///
    /// As cancel-safe as the sink's own send: dropping the future mid-flight may leave the
    /// message in an indeterminate state.
    // On the method for the same reason as on the typed terminal: a bytes publish that never
    // named its destination has to read as the missing `to(..)`, not as a missing method.
    pub async fn publish(self) -> Result<(), PublishError<Sink::Error>>
    where
        Dest: ResolvedName,
    {
        let Self {
            sink,
            body,
            headers,
            dest,
            options,
            ..
        } = self;
        // Bytes declare no name of their own, so the destination is always the supplied one.
        let name = dest.resolve("");
        // The caller owns these bytes and keeps them for the length of the call: a transport
        // that reads them is lent them, and only one that keeps them is handed a buffer.
        let payload = <Sink::Payload as PayloadForm>::Form::from(body.0);
        deliver(sink, name, payload, headers, options.as_ref()).await
    }
}