ETL quarantine

1nika: v12workflow:3  id: etl-quarantine4  description: "CSV batch → schema gate → quarantine the bad · aggregate the good"5​6const:7  batch_csv: "examples/fixtures/orders.csv"8⛨permits:9  ⛨tools: ["nika:convert", "nika:jq", "nika:read", "nika:validate", "nika:write"]10  # The batch comes in at ONE path and leaves at TWO · each named exactly.11  # Note what is NOT granted: `out/**` would have handed the whole artifact12  # tree to a pipeline that only ever writes these two files.13  ⛨fs:14    ⛨read: ["examples/fixtures/orders.csv"]15    ⛨write: ["out/quarantine/orders-rejected.json", "out/reports/daily-totals.json"]16​17tasks:18  # A deterministic empty fallback · the recover target when parsing dies.19  empty_batch:20    ◆invoke:21      tool: "nika:jq"22      args: { input: [], expression: "." }23​24  raw:25    ◆invoke:26      tool: "nika:read"27      args: { path: "${{ const.batch_csv }}" }28​29  rows:30    with:31      csv: ${{ tasks.raw.output }}32    ◆invoke:33      tool: "nika:convert"34      args:35        input: "${{ with.csv }}"36        from: csv37        to: json38        has_header: true39    on_error:40      recover: ${{ tasks.empty_batch.output }}    # malformed CSV → empty batch · pipeline lives41​42  check:43    with:44      rows: ${{ tasks.rows.output }}45    ◆invoke:46      tool: "nika:validate"47      args:48        data: "${{ with.rows }}"49        format: json50        schema:51          type: array52          items:53            type: object54            required: [order_id, amount, currency]55            properties:56              order_id: { type: string }57              amount: { type: string }58              currency: { type: string, enum: [EUR, USD, GBP] }59​60  good:61    with:62      rows: ${{ tasks.rows.output }}63      valid: ${{ tasks.check.output.valid }}64    when: ${{ with.valid == true }}65    ◆invoke:66      tool: "nika:jq"67      args:68        input: "${{ with.rows }}"69        expression: 'group_by(.currency) | map({currency: .[0].currency, orders: length, total: (map(.amount | tonumber) | add)})'70​71  quarantine:72    with:73      valid: ${{ tasks.check.output.valid }}74      errors: ${{ tasks.check.output.errors }}75    when: ${{ with.valid == false }}76    ◆invoke:77      tool: "nika:write"78      args:79        path: out/quarantine/orders-rejected.json80        content: "${{ with.errors }}"81        create_dirs: true82​83  report:84    with:85      totals: ${{ tasks.good.output }}86    when: ${{ with.totals != null && size(with.totals) > 0 }}   # good is SKIPPED (null) on the quarantine path · guard before size()87    ◆invoke:88      tool: "nika:write"89      args:90        path: out/reports/daily-totals.json91        content: "${{ with.totals }}"92        create_dirs: true93​94outputs:95  # Deliberately untyped · on the quarantine path `good` is SKIPPED and96  # this is null — a declared `type: array` would refuse the branch the97  # file exists to demonstrate.98  totals: ${{ tasks.good.output }}99​

✔audited · 7 tasks · 5 waves · permits declarednika 0.106.1