Skip to main content

dataflow_rs/engine/functions/
config.rs

1use crate::engine::error::Result;
2use crate::engine::executor::ArenaContext;
3use crate::engine::functions::filter::FilterConfig;
4use crate::engine::functions::integration::{EnrichConfig, HttpCallConfig, PublishKafkaConfig};
5use crate::engine::functions::log::LogConfig;
6use crate::engine::functions::map::MapConfig;
7use crate::engine::functions::parse::{ParseConfig, execute_parse_json_in_arena, parse_xml_in};
8use crate::engine::functions::path_template::ParamCtx;
9use crate::engine::functions::publish::{PublishConfig, publish_json_in, publish_xml_in};
10use crate::engine::functions::template::Template;
11use crate::engine::functions::validation::ValidationConfig;
12use crate::engine::message::{Change, Message};
13use crate::engine::task_outcome::TaskOutcome;
14use datalogic_rs::Engine;
15use serde::de::DeserializeOwned;
16use serde::{Deserialize, Deserializer};
17use serde_json::Value;
18use std::any::Any;
19use std::sync::Arc;
20
21/// Pre-parsed typed input for a `FunctionConfig::Custom` task. Populated by
22/// the engine at `Engine::new()` time by calling the registered
23/// `AsyncFunctionHandler::parse_input` for the named function. Cached as
24/// `Arc<dyn Any>` so the dispatch path can hand it to the handler with a
25/// single `downcast_ref` (O(1)) and zero per-message deserialization cost.
26///
27/// The wrapper exists because `dyn Any` does not implement `Debug`, which
28/// would otherwise prevent `#[derive(Debug)]` on `FunctionConfig`.
29#[derive(Clone)]
30pub struct CompiledCustomInput(pub Arc<dyn Any + Send + Sync>);
31
32impl CompiledCustomInput {
33    /// Borrow the inner value as `&(dyn Any + Send + Sync)` for handoff to
34    /// `DynAsyncFunctionHandler::dyn_execute`.
35    #[inline]
36    pub fn as_any(&self) -> &(dyn Any + Send + Sync) {
37        &*self.0
38    }
39}
40
41impl std::fmt::Debug for CompiledCustomInput {
42    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
43        f.write_str("CompiledCustomInput(<opaque>)")
44    }
45}
46
47/// How a task names its connector.
48///
49/// `connector` is JSONLogic like every other parameter, so it may be a literal
50/// the host can read at authoring time or an expression that only resolves per
51/// message. Splitting the two is what keeps a computed connector from vanishing
52/// out of [`crate::Workflow::connector_refs`] — a host pre-warming connection
53/// pools needs to know it exists even when it cannot know its name yet.
54#[derive(Debug, Clone, Copy, PartialEq, Eq)]
55pub enum ConnectorName<'a> {
56    /// Authored as a literal string. Known without a message.
57    Static(&'a str),
58    /// Authored as an expression, carrying the authored JSON. Resolve it per
59    /// message with the config's `resolve_connector`.
60    Computed(&'a Value),
61}
62
63impl<'a> ConnectorName<'a> {
64    /// Classify an authored `connector` parameter.
65    fn of(template: &'a Template) -> Self {
66        match template.as_json() {
67            Value::String(s) => Self::Static(s),
68            other => Self::Computed(other),
69        }
70    }
71
72    /// The literal name, or `None` when the connector is computed.
73    ///
74    /// The narrowing accessor for callers that genuinely only handle static
75    /// connectors. Prefer matching the enum, so the computed case is a decision
76    /// rather than an omission.
77    pub fn as_static(&self) -> Option<&'a str> {
78        match self {
79            Self::Static(s) => Some(s),
80            Self::Computed(_) => None,
81        }
82    }
83}
84
85/// Enum containing all possible function configurations.
86///
87/// Deserialization dispatches on the `name` field: each known built-in
88/// (`map`, `validate`, `parse_json`, …) parses its `input` strictly into the
89/// matching typed config and errors with a clear envelope (`config for
90/// function 'map': missing field 'mappings'`). Unknown names fall through
91/// to [`FunctionConfig::Custom`], which preserves the raw input for a
92/// user-registered handler to consume at engine construction time.
93#[derive(Debug, Clone)]
94pub enum FunctionConfig {
95    Map {
96        name: MapName,
97        input: MapConfig,
98    },
99    Validation {
100        name: ValidationName,
101        input: ValidationConfig,
102    },
103    ParseJson {
104        name: ParseJsonName,
105        input: ParseConfig,
106    },
107    ParseXml {
108        name: ParseXmlName,
109        input: ParseConfig,
110    },
111    PublishJson {
112        name: PublishJsonName,
113        input: PublishConfig,
114    },
115    PublishXml {
116        name: PublishXmlName,
117        input: PublishConfig,
118    },
119    Filter {
120        name: FilterName,
121        input: FilterConfig,
122    },
123    Log {
124        name: LogName,
125        input: LogConfig,
126    },
127    HttpCall {
128        name: HttpCallName,
129        input: HttpCallConfig,
130    },
131    Enrich {
132        name: EnrichName,
133        input: EnrichConfig,
134    },
135    PublishKafka {
136        name: PublishKafkaName,
137        input: PublishKafkaConfig,
138    },
139    /// For custom or unknown functions, store raw input and a slot for the
140    /// pre-parsed typed value populated at engine construction time.
141    Custom {
142        name: String,
143        input: Value,
144        /// Pre-parsed `<RegisteredHandler as AsyncFunctionHandler>::Input`,
145        /// boxed as `dyn Any`. Set by the engine after handler registration;
146        /// `None` on initial deserialization. `FunctionConfig` is
147        /// deserialize-only — round-tripping a workflow through JSON
148        /// re-parses on the next `Engine::new()` call.
149        compiled_input: Option<CompiledCustomInput>,
150    },
151}
152
153#[derive(Debug, Clone, Deserialize)]
154#[serde(rename_all = "lowercase")]
155pub enum MapName {
156    Map,
157}
158
159#[derive(Debug, Clone, Deserialize, PartialEq)]
160#[serde(rename_all = "lowercase")]
161pub enum ValidationName {
162    Validation,
163    Validate,
164}
165
166#[derive(Debug, Clone, Deserialize, PartialEq)]
167#[serde(rename_all = "snake_case")]
168pub enum ParseJsonName {
169    ParseJson,
170}
171
172#[derive(Debug, Clone, Deserialize, PartialEq)]
173#[serde(rename_all = "snake_case")]
174pub enum ParseXmlName {
175    ParseXml,
176}
177
178#[derive(Debug, Clone, Deserialize, PartialEq)]
179#[serde(rename_all = "snake_case")]
180pub enum PublishJsonName {
181    PublishJson,
182}
183
184#[derive(Debug, Clone, Deserialize, PartialEq)]
185#[serde(rename_all = "snake_case")]
186pub enum PublishXmlName {
187    PublishXml,
188}
189
190#[derive(Debug, Clone, Deserialize, PartialEq)]
191#[serde(rename_all = "lowercase")]
192pub enum FilterName {
193    Filter,
194}
195
196#[derive(Debug, Clone, Deserialize, PartialEq)]
197#[serde(rename_all = "lowercase")]
198pub enum LogName {
199    Log,
200}
201
202#[derive(Debug, Clone, Deserialize, PartialEq)]
203#[serde(rename_all = "snake_case")]
204pub enum HttpCallName {
205    HttpCall,
206}
207
208#[derive(Debug, Clone, Deserialize, PartialEq)]
209#[serde(rename_all = "snake_case")]
210pub enum EnrichName {
211    Enrich,
212}
213
214#[derive(Debug, Clone, Deserialize, PartialEq)]
215#[serde(rename_all = "snake_case")]
216pub enum PublishKafkaName {
217    PublishKafka,
218}
219
220/// Every function name that [`FunctionConfig`]'s deserializer resolves to a
221/// typed built-in variant instead of [`FunctionConfig::Custom`].
222///
223/// Used in error messages and as the discriminator for [`FunctionConfig`]
224/// deserialization. Kept in one place so adding a new built-in updates the
225/// dispatch, the error suggestion list, and the docs in lockstep.
226///
227/// Public so a service layer that gates workflow authoring on a closed
228/// function set can derive that set rather than copy it. Membership here is
229/// **not** the same fact as "this engine can run it" — see
230/// [`builtin_function_kind`], and prefer it for that question.
231///
232/// # Stability
233///
234/// Names are only added in a minor release and only removed in a major one, so
235/// a caller may treat a name that appears here as durable. **Ordering is not
236/// meaningful** and may change without notice; treat this as a set. Note that
237/// `validation` and `validate` are both present and both resolve to
238/// [`FunctionConfig::Validation`].
239pub const BUILTIN_FUNCTION_NAMES: &[&str] = &[
240    "map",
241    "validation",
242    "validate",
243    "parse_json",
244    "parse_xml",
245    "publish_json",
246    "publish_xml",
247    "filter",
248    "log",
249    "http_call",
250    "enrich",
251    "publish_kafka",
252];
253
254/// How a built-in function reaches an implementation.
255///
256/// This is the programmatic form of the distinction
257/// `docs/src/built-in-functions/integrations.md` draws in prose: some built-ins
258/// this crate executes itself, and for others it only supplies a config schema.
259///
260/// Deliberately **not** `#[non_exhaustive]`. A caller matching on this is
261/// usually deciding whether to accept a workflow definition, and if a third
262/// kind is ever added that decision needs revisiting — a compile error at every
263/// match site is the correct signal, not a silent fall-through to a `_` arm.
264#[derive(Debug, Clone, Copy, PartialEq, Eq)]
265pub enum BuiltinKind {
266    /// Executed by this crate. Needs no registration; an engine can always run it.
267    SelfContained,
268    /// Deserializes into a typed built-in variant — so `Engine::new` accepts it
269    /// without complaint — but dispatches to a handler registered under the same
270    /// name, and fails with [`crate::DataflowError::FunctionNotFound`] on the
271    /// first message if none is registered.
272    ///
273    /// `http_call`, `enrich` and `publish_kafka` are these. A validator that
274    /// treats them like [`BuiltinKind::SelfContained`] will green-light a
275    /// workflow that builds cleanly and then fails every request.
276    RequiresHandler,
277}
278
279/// Classify `name` as a built-in function.
280///
281/// `None` means it is not a built-in: it lands in [`FunctionConfig::Custom`] and
282/// needs `Engine::builder().register(name, handler)`.
283///
284/// Matching is exact, the same as the deserializer dispatch — `"HTTP_CALL"` and
285/// `"htttp_call"` are both `None`.
286///
287/// ```
288/// use dataflow_rs::{BuiltinKind, builtin_function_kind};
289///
290/// assert_eq!(builtin_function_kind("map"), Some(BuiltinKind::SelfContained));
291/// assert_eq!(builtin_function_kind("enrich"), Some(BuiltinKind::RequiresHandler));
292/// assert_eq!(builtin_function_kind("my_handler"), None);
293/// ```
294pub fn builtin_function_kind(name: &str) -> Option<BuiltinKind> {
295    match name {
296        "map" | "validation" | "validate" | "parse_json" | "parse_xml" | "publish_json"
297        | "publish_xml" | "filter" | "log" => Some(BuiltinKind::SelfContained),
298        "http_call" | "enrich" | "publish_kafka" => Some(BuiltinKind::RequiresHandler),
299        _ => None,
300    }
301}
302
303/// Whether `name` deserializes to a typed built-in variant at all.
304///
305/// Equivalent to `builtin_function_kind(name).is_some()`. This answers "is this
306/// a name the crate special-cases", **not** "can this engine run it" — a
307/// [`BuiltinKind::RequiresHandler`] name returns `true` here whether or not a
308/// handler is registered.
309#[inline]
310pub fn is_builtin_function(name: &str) -> bool {
311    builtin_function_kind(name).is_some()
312}
313
314/// One function an engine will actually dispatch.
315///
316/// Yielded by [`crate::Engine::dispatchable_functions`] and
317/// [`crate::EngineBuilder::dispatchable_functions`]. Together with
318/// [`BuiltinKind`] this is the whole authoring-side vocabulary: `kind` says how
319/// the name reaches an implementation, and `aliases` says which other spellings
320/// resolve to the same one.
321#[derive(Debug, Clone, PartialEq, Eq)]
322pub struct DispatchableFunction<'a> {
323    /// The canonical name. Alternative spellings are listed in
324    /// [`Self::aliases`] rather than yielded as separate entries.
325    pub name: &'a str,
326    /// `Some(..)` for a built-in; `None` for a name backed only by a registered
327    /// custom handler.
328    ///
329    /// The `Option` deliberately mirrors [`builtin_function_kind`], where
330    /// `None` already means "not a built-in". [`BuiltinKind`] is not
331    /// `#[non_exhaustive]` on purpose, so widening it with a third variant
332    /// would break every downstream `match`; this carries the same fact
333    /// additively.
334    pub kind: Option<BuiltinKind>,
335    /// Other accepted spellings of this same function.
336    ///
337    /// `validate` carries `["validation"]`; every other name is empty today.
338    /// An alias never appears as its own entry, but
339    /// [`crate::Engine::can_dispatch`] does accept it — a task named
340    /// `validation` really does execute.
341    pub aliases: &'static [&'static str],
342}
343
344/// Aliases of `validate`. Both spellings deserialize to
345/// [`FunctionConfig::Validation`]; `function_name()` reports `"validate"`, which
346/// makes that the canonical one.
347const VALIDATE_ALIASES: &[&str] = &["validation"];
348
349/// The empty alias list, so [`builtin_aliases`] can return a `'static` slice for
350/// every name without allocating.
351const NO_ALIASES: &[&str] = &[];
352
353/// Map a built-in spelling to the canonical one for its function.
354///
355/// Every name is its own canonical form except `validation`, which is an alias
356/// of `validate`. Deliberately a match rather than a table: the alias relation
357/// is the *only* new fact here, and a table of canonical names would be a
358/// second copy of [`BUILTIN_FUNCTION_NAMES`] to keep in sync.
359pub(crate) fn canonical_builtin_name(name: &str) -> &str {
360    match name {
361        "validation" => "validate",
362        other => other,
363    }
364}
365
366/// The alternative spellings of `canonical`, which must already be a canonical
367/// name. Paired with [`canonical_builtin_name`]; the two are pinned to each
368/// other and to [`BUILTIN_FUNCTION_NAMES`] by `aliases_and_canonical_names_agree`.
369pub(crate) fn builtin_aliases(canonical: &str) -> &'static [&'static str] {
370    match canonical {
371        "validate" => VALIDATE_ALIASES,
372        _ => NO_ALIASES,
373    }
374}
375
376/// Whether a registry containing `registry`'s keys will dispatch `name`.
377///
378/// The single definition of "this engine can run it": a
379/// [`BuiltinKind::SelfContained`] built-in always can, and every other name —
380/// [`BuiltinKind::RequiresHandler`] built-ins and custom names alike — can only
381/// if a handler is registered under it. `TaskExecutor::has_function` and the
382/// two public `can_dispatch` methods all route through here so the predicate
383/// the engine dispatches on and the predicate hosts query cannot drift.
384///
385/// Generic over the map's value type so this module needs no dependency on
386/// `BoxedFunctionHandler`.
387pub(crate) fn can_dispatch_in<V>(
388    registry: &std::collections::HashMap<String, V>,
389    name: &str,
390) -> bool {
391    match builtin_function_kind(name) {
392        Some(BuiltinKind::SelfContained) => true,
393        // RequiresHandler and Custom alike: only if a handler was registered.
394        _ => registry.contains_key(name),
395    }
396}
397
398/// Every function a registry with these keys will dispatch.
399///
400/// Built-ins are yielded only when they are their own canonical name, which
401/// performs the alias grouping with no list to maintain. `RequiresHandler`
402/// built-ins appear only when backed by a registration; custom keys appear with
403/// `kind: None`.
404///
405/// A key that names a [`BuiltinKind::SelfContained`] built-in is skipped on the
406/// registry side — it is already yielded as a built-in, and the registration
407/// itself is inert (the deserializer routes `map` to [`FunctionConfig::Map`],
408/// which this crate executes without consulting the registry).
409pub(crate) fn dispatchable_functions_in<V>(
410    registry: &std::collections::HashMap<String, V>,
411) -> impl Iterator<Item = DispatchableFunction<'_>> {
412    let builtins = BUILTIN_FUNCTION_NAMES
413        .iter()
414        .copied()
415        // Skip aliases: `validation` is reported under `validate`.
416        .filter(|name| canonical_builtin_name(name) == *name)
417        .filter_map(move |name| match builtin_function_kind(name) {
418            // Always runnable, registered or not.
419            kind @ Some(BuiltinKind::SelfContained) => Some(DispatchableFunction {
420                name,
421                kind,
422                aliases: builtin_aliases(name),
423            }),
424            // Config schema only — present iff a handler backs it.
425            kind @ Some(BuiltinKind::RequiresHandler) if registry.contains_key(name) => {
426                Some(DispatchableFunction {
427                    name,
428                    kind,
429                    aliases: builtin_aliases(name),
430                })
431            }
432            _ => None,
433        });
434
435    let customs = registry
436        .keys()
437        .map(String::as_str)
438        // Built-in names are handled above; a registration under one is either
439        // already counted (RequiresHandler) or inert (SelfContained).
440        .filter(|name| builtin_function_kind(name).is_none())
441        .map(|name| DispatchableFunction {
442            name,
443            kind: None,
444            aliases: NO_ALIASES,
445        });
446
447    builtins.chain(customs)
448}
449
450/// Parse a `serde_json::Value` into a typed config, wrapping any error in a
451/// "config for function '<func>': …" envelope. Strips the trailing
452/// `" at line 0 column 0"` that `serde_json::from_value` always appends
453/// (since the source `Value` has no source-text location); the outer
454/// deserializer re-attaches the real source location when this error
455/// bubbles up to e.g. `Workflow::from_json`.
456fn parse_function_input<T, E>(func: &str, input: Value) -> std::result::Result<T, E>
457where
458    T: DeserializeOwned,
459    E: serde::de::Error,
460{
461    serde_json::from_value::<T>(input).map_err(|err| {
462        let raw = err.to_string();
463        let trimmed = raw
464            .rsplit_once(" at line ")
465            .map(|(head, _)| head)
466            .unwrap_or(&raw);
467        E::custom(format!("config for function '{func}': {trimmed}"))
468    })
469}
470
471impl<'de> Deserialize<'de> for FunctionConfig {
472    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
473    where
474        D: Deserializer<'de>,
475    {
476        // Tag-only intermediate. Format-agnostic: works for any deserializer
477        // that produces `String`/`serde_json::Value`. The strict typed parse
478        // happens in the dispatch below.
479        #[derive(Deserialize)]
480        struct Raw {
481            name: String,
482            input: Value,
483        }
484
485        let Raw { name, input } = Raw::deserialize(deserializer)?;
486
487        Ok(match name.as_str() {
488            "map" => Self::Map {
489                name: MapName::Map,
490                input: parse_function_input("map", input)?,
491            },
492            "validate" => Self::Validation {
493                name: ValidationName::Validate,
494                input: parse_function_input("validate", input)?,
495            },
496            "validation" => Self::Validation {
497                name: ValidationName::Validation,
498                input: parse_function_input("validation", input)?,
499            },
500            "parse_json" => Self::ParseJson {
501                name: ParseJsonName::ParseJson,
502                input: parse_function_input("parse_json", input)?,
503            },
504            "parse_xml" => Self::ParseXml {
505                name: ParseXmlName::ParseXml,
506                input: parse_function_input("parse_xml", input)?,
507            },
508            "publish_json" => Self::PublishJson {
509                name: PublishJsonName::PublishJson,
510                input: parse_function_input("publish_json", input)?,
511            },
512            "publish_xml" => Self::PublishXml {
513                name: PublishXmlName::PublishXml,
514                input: parse_function_input("publish_xml", input)?,
515            },
516            "filter" => Self::Filter {
517                name: FilterName::Filter,
518                input: parse_function_input("filter", input)?,
519            },
520            "log" => Self::Log {
521                name: LogName::Log,
522                input: parse_function_input("log", input)?,
523            },
524            "http_call" => Self::HttpCall {
525                name: HttpCallName::HttpCall,
526                input: parse_function_input("http_call", input)?,
527            },
528            "enrich" => Self::Enrich {
529                name: EnrichName::Enrich,
530                input: parse_function_input("enrich", input)?,
531            },
532            "publish_kafka" => Self::PublishKafka {
533                name: PublishKafkaName::PublishKafka,
534                input: parse_function_input("publish_kafka", input)?,
535            },
536            _ => Self::Custom {
537                name,
538                input,
539                compiled_input: None,
540            },
541        })
542    }
543}
544
545/// Refresh the arena's `"data"` slot on `Ok`, then return `result` unchanged.
546/// Shared by the three built-ins (`parse_xml`, `publish_json`, `publish_xml`)
547/// that write through `set_nested_value` on the owned context rather than the
548/// arena — the arena cache would otherwise miss that write for the rest of the
549/// sync stretch. On `Err` the context didn't change, so the cache is already
550/// in sync and is left alone.
551fn refresh_data_on_success(
552    message: &Message,
553    arena_ctx: &mut ArenaContext<'_>,
554    result: Result<(TaskOutcome, Vec<Change>)>,
555) -> Result<(TaskOutcome, Vec<Change>)> {
556    if result.is_ok() {
557        arena_ctx.refresh_for_path(&message.context, "data");
558    }
559    result
560}
561
562impl FunctionConfig {
563    /// Get the function name for this configuration
564    pub fn function_name(&self) -> &str {
565        match self {
566            Self::Map { .. } => "map",
567            Self::Validation { .. } => "validate",
568            Self::ParseJson { .. } => "parse_json",
569            Self::ParseXml { .. } => "parse_xml",
570            Self::PublishJson { .. } => "publish_json",
571            Self::PublishXml { .. } => "publish_xml",
572            Self::Filter { .. } => "filter",
573            Self::Log { .. } => "log",
574            Self::HttpCall { .. } => "http_call",
575            Self::Enrich { .. } => "enrich",
576            Self::PublishKafka { .. } => "publish_kafka",
577            Self::Custom { name, .. } => name,
578        }
579    }
580
581    /// Whether this is a synchronous built-in. Synchronous built-ins can share
582    /// a single `ArenaContext` lifetime across consecutive tasks within a
583    /// workflow without crossing any `.await` point.
584    ///
585    /// Must match the variants handled in `Self::try_execute_in_arena`; the
586    /// debug assertion below ties the two together so they can't drift.
587    /// The connector this task references, if any.
588    ///
589    /// The three integration variants return their typed `connector` field
590    /// verbatim — including an empty string. Whether an empty connector name is
591    /// acceptable is a validation question for the host, not this accessor's.
592    ///
593    /// Since 3.9 `connector` is JSONLogic, so the answer is
594    /// [`ConnectorName::Static`] only when it was authored as a literal string.
595    /// A computed connector yields [`ConnectorName::Computed`], which names no
596    /// single connector until a message is in hand. A host enumerating
597    /// connectors to validate or pre-warm them must decide what to do with
598    /// those rather than have them silently disappear from the list — which is
599    /// why this returns an enum instead of `Option<&str>`.
600    ///
601    /// [`FunctionConfig::Custom`] returns `input["connector"]` when that key
602    /// holds a string. That is the convention for service-registered integration
603    /// handlers, mirroring the three built-in schemas; a `Custom` input whose
604    /// `connector` key means something else is a false positive, and the
605    /// convention is the only contract available.
606    ///
607    /// Usable without a [`crate::Task`]: `FunctionConfig` deserializes from a
608    /// bare `{"name": .., "input": ..}` object, so a caller holding only a task's
609    /// `function` value does not need to satisfy `Task`'s required `id` and
610    /// `name`.
611    ///
612    /// The match is exhaustive on purpose — a future connector-bearing config
613    /// cannot be silently omitted.
614    pub fn connector(&self) -> Option<ConnectorName<'_>> {
615        match self {
616            Self::HttpCall { input, .. } => Some(ConnectorName::of(&input.connector)),
617            Self::Enrich { input, .. } => Some(ConnectorName::of(&input.connector)),
618            Self::PublishKafka { input, .. } => Some(ConnectorName::of(&input.connector)),
619            Self::Custom { input, .. } => input
620                .get("connector")
621                .and_then(Value::as_str)
622                .map(ConnectorName::Static),
623            Self::Map { .. }
624            | Self::Validation { .. }
625            | Self::ParseJson { .. }
626            | Self::ParseXml { .. }
627            | Self::PublishJson { .. }
628            | Self::PublishXml { .. }
629            | Self::Filter { .. }
630            | Self::Log { .. } => None,
631        }
632    }
633
634    /// Whether the workflow executor runs this task itself, inside the shared
635    /// arena, rather than breaking the sync stretch to `.await` a handler.
636    ///
637    /// Spelled as the *complement* of the handler-backed set rather than as the
638    /// list of sync built-ins, and that is load-bearing. This must agree with
639    /// `Self::try_execute_in_arena` exactly — a `true` here that meets a
640    /// `None` there is the "engine bug" arm of
641    /// `WorkflowExecutor::execute_sync_task_in_arena`. Adding a variant forces
642    /// an arm in `try_execute_in_arena` (that match is exhaustive), and with
643    /// the negation form a *new sync built-in* then classifies correctly here
644    /// with no second edit. Only a new handler-backed variant needs adding to
645    /// the list below, and
646    /// `is_sync_builtin_agrees_with_arena_dispatch_for_every_builtin` fails if
647    /// it is forgotten.
648    pub fn is_sync_builtin(&self) -> bool {
649        !matches!(
650            self,
651            Self::HttpCall { .. }
652                | Self::Enrich { .. }
653                | Self::PublishKafka { .. }
654                | Self::Custom { .. }
655        )
656    }
657
658    /// If this config is a sync built-in, execute it against the supplied
659    /// arena context and return `Some(result)`. Otherwise return `None` —
660    /// the workflow executor uses that as the signal to break the sync
661    /// stretch and dispatch the task on the async path instead.
662    ///
663    /// `mapping_snapshots` is only consulted by the `Map` variant — when
664    /// `Some`, the map function pushes a `serde_json::Value` snapshot of the
665    /// context before each mapping (for the trace surface). All other
666    /// variants ignore it. Pass `None` from the production path.
667    ///
668    /// This is the single source of truth for the sync-stretch dispatch, and
669    /// the match is exhaustive, so a new variant cannot be added without
670    /// deciding here. A new *sync* built-in needs nothing else:
671    /// [`Self::is_sync_builtin`] is written as the complement of the
672    /// handler-backed set, so it classifies the newcomer correctly on its own.
673    /// A new *handler-backed* one must also join that set, and
674    /// `is_sync_builtin_agrees_with_arena_dispatch_for_every_builtin` fails if
675    /// it does not.
676    pub(crate) fn try_execute_in_arena<'arena>(
677        &'arena self,
678        message: &mut Message,
679        arena_ctx: &mut ArenaContext<'arena>,
680        engine: &Arc<Engine>,
681        mapping_snapshots: Option<&mut Vec<Value>>,
682    ) -> Option<Result<(TaskOutcome, Vec<Change>)>> {
683        match self {
684            Self::Map { input, .. } => {
685                Some(input.execute_in_arena(message, arena_ctx, engine, mapping_snapshots))
686            }
687            Self::Validation { input, .. } => {
688                Some(input.execute_in_arena(message, arena_ctx, engine))
689            }
690            Self::ParseJson { input, .. } => Some(execute_parse_json_in_arena(
691                message, input, engine, arena_ctx,
692            )),
693            Self::ParseXml { input, .. } => {
694                // parse_xml/publish_json/publish_xml all write through
695                // `set_nested_value` on the owned context rather than the
696                // arena, so the arena's "data" slot needs a manual refresh —
697                // but only on success; on error the context didn't change
698                // either, so the arena cache is still in sync.
699                let p = ParamCtx::from_arena(engine, arena_ctx);
700                let result = parse_xml_in(message, input, p);
701                Some(refresh_data_on_success(message, arena_ctx, result))
702            }
703            Self::PublishJson { input, .. } => {
704                let p = ParamCtx::from_arena(engine, arena_ctx);
705                let result = publish_json_in(message, input, p);
706                Some(refresh_data_on_success(message, arena_ctx, result))
707            }
708            Self::PublishXml { input, .. } => {
709                let p = ParamCtx::from_arena(engine, arena_ctx);
710                let result = publish_xml_in(message, input, p);
711                Some(refresh_data_on_success(message, arena_ctx, result))
712            }
713            Self::Filter { input, .. } => Some(input.execute_in_arena(message, arena_ctx, engine)),
714            Self::Log { input, .. } => Some(input.execute_in_arena(message, arena_ctx, engine)),
715            Self::HttpCall { .. }
716            | Self::Enrich { .. }
717            | Self::PublishKafka { .. }
718            | Self::Custom { .. } => None,
719        }
720    }
721}
722
723#[cfg(test)]
724mod tests {
725    use super::*;
726    use serde_json::json;
727
728    fn parse(value: serde_json::Value) -> std::result::Result<FunctionConfig, serde_json::Error> {
729        serde_json::from_value(value)
730    }
731
732    /// The smallest `input` that parses for each built-in name.
733    ///
734    /// Shared by every table-driven test below that walks
735    /// [`BUILTIN_FUNCTION_NAMES`], so a new built-in is described once. A name
736    /// with required fields that is not listed here falls to `{}` and fails
737    /// loudly at the parse in each caller, rather than silently skewing a
738    /// partition.
739    fn minimal_input(name: &str) -> serde_json::Value {
740        match name {
741            "map" => json!({ "mappings": [] }),
742            "validation" | "validate" => json!({ "rules": [] }),
743            "parse_json" | "parse_xml" | "publish_json" | "publish_xml" => {
744                json!({ "source": "data.in", "target": "out" })
745            }
746            "filter" => json!({ "condition": true }),
747            "log" => json!({ "message": "hi" }),
748            "http_call" => json!({ "connector": "c" }),
749            "enrich" => json!({ "connector": "c", "merge_path": "data.out" }),
750            "publish_kafka" => json!({ "connector": "c", "topic": "t" }),
751            _ => json!({}),
752        }
753    }
754
755    #[test]
756    fn map_with_valid_config_deserializes_to_map_variant() {
757        let cfg = parse(json!({
758            "name": "map",
759            "input": {
760                "mappings": [
761                    { "path": "data.x", "logic": { "var": "data.y" } }
762                ]
763            }
764        }))
765        .expect("valid map config should deserialize");
766        assert!(matches!(cfg, FunctionConfig::Map { .. }));
767    }
768
769    #[test]
770    fn map_with_missing_mappings_gives_clear_error() {
771        let err = parse(json!({
772            "name": "map",
773            "input": {}
774        }))
775        .expect_err("map with empty input should fail");
776        let msg = err.to_string();
777        assert!(
778            msg.starts_with("config for function 'map':"),
779            "error should be prefixed with function envelope, got: {msg}"
780        );
781        assert!(
782            msg.contains("mappings"),
783            "error should mention the missing field, got: {msg}"
784        );
785    }
786
787    #[test]
788    fn map_with_wrong_input_shape_gives_clear_error() {
789        let err = parse(json!({
790            "name": "map",
791            "input": { "mappings": "not an array" }
792        }))
793        .expect_err("map with bad mappings type should fail");
794        let msg = err.to_string();
795        assert!(
796            msg.starts_with("config for function 'map':"),
797            "error should be prefixed with function envelope, got: {msg}"
798        );
799    }
800
801    #[test]
802    fn validation_accepts_both_spellings() {
803        for name in ["validate", "validation"] {
804            let cfg = parse(json!({
805                "name": name,
806                "input": { "rules": [] }
807            }))
808            .unwrap_or_else(|e| panic!("'{name}' should deserialize: {e}"));
809            assert!(matches!(cfg, FunctionConfig::Validation { .. }));
810        }
811    }
812
813    #[test]
814    fn unknown_name_falls_through_to_custom() {
815        let cfg = parse(json!({
816            "name": "my_custom_handler",
817            "input": { "anything": "goes" }
818        }))
819        .expect("unknown name should produce Custom");
820        match cfg {
821            FunctionConfig::Custom {
822                name,
823                compiled_input,
824                ..
825            } => {
826                assert_eq!(name, "my_custom_handler");
827                assert!(compiled_input.is_none());
828            }
829            other => panic!("expected Custom, got {other:?}"),
830        }
831    }
832
833    #[test]
834    fn missing_name_field_errors() {
835        let err = parse(json!({ "input": {} })).expect_err("missing name should fail");
836        assert!(err.to_string().contains("name"));
837    }
838
839    #[test]
840    fn missing_input_field_errors() {
841        let err = parse(json!({ "name": "map" })).expect_err("missing input should fail");
842        assert!(err.to_string().contains("input"));
843    }
844
845    #[test]
846    fn http_call_with_missing_connector_gives_clear_error() {
847        let err = parse(json!({
848            "name": "http_call",
849            "input": { "method": "GET" }
850        }))
851        .expect_err("http_call needs connector");
852        let msg = err.to_string();
853        assert!(
854            msg.starts_with("config for function 'http_call':"),
855            "error should be prefixed with function envelope, got: {msg}"
856        );
857        assert!(msg.contains("connector"));
858    }
859
860    #[test]
861    fn builtin_names_never_fall_through_to_custom() {
862        // Every name in BUILTIN_FUNCTION_NAMES must be handled by the
863        // dispatch — either parsing successfully or failing with the
864        // envelope. None should silently land in Custom.
865        for name in BUILTIN_FUNCTION_NAMES {
866            let cfg = parse(json!({
867                "name": name,
868                "input": {}
869            }));
870            match cfg {
871                Ok(c) => assert!(
872                    !matches!(c, FunctionConfig::Custom { .. }),
873                    "name '{name}' silently fell through to Custom"
874                ),
875                Err(e) => assert!(
876                    e.to_string()
877                        .starts_with(&format!("config for function '{name}':")),
878                    "name '{name}' failed without envelope: {e}"
879                ),
880            }
881
882            // The const and the classifier cannot drift: anything listed as a
883            // built-in must classify as one.
884            assert!(
885                builtin_function_kind(name).is_some(),
886                "name '{name}' is in BUILTIN_FUNCTION_NAMES but classifies as None"
887            );
888        }
889    }
890
891    /// Parse an `http_call` task and hand back its typed config.
892    fn parse_http_call(
893        input: serde_json::Value,
894    ) -> std::result::Result<HttpCallConfig, serde_json::Error> {
895        match parse(json!({ "name": "http_call", "input": input }))? {
896            FunctionConfig::HttpCall { input, .. } => Ok(input),
897            other => panic!("expected HttpCall, got {other:?}"),
898        }
899    }
900
901    #[test]
902    fn http_call_response_path_is_read_under_its_own_name() {
903        let cfg = parse_http_call(json!({ "connector": "c", "response_path": "data.x" }))
904            .expect("response_path should parse");
905        assert_eq!(
906            cfg.response_path.as_ref().map(Template::as_json),
907            Some(&json!("data.x"))
908        );
909    }
910
911    #[test]
912    fn http_call_response_path_accepts_the_output_alias() {
913        // This is the case that previously yielded None, silently: the request
914        // was made and the response thrown away.
915        let cfg = parse_http_call(json!({ "connector": "c", "output": "data.x" }))
916            .expect("output should be accepted as an alias");
917        assert_eq!(
918            cfg.response_path.as_ref().map(Template::as_json),
919            Some(&json!("data.x"))
920        );
921    }
922
923    #[test]
924    fn http_call_response_path_is_optional() {
925        let cfg = parse_http_call(json!({ "connector": "c" })).expect("no destination is valid");
926        assert!(cfg.response_path.is_none());
927    }
928
929    #[test]
930    fn http_call_rejects_both_destination_keys_in_either_order() {
931        // Asserting both orderings matters: a single-order test would also pass
932        // on an implementation that had an order-dependent precedence rule.
933        for input in [
934            json!({ "connector": "c", "response_path": "a", "output": "b" }),
935            json!({ "connector": "c", "output": "b", "response_path": "a" }),
936        ] {
937            let err = parse_http_call(input.clone())
938                .expect_err("supplying both destination keys must fail");
939            let msg = err.to_string();
940            assert!(
941                msg.starts_with("config for function 'http_call':"),
942                "error should carry the function envelope, got: {msg}"
943            );
944            assert!(
945                msg.contains("duplicate field"),
946                "error should name the conflict, got: {msg}"
947            );
948        }
949    }
950
951    #[test]
952    fn http_call_rejects_a_misspelled_destination_field() {
953        // The recorded decision: `HttpCallConfig` is `deny_unknown_fields`, so a
954        // near-miss spelling is a parse error naming the field rather than a
955        // silently discarded response. This is the defect this closes.
956        for bad in ["outputs", "Output", "respose_path", "response-path"] {
957            let mut input = serde_json::Map::new();
958            input.insert("connector".to_string(), json!("c"));
959            input.insert(bad.to_string(), json!("data.x"));
960
961            let err = parse_http_call(serde_json::Value::Object(input))
962                .expect_err("a misspelled field must be rejected, not silently discarded");
963            let msg = err.to_string();
964            assert!(
965                msg.starts_with("config for function 'http_call':"),
966                "error should carry the function envelope, got: {msg}"
967            );
968            assert!(
969                msg.contains("unknown field"),
970                "error should say the field is unknown, got: {msg}"
971            );
972            assert!(
973                msg.contains(bad),
974                "error should name the offending field '{bad}', got: {msg}"
975            );
976        }
977    }
978
979    #[test]
980    fn enrich_does_not_accept_the_output_alias() {
981        // The asymmetry is deliberate: only `HttpCallConfig::response_path`
982        // takes the alias. `EnrichConfig`'s destination is `merge_path`, and
983        // `deny_unknown_fields` makes the mistake loud instead of silent.
984        let err = parse(json!({
985            "name": "enrich",
986            "input": { "connector": "c", "output": "data.x" }
987        }))
988        .expect_err("enrich has no `output` field");
989        let msg = err.to_string();
990        assert!(
991            msg.starts_with("config for function 'enrich':"),
992            "error should carry the function envelope, got: {msg}"
993        );
994
995        // And the real spelling still works.
996        let ok = parse(json!({
997            "name": "enrich",
998            "input": { "connector": "c", "merge_path": "data.x" }
999        }))
1000        .expect("merge_path is enrich's destination field");
1001        assert!(matches!(ok, FunctionConfig::Enrich { .. }));
1002    }
1003
1004    #[test]
1005    fn publish_kafka_rejects_unknown_fields() {
1006        let err = parse(json!({
1007            "name": "publish_kafka",
1008            "input": { "connector": "c", "topic": "t", "tpoic": "typo" }
1009        }))
1010        .expect_err("publish_kafka should reject an unknown field");
1011        assert!(err.to_string().contains("unknown field"), "got: {err}");
1012    }
1013
1014    #[test]
1015    fn connector_is_returned_for_the_three_typed_integrations() {
1016        let cases = [
1017            (
1018                json!({ "name": "http_call", "input": { "connector": "user_service" } }),
1019                "user_service",
1020            ),
1021            (
1022                json!({ "name": "enrich",
1023                        "input": { "connector": "ref_data", "merge_path": "data.out" } }),
1024                "ref_data",
1025            ),
1026            (
1027                json!({ "name": "publish_kafka",
1028                        "input": { "connector": "events", "topic": "t" } }),
1029                "events",
1030            ),
1031        ];
1032        for (input, expected) in cases {
1033            let cfg = parse(input.clone()).expect("should parse");
1034            assert_eq!(
1035                cfg.connector().and_then(|c| c.as_static()),
1036                Some(expected),
1037                "for {input}"
1038            );
1039        }
1040    }
1041
1042    #[test]
1043    fn connector_is_none_for_every_non_connector_builtin() {
1044        // Table-driven over BUILTIN_FUNCTION_NAMES minus the three integration
1045        // names, so this cannot go stale when a built-in is added.
1046        for name in BUILTIN_FUNCTION_NAMES {
1047            if matches!(*name, "http_call" | "enrich" | "publish_kafka") {
1048                continue;
1049            }
1050            let cfg = parse(json!({ "name": name, "input": minimal_input(name) }))
1051                .unwrap_or_else(|e| panic!("'{name}' should parse: {e}"));
1052            assert!(cfg.connector().is_none(), "'{name}' names no connector");
1053        }
1054    }
1055
1056    #[test]
1057    fn connector_reads_the_custom_convention() {
1058        let cfg = parse(json!({
1059            "name": "pg_query",
1060            "input": { "connector": "pg_main", "database": "orders" }
1061        }))
1062        .unwrap();
1063        assert_eq!(cfg.connector().and_then(|c| c.as_static()), Some("pg_main"));
1064    }
1065
1066    #[test]
1067    fn connector_is_none_for_a_custom_input_without_a_string_connector() {
1068        // `Custom` accepts arbitrary input, so every one of these is reachable.
1069        for input in [
1070            json!({}),                            // key absent
1071            json!({ "connector": 7 }),            // number
1072            json!({ "connector": true }),         // bool
1073            json!({ "connector": null }),         // null
1074            json!({ "connector": ["a"] }),        // array
1075            json!({ "connector": { "n": "a" } }), // object
1076            json!([]),                            // input is not an object
1077            json!(7),                             // input is a scalar
1078        ] {
1079            let cfg = parse(json!({ "name": "my_handler", "input": input.clone() }))
1080                .unwrap_or_else(|e| panic!("custom should parse {input}: {e}"));
1081            assert_eq!(cfg.connector(), None, "for input {input}");
1082        }
1083    }
1084
1085    #[test]
1086    fn connector_returns_an_empty_name_verbatim() {
1087        // The recorded decision: the accessor reports what was authored and
1088        // never disagrees with itself across the typed and Custom arms. Whether
1089        // an empty connector is acceptable is the host's validation question.
1090        let typed = parse(json!({ "name": "http_call", "input": { "connector": "" } })).unwrap();
1091        assert_eq!(typed.connector().and_then(|c| c.as_static()), Some(""));
1092
1093        let custom = parse(json!({ "name": "x", "input": { "connector": "" } })).unwrap();
1094        assert_eq!(custom.connector().and_then(|c| c.as_static()), Some(""));
1095    }
1096
1097    #[test]
1098    fn connector_returns_a_non_ascii_name_byte_for_byte() {
1099        // A pin against a future "normalize or trim it here" change.
1100        let cfg =
1101            parse(json!({ "name": "http_call", "input": { "connector": "連携先" } })).unwrap();
1102        assert_eq!(cfg.connector().and_then(|c| c.as_static()), Some("連携先"));
1103    }
1104
1105    #[test]
1106    fn builtin_function_kind_is_none_for_non_builtins() {
1107        // Classification is exact-match, same as the deserializer dispatch.
1108        for name in [
1109            "",
1110            "__not_a_builtin__",
1111            "HTTP_CALL",    // case differs
1112            "htttp_call",   // typo
1113            "map ",         // trailing space
1114            "publish_kafk", // truncated
1115        ] {
1116            assert_eq!(
1117                builtin_function_kind(name),
1118                None,
1119                "'{name}' must not classify as a built-in"
1120            );
1121            assert!(!is_builtin_function(name));
1122        }
1123    }
1124
1125    #[test]
1126    fn builtin_kinds_partition_matches_the_sync_builtin_classifier() {
1127        // `builtin_function_kind` and `is_sync_builtin` are two hand-maintained
1128        // views of one partition, so they must agree: a SelfContained name is a
1129        // sync built-in and a RequiresHandler name is not. That
1130        // `is_sync_builtin` in turn agrees with the *dispatch* is the separate
1131        // claim `is_sync_builtin_agrees_with_arena_dispatch_for_every_builtin`
1132        // proves — by calling it, rather than by asserting it in prose here.
1133        for name in BUILTIN_FUNCTION_NAMES {
1134            let kind = builtin_function_kind(name)
1135                .unwrap_or_else(|| panic!("'{name}' must classify as a built-in"));
1136            let cfg = parse(json!({ "name": name, "input": minimal_input(name) }))
1137                .unwrap_or_else(|e| panic!("'{name}' should parse with minimal input: {e}"));
1138
1139            assert_eq!(
1140                cfg.is_sync_builtin(),
1141                matches!(kind, BuiltinKind::SelfContained),
1142                "'{name}' classifies as {kind:?} but is_sync_builtin() is {}",
1143                cfg.is_sync_builtin()
1144            );
1145        }
1146    }
1147
1148    /// `is_sync_builtin` is what `next_async_boundary` chunks the task list
1149    /// with, and `try_execute_in_arena` is what actually runs those chunks. A
1150    /// `true` that meets a `None` is the "engine bug" arm of
1151    /// `WorkflowExecutor::execute_sync_task_in_arena` — reachable only at
1152    /// runtime, on the first message that touches the task.
1153    ///
1154    /// Nothing checked that before: the partition test above compares
1155    /// `is_sync_builtin` against `builtin_function_kind`, a third hand-written
1156    /// list, and merely *asserts in a comment* that the dispatch agrees. This
1157    /// calls the dispatch instead, for every built-in plus a custom name.
1158    #[test]
1159    fn is_sync_builtin_agrees_with_arena_dispatch_for_every_builtin() {
1160        use crate::engine::compiler::LogicCompiler;
1161        use crate::engine::executor::with_arena;
1162        use crate::engine::workflow::Workflow;
1163
1164        // Every built-in, plus an unregistered custom name — the fourth
1165        // handler-backed variant, which `BUILTIN_FUNCTION_NAMES` cannot cover.
1166        let names: Vec<&str> = BUILTIN_FUNCTION_NAMES
1167            .iter()
1168            .copied()
1169            .chain(std::iter::once("some_custom_handler"))
1170            .collect();
1171
1172        for name in names {
1173            // Compile through the real pass, so every `Template` and condition
1174            // is populated exactly as it is on the live path. `LogicCompiler`
1175            // does not resolve handlers, so a config-only integration and an
1176            // unregistered custom name both compile here.
1177            let workflow = Workflow::from_json(&format!(
1178                r#"{{"id": "w", "name": "w", "priority": 0, "tasks": [
1179                    {{"id": "t", "name": "t", "function": {{"name": "{name}", "input": {}}}}}
1180                ]}}"#,
1181                minimal_input(name)
1182            ))
1183            .unwrap_or_else(|e| panic!("'{name}' should parse into a workflow: {e}"));
1184
1185            let compiler = LogicCompiler::new();
1186            let compiled = compiler
1187                .compile_workflows(vec![workflow])
1188                .unwrap_or_else(|e| panic!("'{name}' should compile: {e}"));
1189            let engine = compiler.into_engine();
1190            let function = &compiled[0].tasks[0].function;
1191
1192            let mut message = Message::from_value(&json!({}));
1193            let dispatches = with_arena(|arena| {
1194                let mut arena_ctx = ArenaContext::from_owned(&message.context, arena);
1195                // Only the `Some`/`None` decision matters. The inner `Result`
1196                // is whatever running against an empty message produces, and
1197                // is deliberately not asserted on.
1198                function
1199                    .try_execute_in_arena(&mut message, &mut arena_ctx, &engine, None)
1200                    .is_some()
1201            });
1202
1203            assert_eq!(
1204                dispatches,
1205                function.is_sync_builtin(),
1206                "'{name}': is_sync_builtin() is {} but try_execute_in_arena() \
1207                 {} — the sync stretch would hit the engine-bug arm",
1208                function.is_sync_builtin(),
1209                if dispatches { "dispatched" } else { "declined" }
1210            );
1211        }
1212    }
1213
1214    #[test]
1215    fn requires_handler_kind_covers_exactly_the_config_only_integrations() {
1216        // The three names documented as shipping config-only.
1217        for name in ["http_call", "enrich", "publish_kafka"] {
1218            assert_eq!(
1219                builtin_function_kind(name),
1220                Some(BuiltinKind::RequiresHandler),
1221                "'{name}' ships as config only and needs a registered handler"
1222            );
1223        }
1224
1225        // Both accepted spellings of validation are self-contained. The
1226        // deserializer takes either; `function_name()` only ever returns
1227        // "validate", so checking one spelling would miss a regression.
1228        for name in [
1229            "map",
1230            "validation",
1231            "validate",
1232            "parse_json",
1233            "parse_xml",
1234            "publish_json",
1235            "publish_xml",
1236            "filter",
1237            "log",
1238        ] {
1239            assert_eq!(
1240                builtin_function_kind(name),
1241                Some(BuiltinKind::SelfContained),
1242                "'{name}' is executed by this crate"
1243            );
1244        }
1245    }
1246}
1247
1248#[cfg(test)]
1249mod dispatch_vocabulary_tests {
1250    use super::*;
1251    use std::collections::HashMap;
1252
1253    /// A registry whose values are irrelevant — every function here keys off
1254    /// name membership only.
1255    fn registry(names: &[&str]) -> HashMap<String, ()> {
1256        names.iter().map(|n| ((*n).to_string(), ())).collect()
1257    }
1258
1259    fn names(registry: &HashMap<String, ()>) -> Vec<&str> {
1260        let mut out: Vec<&str> = dispatchable_functions_in(registry)
1261            .map(|f| f.name)
1262            .collect();
1263        out.sort_unstable();
1264        out
1265    }
1266
1267    /// Acceptance criterion: the enumeration accounts for every name in
1268    /// `BUILTIN_FUNCTION_NAMES` exactly once, aliases grouped.
1269    ///
1270    /// This is the drift net for `canonical_builtin_name` / `builtin_aliases`.
1271    /// Adding a built-in without teaching them about it fails here rather than
1272    /// silently dropping the name from every host's vocabulary.
1273    #[test]
1274    fn aliases_and_canonical_names_agree() {
1275        // Every canonical name is its own canonical form, and every alias
1276        // resolves to a name that is.
1277        for name in BUILTIN_FUNCTION_NAMES {
1278            let canonical = canonical_builtin_name(name);
1279            assert_eq!(
1280                canonical_builtin_name(canonical),
1281                canonical,
1282                "'{name}' resolves to '{canonical}', which must itself be canonical"
1283            );
1284            assert!(
1285                BUILTIN_FUNCTION_NAMES.contains(&canonical),
1286                "'{canonical}' is a canonical name and must be an accepted spelling"
1287            );
1288            // An alias shares its canonical name's kind — both spellings
1289            // deserialize to the same variant.
1290            assert_eq!(
1291                builtin_function_kind(name),
1292                builtin_function_kind(canonical),
1293                "'{name}' and '{canonical}' are one function and must classify alike"
1294            );
1295        }
1296
1297        // Each name is either a canonical entry or an alias of exactly one —
1298        // never both, never neither.
1299        for name in BUILTIN_FUNCTION_NAMES {
1300            let is_canonical = canonical_builtin_name(name) == *name;
1301            let alias_of: Vec<&str> = BUILTIN_FUNCTION_NAMES
1302                .iter()
1303                .copied()
1304                .filter(|c| builtin_aliases(c).contains(name))
1305                .collect();
1306            assert_eq!(
1307                is_canonical,
1308                alias_of.is_empty(),
1309                "'{name}' must be canonical XOR an alias, got canonical={is_canonical} \
1310                 listed-as-alias-of={alias_of:?}"
1311            );
1312            assert!(
1313                alias_of.len() <= 1,
1314                "'{name}' is listed as an alias of more than one function: {alias_of:?}"
1315            );
1316        }
1317
1318        // And the two directions agree: every alias listed is a real spelling.
1319        for canonical in BUILTIN_FUNCTION_NAMES {
1320            for alias in builtin_aliases(canonical) {
1321                assert_eq!(
1322                    canonical_builtin_name(alias),
1323                    *canonical,
1324                    "'{alias}' is listed under '{canonical}' but does not resolve to it"
1325                );
1326            }
1327        }
1328    }
1329
1330    #[test]
1331    fn validate_is_canonical_and_validation_is_its_alias() {
1332        assert_eq!(canonical_builtin_name("validation"), "validate");
1333        assert_eq!(canonical_builtin_name("validate"), "validate");
1334        assert_eq!(builtin_aliases("validate"), &["validation"]);
1335        assert!(builtin_aliases("validation").is_empty());
1336        assert!(builtin_aliases("map").is_empty());
1337    }
1338
1339    #[test]
1340    fn an_empty_registry_dispatches_every_self_contained_builtin() {
1341        assert_eq!(
1342            names(&registry(&[])),
1343            vec![
1344                "filter",
1345                "log",
1346                "map",
1347                "parse_json",
1348                "parse_xml",
1349                "publish_json",
1350                "publish_xml",
1351                "validate",
1352            ],
1353            "self-contained built-ins need no registration; `validation` is \
1354             folded into `validate`, and the three config-only integrations are absent"
1355        );
1356    }
1357
1358    #[test]
1359    fn requires_handler_builtins_appear_only_when_registered() {
1360        let empty = registry(&[]);
1361        assert!(!names(&empty).contains(&"enrich"));
1362        assert!(!can_dispatch_in(&empty, "enrich"));
1363
1364        let backed = registry(&["enrich"]);
1365        assert!(names(&backed).contains(&"enrich"));
1366        assert!(can_dispatch_in(&backed, "enrich"));
1367
1368        // …and it keeps its built-in classification rather than reading as custom.
1369        let entry = dispatchable_functions_in(&backed)
1370            .find(|f| f.name == "enrich")
1371            .expect("registered enrich is enumerated");
1372        assert_eq!(entry.kind, Some(BuiltinKind::RequiresHandler));
1373    }
1374
1375    #[test]
1376    fn custom_names_are_enumerated_with_no_kind() {
1377        let reg = registry(&["shout"]);
1378        let entry = dispatchable_functions_in(&reg)
1379            .find(|f| f.name == "shout")
1380            .expect("a registered custom name is enumerated");
1381        assert_eq!(entry.kind, None, "None is how a custom handler reports");
1382        assert!(entry.aliases.is_empty());
1383        assert!(can_dispatch_in(&reg, "shout"));
1384        assert!(!can_dispatch_in(&registry(&[]), "shout"));
1385    }
1386
1387    #[test]
1388    fn registering_a_self_contained_name_is_inert_and_never_duplicates_it() {
1389        // Registering under `map` does nothing: the deserializer routes `map`
1390        // to FunctionConfig::Map, which the crate executes itself. The name is
1391        // reported once either way.
1392        let shadowed = registry(&["map"]);
1393        assert_eq!(
1394            names(&shadowed),
1395            names(&registry(&[])),
1396            "a shadowing registration changes nothing about the vocabulary"
1397        );
1398        assert_eq!(
1399            dispatchable_functions_in(&shadowed)
1400                .filter(|f| f.name == "map")
1401                .count(),
1402            1,
1403            "`map` is yielded exactly once, not once per source"
1404        );
1405    }
1406
1407    #[test]
1408    fn aliases_dispatch_but_are_not_enumerated() {
1409        let reg = registry(&[]);
1410        assert!(
1411            can_dispatch_in(&reg, "validation"),
1412            "a task named `validation` really does execute"
1413        );
1414        assert!(
1415            !names(&reg).contains(&"validation"),
1416            "but the enumeration reports it under `validate`"
1417        );
1418    }
1419
1420    #[test]
1421    fn can_dispatch_rejects_names_the_crate_does_not_know() {
1422        let reg = registry(&["shout"]);
1423        assert!(!can_dispatch_in(&reg, "SHOUT"), "matching is exact");
1424        assert!(!can_dispatch_in(&reg, "htttp_call"));
1425        assert!(!can_dispatch_in(&reg, ""));
1426    }
1427
1428    /// The predicate and the enumeration must describe the same set — with the
1429    /// one documented exception that `can_dispatch` also accepts aliases.
1430    #[test]
1431    fn every_enumerated_name_is_dispatchable() {
1432        let reg = registry(&["enrich", "shout"]);
1433        for f in dispatchable_functions_in(&reg) {
1434            assert!(
1435                can_dispatch_in(&reg, f.name),
1436                "'{}' is enumerated, so it must dispatch",
1437                f.name
1438            );
1439            for alias in f.aliases {
1440                assert!(
1441                    can_dispatch_in(&reg, alias),
1442                    "alias '{alias}' of '{}' must dispatch too",
1443                    f.name
1444                );
1445            }
1446        }
1447    }
1448}