Skip to main content

faucet_cli/
transforms.rs

1//! Compile YAML/JSON transform declarations into `TransformStage` values.
2//!
3//! The built-in transforms are exposed via config — custom closure
4//! transforms require Rust code and are reserved for the library API.
5
6use crate::config::TransformSpec;
7use crate::error::{CliError, CliResult};
8#[cfg(feature = "transforms")]
9use faucet_core::{
10    CastOnError, CastType, HashAlgorithm, HashEncoding, JsonParseOnError, KeyCaseMode,
11    ValueCaseMode,
12};
13#[cfg(any(feature = "transforms", feature = "transform-cdc-unwrap"))]
14use faucet_core::{JsonSchema, schema_for};
15use faucet_core::{RecordTransform, TransformStage};
16#[cfg(any(feature = "transforms", feature = "transform-cdc-unwrap"))]
17use serde::Deserialize;
18use serde_json::Value;
19#[cfg(feature = "transforms")]
20use std::collections::HashMap;
21
22/// Inline-config schema for the `flatten` transform.
23#[cfg(feature = "transforms")]
24#[derive(Debug, Deserialize, JsonSchema)]
25struct FlattenConfig {
26    /// Separator joining nested keys (default: `"__"`).
27    #[serde(default = "default_separator")]
28    separator: String,
29}
30
31#[cfg(feature = "transforms")]
32fn default_separator() -> String {
33    "__".to_owned()
34}
35
36/// Inline-config schema for the `rename_keys` transform.
37#[cfg(feature = "transforms")]
38#[derive(Debug, Deserialize, JsonSchema)]
39struct RenameKeysConfig {
40    /// Rust regex matched against every key.
41    pattern: String,
42    /// Replacement string. May reference capture groups (`$1`, `${name}`).
43    replacement: String,
44}
45
46#[cfg(feature = "transforms")]
47#[derive(Debug, Deserialize, JsonSchema)]
48struct FieldsConfig {
49    /// Top-level field names to act on.
50    fields: Vec<String>,
51}
52
53#[cfg(feature = "transforms")]
54#[derive(Debug, Deserialize, JsonSchema)]
55struct SetConfig {
56    /// Map of field name → constant value to set on every record.
57    values: serde_json::Map<String, Value>,
58}
59
60#[cfg(feature = "transforms")]
61#[derive(Debug, Deserialize, JsonSchema)]
62struct RenameFieldConfig {
63    /// Map of old field name → new field name.
64    fields: HashMap<String, String>,
65}
66
67#[cfg(feature = "transforms")]
68#[derive(Debug, Deserialize, JsonSchema)]
69struct CastConfig {
70    /// Map of field name → target type.
71    fields: HashMap<String, CastType>,
72    /// What to do when a value cannot be cast. Default: `error`.
73    #[serde(default)]
74    on_error: CastOnError,
75}
76
77#[cfg(feature = "transforms")]
78#[derive(Debug, Deserialize, JsonSchema)]
79struct RedactConfig {
80    /// Top-level field names to overwrite with `mask`.
81    fields: Vec<String>,
82    /// Replacement value. Default: the string `"***"`.
83    #[serde(default = "default_mask")]
84    mask: Value,
85}
86
87#[cfg(feature = "transforms")]
88fn default_mask() -> Value {
89    Value::String("***".to_owned())
90}
91
92#[cfg(feature = "transforms")]
93#[derive(Debug, Deserialize, JsonSchema)]
94struct ValueCaseConfig {
95    /// String-valued fields to re-case.
96    fields: Vec<String>,
97    /// Casing convention to apply to each listed field.
98    mode: ValueCaseMode,
99}
100
101#[cfg(feature = "transforms")]
102#[derive(Debug, Deserialize, JsonSchema)]
103struct SpellSymbolsConfig {
104    /// Extra symbol → word overrides layered on top of the built-in map.
105    #[serde(default)]
106    extra: HashMap<String, String>,
107    /// String inserted between expanded words. Default: a single space.
108    #[serde(default = "default_spell_separator")]
109    separator: String,
110}
111
112#[cfg(feature = "transforms")]
113fn default_spell_separator() -> String {
114    " ".to_owned()
115}
116
117#[cfg(feature = "transforms")]
118#[derive(Debug, Deserialize, JsonSchema)]
119struct KeysCaseConfig {
120    /// Output convention for every key in the record.
121    mode: KeyCaseMode,
122}
123
124#[cfg(feature = "transforms")]
125#[derive(Debug, Deserialize, JsonSchema)]
126struct HashConfig {
127    /// One or more field names to hash. When `into` is set, exactly one.
128    fields: Vec<String>,
129    /// Hash algorithm. Default: `sha256`.
130    #[serde(default)]
131    algorithm: HashAlgorithm,
132    /// Digest encoding. Default: `hex`.
133    #[serde(default)]
134    encoding: HashEncoding,
135    /// Optional salt prepended before hashing. Recommend `${env:...}`.
136    #[serde(default, skip_serializing_if = "Option::is_none")]
137    salt: Option<String>,
138    /// Optional target key for the digest (source left intact). `null` =
139    /// replace in place. Only valid with a single field.
140    #[serde(default, skip_serializing_if = "Option::is_none")]
141    into: Option<String>,
142}
143
144#[cfg(feature = "transforms")]
145#[derive(Debug, Deserialize, JsonSchema)]
146struct JsonParseConfig {
147    /// Field names holding stringified JSON to parse.
148    fields: Vec<String>,
149    /// What to do when a value is a string but not valid JSON. Default: `keep`.
150    #[serde(default)]
151    on_error: JsonParseOnError,
152    /// Optional target key for the parsed value. `null` = replace in place.
153    /// Only valid with a single field.
154    #[serde(default, skip_serializing_if = "Option::is_none")]
155    into: Option<String>,
156}
157
158#[cfg(feature = "transforms")]
159#[derive(Debug, Deserialize, JsonSchema)]
160struct CoalesceConfig {
161    /// Target field to fill when missing or null.
162    field: String,
163    /// Literal default value. Mutually exclusive with `from`.
164    #[serde(default, skip_serializing_if = "Option::is_none")]
165    default: Option<Value>,
166    /// Fallback keys; the first non-null wins. Mutually exclusive with `default`.
167    #[serde(default)]
168    from: Vec<String>,
169    /// Treat an empty string as null. Default: `false`.
170    #[serde(default)]
171    treat_empty_string_as_null: bool,
172}
173
174#[cfg(feature = "transforms")]
175#[derive(Debug, Deserialize, JsonSchema)]
176struct SplitConfig {
177    /// String field to split into an array.
178    field: String,
179    /// Delimiter to split on.
180    delimiter: String,
181    /// Trim whitespace from each element. Default: `false`.
182    #[serde(default)]
183    trim: bool,
184    /// Optional target key for the array. `null` = replace in place.
185    #[serde(default, skip_serializing_if = "Option::is_none")]
186    into: Option<String>,
187}
188
189#[cfg(feature = "transforms")]
190#[derive(Debug, Deserialize, JsonSchema)]
191struct JoinConfig {
192    /// Array field to join into a string.
193    field: String,
194    /// Delimiter placed between elements.
195    delimiter: String,
196    /// Optional target key for the string. `null` = replace in place.
197    #[serde(default, skip_serializing_if = "Option::is_none")]
198    into: Option<String>,
199}
200
201#[cfg(feature = "transform-filter")]
202#[derive(Debug, Deserialize, JsonSchema)]
203struct FilterConfig {
204    /// JSONPath subset: bare key, dot path, or bracketed string key.
205    path: String,
206    /// One of `eq`, `ne`, `exists`, `in`, `not_in`.
207    op: faucet_core::FilterOp,
208    /// Required for `eq`/`ne`/`in`/`not_in`. For `in`/`not_in`, must be an array.
209    #[serde(default, skip_serializing_if = "Option::is_none")]
210    value: Option<Value>,
211}
212
213#[cfg(feature = "transform-explode")]
214#[derive(Debug, Deserialize, JsonSchema)]
215struct ExplodeConfig {
216    /// JSONPath subset: bare key, dot path, or bracketed string key.
217    path: String,
218    /// Prefix prepended to object-element fields. Defaults to the last
219    /// segment of `path`. Empty string = pure LATERAL FLATTEN (no prefix).
220    #[serde(default, skip_serializing_if = "Option::is_none")]
221    prefix: Option<String>,
222    /// Separator between prefix and element field key. Default `"_"`.
223    #[serde(default = "default_explode_separator_cli")]
224    separator: String,
225    /// `passthrough` (default), `drop`, or `error` when path doesn't yield a
226    /// non-empty array.
227    #[serde(default)]
228    on_missing: faucet_core::OnMissing,
229}
230
231#[cfg(feature = "transform-explode")]
232fn default_explode_separator_cli() -> String {
233    "_".to_owned()
234}
235
236#[cfg(feature = "transform-cdc-unwrap")]
237#[derive(Debug, Deserialize, JsonSchema)]
238struct CdcUnwrapConfig {
239    /// Envelope field holding the operation code. Default `"op"`.
240    #[serde(default = "cdc_op_field")]
241    op_field: String,
242    /// Envelope field holding the post-image row. Default `"after"`.
243    #[serde(default = "cdc_after_field")]
244    after_field: String,
245    /// Envelope field holding the pre-image row. Default `"before"`.
246    #[serde(default = "cdc_before_field")]
247    before_field: String,
248    /// Fallback key field for deletes when `before` is absent. Default `"document_key"`.
249    #[serde(default = "cdc_key_field")]
250    key_field: String,
251    /// Field stamped onto every emitted row with the op value. Default `"__op"`.
252    #[serde(default = "cdc_marker_field")]
253    marker_field: String,
254    /// Op values that mean delete. Default `["d", "delete"]`.
255    #[serde(default = "cdc_delete_ops")]
256    delete_ops: Vec<String>,
257    /// Op values that cause the record to be dropped (1→0). Default `["ddl", "truncate"]`.
258    #[serde(default = "cdc_drop_ops")]
259    drop_ops: Vec<String>,
260}
261
262#[cfg(feature = "transform-cdc-unwrap")]
263fn cdc_op_field() -> String {
264    "op".into()
265}
266#[cfg(feature = "transform-cdc-unwrap")]
267fn cdc_after_field() -> String {
268    "after".into()
269}
270#[cfg(feature = "transform-cdc-unwrap")]
271fn cdc_before_field() -> String {
272    "before".into()
273}
274#[cfg(feature = "transform-cdc-unwrap")]
275fn cdc_key_field() -> String {
276    "document_key".into()
277}
278#[cfg(feature = "transform-cdc-unwrap")]
279fn cdc_marker_field() -> String {
280    "__op".into()
281}
282#[cfg(feature = "transform-cdc-unwrap")]
283fn cdc_delete_ops() -> Vec<String> {
284    vec!["d".into(), "delete".into()]
285}
286#[cfg(feature = "transform-cdc-unwrap")]
287fn cdc_drop_ops() -> Vec<String> {
288    vec!["ddl".into(), "truncate".into()]
289}
290
291/// One row in the transform registry — the single source of truth for every
292/// built-in transform's kind, one-line description, JSON Schema, and
293/// `TransformSpec → TransformStage` decoder. `compile_one`,
294/// `transform_descriptions`, and `transform_schema` all read from this list
295/// so adding a new transform means appending one entry (no parallel match
296/// arms to keep in sync).
297struct TransformDef {
298    kind: &'static str,
299    description: &'static str,
300    schema_fn: fn() -> Value,
301    compile_fn: fn(&str, Value) -> CliResult<TransformStage>,
302}
303
304/// Every transform compiled into this build, in display order.
305///
306/// Non-capturing closures coerce to the `fn` pointers held by `TransformDef`,
307/// so each row stays a single self-contained record next to its sibling
308/// entries.
309fn registry() -> Vec<TransformDef> {
310    #[allow(unused_mut)]
311    let mut defs: Vec<TransformDef> = Vec::new();
312    #[cfg(feature = "transforms")]
313    {
314        defs.extend(vec![
315            TransformDef {
316                kind: "flatten",
317                description: "Flatten nested objects into a single level (configurable separator).",
318                schema_fn: || schema::<FlattenConfig>(),
319                compile_fn: |kind, config| {
320                    let cfg = decode::<FlattenConfig>(kind, config)?;
321                    Ok(TransformStage::Map(RecordTransform::Flatten {
322                        separator: cfg.separator,
323                    }))
324                },
325            },
326            TransformDef {
327                kind: "rename_keys",
328                description: "Rewrite every key via a regex pattern + replacement.",
329                schema_fn: || schema::<RenameKeysConfig>(),
330                compile_fn: |kind, config| {
331                    let cfg = decode::<RenameKeysConfig>(kind, config)?;
332                    Ok(TransformStage::Map(RecordTransform::RenameKeys {
333                        pattern: cfg.pattern,
334                        replacement: cfg.replacement,
335                    }))
336                },
337            },
338            TransformDef {
339                kind: "keys_case",
340                description: "Re-case every key (snake / camel / pascal / kebab / screaming_snake).",
341                schema_fn: || schema::<KeysCaseConfig>(),
342                compile_fn: |kind, config| {
343                    let cfg = decode::<KeysCaseConfig>(kind, config)?;
344                    Ok(TransformStage::Map(RecordTransform::KeysCase {
345                        mode: cfg.mode,
346                    }))
347                },
348            },
349            TransformDef {
350                kind: "select",
351                description: "Keep only the listed top-level fields; drop the rest.",
352                schema_fn: || schema::<FieldsConfig>(),
353                compile_fn: |kind, config| {
354                    let cfg = decode::<FieldsConfig>(kind, config)?;
355                    Ok(TransformStage::Map(RecordTransform::Select {
356                        fields: cfg.fields,
357                    }))
358                },
359            },
360            TransformDef {
361                kind: "drop",
362                description: "Remove the listed top-level fields.",
363                schema_fn: || schema::<FieldsConfig>(),
364                compile_fn: |kind, config| {
365                    let cfg = decode::<FieldsConfig>(kind, config)?;
366                    Ok(TransformStage::Map(RecordTransform::Drop {
367                        fields: cfg.fields,
368                    }))
369                },
370            },
371            TransformDef {
372                kind: "set",
373                description: "Set named fields to constant values on every record.",
374                schema_fn: || schema::<SetConfig>(),
375                compile_fn: |kind, config| {
376                    let cfg = decode::<SetConfig>(kind, config)?;
377                    Ok(TransformStage::Map(RecordTransform::Set {
378                        values: cfg.values,
379                    }))
380                },
381            },
382            TransformDef {
383                kind: "rename_field",
384                description: "Rename specific top-level fields by name.",
385                schema_fn: || schema::<RenameFieldConfig>(),
386                compile_fn: |kind, config| {
387                    let cfg = decode::<RenameFieldConfig>(kind, config)?;
388                    Ok(TransformStage::Map(RecordTransform::RenameField {
389                        fields: cfg.fields,
390                    }))
391                },
392            },
393            TransformDef {
394                kind: "cast",
395                description: "Coerce named fields to int / float / bool / string / timestamp.",
396                schema_fn: || schema::<CastConfig>(),
397                compile_fn: |kind, config| {
398                    let cfg = decode::<CastConfig>(kind, config)?;
399                    Ok(TransformStage::Map(RecordTransform::Cast {
400                        fields: cfg.fields,
401                        on_error: cfg.on_error,
402                    }))
403                },
404            },
405            TransformDef {
406                kind: "redact",
407                description: "Overwrite the listed fields with a mask value (default `***`).",
408                schema_fn: || schema::<RedactConfig>(),
409                compile_fn: |kind, config| {
410                    let cfg = decode::<RedactConfig>(kind, config)?;
411                    Ok(TransformStage::Map(RecordTransform::Redact {
412                        fields: cfg.fields,
413                        mask: cfg.mask,
414                    }))
415                },
416            },
417            TransformDef {
418                kind: "value_case",
419                description: "Lowercase, uppercase, or trim the value of named string fields.",
420                schema_fn: || schema::<ValueCaseConfig>(),
421                compile_fn: |kind, config| {
422                    let cfg = decode::<ValueCaseConfig>(kind, config)?;
423                    Ok(TransformStage::Map(RecordTransform::ValueCase {
424                        fields: cfg.fields,
425                        mode: cfg.mode,
426                    }))
427                },
428            },
429            TransformDef {
430                kind: "spell_symbols",
431                description: "Replace punctuation/symbols in string values with their spelled-out words.",
432                schema_fn: || schema::<SpellSymbolsConfig>(),
433                compile_fn: |kind, config| {
434                    let cfg = decode::<SpellSymbolsConfig>(kind, config)?;
435                    Ok(TransformStage::Map(RecordTransform::SpellSymbols {
436                        extra: cfg.extra,
437                        separator: cfg.separator,
438                    }))
439                },
440            },
441            TransformDef {
442                kind: "hash",
443                description: "Hash listed fields (SHA-256 / BLAKE3) into stable, join-able tokens.",
444                schema_fn: || schema::<HashConfig>(),
445                compile_fn: |kind, config| {
446                    let cfg = decode::<HashConfig>(kind, config)?;
447                    let stage = TransformStage::Map(RecordTransform::Hash {
448                        fields: cfg.fields,
449                        algorithm: cfg.algorithm,
450                        encoding: cfg.encoding,
451                        salt: cfg.salt,
452                        into: cfg.into,
453                    });
454                    validate_stage(kind, &stage)?;
455                    Ok(stage)
456                },
457            },
458            TransformDef {
459                kind: "json_parse",
460                description: "Parse a stringified-JSON field into a real nested JSON value.",
461                schema_fn: || schema::<JsonParseConfig>(),
462                compile_fn: |kind, config| {
463                    let cfg = decode::<JsonParseConfig>(kind, config)?;
464                    let stage = TransformStage::Map(RecordTransform::JsonParse {
465                        fields: cfg.fields,
466                        on_error: cfg.on_error,
467                        into: cfg.into,
468                    });
469                    validate_stage(kind, &stage)?;
470                    Ok(stage)
471                },
472            },
473            TransformDef {
474                kind: "coalesce",
475                description: "Fill a missing/null field from a default or first non-null fallback key.",
476                schema_fn: || schema::<CoalesceConfig>(),
477                compile_fn: |kind, config| {
478                    let cfg = decode::<CoalesceConfig>(kind, config)?;
479                    let stage = TransformStage::Map(RecordTransform::Coalesce {
480                        field: cfg.field,
481                        default: cfg.default,
482                        from: cfg.from,
483                        treat_empty_string_as_null: cfg.treat_empty_string_as_null,
484                    });
485                    validate_stage(kind, &stage)?;
486                    Ok(stage)
487                },
488            },
489            TransformDef {
490                kind: "split",
491                description: "Split a string field into an array on a delimiter.",
492                schema_fn: || schema::<SplitConfig>(),
493                compile_fn: |kind, config| {
494                    let cfg = decode::<SplitConfig>(kind, config)?;
495                    Ok(TransformStage::Map(RecordTransform::Split {
496                        field: cfg.field,
497                        delimiter: cfg.delimiter,
498                        trim: cfg.trim,
499                        into: cfg.into,
500                    }))
501                },
502            },
503            TransformDef {
504                kind: "join",
505                description: "Join an array field into a string with a delimiter.",
506                schema_fn: || schema::<JoinConfig>(),
507                compile_fn: |kind, config| {
508                    let cfg = decode::<JoinConfig>(kind, config)?;
509                    Ok(TransformStage::Map(RecordTransform::Join {
510                        field: cfg.field,
511                        delimiter: cfg.delimiter,
512                        into: cfg.into,
513                    }))
514                },
515            },
516            #[cfg(feature = "transform-filter")]
517            TransformDef {
518                kind: "filter",
519                description: "Keep records where a JSONPath predicate is true.",
520                schema_fn: || schema::<FilterConfig>(),
521                compile_fn: |kind, config| {
522                    let cfg = decode::<FilterConfig>(kind, config)?;
523                    // Re-use stage's compile-time validation so error messages match.
524                    let stage = TransformStage::Filter(faucet_core::FilterSpec {
525                        path: cfg.path,
526                        op: cfg.op,
527                        value: cfg.value,
528                    });
529                    faucet_core::compile_stage(&stage).map_err(|e| match e {
530                        faucet_core::FaucetError::Transform(msg) => CliError::InvalidTransform {
531                            name: kind.to_owned(),
532                            message: msg,
533                        },
534                        other => CliError::InvalidTransform {
535                            name: kind.to_owned(),
536                            message: format!("{other}"),
537                        },
538                    })?;
539                    Ok(stage)
540                },
541            },
542            #[cfg(feature = "transform-explode")]
543            TransformDef {
544                kind: "explode",
545                description: "Expand an array field into one record per element.",
546                schema_fn: || schema::<ExplodeConfig>(),
547                compile_fn: |kind, config| {
548                    let cfg = decode::<ExplodeConfig>(kind, config)?;
549                    let stage = TransformStage::Explode(faucet_core::ExplodeSpec {
550                        path: cfg.path,
551                        prefix: cfg.prefix,
552                        separator: cfg.separator,
553                        on_missing: cfg.on_missing,
554                    });
555                    faucet_core::compile_stage(&stage).map_err(|e| match e {
556                        faucet_core::FaucetError::Transform(msg) => CliError::InvalidTransform {
557                            name: kind.to_owned(),
558                            message: msg,
559                        },
560                        other => CliError::InvalidTransform {
561                            name: kind.to_owned(),
562                            message: format!("{other}"),
563                        },
564                    })?;
565                    Ok(stage)
566                },
567            },
568        ]);
569    }
570    #[cfg(feature = "transform-sql")]
571    {
572        defs.push(TransformDef {
573            kind: "sql",
574            description: "Run DuckDB SQL over the whole page; records are the `batch` relation.",
575            schema_fn: || schema_sql(),
576            compile_fn: |kind, config| {
577                let cfg: faucet_transform_sql::SqlTransformConfig = decode_sql(kind, config)?;
578                let transform = faucet_transform_sql::SqlTransform::compile(&cfg).map_err(|e| {
579                    let message = match &e {
580                        faucet_core::FaucetError::Transform(m)
581                        | faucet_core::FaucetError::Config(m) => m.clone(),
582                        other => format!("{other}"),
583                    };
584                    CliError::InvalidTransform {
585                        name: kind.to_owned(),
586                        message,
587                    }
588                })?;
589                Ok(transform.into_page_stage())
590            },
591        });
592    }
593    #[cfg(feature = "transform-wasm")]
594    {
595        defs.push(TransformDef {
596            kind: "wasm",
597            description: "Run a user-provided sandboxed .wasm module over each record (wasmtime).",
598            schema_fn: || schema_wasm(),
599            compile_fn: |kind, config| {
600                let cfg: faucet_transform_wasm::WasmTransformConfig = decode_wasm(kind, config)?;
601                let transform =
602                    faucet_transform_wasm::WasmTransform::compile(&cfg).map_err(|e| {
603                        let message = match &e {
604                            faucet_core::FaucetError::Transform(m)
605                            | faucet_core::FaucetError::Config(m) => m.clone(),
606                            other => format!("{other}"),
607                        };
608                        CliError::InvalidTransform {
609                            name: kind.to_owned(),
610                            message,
611                        }
612                    })?;
613                Ok(transform.into_page_stage())
614            },
615        });
616    }
617    #[cfg(feature = "transform-cdc-unwrap")]
618    {
619        defs.push(TransformDef {
620            kind: "cdc_unwrap",
621            description: "Normalize a CDC envelope into a flat row + delete marker (for upsert sinks).",
622            schema_fn: || schema::<CdcUnwrapConfig>(),
623            compile_fn: |kind, config| {
624                let cfg = decode::<CdcUnwrapConfig>(kind, config)?;
625                Ok(faucet_core::TransformStage::CdcUnwrap(faucet_core::CdcUnwrapSpec {
626                    op_field: cfg.op_field,
627                    after_field: cfg.after_field,
628                    before_field: cfg.before_field,
629                    key_field: cfg.key_field,
630                    marker_field: cfg.marker_field,
631                    delete_ops: cfg.delete_ops,
632                    drop_ops: cfg.drop_ops,
633                }))
634            },
635        });
636    }
637    defs
638}
639
640/// Compile a list of [`TransformSpec`]s into [`TransformStage`]s in the
641/// declared order. Most built-ins compile to a [`TransformStage::Map`];
642/// richer stages (e.g. `filter`, future fan-outs) compile to other
643/// variants. Unknown or malformed entries surface as a `CliError`.
644pub fn compile_transforms(specs: &[TransformSpec]) -> CliResult<Vec<TransformStage>> {
645    let mut out = Vec::with_capacity(specs.len());
646    for s in specs {
647        out.push(compile_one(s)?);
648    }
649    Ok(out)
650}
651
652fn compile_one(spec: &TransformSpec) -> CliResult<TransformStage> {
653    match registry().into_iter().find(|t| t.kind == spec.kind) {
654        Some(def) => (def.compile_fn)(&spec.kind, spec.config.clone()),
655        None => Err(unknown_transform(&spec.kind)),
656    }
657}
658
659/// Like [`compile_transforms`] but also returns each stage's Arrow
660/// `RecordBatch` form (parallel to the stages), so the executor can build a
661/// columnar-capable `TransformingSource` (#375). Only the `sql` transform
662/// supplies a batch form today; every other stage's entry is `None`, which
663/// keeps the whole chain on the `Value` path unless every stage is columnar.
664#[cfg(feature = "arrow")]
665pub fn compile_transforms_columnar(
666    specs: &[TransformSpec],
667) -> CliResult<(
668    Vec<TransformStage>,
669    Vec<Option<faucet_core::stage::PageFnBatchBox>>,
670)> {
671    let mut stages = Vec::with_capacity(specs.len());
672    let mut batches = Vec::with_capacity(specs.len());
673    for s in specs {
674        #[cfg(feature = "transform-sql")]
675        if s.kind == "sql" {
676            let cfg = decode_sql("sql", s.config.clone())?;
677            let transform = faucet_transform_sql::SqlTransform::compile(&cfg).map_err(|e| {
678                let message = match &e {
679                    faucet_core::FaucetError::Transform(m)
680                    | faucet_core::FaucetError::Config(m) => m.clone(),
681                    other => format!("{other}"),
682                };
683                CliError::InvalidTransform {
684                    name: "sql".to_owned(),
685                    message,
686                }
687            })?;
688            let (stage, batch) = transform.into_columnar_stage();
689            stages.push(stage);
690            batches.push(Some(batch));
691            continue;
692        }
693        stages.push(compile_one(s)?);
694        batches.push(None);
695    }
696    Ok((stages, batches))
697}
698
699/// One-line summary of every transform compiled into this build. Used by
700/// `faucet list`.
701pub fn transform_descriptions() -> Vec<(&'static str, &'static str)> {
702    registry()
703        .into_iter()
704        .map(|t| (t.kind, t.description))
705        .collect()
706}
707
708/// Names of every transform compiled into this build.
709pub fn available_transforms() -> Vec<&'static str> {
710    registry().into_iter().map(|t| t.kind).collect()
711}
712
713// Keep in sync with faucet_core::{RecordCheck, BatchCheck} — one entry per check variant.
714/// One-line descriptions of the available quality checks, for `faucet list`.
715/// The `json_schema` entry only appears when the `quality-jsonschema` feature
716/// is enabled, mirroring `faucet schema quality` so `list` and `schema` agree.
717#[cfg(feature = "quality")]
718pub fn quality_descriptions() -> Vec<(&'static str, &'static str)> {
719    let mut checks = vec![
720        ("not_null", "field present and non-null"),
721        ("not_empty", "string non-empty after trim"),
722        ("regex_match", "string matches a regex"),
723        ("value_in_set", "value is in an allowed set"),
724        ("not_in_set", "value is not in a forbidden set"),
725        ("compare", "numeric/scalar comparison (gt/gte/lt/lte/eq/ne)"),
726        ("type_is", "value is of an expected JSON type"),
727        ("string_length", "string length within [min,max]"),
728    ];
729    #[cfg(feature = "quality-jsonschema")]
730    checks.push((
731        "json_schema",
732        "record validates against a JSON Schema (feature-gated)",
733    ));
734    checks.extend([
735        ("row_count", "batch row count within [min,max]"),
736        ("null_rate", "batch null rate of a field <= max"),
737        ("unique", "composite key unique within the batch"),
738        (
739            "distinct_count",
740            "distinct values of a field within [min,max]",
741        ),
742    ]);
743    checks
744}
745
746/// Return the JSON Schema for the named transform's config. Mirrors
747/// `registry::source_schema` / `sink_schema` so `faucet schema transform <name>`
748/// reads symmetrically with the connector variants.
749pub fn transform_schema(name: &str) -> CliResult<Value> {
750    registry()
751        .into_iter()
752        .find(|t| t.kind == name)
753        .map(|t| (t.schema_fn)())
754        .ok_or_else(|| unknown_transform(name))
755}
756
757fn unknown_transform(name: &str) -> CliError {
758    let available = available_transforms();
759    CliError::UnknownTransform {
760        name: name.to_owned(),
761        available: if available.is_empty() {
762            "(none — rebuild faucet-cli with the `transforms` feature enabled)".to_owned()
763        } else {
764            available.join(", ")
765        },
766    }
767}
768
769#[cfg(any(feature = "transforms", feature = "transform-cdc-unwrap"))]
770fn schema<T: JsonSchema>() -> Value {
771    serde_json::to_value(schema_for!(T)).unwrap_or_else(|_| serde_json::json!({"type": "object"}))
772}
773
774#[cfg(any(feature = "transforms", feature = "transform-cdc-unwrap"))]
775fn decode<T: serde::de::DeserializeOwned>(name: &str, config: Value) -> CliResult<T> {
776    serde_json::from_value(config).map_err(|e| CliError::InvalidTransform {
777        name: name.to_owned(),
778        message: e.to_string(),
779    })
780}
781
782/// Compile-validate a `Map` stage so its config errors (empty `fields`, `into`
783/// with multiple fields, `coalesce` requiring exactly one source) surface at
784/// load time as `CliError::InvalidTransform` — mirroring how `filter` /
785/// `explode` validate their specs at compile time.
786#[cfg(feature = "transforms")]
787fn validate_stage(kind: &str, stage: &TransformStage) -> CliResult<()> {
788    faucet_core::compile_stage(stage).map(|_| ()).map_err(|e| {
789        let message = match e {
790            faucet_core::FaucetError::Transform(m) | faucet_core::FaucetError::Config(m) => m,
791            other => format!("{other}"),
792        };
793        CliError::InvalidTransform {
794            name: kind.to_owned(),
795            message,
796        }
797    })
798}
799
800#[cfg(feature = "transform-sql")]
801fn schema_sql() -> Value {
802    serde_json::to_value(faucet_core::schema_for!(
803        faucet_transform_sql::SqlTransformConfig
804    ))
805    .unwrap_or(Value::Null)
806}
807
808#[cfg(feature = "transform-sql")]
809fn decode_sql(kind: &str, config: Value) -> CliResult<faucet_transform_sql::SqlTransformConfig> {
810    serde_json::from_value(config).map_err(|e| CliError::InvalidTransform {
811        name: kind.to_owned(),
812        message: e.to_string(),
813    })
814}
815
816#[cfg(feature = "transform-wasm")]
817fn schema_wasm() -> Value {
818    serde_json::to_value(faucet_core::schema_for!(
819        faucet_transform_wasm::WasmTransformConfig
820    ))
821    .unwrap_or(Value::Null)
822}
823
824#[cfg(feature = "transform-wasm")]
825fn decode_wasm(kind: &str, config: Value) -> CliResult<faucet_transform_wasm::WasmTransformConfig> {
826    serde_json::from_value(config).map_err(|e| CliError::InvalidTransform {
827        name: kind.to_owned(),
828        message: e.to_string(),
829    })
830}
831
832#[cfg(test)]
833mod tests {
834    use super::*;
835    use serde_json::json;
836
837    #[test]
838    fn empty_list_compiles_to_empty() {
839        let out = compile_transforms(&[]).unwrap();
840        assert!(out.is_empty());
841    }
842
843    #[cfg(feature = "transforms")]
844    #[test]
845    fn compiles_keys_case_and_flatten() {
846        let specs = vec![
847            TransformSpec {
848                kind: "keys_case".into(),
849                config: json!({"mode": "snake"}),
850            },
851            TransformSpec {
852                kind: "flatten".into(),
853                config: json!({"separator": "."}),
854            },
855        ];
856        let out = compile_transforms(&specs).unwrap();
857        assert_eq!(out.len(), 2);
858    }
859
860    #[cfg(feature = "transforms")]
861    #[test]
862    fn keys_case_rejects_unknown_mode() {
863        let specs = vec![TransformSpec {
864            kind: "keys_case".into(),
865            config: json!({"mode": "spongebob"}),
866        }];
867        let err = compile_transforms(&specs).unwrap_err();
868        match err {
869            CliError::InvalidTransform { name, .. } => assert_eq!(name, "keys_case"),
870            other => panic!("expected InvalidTransform, got {other:?}"),
871        }
872    }
873
874    #[cfg(feature = "transforms")]
875    #[test]
876    fn keys_case_requires_mode() {
877        let specs = vec![TransformSpec {
878            kind: "keys_case".into(),
879            config: json!({}),
880        }];
881        let err = compile_transforms(&specs).unwrap_err();
882        match err {
883            CliError::InvalidTransform { name, .. } => assert_eq!(name, "keys_case"),
884            other => panic!("expected InvalidTransform, got {other:?}"),
885        }
886    }
887
888    #[cfg(feature = "transforms")]
889    #[test]
890    fn snake_case_kind_is_no_longer_recognized() {
891        // Removed in favour of `keys_case { mode: snake }`.
892        let specs = vec![TransformSpec {
893            kind: "snake_case".into(),
894            config: json!({}),
895        }];
896        let err = compile_transforms(&specs).unwrap_err();
897        match err {
898            CliError::UnknownTransform { name, .. } => assert_eq!(name, "snake_case"),
899            other => panic!("expected UnknownTransform, got {other:?}"),
900        }
901    }
902
903    #[cfg(feature = "transforms")]
904    #[test]
905    fn rename_keys_requires_pattern_and_replacement() {
906        let specs = vec![TransformSpec {
907            kind: "rename_keys".into(),
908            config: json!({"pattern": "^_"}),
909        }];
910        let err = compile_transforms(&specs).unwrap_err();
911        match err {
912            CliError::InvalidTransform { name, .. } => assert_eq!(name, "rename_keys"),
913            other => panic!("expected InvalidTransform, got {other:?}"),
914        }
915    }
916
917    #[test]
918    fn unknown_transform_errors() {
919        let specs = vec![TransformSpec {
920            kind: "make_uppercase".into(),
921            config: json!({}),
922        }];
923        let err = compile_transforms(&specs).unwrap_err();
924        match err {
925            CliError::UnknownTransform { name, .. } => assert_eq!(name, "make_uppercase"),
926            other => panic!("expected UnknownTransform, got {other:?}"),
927        }
928    }
929
930    #[cfg(feature = "transforms")]
931    #[test]
932    fn compiles_select_and_drop() {
933        let specs = vec![
934            TransformSpec {
935                kind: "select".into(),
936                config: json!({"fields": ["id", "name"]}),
937            },
938            TransformSpec {
939                kind: "drop".into(),
940                config: json!({"fields": ["secret"]}),
941            },
942        ];
943        let out = compile_transforms(&specs).unwrap();
944        assert_eq!(out.len(), 2);
945    }
946
947    #[cfg(feature = "transforms")]
948    #[test]
949    fn compiles_set_with_object_values() {
950        let specs = vec![TransformSpec {
951            kind: "set".into(),
952            config: json!({"values": {"_source": "api", "version": 1}}),
953        }];
954        let out = compile_transforms(&specs).unwrap();
955        assert_eq!(out.len(), 1);
956    }
957
958    #[cfg(feature = "transforms")]
959    #[test]
960    fn compiles_rename_field() {
961        let specs = vec![TransformSpec {
962            kind: "rename_field".into(),
963            config: json!({"fields": {"old": "new"}}),
964        }];
965        let out = compile_transforms(&specs).unwrap();
966        assert_eq!(out.len(), 1);
967    }
968
969    #[cfg(feature = "transforms")]
970    #[test]
971    fn compiles_cast_with_default_on_error() {
972        let specs = vec![TransformSpec {
973            kind: "cast".into(),
974            config: json!({"fields": {"age": "int", "price": "float"}}),
975        }];
976        let out = compile_transforms(&specs).unwrap();
977        assert_eq!(out.len(), 1);
978    }
979
980    #[cfg(feature = "transforms")]
981    #[test]
982    fn cast_rejects_unknown_target_type() {
983        let specs = vec![TransformSpec {
984            kind: "cast".into(),
985            config: json!({"fields": {"x": "uuid"}}),
986        }];
987        let err = compile_transforms(&specs).unwrap_err();
988        match err {
989            CliError::InvalidTransform { name, .. } => assert_eq!(name, "cast"),
990            other => panic!("expected InvalidTransform, got {other:?}"),
991        }
992    }
993
994    #[cfg(feature = "transforms")]
995    #[test]
996    fn cast_rejects_unknown_on_error_mode() {
997        let specs = vec![TransformSpec {
998            kind: "cast".into(),
999            config: json!({"fields": {"x": "int"}, "on_error": "explode"}),
1000        }];
1001        let err = compile_transforms(&specs).unwrap_err();
1002        match err {
1003            CliError::InvalidTransform { name, .. } => assert_eq!(name, "cast"),
1004            other => panic!("expected InvalidTransform, got {other:?}"),
1005        }
1006    }
1007
1008    #[cfg(feature = "transforms")]
1009    #[test]
1010    fn redact_uses_default_mask_when_omitted() {
1011        let specs = vec![TransformSpec {
1012            kind: "redact".into(),
1013            config: json!({"fields": ["ssn"]}),
1014        }];
1015        let out = compile_transforms(&specs).unwrap();
1016        assert_eq!(out.len(), 1);
1017    }
1018
1019    #[cfg(feature = "transforms")]
1020    #[test]
1021    fn compiles_hash_with_defaults_and_options() {
1022        let specs = vec![
1023            TransformSpec {
1024                kind: "hash".into(),
1025                config: json!({"fields": ["email"]}),
1026            },
1027            TransformSpec {
1028                kind: "hash".into(),
1029                config: json!({
1030                    "fields": ["user_id"],
1031                    "algorithm": "blake3",
1032                    "encoding": "base64",
1033                    "salt": "pepper",
1034                    "into": "user_id_hash"
1035                }),
1036            },
1037        ];
1038        let out = compile_transforms(&specs).unwrap();
1039        assert_eq!(out.len(), 2);
1040    }
1041
1042    #[cfg(feature = "transforms")]
1043    #[test]
1044    fn hash_empty_fields_is_rejected_at_load() {
1045        let specs = vec![TransformSpec {
1046            kind: "hash".into(),
1047            config: json!({"fields": []}),
1048        }];
1049        match compile_transforms(&specs).unwrap_err() {
1050            CliError::InvalidTransform { name, .. } => assert_eq!(name, "hash"),
1051            other => panic!("expected InvalidTransform, got {other:?}"),
1052        }
1053    }
1054
1055    #[cfg(feature = "transforms")]
1056    #[test]
1057    fn hash_into_with_multiple_fields_is_rejected_at_load() {
1058        let specs = vec![TransformSpec {
1059            kind: "hash".into(),
1060            config: json!({"fields": ["a", "b"], "into": "x"}),
1061        }];
1062        match compile_transforms(&specs).unwrap_err() {
1063            CliError::InvalidTransform { name, message } => {
1064                assert_eq!(name, "hash");
1065                assert!(message.contains("into"), "{message}");
1066            }
1067            other => panic!("expected InvalidTransform, got {other:?}"),
1068        }
1069    }
1070
1071    #[cfg(feature = "transforms")]
1072    #[test]
1073    fn compiles_json_parse_with_on_error() {
1074        let specs = vec![TransformSpec {
1075            kind: "json_parse".into(),
1076            config: json!({"fields": ["payload"], "on_error": "null", "into": "parsed"}),
1077        }];
1078        let out = compile_transforms(&specs).unwrap();
1079        assert_eq!(out.len(), 1);
1080    }
1081
1082    #[cfg(feature = "transforms")]
1083    #[test]
1084    fn json_parse_into_with_multiple_fields_is_rejected_at_load() {
1085        let specs = vec![TransformSpec {
1086            kind: "json_parse".into(),
1087            config: json!({"fields": ["a", "b"], "into": "x"}),
1088        }];
1089        match compile_transforms(&specs).unwrap_err() {
1090            CliError::InvalidTransform { name, .. } => assert_eq!(name, "json_parse"),
1091            other => panic!("expected InvalidTransform, got {other:?}"),
1092        }
1093    }
1094
1095    #[cfg(feature = "transforms")]
1096    #[test]
1097    fn compiles_coalesce_with_default() {
1098        let specs = vec![TransformSpec {
1099            kind: "coalesce".into(),
1100            config: json!({"field": "status", "default": "unknown"}),
1101        }];
1102        let out = compile_transforms(&specs).unwrap();
1103        assert_eq!(out.len(), 1);
1104    }
1105
1106    #[cfg(feature = "transforms")]
1107    #[test]
1108    fn compiles_coalesce_with_from() {
1109        let specs = vec![TransformSpec {
1110            kind: "coalesce".into(),
1111            config: json!({"field": "status", "from": ["status", "state"]}),
1112        }];
1113        let out = compile_transforms(&specs).unwrap();
1114        assert_eq!(out.len(), 1);
1115    }
1116
1117    #[cfg(feature = "transforms")]
1118    #[test]
1119    fn coalesce_both_default_and_from_is_rejected_at_load() {
1120        let specs = vec![TransformSpec {
1121            kind: "coalesce".into(),
1122            config: json!({"field": "s", "default": "x", "from": ["y"]}),
1123        }];
1124        match compile_transforms(&specs).unwrap_err() {
1125            CliError::InvalidTransform { name, .. } => assert_eq!(name, "coalesce"),
1126            other => panic!("expected InvalidTransform, got {other:?}"),
1127        }
1128    }
1129
1130    #[cfg(feature = "transforms")]
1131    #[test]
1132    fn coalesce_neither_default_nor_from_is_rejected_at_load() {
1133        let specs = vec![TransformSpec {
1134            kind: "coalesce".into(),
1135            config: json!({"field": "s"}),
1136        }];
1137        match compile_transforms(&specs).unwrap_err() {
1138            CliError::InvalidTransform { name, .. } => assert_eq!(name, "coalesce"),
1139            other => panic!("expected InvalidTransform, got {other:?}"),
1140        }
1141    }
1142
1143    #[cfg(feature = "transforms")]
1144    #[test]
1145    fn compiles_split_and_join() {
1146        let specs = vec![
1147            TransformSpec {
1148                kind: "split".into(),
1149                config: json!({"field": "tags", "delimiter": ",", "trim": true}),
1150            },
1151            TransformSpec {
1152                kind: "join".into(),
1153                config: json!({"field": "tags", "delimiter": ",", "into": "csv"}),
1154            },
1155        ];
1156        let out = compile_transforms(&specs).unwrap();
1157        assert_eq!(out.len(), 2);
1158    }
1159
1160    #[cfg(feature = "transforms")]
1161    #[test]
1162    fn value_case_requires_mode() {
1163        let specs = vec![TransformSpec {
1164            kind: "value_case".into(),
1165            config: json!({"fields": ["email"]}),
1166        }];
1167        let err = compile_transforms(&specs).unwrap_err();
1168        match err {
1169            CliError::InvalidTransform { name, .. } => assert_eq!(name, "value_case"),
1170            other => panic!("expected InvalidTransform, got {other:?}"),
1171        }
1172    }
1173
1174    #[cfg(feature = "transforms")]
1175    #[test]
1176    fn available_transforms_lists_every_kind() {
1177        let names = available_transforms();
1178        for expected in [
1179            "flatten",
1180            "rename_keys",
1181            "keys_case",
1182            "select",
1183            "drop",
1184            "set",
1185            "rename_field",
1186            "cast",
1187            "redact",
1188            "value_case",
1189            "spell_symbols",
1190            "hash",
1191            "json_parse",
1192            "coalesce",
1193            "split",
1194            "join",
1195            "filter",
1196            "explode",
1197        ] {
1198            assert!(names.contains(&expected), "missing {expected}");
1199        }
1200        assert!(
1201            !names.contains(&"snake_case"),
1202            "snake_case must be removed in favour of keys_case"
1203        );
1204    }
1205
1206    #[cfg(feature = "transforms")]
1207    #[test]
1208    fn transform_descriptions_covers_every_compiled_kind() {
1209        // descriptions and available_transforms must never drift — `faucet list`
1210        // and the `UnknownTransform` "Available:" line both read from this.
1211        let names = available_transforms();
1212        let desc_names: Vec<&'static str> = transform_descriptions()
1213            .into_iter()
1214            .map(|(n, _)| n)
1215            .collect();
1216        assert_eq!(names, desc_names);
1217        for (_, desc) in transform_descriptions() {
1218            assert!(!desc.is_empty(), "every transform needs a description");
1219        }
1220    }
1221
1222    #[cfg(feature = "transforms")]
1223    #[test]
1224    fn transform_schema_returns_object_for_every_kind() {
1225        for name in available_transforms() {
1226            let schema = transform_schema(name).unwrap_or_else(|e| {
1227                panic!("schema lookup failed for {name}: {e}");
1228            });
1229            assert!(schema.is_object(), "schema for {name} must be an object");
1230        }
1231    }
1232
1233    #[cfg(feature = "transforms")]
1234    #[test]
1235    fn transform_schema_select_and_drop_share_shape() {
1236        // Both accept `{ fields: Vec<String> }` — the schema is the same object,
1237        // just titled `FieldsConfig`.
1238        let select = transform_schema("select").unwrap();
1239        let drop = transform_schema("drop").unwrap();
1240        assert_eq!(select, drop);
1241    }
1242
1243    #[test]
1244    fn transform_schema_unknown_errors_with_available_list() {
1245        let err = transform_schema("make_uppercase").unwrap_err();
1246        match err {
1247            CliError::UnknownTransform { name, available } => {
1248                assert_eq!(name, "make_uppercase");
1249                #[cfg(feature = "transforms")]
1250                assert!(available.contains("flatten"), "{available}");
1251                #[cfg(not(feature = "transforms"))]
1252                assert!(available.contains("rebuild"), "{available}");
1253            }
1254            other => panic!("expected UnknownTransform, got {other:?}"),
1255        }
1256    }
1257
1258    #[cfg(feature = "transform-filter")]
1259    #[test]
1260    fn compiles_filter_eq() {
1261        let specs = vec![TransformSpec {
1262            kind: "filter".into(),
1263            config: json!({"path": "status", "op": "eq", "value": "active"}),
1264        }];
1265        let out = compile_transforms(&specs).unwrap();
1266        assert_eq!(out.len(), 1);
1267        assert!(matches!(out[0], TransformStage::Filter(_)));
1268    }
1269
1270    #[cfg(feature = "transform-filter")]
1271    #[test]
1272    fn filter_rejects_in_with_non_array_value() {
1273        let specs = vec![TransformSpec {
1274            kind: "filter".into(),
1275            config: json!({"path": "v", "op": "in", "value": "scalar"}),
1276        }];
1277        let err = compile_transforms(&specs).unwrap_err();
1278        match err {
1279            CliError::InvalidTransform { name, message } => {
1280                assert_eq!(name, "filter");
1281                assert!(message.contains("requires an array"), "{message}");
1282            }
1283            other => panic!("expected InvalidTransform, got {other:?}"),
1284        }
1285    }
1286
1287    #[cfg(feature = "transform-filter")]
1288    #[test]
1289    fn filter_rejects_exists_with_value() {
1290        let specs = vec![TransformSpec {
1291            kind: "filter".into(),
1292            config: json!({"path": "v", "op": "exists", "value": "x"}),
1293        }];
1294        let err = compile_transforms(&specs).unwrap_err();
1295        match err {
1296            CliError::InvalidTransform { name, .. } => assert_eq!(name, "filter"),
1297            other => panic!("expected InvalidTransform, got {other:?}"),
1298        }
1299    }
1300
1301    #[cfg(feature = "transform-filter")]
1302    #[test]
1303    fn filter_rejects_bad_path() {
1304        let specs = vec![TransformSpec {
1305            kind: "filter".into(),
1306            config: json!({"path": "$..items", "op": "exists"}),
1307        }];
1308        let err = compile_transforms(&specs).unwrap_err();
1309        match err {
1310            CliError::InvalidTransform { name, .. } => assert_eq!(name, "filter"),
1311            other => panic!("expected InvalidTransform, got {other:?}"),
1312        }
1313    }
1314
1315    #[cfg(feature = "transform-explode")]
1316    #[test]
1317    fn compiles_explode_with_defaults() {
1318        let specs = vec![TransformSpec {
1319            kind: "explode".into(),
1320            config: json!({"path": "items"}),
1321        }];
1322        let out = compile_transforms(&specs).unwrap();
1323        assert_eq!(out.len(), 1);
1324        assert!(matches!(out[0], TransformStage::Explode(_)));
1325    }
1326
1327    #[cfg(feature = "transform-explode")]
1328    #[test]
1329    fn compiles_explode_with_custom_prefix_and_on_missing() {
1330        let specs = vec![TransformSpec {
1331            kind: "explode".into(),
1332            config: json!({
1333                "path": "items",
1334                "prefix": "item",
1335                "separator": "_",
1336                "on_missing": "drop"
1337            }),
1338        }];
1339        let out = compile_transforms(&specs).unwrap();
1340        assert_eq!(out.len(), 1);
1341    }
1342
1343    #[cfg(feature = "transform-explode")]
1344    #[test]
1345    fn explode_rejects_bad_path() {
1346        let specs = vec![TransformSpec {
1347            kind: "explode".into(),
1348            config: json!({"path": "$..items"}),
1349        }];
1350        let err = compile_transforms(&specs).unwrap_err();
1351        match err {
1352            CliError::InvalidTransform { name, .. } => assert_eq!(name, "explode"),
1353            other => panic!("expected InvalidTransform, got {other:?}"),
1354        }
1355    }
1356
1357    #[cfg(feature = "transform-explode")]
1358    #[test]
1359    fn explode_rejects_invalid_on_missing() {
1360        let specs = vec![TransformSpec {
1361            kind: "explode".into(),
1362            config: json!({"path": "items", "on_missing": "explode_harder"}),
1363        }];
1364        let err = compile_transforms(&specs).unwrap_err();
1365        match err {
1366            CliError::InvalidTransform { name, .. } => assert_eq!(name, "explode"),
1367            other => panic!("expected InvalidTransform, got {other:?}"),
1368        }
1369    }
1370
1371    #[cfg(feature = "transform-sql")]
1372    #[test]
1373    fn compiles_sql_to_page_fn() {
1374        let specs = vec![TransformSpec {
1375            kind: "sql".into(),
1376            config: json!({"query": "SELECT * FROM batch"}),
1377        }];
1378        let out = compile_transforms(&specs).unwrap();
1379        assert_eq!(out.len(), 1);
1380        assert!(matches!(out[0], faucet_core::TransformStage::PageFn(_)));
1381    }
1382
1383    #[cfg(feature = "transform-sql")]
1384    #[test]
1385    fn sql_bad_query_is_invalid_transform() {
1386        let specs = vec![TransformSpec {
1387            kind: "sql".into(),
1388            config: json!({"query": "SELEKT bad"}),
1389        }];
1390        let err = compile_transforms(&specs).unwrap_err();
1391        match err {
1392            CliError::InvalidTransform { name, .. } => assert_eq!(name, "sql"),
1393            other => panic!("expected InvalidTransform, got {other:?}"),
1394        }
1395    }
1396
1397    #[cfg(feature = "transform-sql")]
1398    #[test]
1399    fn sql_schema_and_listing_present() {
1400        assert!(transform_schema("sql").is_ok());
1401        assert!(available_transforms().contains(&"sql"));
1402    }
1403
1404    #[cfg(feature = "transform-wasm")]
1405    #[test]
1406    fn wasm_schema_and_listing_present() {
1407        assert!(transform_schema("wasm").is_ok());
1408        assert!(available_transforms().contains(&"wasm"));
1409    }
1410
1411    #[cfg(feature = "transform-wasm")]
1412    #[test]
1413    fn wasm_missing_module_field_is_invalid_transform() {
1414        let specs = vec![TransformSpec {
1415            kind: "wasm".into(),
1416            config: json!({}),
1417        }];
1418        let err = compile_transforms(&specs).unwrap_err();
1419        match err {
1420            CliError::InvalidTransform { name, .. } => assert_eq!(name, "wasm"),
1421            other => panic!("expected InvalidTransform, got {other:?}"),
1422        }
1423    }
1424
1425    #[cfg(feature = "transform-wasm")]
1426    #[test]
1427    fn wasm_nonexistent_module_is_invalid_transform() {
1428        let specs = vec![TransformSpec {
1429            kind: "wasm".into(),
1430            config: json!({"module": "/no/such/path/mod.wasm"}),
1431        }];
1432        let err = compile_transforms(&specs).unwrap_err();
1433        match err {
1434            CliError::InvalidTransform { name, message } => {
1435                assert_eq!(name, "wasm");
1436                assert!(message.contains("cannot read module"), "{message}");
1437            }
1438            other => panic!("expected InvalidTransform, got {other:?}"),
1439        }
1440    }
1441
1442    #[cfg(feature = "transform-cdc-unwrap")]
1443    #[test]
1444    fn compiles_cdc_unwrap_with_defaults() {
1445        let specs = vec![TransformSpec {
1446            kind: "cdc_unwrap".into(),
1447            config: json!({}),
1448        }];
1449        let out = compile_transforms(&specs).unwrap();
1450        assert_eq!(out.len(), 1);
1451        assert!(matches!(out[0], faucet_core::TransformStage::CdcUnwrap(_)));
1452    }
1453
1454    #[cfg(feature = "transform-cdc-unwrap")]
1455    #[test]
1456    fn cdc_unwrap_schema_and_listing_present() {
1457        assert!(transform_schema("cdc_unwrap").is_ok());
1458        assert!(available_transforms().contains(&"cdc_unwrap"));
1459    }
1460
1461    #[cfg(feature = "quality")]
1462    #[test]
1463    fn quality_descriptions_has_one_entry_per_check() {
1464        // 8 always-on per-record checks + 4 per-batch checks = 12; the
1465        // `json_schema` per-record check is only present (→ 13) when the
1466        // `quality-jsonschema` feature is enabled, matching `faucet schema
1467        // quality`. If you add a RecordCheck/BatchCheck variant in faucet-core,
1468        // add its description here too.
1469        #[cfg(feature = "quality-jsonschema")]
1470        assert_eq!(quality_descriptions().len(), 13);
1471        #[cfg(not(feature = "quality-jsonschema"))]
1472        assert_eq!(quality_descriptions().len(), 12);
1473    }
1474}