ruststream 0.7.0-rc.12

//! The plain cells of the matrix: a sealed definition with no reply and no injections mounts
//! through the ordinary subscriber and batch machinery, with the body adapted to the dispatch
//! contracts here.

use std::marker::PhantomData;

use tracing::warn;

use crate::runtime::batch::{BatchDef, BatchResult, SliceHandler};
use crate::runtime::context::Context;
use crate::runtime::failure::FailurePolicy;
use crate::runtime::handler::{Handler, HandlerOutcome};
use crate::runtime::router::IncludeDef;
use crate::runtime::subscriber_def::SubscriberDef;
use crate::{Name, Unnamed};

use super::Handle;
use super::axis::{
    Axis, AxisDocs, Batch, BatchDeserialized, BatchPair, BatchedAxis, Deserialized, Input, Message,
    Solo, SoloAxis, SoloDeserialized, SoloPair,
};
use super::value::{HandleValue, Sealed};

/// Constructs one [`Deserialized`] input from a delivery's payload; a failed construction is
/// settled by the subscriber's decode policy, exactly as a codec decode failure is.
pub(crate) fn construct<'p, F, C, S>(
    payload: &'p [u8],
    ctx: &mut Context<'_, C, S>,
) -> Result<F::Output<'p>, HandlerOutcome>
where
    F: Deserialized,
{
    match F::from_payload(payload) {
        Ok(value) => Ok(value),
        Err(err) => {
            warn!(
                target: "ruststream::dispatch",
                subscription = %ctx.name(),
                message_type = std::any::type_name::<F>(),
                error = %err,
                "payload construction failed",
            );
            #[cfg(any(feature = "testing", feature = "otel"))]
            ctx.mark_decode_failed();
            Err(match ctx.decode_policy() {
                FailurePolicy::FailFast => {
                    ctx.fail_fast(&format!("payload construction failed: {err}"));
                    HandlerOutcome::drop()
                }
                other => other
                    .settlement()
                    .map_or_else(HandlerOutcome::drop, Into::into),
            })
        }
    }
}

/// The dispatch adapter of a single-delivery body: awaits the verdict and settles by it.
///
/// The state axis is not part of the type: the [`Handler`] impls quantify over it, so a body
/// whose `Handle` impl is generic over the state mounts on an app with any state, and one
/// naming a concrete state mounts only there.
pub struct SoloBody<A, C, H> {
    body: H,
    _axes: SoloAxes<A, C>,
}

/// The phantom carrying a solo adapter's axes.
type SoloAxes<A, C> = PhantomData<fn() -> (A, C)>;

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

/// Settles by one solo verdict: `Ok` acks, `Err` is the outcome.
pub(super) fn settle_solo(verdict: Result<(), HandlerOutcome>) -> HandlerOutcome {
    match verdict {
        Ok(()) => HandlerOutcome::ack(),
        Err(outcome) => outcome,
    }
}

impl<T, C, S, H> Handler<T, C, S> for SoloBody<Solo<T>, C, H>
where
    T: Input<Axis = Solo<T>> + Send + Sync + 'static,
    C: Send + Sync,
    S: Send + Sync,
    H: Handle<T, (), (), C, S>,
{
    async fn handle(&self, msg: &T, ctx: &mut Context<'_, C, S>) -> HandlerOutcome {
        settle_solo(self.body.handle(msg, &(), ctx).await)
    }
}

impl<F, C, S, H> Handler<[u8], C, S> for SoloBody<SoloDeserialized<F>, C, H>
where
    F: Deserialized + Send + Sync + 'static,
    // Pinning the axis is what normalizes the generic output's verdict family.
    for<'p> F::Output<'p>: Input<Axis = SoloDeserialized<F>>,
    C: Send + Sync,
    S: Send + Sync,
    H: for<'p> Handle<F::Output<'p>, (), (), C, S>,
{
    async fn handle(&self, msg: &[u8], ctx: &mut Context<'_, C, S>) -> HandlerOutcome {
        let input = match construct::<F, C, S>(msg, ctx) {
            Ok(input) => input,
            Err(outcome) => return outcome,
        };
        settle_solo(self.body.handle(&input, &(), ctx).await)
    }
}

impl<Hd, P, C, S, H> Handler<Message<Hd, P>, C, S> for SoloBody<SoloPair<Hd, P>, C, H>
where
    Message<Hd, P>: Input<Axis = SoloPair<Hd, P>>,
    Hd: Send + Sync + 'static,
    P: Send + Sync + 'static,
    C: Send + Sync,
    S: Send + Sync,
    H: Handle<Message<Hd, P>, (), (), C, S>,
{
    async fn handle(&self, msg: &Message<Hd, P>, ctx: &mut Context<'_, C, S>) -> HandlerOutcome {
        settle_solo(self.body.handle(msg, &(), ctx).await)
    }
}

impl<A, R, C, H, Doc> IncludeDef for Sealed<HandleValue<A, R, (), C, H, Doc>>
where
    A: Axis,
{
    type Form = A::EagerForm;
}

impl<A, C, H, Doc> SubscriberDef for Sealed<HandleValue<A, (), (), C, H, Doc>>
where
    A: SoloAxis,
    Doc: AxisDocs<A>,
{
    type Input = A::Kind;
    type Context = C;
    type Handler = SoloBody<A, C, H>;
    // The sealed value never builds a source: the settings builder wrapping it carries the real
    // one, and this placeholder is no `SubscriptionSource` at all, so a bare mount is a compile
    // error.
    type Source = Unnamed<Name>;

    fn source(&self) -> Unnamed<Name> {
        Unnamed::new()
    }

    fn description(&self) -> Option<&str> {
        self.0.docs.description()
    }

    fn input_schema(&self) -> Option<String> {
        self.0
            .docs
            .input_schema
            .clone()
            .or_else(Doc::payload_schema)
    }

    fn headers_schema(&self) -> Option<String> {
        self.0
            .docs
            .headers_schema
            .clone()
            .or_else(Doc::headers_schema)
    }

    fn message_name(&self) -> Option<&'static str> {
        self.0.docs.message_name
    }

    fn message_description(&self) -> Option<&'static str> {
        self.0.docs.message_description
    }

    fn into_handler(self) -> SoloBody<A, C, H> {
        SoloBody {
            body: self.0.body,
            _axes: PhantomData,
        }
    }
}

/// The dispatch adapter of a batch body.
///
/// Awaits the verdict, checks the per-element contract, and settles the batch by it. The
/// batch's size was named by [`batch`](crate::runtime::SubscriberSettings::batch) and applied by
/// the broker, so nothing here resizes it.
pub struct BatchBody<A, H> {
    body: H,
    _axes: PhantomData<fn() -> A>,
}

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

/// Applies the batch contract to one verdict: `Ok` acks the batch, an `Err` vector must be
/// exactly batch-length (a mismatch is a bug in the handler and panics under the subscriber's
/// panic policy).
pub(super) fn settle_batch(
    verdict: Result<(), Vec<HandlerOutcome>>,
    batch_len: usize,
    subscription: &str,
) -> BatchResult {
    match verdict {
        Ok(()) => BatchResult::Uniform(HandlerOutcome::ack()),
        Err(outcomes) => {
            assert!(
                outcomes.len() == batch_len,
                "subscriber '{subscription}' returned {} per-element outcomes for a batch of {}",
                outcomes.len(),
                batch_len,
            );
            BatchResult::PerElement(outcomes)
        }
    }
}

/// Runs a batch body over one delivered batch and settles by its verdict.
///
/// The batch arrives at the size the registration asked the broker for, and reaches the body
/// exactly as it arrived: the settling batch forms all run through here - plain and
/// slot-carrying alike - and none of them splits it.
pub(super) async fn run_batch<T, O, C, S, H>(
    body: &H,
    outs: &O,
    batch: &[T],
    ctx: &mut Context<'_, C, S>,
) -> BatchResult
where
    [T]: Input<Axis: BatchedAxis>,
    H: Handle<[T], (), O, C, S>,
    T: Send + Sync,
    O: Send + Sync,
    C: Send + Sync,
    S: Send + Sync,
{
    let verdict = body.handle(batch, outs, ctx).await;
    settle_batch(verdict, batch.len(), ctx.name())
}

impl<T, C, S, H> SliceHandler<T, C, S> for BatchBody<Batch<T>, H>
where
    [T]: Input<Axis = Batch<T>>,
    T: Send + Sync + 'static,
    C: Send + Sync,
    S: Send + Sync,
    H: Handle<[T], (), (), C, S>,
{
    async fn handle_slice(&self, batch: &[T], ctx: &mut Context<'_, C, S>) -> BatchResult {
        run_batch(&self.body, &(), batch, ctx).await
    }
}

impl<Hd, P, C, S, H> SliceHandler<Message<Hd, P>, C, S> for BatchBody<BatchPair<Hd, P>, H>
where
    [Message<Hd, P>]: Input<Axis = BatchPair<Hd, P>>,
    Hd: Send + Sync + 'static,
    P: Send + Sync + 'static,
    C: Send + Sync,
    S: Send + Sync,
    H: Handle<[Message<Hd, P>], (), (), C, S>,
{
    async fn handle_slice(
        &self,
        batch: &[Message<Hd, P>],
        ctx: &mut Context<'_, C, S>,
    ) -> BatchResult {
        run_batch(&self.body, &(), batch, ctx).await
    }
}

// The elements were already constructed by the dispatch adapter, borrowing the deliveries'
// payloads, so this cell only runs and settles like the decoded one. The element is a fresh
// parameter (`T`, one lifetime instantiation of the family's output) because a projection with
// a free lifetime cannot head an impl; the pinned-axis bound is what ties it back to `F` and
// normalizes the verdict family.
impl<T, F, C, S, H> SliceHandler<T, C, S> for BatchBody<BatchDeserialized<F>, H>
where
    T: Send + Sync,
    F: Deserialized + Send + Sync + 'static,
    [T]: Input<Axis = BatchDeserialized<F>>,
    C: Send + Sync,
    S: Send + Sync,
    H: Handle<[T], (), (), C, S>,
{
    async fn handle_slice(&self, batch: &[T], ctx: &mut Context<'_, C, S>) -> BatchResult {
        run_batch(&self.body, &(), batch, ctx).await
    }
}

impl<A, C, H, Doc> BatchDef for Sealed<HandleValue<A, (), (), C, H, Doc>>
where
    A: BatchedAxis,
    Doc: AxisDocs<A>,
{
    type Input = A::Kind;
    type Context = C;
    type Handler = BatchBody<A, H>;
    // See `SubscriberDef::Source` above: the builder carries the real source.
    type Source = Unnamed<Name>;

    fn source(&self) -> Unnamed<Name> {
        Unnamed::new()
    }

    fn description(&self) -> Option<&str> {
        self.0.docs.description()
    }

    fn input_schema(&self) -> Option<String> {
        self.0
            .docs
            .input_schema
            .clone()
            .or_else(Doc::payload_schema)
    }

    fn headers_schema(&self) -> Option<String> {
        self.0
            .docs
            .headers_schema
            .clone()
            .or_else(Doc::headers_schema)
    }

    fn message_name(&self) -> Option<&'static str> {
        self.0.docs.message_name
    }

    fn message_description(&self) -> Option<&'static str> {
        self.0.docs.message_description
    }

    fn into_handler(self) -> BatchBody<A, H> {
        BatchBody {
            body: self.0.body,
            _axes: PhantomData,
        }
    }
}