nest-rs-queue-macros 0.3.0

//! `#[processor]` — orchestrator on a provider's `impl` block. Walks the
//! methods; for each one tagged with `#[process(queue = …, concurrency, retries)]`
//! emits a type-erased handler `fn` and a `ProcessMethod` inventory submission
//! the active queue backend (e.g. Redis via `nest-rs-redis`) drains at boot.
//!
//! Like `#[scheduled]`, this does NOT emit `Discoverable` for the host
//! struct — the user's own `#[injectable]` owns it. Inventory is the seam.
//!
//! The handler is emitted as a `nest_rs_queue::JobHandler` — a fn pointer
//! that takes the raw JSON payload + a `Container`, deserializes to the
//! method's job type, resolves the provider, and dispatches. Every reference
//! is to `::nest_rs_queue::*` (the abstractions crate, which also re-exports
//! this macro and `serde_json`), so the call site reaches the macro and the
//! emission targets through the same import root regardless of which
//! backend integration (nest-rs-redis, …) is wired in.

use nest_rs_codegen::{impl_self_ident, snake_case};
use proc_macro::TokenStream;
use proc_macro2::TokenStream as TokenStream2;
use quote::{format_ident, quote};
use syn::parse::{Parse, ParseStream};
use syn::spanned::Spanned;
use syn::{
    FnArg, Ident, ImplItem, ItemImpl, LitInt, LitStr, PatType, Token, Type, parse_macro_input,
};

pub(crate) fn processor(_args: TokenStream, input: TokenStream) -> TokenStream {
    let mut item = parse_macro_input!(input as ItemImpl);
    let self_ty = item.self_ty.clone();
    let provider_ident = match impl_self_ident(&self_ty, "#[processor]") {
        Ok(ident) => ident,
        Err(err) => return err.to_compile_error().into(),
    };
    let provider_name = provider_ident.to_string();

    let mut emissions: Vec<TokenStream2> = Vec::new();

    for impl_item in item.items.iter_mut() {
        let ImplItem::Fn(method) = impl_item else {
            continue;
        };

        let attr_idx = method
            .attrs
            .iter()
            .position(|attr| attr.path().is_ident("process"));
        let Some(idx) = attr_idx else { continue };
        let attr = method.attrs.remove(idx);

        let args = match attr.parse_args::<ProcessArgs>() {
            Ok(a) => a,
            Err(err) => return err.to_compile_error().into(),
        };
        let ProcessArgs {
            queue,
            concurrency,
            retries,
        } = args;

        let job_ty = match extract_job_type(method) {
            Ok(ty) => ty,
            Err(err) => return err.to_compile_error().into(),
        };
        // A `Piped<P, T>` / `Valid<T>` job argument is a per-argument pipe: the
        // wire payload is `T`, the pipe runs after deserialization, and the
        // handler receives the carrier. Matches the HTTP / GraphQL forms.
        let (deser_ty, job_wrap) = pipe_binding(&job_ty);

        // The queue name is either a raw string literal (legacy form) or a
        // `QueueName` type path (`#[process(queue = AudioQueue)]`). The type
        // form yields `<Q as QueueName>::NAME` — a `&'static str` const usable
        // everywhere the literal was — and additionally asserts, at compile
        // time, that the method's wire payload is exactly `<Q as
        // QueueName>::Job`. A mismatch is an error naming both types.
        let queue_str: TokenStream2 = match &queue {
            QueueId::Literal(s) => quote!(#s),
            QueueId::Type(ty) => {
                let ty: &Type = ty;
                quote!(<#ty as ::nest_rs_queue::QueueName>::NAME)
            }
        };
        let queue_assert: TokenStream2 = match &queue {
            QueueId::Literal(_) => quote!(),
            QueueId::Type(ty) => {
                let ty: &Type = ty;
                quote! {
                const _: () = {
                    // Requires `<#ty as QueueName>::Job == #deser_ty`; a
                    // mismatch fails here naming both the queue's `Job` and the
                    // handler's argument type.
                    fn __nestrs_assert_queue_job<__Q>()
                    where
                        __Q: ::nest_rs_queue::QueueName<Job = #deser_ty>,
                    {
                    }
                    let _ = __nestrs_assert_queue_job::<#ty>;
                };
                }
            }
        };

        let method_ident = method.sig.ident.clone();
        let method_name = method_ident.to_string();
        let qualified_name = format!("{provider_name}::{method_name}");

        let provider_snake = snake_case(&provider_name);
        let method_snake = snake_case(&method_name);
        let handler_ident = format_ident!(
            "__nestrs_process_handler_{}_{}",
            provider_snake,
            method_snake
        );

        let concurrency_lit = LitInt::new(&concurrency.to_string(), proc_macro2::Span::call_site());
        let retries_lit = LitInt::new(&retries.to_string(), proc_macro2::Span::call_site());

        emissions.push(quote! {
            #queue_assert

            #[doc(hidden)]
            #[allow(non_snake_case)]
            fn #handler_ident(
                __payload: ::nest_rs_queue::serde_json::Value,
                __container: ::nest_rs_core::Container,
            ) -> ::std::pin::Pin<
                ::std::boxed::Box<
                    dyn ::std::future::Future<
                        Output = ::std::result::Result<
                            (),
                            ::std::boxed::Box<
                                dyn ::std::error::Error + ::std::marker::Send + ::std::marker::Sync,
                            >,
                        >,
                    > + ::std::marker::Send,
                >,
            > {
                ::std::boxed::Box::pin(async move {
                    // Unwrap the wire envelope `{ "v": <n>, "payload": <…> }`.
                    // Detection is strict to avoid mis-classifying a user `Job`
                    // struct that happens to have `v`+`payload` fields plus
                    // anything else:
                    //   - the object MUST have exactly two top-level keys, and
                    //     they MUST be `v` and `payload`;
                    //   - `v` MUST be a JSON Number with a non-negative integer
                    //     value (accepting both `1` and `1.0` — a hand-rolled
                    //     producer may serialize as a float).
                    // Anything else falls through to the legacy raw-decode path
                    // (with a warning), so jobs left in Redis from a prior
                    // deploy still drain.
                    let __is_envelope = match &__payload {
                        ::nest_rs_queue::serde_json::Value::Object(__obj) => {
                            __obj.len() == 2
                                && __obj.contains_key("v")
                                && __obj.contains_key("payload")
                                && match __obj.get("v") {
                                    ::std::option::Option::Some(
                                        ::nest_rs_queue::serde_json::Value::Number(__n),
                                    ) => {
                                        __n.as_u64().is_some()
                                            || __n.as_f64().is_some_and(|__f| {
                                                __f.is_finite()
                                                    && __f >= 0.0
                                                    && __f.fract() == 0.0
                                            })
                                    }
                                    _ => false,
                                }
                        }
                        _ => false,
                    };
                    let __raw: ::nest_rs_queue::serde_json::Value = if __is_envelope {
                        let ::nest_rs_queue::serde_json::Value::Object(mut __obj) = __payload else {
                            ::std::unreachable!("__is_envelope guarantees an Object");
                        };
                        let __v_value = __obj.remove("v").unwrap_or(
                            ::nest_rs_queue::serde_json::Value::Null,
                        );
                        let __v = match &__v_value {
                            ::nest_rs_queue::serde_json::Value::Number(__n) => __n
                                .as_u64()
                                .or_else(|| __n.as_f64().map(|__f| __f as u64))
                                .unwrap_or(u64::MAX),
                            _ => u64::MAX,
                        };
                        if __v != ::nest_rs_queue::WIRE_FORMAT_VERSION as u64 {
                            let __msg = if __v > ::nest_rs_queue::WIRE_FORMAT_VERSION as u64 {
                                ::std::format!(
                                    "unsupported job wire-format version {} on queue `{}`; \
                                     the producer is from a newer release; either roll back \
                                     this consumer or wait for the producer to drain",
                                    __v,
                                    #queue_str,
                                )
                            } else {
                                ::std::format!(
                                    "unsupported job wire-format version {0} on queue `{1}`; \
                                     the producer is from an older release; either drain \
                                     the queue or pin the consumer at version {0}",
                                    __v,
                                    #queue_str,
                                )
                            };
                            return ::std::result::Result::Err(::std::boxed::Box::<
                                dyn ::std::error::Error + ::std::marker::Send + ::std::marker::Sync,
                            >::from(__msg));
                        }
                        __obj.remove("payload").unwrap_or(
                            ::nest_rs_queue::serde_json::Value::Null,
                        )
                    } else {
                        ::nest_rs_queue::tracing::warn!(
                            target: "nest_rs::queue",
                            queue = #queue_str,
                            hint = "producer predates the wire envelope; drain the queue to clear legacy jobs",
                            "processed an unversioned job payload",
                        );
                        __payload
                    };
                    let __deser: #deser_ty = match ::nest_rs_queue::serde_json::from_value(__raw) {
                        ::std::result::Result::Ok(j) => j,
                        ::std::result::Result::Err(e) => {
                            return ::std::result::Result::Err(::std::boxed::Box::<
                                dyn ::std::error::Error + ::std::marker::Send + ::std::marker::Sync,
                            >::from(::std::format!(
                                "failed to deserialize job for queue `{}`: {e}",
                                #queue_str,
                            )));
                        }
                    };
                    // Identity when the argument is a plain job type; runs the
                    // pipe (surfacing a `PipeError` as the boxed job error) for a
                    // `Piped<P, T>` / `Valid<T>` argument.
                    let __job = #job_wrap;
                    let __provider = match ::nest_rs_core::Container::get::<#self_ty>(&__container) {
                        ::std::option::Option::Some(p) => p,
                        ::std::option::Option::None => {
                            return ::std::result::Result::Err(::std::boxed::Box::<
                                dyn ::std::error::Error + ::std::marker::Send + ::std::marker::Sync,
                            >::from(::std::format!(
                                "queue processor provider `{}` not registered in the running \
                                 container — add it to a reachable module's `providers = [...]`",
                                ::std::any::type_name::<#self_ty>(),
                            )));
                        }
                    };
                    let __job_context = ::nest_rs_core::Container::get_dyn::<
                        dyn ::nest_rs_worker::JobContext,
                    >(&__container);
                    ::nest_rs_worker::run_in_job_context(
                        __job_context.as_ref(),
                        async move { <#self_ty>::#method_ident(&__provider, __job).await },
                    )
                    .await
                    .map_err(::std::convert::Into::into)
                })
            }

            ::nest_rs_core::inventory::submit! {
                ::nest_rs_queue::ProcessMethod {
                    name: #qualified_name,
                    queue: #queue_str,
                    concurrency: #concurrency_lit,
                    retries: #retries_lit,
                    provider_type_id: || ::std::any::TypeId::of::<#self_ty>(),
                    handler: #handler_ident,
                }
            }
        });
    }

    let out = quote! {
        #item
        #(#emissions)*
    };
    out.into()
}

/// Extract the second parameter's type — the job payload. Errors out crisply
/// when the signature is wrong (no `&self`, or no job arg).
fn extract_job_type(method: &syn::ImplItemFn) -> syn::Result<Type> {
    let mut iter = method.sig.inputs.iter();
    match iter.next() {
        Some(FnArg::Receiver(_)) => {}
        Some(other) => {
            return Err(syn::Error::new(
                other.span(),
                "a `#[process]` method must take `&self` as its first argument",
            ));
        }
        None => {
            return Err(syn::Error::new(
                method.sig.span(),
                "a `#[process]` method must take `&self` and one job argument",
            ));
        }
    }
    let Some(arg) = iter.next() else {
        return Err(syn::Error::new(
            method.sig.span(),
            "a `#[process]` method needs a job argument: `async fn(&self, job: T)`",
        ));
    };
    match arg {
        FnArg::Typed(PatType { ty, .. }) => Ok((**ty).clone()),
        FnArg::Receiver(r) => Err(syn::Error::new(
            r.span(),
            "a `#[process]` method takes exactly one `&self` receiver",
        )),
    }
}

/// Split a job argument into (type to deserialize from the wire, expression that
/// yields the value the handler receives). For a plain type both are trivial:
/// deserialize `T`, hand over `__deser`. For a per-argument pipe `Piped<P, T>` /
/// `Valid<T>` the wire type is `T`, and the expression runs the pipe over
/// `__deser`, surfacing a `PipeError` as the queue's boxed error.
fn pipe_binding(job_ty: &Type) -> (Type, TokenStream2) {
    let box_err = quote! {
        |__e: ::nest_rs_pipes::PipeError| ::std::boxed::Box::<
            dyn ::std::error::Error + ::std::marker::Send + ::std::marker::Sync,
        >::from(__e.message().to_string())
    };
    if let Type::Path(tp) = job_ty
        && let Some(seg) = tp.path.segments.last()
        && let syn::PathArguments::AngleBracketed(ab) = &seg.arguments
    {
        let tys: Vec<&Type> = ab
            .args
            .iter()
            .filter_map(|a| match a {
                syn::GenericArgument::Type(t) => Some(t),
                _ => None,
            })
            .collect();
        if seg.ident == "Piped" && tys.len() == 2 {
            let (pipe, inner) = (tys[0], tys[1]);
            return (
                inner.clone(),
                quote! {
                    ::nest_rs_pipes::Piped::<#pipe, #inner>::apply(__deser).map_err(#box_err)?
                },
            );
        }
        if seg.ident == "Valid" && tys.len() == 1 {
            let inner = tys[0];
            return (
                inner.clone(),
                quote! {
                    ::nest_rs_pipes::Valid::<#inner>::apply(__deser).map_err(#box_err)?
                },
            );
        }
    }
    (job_ty.clone(), quote!(__deser))
}

/// How a `#[process]` names its queue: the legacy raw string, or a `QueueName`
/// type path (`#[process(queue = AudioQueue)]`) that links the name and payload
/// type to the shared handle at the feature port.
enum QueueId {
    Literal(LitStr),
    // Boxed: `syn::Type` is a large enum, so an unboxed variant would bloat
    // every `QueueId` to its size (clippy::large_enum_variant).
    Type(Box<Type>),
}

impl Parse for QueueId {
    fn parse(input: ParseStream) -> syn::Result<Self> {
        if input.peek(LitStr) {
            Ok(QueueId::Literal(input.parse()?))
        } else {
            Ok(QueueId::Type(Box::new(input.parse()?)))
        }
    }
}

struct ProcessArgs {
    queue: QueueId,
    concurrency: usize,
    retries: usize,
}

impl Parse for ProcessArgs {
    fn parse(input: ParseStream) -> syn::Result<Self> {
        let mut queue: Option<QueueId> = None;
        let mut concurrency: usize = 1;
        let mut retries: usize = 0;

        while !input.is_empty() {
            let key: Ident = input.parse()?;
            input.parse::<Token![=]>()?;
            match key.to_string().as_str() {
                "queue" => queue = Some(input.parse()?),
                "concurrency" => concurrency = input.parse::<LitInt>()?.base10_parse()?,
                "retries" => retries = input.parse::<LitInt>()?.base10_parse()?,
                other => {
                    return Err(syn::Error::new(
                        key.span(),
                        format!(
                            "unknown #[process] key `{other}` \
                             (expected `queue`, `concurrency`, or `retries`)"
                        ),
                    ));
                }
            }
            if !input.is_empty() {
                input.parse::<Token![,]>()?;
            }
        }

        let queue = queue.ok_or_else(|| {
            syn::Error::new(
                input.span(),
                "#[process] requires a `queue = \"...\"` (or `queue = <QueueName type>`) argument",
            )
        })?;

        Ok(Self {
            queue,
            concurrency,
            retries,
        })
    }
}