ruststream 0.6.1

//! Startup injections for batch handlers: the batch counterpart of [`inject`](super::inject).
//!
//! `#[subscriber(batch(..))] async fn f(batch: &[T], Out(out): Out<P>, Seek(seeker): Seek<K>)`
//! declares the same startup-resolved parameters as a single-message handler: the runtime
//! resolves the [`FromStartup`](super::FromStartup) tuple after the subscription opens and
//! before the first batch, so injected values are live by construction. The batch shape needs
//! its own definition pair because the handler consumes a slice and returns a
//! [`BatchResult`], not a per-message [`Settle`](super::Settle).

use std::future::Future;

use crate::IncomingMessage;

use super::batch::{BatchHandler, BatchResult, decode_batch, settle_batch};
use super::context::Context;
use super::dispatch::Workers;
use super::failure::{FailurePolicies, FailurePolicy};
use super::input::{DecodeWith, InputKind};
use super::metadata::HandlerMetadata;

/// A batch definition whose handler takes startup-injected parameters.
///
/// Generated by `#[subscriber(batch(..))]` when the signature carries
/// [`Out`](super::Out) / [`Seek`](super::Seek) parameters; [`Self::Injections`] is their
/// tuple, in declaration order. The metadata surface mirrors [`BatchDef`](super::BatchDef).
pub trait BatchInjectDef: Send + Sync {
    /// The input kind of one batch element (see [`BatchDef::Input`](super::BatchDef::Input)).
    type Input: InputKind;

    /// The subscription source this handler binds to.
    type Source;

    /// The tuple of startup-injected parameters ([`Out`](super::Out), [`Seek`](super::Seek),
    /// ...).
    type Injections;

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

    /// The concurrency policy for this subscriber's dispatch loop (how many batches are in
    /// flight at once).
    fn workers(&self) -> Workers {
        Workers::sequential()
    }

    /// The failure policy for a batch-handler panic and a per-element 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 element type's serialized JSON Schema, when available.
    fn input_schema(&self) -> Option<String> {
        None
    }

    /// The element type's [`Message`](crate::Message) name, when it implements that trait.
    fn message_name(&self) -> Option<&'static str> {
        None
    }

    /// The element type's [`Message`](crate::Message) description, when it implements that
    /// trait.
    fn message_description(&self) -> Option<&'static str> {
        None
    }
}

/// Runs a [`BatchInjectDef`]'s handler body over an app state of type `S`.
///
/// The same state-generic shape as [`SliceHandler`](super::SliceHandler); see
/// [`PublishingCall`](super::PublishingCall) for the rationale.
pub trait BatchInjectCall<S>: BatchInjectDef {
    /// Runs the handler body on one decoded batch, with the resolved injections.
    fn call(
        &self,
        batch: &[<Self::Input as InputKind>::Owned],
        injections: &Self::Injections,
        ctx: &mut Context<'_, (), S>,
    ) -> impl Future<Output = BatchResult> + Send;
}

/// Builds the registration metadata for an injected batch definition mounted under `name`.
pub(crate) fn batch_inject_metadata<D: BatchInjectDef>(name: String, def: &D) -> HandlerMetadata {
    let mut meta = HandlerMetadata::raw(name).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 batch handler built from a [`BatchInjectDef`] once its injections resolved: decode
/// the batch, run the body with them, settle every delivery.
pub struct BatchInjectHandler<Def: BatchInjectDef, DecodeCodec> {
    pub(crate) def: Def,
    pub(crate) codec: DecodeCodec,
    pub(crate) injections: Def::Injections,
    pub(crate) decode: FailurePolicy,
}

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

impl<Msg, Def, DecodeCodec, State> BatchHandler<Msg, State> for BatchInjectHandler<Def, DecodeCodec>
where
    Msg: IncomingMessage,
    Def: BatchInjectCall<State>,
    Def::Input: DecodeWith<DecodeCodec>,
    Def::Injections: Send + Sync,
    DecodeCodec: Send + Sync,
    State: Send + Sync,
{
    async fn handle_batch(&self, batch: Vec<Msg>, ctx: &mut Context<'_, (), State>) {
        let subscription = ctx.name().to_owned();
        let (values, accepted) = decode_batch::<Msg, Def::Input, DecodeCodec, State>(
            batch,
            &self.codec,
            self.decode,
            ctx,
        )
        .await;
        if accepted.is_empty() {
            return;
        }
        let tasks = ctx.tasks().clone();
        let result = self.def.call(&values, &self.injections, ctx).await;
        settle_batch(accepted, result, &subscription, &tasks).await;
    }
}