# Recipes Reference

> Auto-generated from docs.expanso.io documentation

## Build by Job

# Build by Job

Start from the job you were asked to do. Each row names the exact components
to use and links to a complete job spec. Every linked job was run end to end
and then checked at its destination: the records were counted where they
landed, not inferred from a job status. Each row says where it ran.

## Task-to-components matrix

| If you are asked to build… | Use these components | Complete job | What the run proved |
|---|---|---|---|
| An **RSS feed engine**: poll feeds, parse items, never emit an item twice | `generate` → `http` → `mapping` (fetch error check) → `xml` → `mapping` → `unarchive` → `mapping` → `dedupe` (with a `file` cache) → `file` output | [RSS feed engine](/recipes/rss-feed-engine) | On 2026-09-18, a live NASA feed (10 items) and BBC Technology feed (21 items) each matched an independent fetch. A duplicate GUID was emitted once, and a restarted job emitted 0 items it had already seen. The fetch error check was added later and proved on a local-mode node only |
| A **data migration engine**: copy a legacy table into a new schema | `sql_select` → `mapping` → `switch` output: `sql_insert` for good rows, `file` for rejects | [Data migration engine](/recipes/data-migration) | 1,000 legacy rows became 994 target rows plus 6 rejects, each with its reason. The target matched an independent SQL transform, and a second run changed nothing |
| A **notification engine**: receive events, send a message when one matters | `http_server` → `dedupe` → `mapping` → `mapping` → `http_client` output with retries | [Notification engine](/recipes/notification-engine) | 8 events in, exactly the 3 critical or high ones delivered, a duplicate delivered once, and two `503` responses retried until they succeeded |
| **Document embedding for retrieval (RAG)**: fetch documents, chunk, embed, store, search | Ingest: `generate` → `http` → `unarchive` → `http` → `mapping` (paragraph chunks) → `unarchive` → `branch` + `ollama_embeddings` → `mapping` → `http_client` to the Qdrant REST API. Search: `file` → `branch` + `ollama_embeddings` → `branch` + `qdrant` processor → `mapping` → `file` | [RAG: embed and retrieve](/recipes/rag-embed-retrieve) | 4 documents fetched over HTTP became 16 chunks, each carrying its source URL. Stored vectors equal a direct embedding of the same text (cosine 1.000000). All 4 test questions returned the right chunk first |
| A **webhook fan-out**: take one webhook and deliver every event to several endpoints | `http_server` → `mapping` → `broker` output (`pattern: fan_out`) over several `http_client` outputs, each with retries | [Webhook fan-out](/recipes/webhook-fan-out) | 10 events in, and each of 3 receivers got all 10 exactly once, including one that answered `503` twice. **Local-mode run**, not through Cloud |
| **Log reduction**: cut access-log volume but keep every error | `broker` input over a `file` input → `mapping` (regex parse, drop health checks) → `switch` processor: errors kept, successes `group_by_value` → `archive` → `mapping` rollup → `file` | [Log reduction](/recipes/log-reduction) | 5,000 synthetic lines became 330 events: all 250 errors kept individually and 80 per-minute rollups that matched an independent recount. 92.7% fewer bytes on that synthetic log, not a benchmark. **Local-mode run** |
| **Sensor or industrial telemetry over MQTT**: subscribe, normalise, alert | `mqtt` input (wildcard topic) → `mapping` → `switch` output to rejects, alerts and telemetry `file` outputs | [Sensor telemetry over MQTT](/recipes/sensor-telemetry-mqtt) | 30 synthetic readings: 28 valid in telemetry, 3 alerts, 2 malformed rejected with the reason. MQTT only, no hardware or OPC UA. **Local-mode run** |
| **CI test fixtures**: the same synthetic dataset on every run | `generate` → `file`, with `pipeline.threads: 1`; `expanso-edge validate` as the config gate | [CI test fixtures](/recipes/ci-fixtures) | Two runs wrote byte-identical files of 200 schema-valid records; a typo in the config made validation exit 1. **Local-mode run** |

How those runs were done: `expanso-edge` v2.1.21. The first four rows were
each submitted to an Expanso Cloud workspace with a label selector, and each
execution was assigned to the one node carrying that label. The last four
rows ran on a local-mode node, not through Cloud. The RSS, data migration and
RAG jobs were revised after their Cloud runs, and the published versions were
proved on a local-mode node (see [Proving your own run](#proving-your-own-run)).
The Cloud node was a machine registered
to the workspace, not a hosted runner, and the databases, model server and vector
store ran on the same host.

Only the components named in a row were exercised by its run. Each page states
the dependencies the job needs and the limits of what was proved.

## One library, many jobs

The rows above are a small slice of what the same job format covers. The
component reference currently lists **50 inputs, 80 processors and 55
outputs**, plus 5 buffers, 13 caches, 2 rate limits and 11 scanners: 216
components in all. Some examples by exact config name:

- **HTTP and webhooks:** `http_client` and `http_server` (input and output), and the `http` processor
- **Messaging:** `mqtt`, `nats`, `nats_jetstream`, `aws_sqs`, `aws_sns` (output), and the `kafka` output (the `kafka` input is deprecated)
- **Object storage:** `aws_s3`, `gcp_cloud_storage`, `azure_blob_storage`
- **Databases and change data capture:** `sql_select`, `sql_insert`, `sql_raw`, `postgres_cdc`, `mysql_cdc`, `aws_dynamodb_cdc`, `redis_streams` and `redis_list`
- **Industrial:** `opcua` (input and output)
- **Search and vectors:** `opensearch` (output) and `qdrant`
- **Models:** `ollama_chat`, `ollama_embeddings`, `aws_bedrock_chat`, `aws_bedrock_embeddings`, `gcp_vertex_ai_chat`, `gcp_vertex_ai_embeddings`, `openai_chat_completion`, `openai_embeddings`

These names exist in the catalog. Their presence there is not a claim that any
particular combination has been run; check the component's page and its status
(some are beta or deprecated) before you rely on it. Browse the
[full component list](/components), or fetch the
[plain-text catalog](pathname:///llms/components.txt) for scripts and AI tools.

## Running near the data

Jobs run on nodes you choose, selected by label. That lets a pipeline sit next
to the data it reads, so it can filter, reduce, redact or embed records before
they cross a network. This is where Expanso is distinctively strong. It is not
a boundary: the jobs above work the same way wherever the node runs.

## Where Expanso is not the right answer

- **Not a workflow orchestrator, batch query engine or data warehouse.**
  Expanso moves and transforms records. It does not run arbitrary task graphs,
  answer ad hoc queries, or store data for analysis.
- **No lakehouse table writers.** There is no component for Apache Iceberg,
  Delta Lake or Apache Hudi tables.
- **No email, Slack, Discord or PagerDuty component.** Those services are
  reached over HTTP with `http_client`. The [notification engine](/recipes/notification-engine)
  was proved against a test receiver only; delivery into Slack or email was not
  tested. Try your real receiver before you depend on it.
- **No X (Twitter) component, and X ingestion is unproven.** A job reading the
  X API over `http` was built, but it could not be run against the live API
  because the test account had no X API credits. Treat it as untested.
- **Change data capture is unproven here.** `postgres_cdc`, `mysql_cdc` and
  `aws_dynamodb_cdc` exist, but no CDC job has completed a verified run yet.
  For a one-time table copy, use the [data migration engine](/recipes/data-migration).
- **No native Anthropic (Claude) model processor.** The model processors in the
  catalog are listed above.
- **No text-aware chunker.** The only chunking scanner, `chunker`, splits by
  byte count. The [RAG recipe](/recipes/rag-embed-retrieve) splits documents
  into paragraphs with a `mapping` instead.

## Proving your own run

A clean deploy, and even a `completed` job, are not proof that data arrived.
Two behaviours seen in these runs on v2.1.21 show why:

- A source that answers with an error, such as `401 Unauthorized` from `http`,
  can end as a `completed` job with nothing written. Fail loudly by adding
  `root = if errored() { throw("fetch failed: " + error().or("unknown")) } else { this }`
  straight after the `http` processor.
- When every write to the native `qdrant` output failed, the job still ended
  `completed` with 0 points stored. Count the records at the destination.

Every bounded job on these pages sets `restart_policy: never` at the top
level of the job spec. With the default policy, a bounded job that failed on
Cloud was re-run every few seconds and its state read `running`. With `never`
it ended `failed` after one execution. All six bounded jobs were run exactly
as published, `never` included, on a local-mode node running v2.1.21: each
passing run completed after one execution, and the RSS job pointed at a feed
answering `401` or `503` failed after one execution. The three listener jobs
(notification engine, webhook fan-out, MQTT telemetry) keep running by design
and keep the default. If you turn a bounded job into a standing one, remove
`never` too, or a single transient error stops it for good.

Use underscores, not hyphens, in the label keys you select on. With a hyphen
in the key, the pipeline failed to build on the node; the same job with an
underscore in the key ran. See [node labels](/references/configuration/edge-config#labels).

## Cost

The first five nodes are free. Checking a job needs no account at all:
`expanso-edge validate job.yaml` and
`expanso-cli job validate job.yaml --offline` run locally and contact nothing.

## Next steps

- [Deploy to Cloud](/getting-started/deploy-to-cloud/): create a workspace, start a node, deploy a job
- [Common mistakes](/getting-started/configuration-formats): the job wrapper, credentials and endpoints
- [Bloblang guide](/guides/bloblang): the mapping language every recipe uses

## CI Test Fixtures

# CI Test Fixtures

Generate a synthetic dataset for tests in which every field is derived from the
record number, with no random values and no clock, so every run writes the same
file, byte for byte. Use `expanso-edge validate` in the same pipeline to fail
the build when a job config has a typo.

**Components:** `generate` input (`count: 200`) → `file` output, with
`pipeline.threads: 1`.

**Needs:** nothing beyond a node. The job reads no external data.

:::note[Proved on a local-mode node]
This job ran end to end on an `expanso-edge` v2.1.21 node in local mode, not
through Expanso Cloud.
:::

## Complete job

```yaml title="ci-fixtures.yaml"
name: ci-fixtures
type: pipeline
# One-shot job: stop after one failed execution instead of retrying.
restart_policy: never
selector:
  match_labels:
    pipeline_role: ci
config:
  input:
    generate:
      count: 200
      interval: ""
      mapping: |
        let n = count("fixtures_RUN_ID")
        let first = [
          "Ada", "Grace", "Alan", "Edsger", "Barbara", "Ken",
          "Frances", "Donald"
        ]
        root.id = "cust-%05d".format($n)
        root.name = "%s Tester%d".format($first.index($n % 8), $n)
        root.email = "tester%d@example.invalid".format($n)
        root.tier = ["gold", "silver", "bronze"].index($n % 3)
        root.balance = "%d.%02d".format(
          ($n * 7919 % 100000 / 100).floor().int64(),
          ($n * 7919 % 100).int64()
        )
        root.created_at = (1767225600 + $n * 86400).ts_format(
          "2006-01-02T15:04:05Z", "UTC"
        )
        root.active = $n % 7 != 0
  pipeline:
    # One processing thread keeps output order equal to generation
    # order. With the default parallelism the same records came out
    # in a different order on each run, so the file was not
    # byte-identical.
    threads: 1
    processors: []
  output:
    file:
      path: "/var/tmp/expanso-fixtures/fixtures.jsonl"
      codec: lines
```

Two settings make the output repeatable:

- **`threads: 1`.** With the default parallelism, the same 200 records came
  out in a different order on each run. One thread keeps them in generation
  order.
- **A new counter name for every run.** A `count()` counter keeps counting for
  as long as the node process runs, across jobs. Replace `RUN_ID` with a value
  that is unique to the run, such as your CI build number, so the ids start at
  `cust-00001` every time.

The `selector` sends the job to a node labelled `pipeline_role: ci`; set that
label on your node, or change the selector to a label it already carries.
With no matching node, the job is stored and never runs.

## Use it in CI

Check the config first: `expanso-edge validate` exits with a non-zero status
on an unknown component or field, which fails the step. Then substitute the
run id and deploy:

```bash
expanso-edge validate ci-fixtures.yaml
sed "s/RUN_ID/$BUILD_ID/" ci-fixtures.yaml > ci-fixtures-run.yaml
expanso-cli job deploy ci-fixtures-run.yaml
```

`BUILD_ID` stands for your CI system's unique build number. The output path
is on the **node that runs the job**; create its directory there first with
`mkdir -p /var/tmp/expanso-fixtures`. Each run in the proof wrote to a fresh
file, so remove the previous `fixtures.jsonl` before the next run, and remove
the finished job with `expanso-cli job delete ci-fixtures --yes`
(without `--yes` the command waits at a confirmation prompt, which fails in a
non-interactive CI step).

A misspelled component name fails like this, from the run:

> (line 10, col 3) [input]. Unknown component or field 'generat'. Check that
> the component name is spelled correctly.

## Check it

Every record matches this JSON Schema. The run checked all 200 against it:

```json title="schema.json"
{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "type": "object",
  "additionalProperties": false,
  "required": [
    "id",
    "name",
    "email",
    "tier",
    "balance",
    "created_at",
    "active"
  ],
  "properties": {
    "id": {
      "type": "string",
      "pattern": "^cust-[0-9]{5}$"
    },
    "name": {
      "type": "string",
      "minLength": 3
    },
    "email": {
      "type": "string",
      "pattern": "^tester[0-9]+@example\\.invalid$"
    },
    "tier": {
      "enum": [
        "gold",
        "silver",
        "bronze"
      ]
    },
    "balance": {
      "type": "string",
      "pattern": "^[0-9]+\\.[0-9]{2}$"
    },
    "created_at": {
      "type": "string",
      "pattern": "^20[0-9]{2}-[0-9]{2}-[0-9]{2}T00:00:00Z$"
    },
    "active": {
      "type": "boolean"
    }
  }
}
```

The first line of the file, pretty-printed:

```json
{
  "active": true,
  "balance": "79.19",
  "created_at": "2026-01-02T00:00:00Z",
  "email": "tester1@example.invalid",
  "id": "cust-00001",
  "name": "Grace Tester1",
  "tier": "silver"
}
```

## What the run proved

- Two runs each wrote 200 records, `cust-00001` to `cust-00200` in order.
- The two files were byte-identical: the same SHA-256.
- Every record validated against the schema above with an independent JSON
  Schema checker.
- A copy of the job with `generate` misspelled as `generat` made
  `expanso-edge validate` exit 1.
- The job exactly as published, `restart_policy: never` included, ran on a
  local-mode node running `expanso-edge` v2.1.21. Each of the two runs
  completed after 1 execution.

## Limits

- Proved on a local-mode node only, not through Expanso Cloud.
- With `restart_policy: never`, a failed run stops after one execution
  instead of being re-run.
  See [Proving your own run](/recipes#proving-your-own-run).
- The dataset is fixed by the mapping. Edit the mapping to change the fields;
  keep every value derived from the counter to stay deterministic.
- The output is a file on the node. To use it in a CI job on another machine,
  copy it from the node or replace the `file` output.

## Related

- [Build by Job](/recipes): the task-to-components matrix
- [`generate` input](/components/inputs/generate) and [Testing & Debugging](/getting-started/testing-debugging)

## Data Migration Engine

# Data Migration Engine

Read every row of a legacy table, reshape it into a new schema, and write it to
the target in batches. Rows that fail validation go to a rejects file with the
reason, instead of into the target. Running the job again is safe: rows already
migrated are skipped.

**Components:** `sql_select` → `mapping` → `switch` output, with `sql_insert`
for valid rows and `file` for rejects.

**Needs:** network access from the node to both databases, and a connection
string (DSN) for each. The runs used PostgreSQL 14 for both. This is a batch
copy of what is in the table when the job starts; it does not follow later
changes.

## Complete job

The mapping splits `"Last, First"` names, normalises email, parses `MM/DD/YYYY`
dates, decodes one-letter status codes, and turns integer cents into a decimal
string. An email without `@` is marked for rejection, and an unknown status
code throws; the output sends both kinds of row to the rejects file. The runs
exercised the invalid-email case only.

```yaml title="data-migration.yaml"
name: data-migration
type: pipeline
# One-shot job: stop after one failed execution instead of retrying.
restart_policy: never
selector:
  match_labels:
    pipeline_role: migration
config:
  input:
    sql_select:
      driver: postgres
      dsn: "postgres://USER:PASS@SRC_HOST:5432/SRC_DB"
      table: customers
      columns: [cust_no, full_name, email, signup, status_code,
        balance_cents]
      suffix: ORDER BY cust_no
  pipeline:
    processors:
      - mapping: |
          let parts = this.full_name.split(",")
          let email = this.email.trim().lowercase()
          let cents = this.balance_cents.number().int64()
          root.id = this.cust_no
          root.last_name = $parts.index(0).trim()
          root.first_name = $parts.index(1).trim()
          root.email = $email
          root.signed_up_on = this.signup.ts_strptime(
            "%m/%d/%Y"
          ).ts_format("2006-01-02", "UTC")
          root.status = match this.status_code {
            "A" => "active"
            "I" => "inactive"
            "S" => "suspended"
            _ => throw("unknown status code " + this.status_code)
          }
          # Signed cents: format the absolute value, then prefix the
          # sign, so -150 becomes -1.50 and -5 becomes -0.05.
          let sign = if $cents < 0 { "-" } else { "" }
          let abs = if $cents < 0 { 0 - $cents } else { $cents }
          root.balance = "%s%d.%02d".format(
            $sign, ($abs / 100).floor().int64(), ($abs % 100).int64()
          )
          root.reject_reason = if !$email.contains("@") {
            "invalid email"
          } else { null }
  output:
    switch:
      cases:
        - check: errored() || this.reject_reason != null
          output:
            file:
              path: "/var/tmp/expanso-migration/out/rejects.jsonl"
              codec: lines
            processors:
              - mapping: |
                  root = this
                  root.error = if errored() { error() } else { null }
        - output:
            sql_insert:
              driver: postgres
              dsn: "postgres://USER:PASS@DST_HOST:5432/DST_DB"
              table: customers
              columns: [id, first_name, last_name, email,
                signed_up_on, status, balance, migrated_from]
              args_mapping: |
                root = [
                  this.id, this.first_name, this.last_name,
                  this.email, this.signed_up_on, this.status,
                  this.balance,
                  "legacy.customers"
                ]
              suffix: ON CONFLICT (id) DO NOTHING
              max_in_flight: 4
              batching:
                count: 100
                period: 1s
```

The `selector` sends the job to a node labelled `pipeline_role: migration`; set
that label on your node, or change the selector to a label it already carries.
With no matching node, the job is stored and never runs.

Replace the two DSNs with your own. The runs used two local databases without
TLS, with `?sslmode=disable` on both DSNs; add it only in that situation.

The rejects path is on the **node that runs the job**; create its directory
there first:

```bash
mkdir -p /var/tmp/expanso-migration/out
```

## Try it on synthetic data

The target table needs a primary key on `id` for `ON CONFLICT (id)` to work.
This is the target schema the runs used:

```sql title="target.sql"
CREATE TABLE customers (
  id            integer PRIMARY KEY,
  first_name    text    NOT NULL,
  last_name     text    NOT NULL,
  email         text    NOT NULL UNIQUE,
  signed_up_on  date    NOT NULL,
  status        text    NOT NULL
    CHECK (status IN ('active','inactive','suspended')),
  balance       numeric(12,2) NOT NULL,
  migrated_from text    NOT NULL,
  migrated_at   timestamptz NOT NULL DEFAULT now()
);
```

And a synthetic source of 1,000 rows, 6 of them with an invalid email on
purpose:

```sql title="legacy.sql"
-- Rows where cust_no % 143 = 0 (6 rows: 143..858) have an
-- invalid email on purpose.
CREATE TABLE customers (
  cust_no       integer PRIMARY KEY,
  full_name     text    NOT NULL,          -- "Last, First"
  email         text    NOT NULL,          -- mixed case, stray spaces
  signup        text    NOT NULL,          -- MM/DD/YYYY
  status_code   char(1) NOT NULL,          -- A / I / S
  balance_cents text    NOT NULL           -- integer cents as text
);
INSERT INTO customers
SELECT n,
       'Surname' || n || ', Given' || n,
       CASE WHEN n % 143 = 0 THEN ' user' || n || '.example.invalid '
            ELSE ' User' || n || '@Example.INVALID ' END,
       to_char(date '2019-01-01' + (n * 3), 'MM/DD/YYYY'),
       (ARRAY['A','I','S'])[1 + n % 3],
       ((n * 7919) % 500000)::text
FROM generate_series(1, 1000) AS n;
```

Load them into two separate databases, with `SOURCE_DSN` and `TARGET_DSN` set
to their connection strings, then deploy:

```bash
psql "$TARGET_DSN" -v ON_ERROR_STOP=1 -f target.sql
psql "$SOURCE_DSN" -v ON_ERROR_STOP=1 -f legacy.sql
expanso-cli job validate data-migration.yaml --offline
expanso-cli job deploy data-migration.yaml
```

## Check it

Count the rows that landed, and the rejects on the node:

```bash
psql "$TARGET_DSN" -Atc "SELECT count(*) FROM customers"
wc -l < /var/tmp/expanso-migration/out/rejects.jsonl
```

With the synthetic data, expect `994` and `6`. Run the job a second time and
the target count stays at `994`.

## What the run proved

- 1,000 source rows produced 994 target rows and exactly 6 rejects, each
  carrying `"reject_reason": "invalid email"`.
- Every target row was identical to the same transform written independently
  in SQL.
- A second run left the target unchanged at 994 rows.
- The job was deployed through Expanso Cloud to a labelled node running
  `expanso-edge` v2.1.21, against two local PostgreSQL 14.23 databases.
  Those Cloud runs used the job without `restart_policy`.
- The job with `restart_policy: never`, before the balance fix below, was
  run again on a local-mode node running v2.1.21: 994 target rows and 6
  invalid-email rejects, the target identical to the independent SQL
  transform, and a second run completed with the target still at 994 rows.
  Each run completed after 1 execution.
- **Negative balances, local-mode node only.** The balance formatter was
  corrected after the runs above, which had only positive balances: the
  earlier one turned -150 cents into `-1.-5`, which PostgreSQL rejects. The
  published job was run on a local-mode node against 9 rows from -99999 to
  12345 cents. All 9 were written as exact decimals (for example `-1.50`,
  `-0.05`, `-999.99`, `0.00`, `123.45`), identical to an independent SQL
  transform, after 1 execution. A second run left the table unchanged; the
  rejects file gets its reject lines again on every run.

## Limits

- Only PostgreSQL to PostgreSQL was run. `sql_select` and `sql_insert` accept
  other drivers (see their component pages); those were not part of the run.
- The rejects file is on the node. Replace that `file` output to send rejects
  somewhere else.
- `ON CONFLICT (id) DO NOTHING` skips rows that already exist; it does not
  update them. Change the `suffix` if you need an upsert.
- The published job, with `restart_policy: never`, was proved on a
  local-mode node only; the Cloud runs used the earlier job without it. With
  `never`, a failed run stops after one execution instead of being re-run.
  See [Proving your own run](/recipes#proving-your-own-run).

## Related

- [Build by Job](/recipes): the task-to-components matrix
- [`sql_select` input](/components/inputs/sql_select) and [`sql_insert` output](/components/outputs/sql_insert)

## Log Reduction

# Log Reduction

Parse an access log, drop health-check lines, keep every error (status 400 and
above) as its own structured event, and collapse the remaining successful
requests into one count per minute, route and status.

**Components:** `broker` input batching a `file` input (`lines` scanner) →
`mapping` (regex parse, health-check drop) → `switch` processor: errors get a
`mapping`, successes go through `group_by_value` → `archive` → `mapping` →
`file` output.

**Needs:** a log file on the node, in combined log format with the request
time in seconds appended as the last field. A line from the run:
`10.0.0.0 - - [17/Sep/2026:12:00:00 +0000] "GET /healthz HTTP/1.1" 200 100 "-" "synthetic-agent/1.0" 0.000`

:::note[Proved on a local-mode node, with synthetic data]
This job ran end to end on an `expanso-edge` v2.1.21 node in local mode, not
through Expanso Cloud, against a generated 5,000-line log.
:::

## Complete job

```yaml title="log-reduction.yaml"
name: log-reduction
type: pipeline
# One-shot job: stop after one failed execution instead of retrying.
restart_policy: never
selector:
  match_labels:
    pipeline_role: logs
config:
  input:
    broker:
      inputs:
        - file:
            paths: ["/var/log/app/access.log"]
            scanner:
              lines: {}
      batching:
        count: 100000
        period: 2s
  pipeline:
    processors:
      - mapping: |
          let m = content().string().re_find_object(
            "^(?P<ip>\\S+) \\S+ \\S+ " +
            "\\[(?P<ts>[^\\]]+)\\] " +
            "\"(?P<method>\\S+) (?P<path>\\S+) " +
            "[^\"]*\" (?P<status>\\d{3}) (?P<bytes>\\d+) " +
            "\"[^\"]*\" \"[^\"]*\" (?P<latency>[0-9.]+)$"
          )
          root = if $m.path == "/healthz" { deleted() } else {
            {
              "minute": $m.ts.slice(0, 17),
              "method": $m.method,
              "path": $m.path,
              "status": $m.status.number(),
              "latency_ms": ($m.latency.number() * 1000).round(),
              "ip": $m.ip
            }
          }
      - switch:
          - check: this.status >= 400
            processors:
              - mapping: 'root = this.merge({"kind": "error"})'
          - processors:
              - group_by_value:
                  value: '${! json("minute") } ${! json("path") }
                    ${! json("status") }'
              - archive:
                  format: json_array
              - mapping: |
                  root = {
                    "kind": "rollup",
                    "minute": this.index(0).minute,
                    "path": this.index(0).path,
                    "status": this.index(0).status,
                    "count": this.length()
                  }
  output:
    file:
      path: "/var/tmp/expanso-logs/out/reduced.jsonl"
      codec: lines
```

The `selector` sends the job to a node labelled `pipeline_role: logs`; set
that label on your node, or change the selector to a label it already carries.
With no matching node, the job is stored and never runs.

Replace `/var/log/app/access.log` with your log. Both paths are on the **node
that runs the job**, and the user running the agent must be able to read the
log and write the output. Create the output directory there first:

```bash
mkdir -p /var/tmp/expanso-logs/out
```

If your log format differs, change the regular expression in the first
`mapping`; the named groups become the fields.

## Deploy and check it

```bash
expanso-cli job deploy log-reduction.yaml
wc -l < /var/log/app/access.log
wc -l < /var/tmp/expanso-logs/out/reduced.jsonl
```

Run the last two commands on the node once the job has completed. The output
holds one event per error and one rollup per minute, route and status. Two
lines from the run, pretty-printed:

```json
{
  "ip": "10.0.4.95",
  "kind": "error",
  "latency_ms": 95,
  "method": "GET",
  "minute": "17/Sep/2026:12:00",
  "path": "/api/users",
  "status": 404
}
{
  "count": 40,
  "kind": "rollup",
  "minute": "17/Sep/2026:12:00",
  "path": "/api/orders",
  "status": 200
}
```

## What the run proved

- 5,000 synthetic lines went in and 330 events came out. The 3,000 health-check
  lines were dropped.
- All 250 lines with status 400 or above were kept, each as its own event.
- The 80 rollups (one per minute, route and status) matched an independent
  recount from the raw log, and every non-health line was accounted for.
- The file shrank from 546,200 to 39,725 bytes, 92.7% smaller. That figure
  belongs to this synthetic log, in which 60% of lines were health checks. It
  is not a benchmark; your reduction depends on your traffic.
- The job exactly as published, `restart_policy: never` included, ran on a
  local-mode node running `expanso-edge` v2.1.21 and completed after 1
  execution.

## Limits

- Proved on a local-mode node only, not through Expanso Cloud.
- With `restart_policy: never`, a failed run stops after one execution
  instead of being re-run.
  See [Proving your own run](/recipes#proving-your-own-run).
- The file is read in batches of up to 100,000 lines or 2 seconds, and rollups
  are counted per batch. The run's 5,000-line file fit in one batch.
- Successful requests are reduced to counts: individual IPs, latencies and
  timestamps for those requests are not kept.
- The output is a file on the node. Replace the `file` output to ship the
  reduced events elsewhere.

## Related

- [Build by Job](/recipes): the task-to-components matrix
- [`group_by_value` processor](/components/processors/group_by_value) and [`archive` processor](/components/processors/archive)

## Notification Engine

# Notification Engine

Receive events on a webhook, drop duplicates, keep only the critical and high
severity ones, format a one-line message, and POST it to a notification
endpoint, retrying when the endpoint fails.

**Components:** `http_server` input → `dedupe` (`memory` cache) → `mapping`
(severity filter) → `mapping` (message format) → `http_client` output with
retries.

**Needs:** an HTTP endpoint you control that accepts a JSON `POST` and answers
`2xx`. The job sends it the formatted message only.

:::warning[Delivery was proved against a test receiver only]
The body is `{"text": ...}` plus two fields, the shape Slack incoming webhooks
accept, but no message was sent to Slack, email or any other real service.
Point the job at your real receiver and check that messages arrive before you
depend on it.
:::

## Complete job

```yaml title="notification-engine.yaml"
name: notification-engine
type: pipeline
selector:
  match_labels:
    pipeline_role: notify
config:
  input:
    http_server:
      address: "127.0.0.1:8088"
      path: /events
      allowed_verbs: [POST]
  pipeline:
    processors:
      - dedupe:
          cache: seen_events
          key: '${! json("id") }'
      - mapping: |
          root = if ["critical", "high"].contains(this.severity) {
            this
          } else { deleted() }
      - mapping: |
          root.text = "[%s] %s: %s (event %s at %s)".format(
            this.severity.uppercase(), this.source, this.message,
            this.id, this.ts
          )
          root.event_id = this.id
          root.severity = this.severity
  cache_resources:
    - label: seen_events
      memory: {}
  output:
    http_client:
      url: "https://RECEIVER_HOST/notify"
      verb: POST
      headers:
        Content-Type: application/json
      retries: 5
      retry_period: 200ms
      max_in_flight: 1
```

The `selector` sends the job to a node labelled `pipeline_role: notify`; set
that label on your node, or change the selector to a label it already carries.
With no matching node, the job is stored and never runs.

Replace `https://RECEIVER_HOST/notify` with your endpoint. The webhook listens
on `127.0.0.1:8088` of the **node that runs the job**, so only processes on
that node can reach it; change `address` to accept events from elsewhere.

## Send it test events

Save these eight events on the node as `events.json`. Two are critical, one is
high, one repeats an earlier `id`, and the rest are below the threshold:

```json title="events.json"
[
  {
    "id": "evt-001",
    "severity": "critical",
    "source": "pump-4",
    "message": "Discharge pressure above limit",
    "ts": "2026-09-17T12:00:00Z"
  },
  {
    "id": "evt-002",
    "severity": "info",
    "source": "pump-4",
    "message": "Heartbeat",
    "ts": "2026-09-17T12:00:05Z"
  },
  {
    "id": "evt-003",
    "severity": "high",
    "source": "chiller-2",
    "message": "Coolant temperature rising",
    "ts": "2026-09-17T12:01:00Z"
  },
  {
    "id": "evt-004",
    "severity": "low",
    "source": "gateway-1",
    "message": "Clock skew 40ms",
    "ts": "2026-09-17T12:01:30Z"
  },
  {
    "id": "evt-001",
    "severity": "critical",
    "source": "pump-4",
    "message": "Discharge pressure above limit",
    "ts": "2026-09-17T12:00:00Z"
  },
  {
    "id": "evt-005",
    "severity": "warning",
    "source": "chiller-2",
    "message": "Filter change due",
    "ts": "2026-09-17T12:02:00Z"
  },
  {
    "id": "evt-006",
    "severity": "critical",
    "source": "gateway-1",
    "message": "Uplink lost",
    "ts": "2026-09-17T12:03:00Z"
  },
  {
    "id": "evt-007",
    "severity": "info",
    "source": "gateway-1",
    "message": "Uplink restored",
    "ts": "2026-09-17T12:04:00Z"
  }
]
```

Deploy the job, then post each event to the webhook from the node. This needs
`curl` and `jq`:

```bash
expanso-cli job deploy notification-engine.yaml
jq -c '.[]' events.json | while IFS= read -r ev; do
  curl -s -o /dev/null -w '%{http_code}\n' -X POST \
    -H 'Content-Type: application/json' --data "$ev" \
    http://127.0.0.1:8088/events
done
```

Each post should print `200`.

## Check it

Your receiver should get exactly three messages. Each body is a JSON object
with `event_id`, `severity` and `text`. The `text` values the test receiver
recorded were:

- `[CRITICAL] pump-4: Discharge pressure above limit (event evt-001 at 2026-09-17T12:00:00Z)`
- `[HIGH] chiller-2: Coolant temperature rising (event evt-003 at 2026-09-17T12:01:00Z)`
- `[CRITICAL] gateway-1: Uplink lost (event evt-006 at 2026-09-17T12:03:00Z)`

This job keeps running and listening until you stop it.

## What the run proved

- 8 events were accepted. Exactly the 3 critical or high events were
  delivered, and the repeated event was delivered once.
- The receiver answered `503` to the first two attempts. Both were retried, for
  5 attempts in total, and every message arrived.
- The job was deployed through Expanso Cloud to a labelled node running
  `expanso-edge` v2.1.21. The receiver was a local test server on the same
  host.

## Limits

- Slack, email, Discord and PagerDuty were not contacted. There is no
  dedicated component for any of them; they are reached with `http_client`.
- The `memory` dedupe cache is lost when the job restarts, so an event resent
  after a restart is delivered again. The
  [RSS feed engine](/recipes/rss-feed-engine) shows a `file` cache that
  survives restarts.

## Related

- [Build by Job](/recipes): the task-to-components matrix
- [`http_server` input](/components/inputs/http_server) and [`http_client` output](/components/outputs/http_client)

## RAG: Embed and Retrieve Documents

# RAG: Embed and Retrieve Documents

Two jobs. The **ingest** job fetches documents over HTTP, splits each into
paragraph chunks that keep the URL they came from, embeds every chunk, and
writes the vectors to Qdrant. The **search** job embeds each question with the
same model and returns the closest chunks, each traced back to its source.

**Ingest components:** `generate` (one pass) → `http` (document index) →
`mapping` → `unarchive` → `mapping` → `http` (each document) → `mapping` →
`mapping` (paragraph chunks) → `unarchive` → `branch` with
`ollama_embeddings` → `mapping` → `http_client` output to the Qdrant REST
API.

**Search components:** `file` input → `branch` with `ollama_embeddings` →
`branch` with the `qdrant` processor → `mapping` → `file` output.

**Needs:**

- An [Ollama](https://ollama.com) server with the `nomic-embed-text` model,
  reachable from the node. The runs used Ollama 0.33.2 on the node's host.
- A [Qdrant](https://qdrant.tech) server, reachable from the node on its REST
  port (`6333`) for writes and its gRPC port (`6334`) for search. The runs used
  Qdrant 1.19.1.
- A collection whose vector size matches the model: 768, cosine distance.
- A document index: a URL returning
  `{"documents": ["https://…/a.md", "https://…/b.md"]}`.

**Where data goes:** the node fetches the index and each document from their
host. Chunk text goes to the Ollama server and vectors plus chunk text go to
Qdrant. With both on the node's own host, as in the runs, nothing else leaves
it.

:::warning[Write vectors with http_client, not the qdrant output]
On `expanso-edge` v2.1.21 the native `qdrant` output could not read message
fields in its `id` mapping. Every write failed, and the job still reported
`completed` with 0 points stored. The ingest job below builds the Qdrant upsert
body in a `mapping` and sends it with `http_client`, which worked. The `qdrant`
**processor**, used for search, works.
:::

## Set up the dependencies

On the host that will serve the model and the vector store:

```bash
ollama pull nomic-embed-text
docker run -d --name qdrant -p 6333:6333 -p 6334:6334 \
  qdrant/qdrant
curl -s -X PUT http://127.0.0.1:6333/collections/docs \
  -H 'Content-Type: application/json' \
  -d '{"vectors":{"size":768,"distance":"Cosine"}}'
```

The last command should print `"result":true`.

## Ingest job

Each document is split on blank lines, and any paragraph that starts with
`#` is dropped whole, including body text directly under a heading with no
blank line between them. The first line becomes the title, and every chunk
carries `source_uri`, `doc`, `title`, `chunk_index`, `chunk_id` and `text`.
`chunk_id` is the full source URL plus the chunk index, and the point id is
derived from it. Documents with the same file name at different URLs get
separate points, and ingesting the same document again overwrites its points
instead of adding duplicates.

```yaml title="rag-ingest.yaml"
name: rag-ingest
type: pipeline
# One-shot job: stop after one failed execution instead of retrying.
restart_policy: never
selector:
  match_labels:
    pipeline_role: rag
config:
  input:
    generate:
      # One fetch pass. For continuous ingestion set an interval and
      # drop count, and remove `restart_policy: never` above: one
      # transient fetch error would stop a long-running job for good.
      # The default policy restarts it on failure.
      count: 1
      interval: ""
      mapping: root = {}
  pipeline:
    processors:
      # 1. Fetch the document index:
      #    {"documents": ["https://.../a.md", ...]}
      - http:
          url: "https://DOCS_HOST/docs/index.json"
          verb: GET
          retries: 3
      - mapping: |
          root = if errored() {
            throw("index fetch failed: " + error().or("unknown"))
          } else { this.documents }
      - unarchive:
          format: json_array
      # 2. Fetch each document, remembering where it came from.
      - mapping: |
          meta source_uri = this
          root = ""
      - http:
          url: '${! meta("source_uri") }'
          verb: GET
          retries: 3
      - mapping: |
          root = if errored() {
            throw("document fetch failed: " + error().or("unknown"))
          } else { content() }
      # 3. Chunk by paragraph; every chunk carries its source URI.
      - mapping: |
          let src = @source_uri
          let doc = $src.filepath_split().index(1)
          let body = content().string()
          let title = $body.split("\n").index(0).trim_prefix("# ")
          let paras = $body.split("\n\n").map_each(
            p -> p.trim()
          ).filter(p -> p != "" && !p.has_prefix("#"))
          root = $paras.enumerated().map_each(e -> {
            "source_uri": $src,
            "doc": $doc,
            "title": $title,
            "chunk_index": e.index,
            "chunk_id": $src + "#" + e.index.string(),
            "text": e.value
          })
      - unarchive:
          format: json_array
      - branch:
          # nomic-embed-text expects a task prefix on the input text.
          request_map: 'root = "search_document: " + this.text'
          processors:
            - ollama_embeddings:
                model: "nomic-embed-text"
                server_address: "http://127.0.0.1:11434"
          result_map: root.vector = this
      # Deterministic UUID from chunk_id: re-ingesting overwrites,
      # never duplicates.
      - mapping: |
          let h = this.chunk_id.hash("sha256").encode("hex")
          root.points = [{
            "id": "%s-%s-%s-%s-%s".format(
              $h.slice(0, 8), $h.slice(8, 12), $h.slice(12, 16),
              $h.slice(16, 20), $h.slice(20, 32)
            ),
            "vector": this.vector,
            "payload": this.without("vector")
          }]
  output:
    http_client:
      url: "http://127.0.0.1:6333/collections/docs/points?wait=true"
      verb: PUT
      headers:
        Content-Type: application/json
      retries: 3
      retry_period: 500ms
      max_in_flight: 4
```

The `selector` sends the job to a node labelled `pipeline_role: rag`; set
that label on your node, or change the selector to a label it already carries.
With no matching node, the job is stored and never runs.

Replace `https://DOCS_HOST/docs/index.json` with your document index. If Ollama
or Qdrant is not on the node's own host, change `server_address` and the
`http_client` URL.

A fetch that fails, for the index or any document, throws, so the job fails
instead of completing with nothing stored.

## Search job

The questions file is on the **node that runs the job**, one JSON object per
line with a `question` field. The job also copies an optional `expected_doc`
field through, which is useful for testing.

```yaml title="rag-query.yaml"
name: rag-query
type: pipeline
# One-shot job: stop after one failed execution instead of retrying.
restart_policy: never
selector:
  match_labels:
    pipeline_role: rag
config:
  input:
    file:
      paths: ["/var/tmp/expanso-rag/questions.jsonl"]
      scanner:
        lines: {}
  pipeline:
    processors:
      - branch:
          request_map: 'root = "search_query: " + this.question'
          processors:
            - ollama_embeddings:
                model: "nomic-embed-text"
                server_address: "http://127.0.0.1:11434"
          result_map: root.vector = this
      - branch:
          processors:
            - qdrant:
                grpc_host: "127.0.0.1:6334"
                collection_name: "docs"
                vector_mapping: root = this.vector
                # payload_fields defaults to [] (no payload); ask
                # for the source-tracing fields explicitly.
                payload_fields: [source_uri, doc, chunk_id,
                  chunk_index, text]
                payload_filter: include
                limit: 3
          result_map: root.hits = this
      # The qdrant processor returns proto3-JSON hits
      # ({"stringValue": ...}); flatten them into plain fields for
      # whatever consumes the answers.
      - mapping: |
          root.question = this.question
          root.expected_doc = this.expected_doc
          root.hits = this.hits.map_each(h -> {
            "id": h.id.uuid,
            "score": h.score,
            "source_uri": h.payload.source_uri.stringValue.or(null),
            "doc": h.payload.doc.stringValue,
            "chunk_id": h.payload.chunk_id.stringValue,
            "chunk_index":
              h.payload.chunk_index.integerValue.number(),
            "text": h.payload.text.stringValue
          })
  output:
    file:
      path: "/var/tmp/expanso-rag/out/answers.jsonl"
      codec: lines
```

The `qdrant` processor returns no payload unless `payload_fields` lists the
fields, and returns hits in a typed form such as `{"stringValue": "..."}`. The
last `mapping` flattens them.

## Run and check it

Deploy the ingest job, then count the stored points:

```bash
expanso-cli job deploy rag-ingest.yaml
curl -s http://127.0.0.1:6333/collections/docs \
  | grep -o '"points_count":[0-9]*'
```

The count equals the number of chunks across your documents. Deploying the
ingest job again leaves it unchanged. Then, on the node, write a question and
run the search:

```bash
mkdir -p /var/tmp/expanso-rag/out
printf '%s\n' '{"question":"When should the battery be swapped?"}' \
  > /var/tmp/expanso-rag/questions.jsonl
expanso-cli job deploy rag-query.yaml
cat /var/tmp/expanso-rag/out/answers.jsonl
```

Each line of `answers.jsonl` holds the question and its top 3 hits, each with
`score`, `source_uri`, `doc`, `chunk_id`, `chunk_index` and `text`.

## What the run proved

The runs used four synthetic Markdown documents served over HTTP, such as a
field-sensor battery guide and a data-center cooling runbook.

- Each ingest run fetched the index once and each of the 4 documents once.
- The 4 documents became 16 chunks. Every stored payload matched its source
  URL, document, chunk index and text.
- Every vector had 768 dimensions. A stored vector and a direct Ollama
  embedding of the same text had cosine similarity 1.000000.
- A second ingest run left the count at 16.
- For each of 4 questions, the top hit came from the expected document:

| Question | Top hit | Score |
|---|---|---|
| How often do the hoist wire ropes on the port cranes need a visual check? | `harbor-crane-maintenance.md#1` | 0.817 |
| When do we switch on sprinklers to keep the apple buds from freezing? | `orchard-irrigation-policy.md#2` | 0.752 |
| At what reading should the sensor battery be swapped out? | `field-sensor-battery-guide.md#1` | 0.806 |
| What inlet temperature sets off the critical alarm for the server room? | `data-center-cooling-runbook.md#1` | 0.823 |

Both jobs were deployed through Expanso Cloud to a labelled node running
`expanso-edge` v2.1.21. Ollama, Qdrant and the document server ran on the same
host as the node. Those runs built `chunk_id` from the file name, which is
why the table shows ids such as `harbor-crane-maintenance.md#1`, and did not
set `restart_policy`.

Two corrections were added later and proved on a local-mode node only, with
the same Ollama model and Qdrant version:

- **Source-URL chunk ids.** With two documents both named `guide.md` at
  different URLs plus a third document, the file-name id stored 3 points
  because the two guides overwrote each other. The source-URL id stored all 5
  chunks, a second ingest left the count at 5, and a question about each guide
  returned a chunk from the right URL first.
- **`restart_policy: never`.** The ingest job with it completed with 5
  points. The same job pointed at a bad index failed after exactly one
  execution.

Both jobs exactly as published, source-URL chunk ids and
`restart_policy: never` included, were then run on a local-mode node against
the four synthetic documents. Ingest stored 16 chunks, each `chunk_id` the
source URL plus `#` plus the chunk index. The query job answered all 4
questions with the top hit from the expected document. Each job completed
after 1 execution.

## Limits

- Only `ollama_embeddings` with `nomic-embed-text` and Qdrant were run. The
  catalog also has `aws_bedrock_embeddings`, `gcp_vertex_ai_embeddings` and
  `openai_embeddings`, and chat processors such as `ollama_chat`; none of them
  was part of these runs.
- Chunks are paragraphs. There is no token-aware or overlapping splitter, and
  a very long paragraph becomes one long chunk.
- A paragraph that starts with `#` is dropped whole, so text written directly
  under a heading is never embedded and nothing reports it. Put a blank line
  after each heading, or change the chunk filter (a change that was not run).
- The ingest job does one pass. For continuous ingestion, the job's comment
  says to set an `interval` and drop `count`; that was not run. Such a
  long-running job should not keep `restart_policy: never`, or a single
  transient fetch error stops it for good. Choose a policy suited to a
  long-running job; the default restarts it on failure.
- `nomic-embed-text` expects the `search_document: ` and `search_query: `
  prefixes the jobs add. Another model may expect different input.
- The published jobs, with source-URL chunk ids and
  `restart_policy: never`, were proved on a local-mode node only; the Cloud
  runs used the earlier jobs. See What the run proved.

## Related

- [Build by Job](/recipes): the task-to-components matrix
- [`ollama_embeddings` processor](/components/processors/ollama_embeddings) and [`qdrant` processor](/components/processors/qdrant)

## RSS Feed Engine

# RSS Feed Engine

Poll an RSS 2.0 feed, turn each item into a JSON record, and keep a
file-backed record of every item already seen, so no item is emitted twice,
even after the job restarts.

**Components:** `generate` (the poll schedule) → `http` → `mapping` (fetch
error check) → `xml` (`to_json`) → `mapping` → `unarchive` → `mapping` →
`dedupe` with a `file` cache → `file` output.

**Needs:** a node with outbound HTTPS access to the feed. No credentials.

## Complete job

The job below polls the NASA news-release feed once. The `selector` sends it
to a node labelled `pipeline_role: feeds`; set that label on your node, or
change the selector to a label it already carries. With no matching node, the
job is stored and never runs.

The feed URL appears twice in the job: in `http.url` and in the mapping's
`root.feed`. To switch feeds, change both together.

```yaml title="rss-feed-engine.yaml"
name: rss-feed-engine
type: pipeline
# One-shot job: stop after one failed execution instead of retrying.
restart_policy: never
selector:
  match_labels:
    pipeline_role: feeds
config:
  input:
    generate:
      # One bounded poll. For a standing engine, drop `count` and set
      # `interval` to the feed's polling period, e.g. "5m". Also
      # remove `restart_policy: never` above: one transient fetch
      # error would stop a long-running job for good. The default
      # policy restarts it on failure.
      count: 1
      interval: 1s
      mapping: root = {}
  pipeline:
    processors:
      - http:
          url: "https://www.nasa.gov/news-release/feed/"
          verb: GET
          retries: 3
      # Stop here with the fetch error before XML parsing.
      - mapping: |
          root = if errored() {
            throw("feed fetch failed: " + error().or("unknown"))
          } else { content() }
      - xml:
          operator: to_json
      - mapping: |
          let items = this.rss.channel.item
          root = if $items.type() == "array" { $items } else {
            [$items]
          }
      - unarchive:
          format: json_array
      - mapping: |
          let g = this.guid
          root.guid = if $g.type() == "object" { $g."#text" } else {
            $g
          }
          root.title = this.title
          root.link = this.link
          root.category = this.category
          root.summary = this.description
          # RSS 2.0 dates are RFC 822: numeric zone (+0000) or a
          # name (GMT). Bind to a variable: inside .catch(...)
          # `this` is not the item.
          let pd = this.pubDate
          root.published_at = $pd.ts_strptime(
            "%a, %d %b %Y %H:%M:%S %z"
          ).catch($pd.ts_strptime(
            "%a, %d %b %Y %H:%M:%S %Z"
          )).ts_format("2006-01-02T15:04:05Z07:00", "UTC")
          root.feed = "https://www.nasa.gov/news-release/feed/"
      - dedupe:
          cache: seen_guids
          key: '${! json("guid").hash("sha256").encode("hex") }'
  cache_resources:
    - label: seen_guids
      file:
        directory: "/var/tmp/expanso-rss/state/seen-guids"
  output:
    file:
      path: "/var/tmp/expanso-rss/out/rss-items.jsonl"
      codec: lines
```

The two paths are on the **node that runs the job**, not the machine you
deploy from, and the user running the agent must be able to write to them.
Create the directories on that node first:

```bash
mkdir -p /var/tmp/expanso-rss/state/seen-guids \
  /var/tmp/expanso-rss/out
```

## Deploy and check it

```bash
expanso-cli job validate rss-feed-engine.yaml --offline
expanso-cli job deploy rss-feed-engine.yaml
expanso-cli job describe rss-feed-engine
```

Then, on the node, count what arrived and compare it with the feed:

```bash
wc -l < /var/tmp/expanso-rss/out/rss-items.jsonl
curl -sL https://www.nasa.gov/news-release/feed/ \
  | grep -o '<item>' | wc -l
```

A live feed changes over time, so compare the two counts close together, and
expect a difference if the feed published or dropped items in between. Run the
job again with the same cache directory and the file gains lines only for
items it has not seen before.

Each line is one item with the fields `guid`, `title`, `link`, `category`,
`summary`, `published_at` and `feed`. One record from the run on 2026-09-18
(UTC), with its long `link`, `category` and `summary` values left out; today's
feed will carry different items:

```json
{
  "guid": "https://www.nasa.gov/?post_type=press-release&p=1048343",
  "title":
    "NASA Awards Launch Services for StarBurst Gamma-Ray Detector",
  "published_at": "2026-09-17T20:37:09Z",
  "feed": "https://www.nasa.gov/news-release/feed/"
}
```

## What the run proved

- **Live feeds, 2026-09-18.** One poll of the NASA feed produced 10 records and one poll of
  the BBC Technology feed produced 21. In both cases the GUIDs matched an
  independent fetch of the same feed taken before and after the run. BBC dates
  use `GMT` and NASA dates use `+0000`; the date mapping handles both.
- **Deduplication across restarts.** Against a local test feed containing a
  duplicate GUID, the job emitted 4 distinct items. A second run with the same
  cache directory emitted 0.
- **Where it ran.** Each job above was deployed through Expanso Cloud to a
  labelled node running `expanso-edge` v2.1.21. Those runs used an earlier
  job without the fetch error check and without `restart_policy`.
- **The published job, local-mode node only.** The job exactly as published,
  fetch error check and `restart_policy: never` included, was run on a
  local-mode node running v2.1.21. Against a local test feed it completed
  after 1 execution with 4 distinct items, and a second run with the same
  cache emitted 0. A `401` and a `503` response each failed the job after
  exactly 1 execution with 0 records, and the error named the fetch. A live
  poll of the NASA feed completed after 1 execution, and its GUIDs equalled
  an independent fetch. Without the check, a `401` also failed the job with
  0 records, at the `xml` step instead: the check makes the failure name the
  fetch, it does not change whether the job fails.

## Limits

- The runs used bounded polls (`count: 1` against the live feeds, `count: 3`
  against the test feed). A standing engine drops `count` and sets `interval`,
  as the comment in the job says; that schedule was not part of the runs.
  Such a long-running job should not keep `restart_policy: never`, or a
  single transient fetch error stops it for good. Choose a policy suited to
  a long-running job; the default restarts it on failure.
- The output is a file on the node. To send items elsewhere, replace the
  `file` output with another output, such as `http_client`.
- The mapping reads RSS 2.0 (`rss.channel.item`). An Atom feed uses a
  different structure (`feed.entry`) and needs its own mapping.
- The dedupe key is the item's `guid`, which RSS 2.0 makes optional. Items
  without one are never emitted: hashing the missing `guid` fails, so every
  such item is dropped and the run is marked failed. If your feed omits
  `guid`, key on another field such as `link` instead; that variant was not
  run.
- The published job, with `restart_policy: never`, was proved on a
  local-mode node only; the Cloud runs used the earlier job without it. With
  `never`, a failed run stops after one execution instead of being re-run.
  See [Proving your own run](/recipes#proving-your-own-run).

## Related

- [Build by Job](/recipes): the task-to-components matrix
- [`xml` processor](/components/processors/xml) and [`dedupe` processor](/components/processors/dedupe)

## Sensor Telemetry over MQTT

# Sensor Telemetry over MQTT

Subscribe to every sensor topic on an MQTT broker, take the line and sensor
from the topic name, convert Fahrenheit to Celsius, flag readings above 80 °C
as alerts, and keep malformed readings out of the telemetry stream with the
reason they were rejected.

**Components:** `mqtt` input (wildcard topic, QoS 1) → `mapping` → `switch`
output to three `file` outputs: rejects, alerts, and telemetry.

**Needs:** an MQTT broker the node can reach, with devices publishing JSON
readings such as `{"seq": 1, "ts": "2026-09-17T12:01:00Z", "temp_f": 69.0,
"pressure_kpa": 102}` to topics shaped `plant/<line>/<sensor>/telemetry`.

:::note[Proved over MQTT with synthetic readings only]
This job ran end to end on an `expanso-edge` v2.1.21 node in local mode, not
through Expanso Cloud, with a local Mosquitto broker and a script publishing
synthetic readings. No sensor hardware, OPC UA, Modbus or protocol gateway was
involved.
:::

## Complete job

```yaml title="sensor-telemetry.yaml"
name: sensor-telemetry
type: pipeline
selector:
  match_labels:
    pipeline_role: telemetry
config:
  input:
    mqtt:
      urls: ["tcp://127.0.0.1:1883"]
      topics: ["plant/+/+/telemetry"]
      client_id: "sensor-telemetry"
      qos: 1
  pipeline:
    processors:
      - mapping: |
          let parts = @mqtt_topic.split("/")
          root.site = "plant"
          root.line = $parts.index(1)
          root.sensor = $parts.index(2)
          root.seq = this.seq
          root.ts = this.ts
          root.temp_c = if this.temp_f == null {
            throw("missing temp_f")
          } else {
            ((this.temp_f - 32) * 5 / 9).round()
          }
          root.pressure_kpa = this.pressure_kpa
          root.alert = root.temp_c > 80
  output:
    switch:
      cases:
        - check: errored()
          output:
            file:
              path: "/var/tmp/expanso-telemetry/rejects.jsonl"
              codec: lines
            processors:
              - mapping: 'root = {"error": error(),
                  "topic": @mqtt_topic, "raw": content().string()}'
        - check: this.alert
          continue: true
          output:
            file:
              path: "/var/tmp/expanso-telemetry/alerts.jsonl"
              codec: lines
        - output:
            file:
              path: "/var/tmp/expanso-telemetry/telemetry.jsonl"
              codec: lines
```

The `selector` sends the job to a node labelled `pipeline_role: telemetry`;
set that label on your node, or change the selector to a label it already
carries. With no matching node, the job is stored and never runs.

Replace `tcp://127.0.0.1:1883` with your broker. The three output paths are on
the **node that runs the job**; create their directory there first:

```bash
mkdir -p /var/tmp/expanso-telemetry
```

A reading is an alert when `temp_c` is above 80. Alerts go to `alerts.jsonl`
and, because that case sets `continue: true`, also to `telemetry.jsonl`.

## Try it with a local broker

On the node, start a Mosquitto broker in Docker, deploy the job, and publish
three readings: one normal, one hot, one with no temperature:

```bash
printf 'listener 1883\nallow_anonymous true\n' > mosquitto.conf
docker run -d --name mqtt -p 127.0.0.1:1883:1883 \
  -v "$PWD/mosquitto.conf:/mosquitto/config/mosquitto.conf:ro" \
  eclipse-mosquitto:2
expanso-cli job deploy sensor-telemetry.yaml
pub() { docker exec mqtt mosquitto_pub -q 1 -t "$1" -m "$2"; }
pub plant/line1/sensor-a/telemetry \
  '{"seq":1,"ts":"2026-09-17T12:01:00Z",'\
'"temp_f":69.0,"pressure_kpa":102}'
pub plant/line1/sensor-b/telemetry \
  '{"seq":3,"ts":"2026-09-17T12:03:00Z",'\
'"temp_f":220.0,"pressure_kpa":104}'
pub plant/line2/sensor-c/telemetry \
  '{"seq":5,"ts":"2026-09-17T12:05:00Z",'\
'"pressure_kpa":106}'
```

Then check the three files on the node:

```bash
cd /var/tmp/expanso-telemetry
wc -l telemetry.jsonl alerts.jsonl rejects.jsonl
```

Expect 2 lines in telemetry, 1 alert, and 1 reject. The reject records the
reason, the topic and the raw payload. From the run, with `raw` shown
separately:

```json
{
  "error": "failed assignment (line 7): missing temp_f",
  "topic": "plant/line2/sensor-c/telemetry",
  "raw": "..."
}
```

Its `raw` value was the payload exactly as received:
`{"seq": 5, "ts": "2026-09-17T12:05:00Z", "pressure_kpa": 106}`.

The job keeps its subscription open until you stop it.

## What the run proved

- 30 synthetic readings from 3 sensors were published at QoS 1.
- 28 valid readings reached telemetry, with line and sensor taken from the
  topic and each Fahrenheit value converted correctly.
- Exactly 3 readings, at 104 °C, were flagged as alerts, and each also
  appeared in telemetry.
- The 2 readings without `temp_f` were rejected with that reason.

## Limits

- Proved on a local-mode node only, not through Expanso Cloud.
- MQTT only. The catalog has `opcua` input and output components, but no OPC
  UA server, industrial gateway or real sensor hardware was part of this run.
- `site` is fixed to `"plant"` in the mapping; the topic supplies only line
  and sensor.
- The outputs are files on the node. Replace them to send telemetry and alerts
  elsewhere.

## Related

- [Build by Job](/recipes): the task-to-components matrix
- [`mqtt` input](/components/inputs/mqtt) and [`switch` output](/components/outputs/switch)

## Webhook Fan-Out

# Webhook Fan-Out

Accept events on one webhook and deliver every event to several downstream
endpoints. Each destination retries on its own, so a receiver that fails for a
moment still gets every event, and the others are not held back or dropped.

**Components:** `http_server` input → `mapping` (adds `received_at`) →
`broker` output with `pattern: fan_out` over three `http_client` outputs, each
with retries.

**Needs:** HTTP endpoints you control that accept a JSON `POST` and answer
`2xx`. Each receives the full event.

:::note[Proved on a local-mode node]
This job ran end to end on an `expanso-edge` v2.1.21 node in local mode, with
three local test receivers. It was not deployed through Expanso Cloud, and no
third-party service received the events.
:::

## Complete job

```yaml title="webhook-fan-out.yaml"
name: webhook-fan-out
type: pipeline
selector:
  match_labels:
    pipeline_role: webhooks
config:
  input:
    http_server:
      address: "127.0.0.1:8089"
      path: /webhook
      allowed_verbs: [POST]
  pipeline:
    processors:
      - mapping: |
          root = this
          root.received_at = now()
  output:
    broker:
      pattern: fan_out
      outputs:
        - http_client:
            url: "https://RECEIVER_A_HOST/notify"
            verb: POST
            headers: {Content-Type: application/json}
            retries: 5
            retry_period: 200ms
        - http_client:
            url: "https://RECEIVER_B_HOST/notify"
            verb: POST
            headers: {Content-Type: application/json}
            retries: 5
            retry_period: 200ms
        - http_client:
            url: "https://RECEIVER_C_HOST/notify"
            verb: POST
            headers: {Content-Type: application/json}
            retries: 5
            retry_period: 200ms
```

The `selector` sends the job to a node labelled `pipeline_role: webhooks`; set
that label on your node, or change the selector to a label it already carries.
With no matching node, the job is stored and never runs.

Replace the three receiver URLs with yours, and add or remove `http_client`
entries to match the number of destinations. The webhook listens on
`127.0.0.1:8089` of the **node that runs the job**; change `address` to accept
events from other hosts.

## Send it test events

Deploy the job, then post events to the webhook from the node:

```bash
expanso-cli job deploy webhook-fan-out.yaml
for n in 1 2 3; do
  curl -s -o /dev/null -w '%{http_code}\n' -X POST \
    -H 'Content-Type: application/json' \
    --data "{\"id\":\"order-00$n\",\"amount_cents\":${n}000}" \
    http://127.0.0.1:8089/webhook
done
```

Each post should print `200`, and each receiver should get all three events,
each with a `received_at` field added. The job keeps listening until you stop
it.

## What the run proved

- The webhook accepted 10 of 10 events.
- Each of the three receivers got each of the 10 events exactly once, with the
  payload unchanged.
- One receiver answered `503` to two attempts. Those were retried, and it
  still received all 10 events.

## Limits

- Proved on a local-mode node only, not through Expanso Cloud.
- The receivers were local test servers. Delivery into any particular SaaS
  product was not tested.
- The job adds `received_at` and otherwise forwards the event as received. It
  does not verify webhook signatures; see
  [HTTP Webhook Receiver](/examples/http-webhook) for that pattern.

## Related

- [Build by Job](/recipes): the task-to-components matrix
- [`broker` output](/components/outputs/broker) and [`http_client` output](/components/outputs/http_client)


## Full Reference

See https://docs.expanso.io/recipes for complete documentation.