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#[derive(Clone)]
30pub struct CompiledCustomInput(pub Arc<dyn Any + Send + Sync>);
31
32impl CompiledCustomInput {
33 #[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#[derive(Debug, Clone, Copy, PartialEq, Eq)]
55pub enum ConnectorName<'a> {
56 Static(&'a str),
58 Computed(&'a Value),
61}
62
63impl<'a> ConnectorName<'a> {
64 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 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#[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 Custom {
142 name: String,
143 input: Value,
144 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
220pub 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#[derive(Debug, Clone, Copy, PartialEq, Eq)]
265pub enum BuiltinKind {
266 SelfContained,
268 RequiresHandler,
277}
278
279pub 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#[inline]
310pub fn is_builtin_function(name: &str) -> bool {
311 builtin_function_kind(name).is_some()
312}
313
314#[derive(Debug, Clone, PartialEq, Eq)]
322pub struct DispatchableFunction<'a> {
323 pub name: &'a str,
326 pub kind: Option<BuiltinKind>,
335 pub aliases: &'static [&'static str],
342}
343
344const VALIDATE_ALIASES: &[&str] = &["validation"];
348
349const NO_ALIASES: &[&str] = &[];
352
353pub(crate) fn canonical_builtin_name(name: &str) -> &str {
360 match name {
361 "validation" => "validate",
362 other => other,
363 }
364}
365
366pub(crate) fn builtin_aliases(canonical: &str) -> &'static [&'static str] {
370 match canonical {
371 "validate" => VALIDATE_ALIASES,
372 _ => NO_ALIASES,
373 }
374}
375
376pub(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 _ => registry.contains_key(name),
395 }
396}
397
398pub(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 .filter(|name| canonical_builtin_name(name) == *name)
417 .filter_map(move |name| match builtin_function_kind(name) {
418 kind @ Some(BuiltinKind::SelfContained) => Some(DispatchableFunction {
420 name,
421 kind,
422 aliases: builtin_aliases(name),
423 }),
424 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 .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
450fn 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 #[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
545fn 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 for input in [
1070 json!({}), json!({ "connector": 7 }), json!({ "connector": true }), json!({ "connector": null }), json!({ "connector": ["a"] }), json!({ "connector": { "n": "a" } }), json!([]), json!(7), ] {
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 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 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 for name in [
1109 "",
1110 "__not_a_builtin__",
1111 "HTTP_CALL", "htttp_call", "map ", "publish_kafk", ] {
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 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 #[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 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 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 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 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 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 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 #[test]
1274 fn aliases_and_canonical_names_agree() {
1275 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 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 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 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(®istry(&[])),
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 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(®)
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(®, "shout"));
1384 assert!(!can_dispatch_in(®istry(&[]), "shout"));
1385 }
1386
1387 #[test]
1388 fn registering_a_self_contained_name_is_inert_and_never_duplicates_it() {
1389 let shadowed = registry(&["map"]);
1393 assert_eq!(
1394 names(&shadowed),
1395 names(®istry(&[])),
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(®, "validation"),
1412 "a task named `validation` really does execute"
1413 );
1414 assert!(
1415 !names(®).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(®, "SHOUT"), "matching is exact");
1424 assert!(!can_dispatch_in(®, "htttp_call"));
1425 assert!(!can_dispatch_in(®, ""));
1426 }
1427
1428 #[test]
1431 fn every_enumerated_name_is_dispatchable() {
1432 let reg = registry(&["enrich", "shout"]);
1433 for f in dispatchable_functions_in(®) {
1434 assert!(
1435 can_dispatch_in(®, 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(®, alias),
1442 "alias '{alias}' of '{}' must dispatch too",
1443 f.name
1444 );
1445 }
1446 }
1447 }
1448}