ruststream 0.6.1

//! Publishing subscribers: a handler whose return value is encoded and published.
//!
//! Generated by `#[subscriber("name", publish("reply-name"))]` and mounted with
//! [`BrokerScope::include`](super::BrokerScope::include), chaining
//! [`.publisher(..)`](super::IncludePublishing::publisher) to attach the reply publish policy
//! (without it, the statement commits with the broker's default policy); a
//! [`Router`](super::Router) keeps the positional
//! [`include_publishing`](super::Router::include_publishing). At startup the policy pairs into a
//! [`TypedPublisher`] (the live connection + reply codec). The destination name comes from the
//! macro; the publisher and codec come from wiring.

use std::future::Future;

use serde::Serialize;
use tracing::warn;

use crate::codec::Codec;
use crate::{IncomingMessage, OutgoingMessage, Publisher};

use super::context::Context;
use super::dispatch::Workers;
use super::failure::{FailurePolicies, FailurePolicy};
use super::handler::{Handler, HandlerResult, Settle};
use super::input::{DecodeWith, InputKind};
use super::metadata::HandlerMetadata;
use super::publish::{
    PublishContext, PublishIdentity, PublishPipeline, PublishTransform, TypedPublisher,
};

/// The reply-wiring axis: how a handler's reply value leaves the service.
///
/// Selected by the type the include site attaches: a [`TypedPublisher`] stack encodes the
/// reply with its codec and runs its transforms and the scope's publish pipeline, while a bare
/// [`Publisher`] sends the reply's bytes as-is (no codec, no transforms, no pipeline). One
/// [`PublishingHandler`] serves both, so the encoded and the byte reply forms differ only in
/// what pairs at the include site.
pub trait ReplySink<Reply, DeliveryCx, Pipeline>: Send + Sync {
    /// The error surfaced when the reply cannot be published.
    type Error: std::fmt::Display;

    /// Publishes `reply` to `name`.
    fn deliver(
        &self,
        name: &str,
        reply: &Reply,
        pipeline: &Pipeline,
        cx: &PublishContext<'_, DeliveryCx>,
    ) -> impl Future<Output = Result<(), Self::Error>> + Send;
}

/// The encoded wiring: the reply serializes through the stack's reply codec, then travels the
/// stack's transforms and the scope's publish pipeline.
impl<Reply, DeliveryCx, Pipeline, Leaf, ReplyCodec, Transforms>
    ReplySink<Reply, DeliveryCx, Pipeline> for TypedPublisher<Leaf, ReplyCodec, Transforms>
where
    Reply: Serialize + Sync,
    DeliveryCx: Sync,
    Pipeline: PublishPipeline,
    Leaf: Publisher,
    ReplyCodec: Codec,
    Transforms: PublishTransform<DeliveryCx>,
{
    type Error = Box<dyn std::error::Error + Send + Sync>;

    async fn deliver(
        &self,
        name: &str,
        reply: &Reply,
        pipeline: &Pipeline,
        cx: &PublishContext<'_, DeliveryCx>,
    ) -> Result<(), Self::Error> {
        self.publish(name, reply, pipeline, cx).await
    }
}

/// The byte wiring: a bare [`Publisher`] sends an `AsRef<[u8]>` reply unencoded.
impl<Reply, DeliveryCx, Pipeline, Bare> ReplySink<Reply, DeliveryCx, Pipeline> for Bare
where
    Reply: AsRef<[u8]> + Sync,
    DeliveryCx: Sync,
    Pipeline: Send + Sync,
    Bare: Publisher,
{
    type Error = Bare::Error;

    async fn deliver(
        &self,
        name: &str,
        reply: &Reply,
        _pipeline: &Pipeline,
        _cx: &PublishContext<'_, DeliveryCx>,
    ) -> Result<(), Self::Error> {
        self.publish(OutgoingMessage::new(name, reply.as_ref()))
            .await
    }
}

/// A subscriber definition that produces a reply to publish.
///
/// The generated type carries the subscribe name, the reply name, and the reply type. *How* the
/// reply is encoded (codec) and *through which* connection it is sent come from the
/// [`TypedPublisher`] passed at wiring time.
pub trait PublishingDef: Send + Sync {
    /// The input kind the handler consumes ([`Decoded<T>`](super::Decoded) for a typed `&T`
    /// parameter, [`RawBytes`](super::RawBytes) for a raw `&[u8]` one).
    type Input: InputKind;

    /// The tuple of startup-injected parameters ([`Out`](super::Out), [`Seek`](super::Seek),
    /// ...; `()` when the signature carries none), resolved like
    /// [`InjectDef::Injections`](super::InjectDef::Injections).
    type Injections;

    /// The reply type the handler produces, encoded and published.
    type Reply;

    /// The broker's typed per-delivery context the handler reads by key, mirroring
    /// [`SubscriberDef::Context`](super::SubscriberDef::Context) (`()` when the handler names
    /// none). It is threaded onto the reply's static
    /// [`PublishTransform`](super::PublishTransform) so a publish transform can stamp the delivery's trace
    /// or correlation id on the reply.
    type Context;

    /// The subscription source this handler binds to (see
    /// [`SubscriberDef::Source`](super::SubscriberDef::Source)).
    type Source;

    /// Builds the subscription source (fresh each call).
    fn source(&self) -> Self::Source;

    /// The name (subject / channel) the reply is published to.
    fn reply_name(&self) -> &str;

    /// The concurrency policy for this subscriber's dispatch loop. The macro fills this in from
    /// the `workers(..)` argument; the default is sequential dispatch.
    fn workers(&self) -> Workers {
        Workers::sequential()
    }

    /// The failure policy for a handler panic and a decode failure. The macro fills this in from
    /// the `on_failure(panic = .., decode = ..)` argument; the default fails fast on a panic and
    /// drops on a decode failure.
    fn failure_policies(&self) -> FailurePolicies {
        FailurePolicies::default()
    }

    /// An optional human description (from the handler's doc comment), for `AsyncAPI`.
    fn description(&self) -> Option<&str> {
        None
    }

    /// The input type's serialized JSON Schema, when it implements [`schemars::JsonSchema`] and the
    /// `asyncapi` feature is on. The macro fills this in; the default omits it.
    fn input_schema(&self) -> Option<String> {
        None
    }

    /// The input type's [`Message`](crate::Message) name, when it implements that trait. The macro
    /// fills this in; the default omits it.
    fn message_name(&self) -> Option<&'static str> {
        None
    }

    /// The input type's [`Message`](crate::Message) description, when it implements that trait.
    /// The macro fills this in; the default omits it.
    fn message_description(&self) -> Option<&'static str> {
        None
    }
}

/// Runs a [`PublishingDef`]'s handler body over an app state of type `S`.
///
/// Split from [`PublishingDef`] so a handler that ignores the app state is generic over `S` (mounts
/// on any app), while one that reads it via [`Context::state`](super::Context::state) implements
/// this only for its declared `S` - the same shape as [`Handler<M, C, S>`](super::Handler), so the
/// state match is checked at compile time without pinning a single `State` on the def.
pub trait PublishingCall<S>: PublishingDef {
    /// Runs the handler body.
    ///
    /// `Ok(reply)` is encoded and published to [`reply_name`](PublishingDef::reply_name), then the
    /// incoming message is acked. `Err(result)` skips publishing and the dispatcher acts on the
    /// returned [`HandlerResult`] (for example [`HandlerResult::retry`] to ask for redelivery).
    fn call(
        &self,
        input: &<Self::Input as InputKind>::Target,
        injections: &Self::Injections,
        ctx: &mut Context<'_, Self::Context, S>,
    ) -> impl Future<Output = Result<Self::Reply, HandlerResult>> + Send;
}

/// Builds the registration metadata for a publishing definition mounted under `name`.
pub(crate) fn publishing_metadata<D: PublishingDef>(name: String, def: &D) -> HandlerMetadata {
    let mut meta = HandlerMetadata::raw(name)
        .with_output_type(std::any::type_name::<D::Reply>())
        .with_def_details(
            def.description(),
            def.input_schema(),
            def.message_name(),
            def.message_description(),
        );
    meta.input_type = <D::Input as InputKind>::input_label();
    meta
}

/// The [`Handler`] built from a [`PublishingDef`]: decode, run, deliver the reply, ack.
///
/// `DecodeCodec` decodes the incoming message; the reply leaves through the [`ReplySink`]
/// `Wiring` (encoded by a [`TypedPublisher`] stack, or byte-for-byte through a bare publisher) to the
/// definition's [`reply_name`](PublishingDef::reply_name). A handler returning `Err(result)`
/// skips the publish; a failed reply publish nacks the incoming message with `requeue = true`,
/// so the broker redelivers it instead of silently losing the reply.
pub struct PublishingHandler<Def: PublishingDef, DecodeCodec, Wiring, Pipeline = PublishIdentity> {
    pub(crate) def: Def,
    pub(crate) codec: DecodeCodec,
    pub(crate) publisher: Wiring,
    pub(crate) pipeline: Pipeline,
    pub(crate) injections: Def::Injections,
    pub(crate) decode: FailurePolicy,
}

impl<Def: PublishingDef, DecodeCodec, Wiring, Pipeline> std::fmt::Debug
    for PublishingHandler<Def, DecodeCodec, Wiring, Pipeline>
{
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("PublishingHandler").finish_non_exhaustive()
    }
}

impl<Msg, Def, DecodeCodec, Wiring, Pipeline, State> Handler<Msg, Def::Context, State>
    for PublishingHandler<Def, DecodeCodec, Wiring, Pipeline>
where
    Msg: IncomingMessage,
    Def: PublishingCall<State>,
    Def::Input: DecodeWith<DecodeCodec>,
    Def::Injections: Send + Sync,
    Def::Reply: Send + Sync,
    Def::Context: Send + Sync,
    DecodeCodec: Codec,
    Wiring: ReplySink<Def::Reply, Def::Context, Pipeline>,
    Pipeline: Send + Sync,
    State: Send + Sync,
{
    async fn handle(&self, msg: &Msg, ctx: &mut Context<'_, Def::Context, State>) -> Settle {
        // The publishing path settles by a bare outcome (no per-element continuation): decode,
        // run, publish the reply, then ack. It converts to `Settle` with no `and_after`. The
        // decode product lives on this stack frame and the handler borrows its view.
        let owned =
            match <Def::Input as DecodeWith<DecodeCodec>>::decode(&self.codec, msg.payload()) {
                Ok(value) => value,
                Err(err) => {
                    warn!(
                        target: "ruststream::dispatch",
                        subscription = %ctx.name(),
                        message_type = <Def::Input as InputKind>::input_label(),
                        error = %err,
                        "codec decode failed",
                    );
                    #[cfg(any(feature = "testing", feature = "otel"))]
                    ctx.mark_decode_failed();
                    return match self.decode {
                        FailurePolicy::FailFast => {
                            ctx.fail_fast(&format!("decode failed: {err}"));
                            HandlerResult::drop().into()
                        }
                        other => other
                            .settlement()
                            .unwrap_or_else(HandlerResult::drop)
                            .into(),
                    };
                }
            };
        let view = <Def::Input as InputKind>::view(&owned, msg.payload());
        let reply = match self.def.call(view, &self.injections, ctx).await {
            Ok(reply) => reply,
            Err(result) => return result.into(),
        };
        let name = self.def.reply_name();
        let pubcx = PublishContext::new(ctx.name(), ctx.headers(), ctx.cx_ref());
        let publish = self.publisher.deliver(name, &reply, &self.pipeline, &pubcx);
        if let Err(err) = publish.await {
            warn!(
                target: "ruststream::dispatch",
                subscription = %ctx.name(),
                reply = %name,
                reply_type = std::any::type_name::<Def::Reply>(),
                error = %err,
                "reply publish failed",
            );
            return HandlerResult::retry().into();
        }
        HandlerResult::Ack.into()
    }
}

#[cfg(test)]
mod tests {
    use super::{PublishingCall, PublishingDef, publishing_metadata};
    use crate::Headers;
    use crate::Name;
    use crate::runtime::context::Context;
    use crate::runtime::dispatch::{Delivery, Workers};
    use crate::runtime::handler::HandlerResult;

    /// A hand-written publishing def overriding nothing optional, pinning the trait defaults that
    /// the macro always fills in.
    struct ManualPub;

    impl PublishingDef for ManualPub {
        type Input = crate::runtime::Decoded<u32>;
        type Injections = ();
        type Reply = u32;
        type Context = ();
        type Source = Name;

        fn source(&self) -> Name {
            Name::new("in")
        }

        // The trait signature returns `&str` (tied to `&self`); the macro-generated impls do the
        // same, so this hand-written one cannot narrow to `&'static str` without diverging.
        #[allow(clippy::unnecessary_literal_bound)]
        fn reply_name(&self) -> &str {
            "out"
        }
    }

    // Ignores the app state, so it is generic over it (mounts on any app).
    impl<S: Send + Sync> PublishingCall<S> for ManualPub {
        async fn call(
            &self,
            input: &u32,
            (): &(),
            _ctx: &mut Context<'_, (), S>,
        ) -> Result<u32, HandlerResult> {
            Ok(*input)
        }
    }

    #[tokio::test]
    async fn defaults_metadata_and_call() {
        let def = ManualPub;
        assert_eq!(def.workers(), Workers::sequential());
        assert!(def.description().is_none());
        assert!(def.input_schema().is_none());
        assert!(def.message_name().is_none());
        assert!(def.message_description().is_none());
        assert_eq!(def.reply_name(), "out");
        let _source = def.source();

        let meta = publishing_metadata("in".to_owned(), &def);
        assert_eq!(meta.name, "in");

        let state = ();
        let delivery = Delivery::empty();
        let headers = Headers::new();
        let mut ctx = Context::new("in", &headers, &state, (), &delivery);
        assert_eq!(def.call(&5, &(), &mut ctx).await.unwrap(), 5);
    }
}