Machine telemetry keeps its own canonical shape (`CanonicalMachineData`, see the
[data contract](/docs/data-contract)). Everything else that crosses the middleware (a payment
confirmed, a ticket sold, a PO approved) is a **business event**: a free-form JSON record from a
[webhook](/docs/flows#signed-webhooks), the [ingest API](/docs/rest-api) or a
[schedule](/docs/sources#schedule-source) source. Three flow nodes and two stores make those records
first class:

| Piece | What it does |
|-------|--------------|
| **`event` node** | Wraps a record in the business-event envelope. |
| **`validate` node** | Checks a record against an event type's JSON Schema; an invalid record is quarantined with the reasons. |
| **`lookup` node** | Enriches a record from a code list or cross-reference table (outlet code to cost centre, system A guest id to canonical guest id). |
| **Event catalogue** | Every event type with schema, owner and consumers, ships with defaults for common business objects, exports as AsyncAPI. |
| **Lookup tables** | Customer-maintained, edited in the dashboard, by API or by CSV import. |

Typical flow: `webhook:payments` -> **validate** `payment.confirmed` -> **lookup** `outlets`
(`${outlet_code}` -> `cost_centre`) -> **event** `payment.confirmed` -> **http** (ERP) and **sink**
`storage` (history).

## The envelope

```json
{
  "event_id": "6f1c1d4e-0f0e-4c5e-9a53-2d6a7b1e8f10",
  "type": "payment.confirmed",
  "source": "webhook:doku",
  "occurred_at": "2026-03-02T02:31:12.000Z",
  "subject": "INV-2026-000123",
  "correlation_id": "ORD-77",
  "data": { "payment_id": "PAY-1", "order_id": "INV-2026-000123", "amount": 480000, "currency": "IDR" }
}
```

| Field | Meaning |
|-------|---------|
| `event_id` | Unique id (a UUID unless the flow derives one from the record). The event history ignores a second event with the same id. |
| `type` | Catalogue type: lower-case words joined by dots, at least one dot (`payment.confirmed`, `sales.daily_summary`). |
| `source` | Producing system or flow. Default: the route key of the record, e.g. `webhook:doku`. |
| `occurred_at` | When it happened in the business. Default: when the record entered the flow. |
| `subject` | Id of the business object (an invoice number, a member id). Optional. |
| `correlation_id` | Links the events of one process (order, payment, posting). Optional. |
| `data` | The business object. Its schema is the catalogue schema of `type`. |

## `event` node

```json
{ "id": "wrap", "type": "event", "event_type": "payment.confirmed",
  "subject": "${order_id}", "correlation_id": "${order.id}", "occurred_at": "${paid_at}" }
```

| Field | Meaning |
|-------|---------|
| `event_type` | Required. Must exist in the [catalogue](#the-event-catalogue); an unknown type is a deploy error. (The field is not called `type`: that is the node kind.) |
| `source`, `event_id`, `occurred_at`, `subject`, `correlation_id` | Optional `${path.to.field}` templates over the incoming record (same rules as the [dedupe key](/docs/flows#dedupe-idempotency)). `occurred_at` accepts RFC 3339, `YYYY-MM-DD` or unix seconds / milliseconds. A template that does not resolve fails the record and names the path. |
| `data_from` | Dot-path of the object to use as `data`. Default: the whole record without its top-level keys that start with `_` (`_hook`, `_schedule` describe how the record arrived, not what it says). |

A record that already is an envelope is refused (no double wrapping).

## `validate` node

```json
{ "id": "check", "type": "validate", "schema": "payment.confirmed", "on_invalid": "quarantine" }
```

| Field | Meaning |
|-------|---------|
| `schema` | Required. Event type whose JSON Schema the record must match. Unknown type = deploy error. |
| `path` | Optional dot-path of the sub-document to check. Default: the record, or its `data` when the record is an envelope (an envelope of another `type` is refused). |
| `on_invalid` | `quarantine` (default): the node **fails** the record with every rule it broke. `drop`: discard it, counted in the node's drop figure. |

An invalid record never continues. With `quarantine` the failure reason reads like

```text
does not match event type 'payment.confirmed': /amount: -5 is less than or equal to the minimum of 0;
/method: "barter" is not one of ["card","qris","virtual_account","ewallet","bank_transfer","cash"];
"order_id" is a required property
```

The record is parked in the [dead-letter queue](/docs/operations) with that reason (inspect, correct
the value, replay), or follows the node's `error` edge when you wire one. A webhook or ingest caller
sees `422` with the reason. The schema is read from the catalogue at record time: editing an event
type applies to running flows at once, without a redeploy. Test (dry run) shows `valid` and the
`violations` as a side effect of the node.

## `lookup` node

```json
{ "id": "cc", "type": "lookup", "table": "outlets", "key": "${outlet_code}",
  "set": { "cost_centre": "cost_centre", "outlet_name": "name" },
  "on_missing": "fail" }
```

| Field | Meaning |
|-------|---------|
| `table` | Required. Name of a [lookup table](#lookup-tables). |
| `key` | Required `${path}` template, e.g. `${outlet_code}` or `pos:${guest.pos_id}` (cross-reference keys are `{system}:{id}`). Literal text without a placeholder is allowed (a constant key). |
| `set` | `{ "<record path>": "<table column>" }`: copy those columns into the record. Paths are dot-paths; missing objects are created. |
| `into` | Record path that receives the whole entry as an object. |
| `on_missing` | What a record without a usable entry does: `fail` (default) terminates it with the reason, `pass` lets it through unchanged, `default` fills it from `default`. |
| `default` | Object of column values, required with `on_missing: default`; it must cover every column `set` copies. |

At least one of `set` / `into` is required. "No usable entry" means: no entry under the key, a column
`set` asks for is absent from the entry (a half-enriched record is worse than a miss), or a placeholder
of `key` is missing from the record. A table that does not exist always fails, whatever `on_missing` says.
A lookup against a table that does not exist yet, or lacks a `set` column, is a deploy **warning**
(create the table afterwards); edits to a table apply to running flows at once.

## The event catalogue

Open **Event catalogue** in the dashboard, or `GET /api/events`. Each entry has a name, description, an
**owner** (the accountable team), a version, a JSON Schema (draft 2020-12 unless `$schema` says
otherwise; formats such as `date-time`, `email` and `uuid` are checked), an example, and:

- **producers**: flow nodes of type `event` with that `event_type`,
- **consumers**: flow nodes of type `validate` with that `schema`,

each with flow, node and the `match` of the source that feeds it. They are derived from the deployed
flow every time the catalogue is read, so they never go stale.

Shipped defaults cover the RFP business objects:

| Object | Event types |
|--------|-------------|
| guest | `guest.updated` |
| member / pass | `member.renewed` |
| ticket / order | `ticket.sold` |
| payment | `payment.confirmed` |
| sales transaction | `sales.transaction_recorded`, `sales.daily_summary` |
| financial posting | `gl.posting_created` |
| item | `item.updated` |
| vendor / PO | `vendor.updated`, `po.approved` |
| employee | `employee.updated` |
| asset | `asset.updated` |
| feedback | `feedback.received` |

Add your own types, or replace a default, with **New event type** or
`PUT /api/events/{name}` `{description, owner, schema, title?, object?, version?, example?}`. Each is
saved as `<data_dir>/events/{name}.json` (the file name is the type name), so the catalogue can be kept
in the customer-owned repository and dropped onto a new edge; a hand-edited file shows up when the
catalogue is next read. A schema that does not compile, a missing owner or description, or an `example`
that does not match the schema is refused with the reason. `DELETE /api/events/{name}` removes a custom
type (`409` while flow nodes use it) or reverts an override to the shipped default; a shipped default
cannot be deleted.

### Export (AsyncAPI)

`GET /api/events/asyncapi` (button **Export AsyncAPI**) returns an AsyncAPI 3.0 document: one channel
and message per event type (payload = the envelope with `type` pinned and `data` = the catalogue schema),
`x-owner`, `x-version`, `x-origin`, plus one `send` operation per producer node and one `receive`
operation per consumer node (`x-opex-flow` names flow, node and source). The catalogue directory,
`flow.json` and `lookups.db` together are everything needed to move the integration. Properties marked
`"x-sensitive": true` in a schema are personal data: they stay tagged in the exported schema, every
message lists them in `x-sensitive-fields` (`data.email`, ...) and carries a `personal-data` tag, and they
are masked in the event log and live feed; see [Data protection](/docs/data-protection#masking-personal-data).

### Checking a record

`POST /api/events/{name}/validate` `{record, path?}` returns `{valid, errors:[{path, message}]}`; the
dashboard has a **Test a record** tab.

## Lookup tables

A table is a **code list** (`codelist`: outlets, parks, cost centres) or a **cross-reference**
(`xref`: the ids of one guest, item or vendor in different systems). Each entry has a **key** and
named **columns** holding strings, numbers or booleans. They live in SQLite `<data_dir>/lookups.db`.
A fresh database starts with empty tables `outlets`, `parks`, `cost_centres`, `xref_guest`,
`xref_item` and `xref_vendor` so the operators see where to put their data; a table you delete stays
deleted.

Cross-reference keys are `{system}:{id}` and one canonical id can have many keys:

| key | canonical_id |
|-----|--------------|
| `pos:G-77` | `GUEST-0001` |
| `ticketing:5521` | `GUEST-0001` |
| `crm:c-9` | `GUEST-0001` |

then `{ "type": "lookup", "table": "xref_guest", "key": "pos:${guest.pos_id}", "set": { "guest.canonical_id": "canonical_id" } }`.

Edit in the dashboard (**Lookup tables**: a grid editor with search, add row / column, delete, CSV
import with a dry run first, CSV export) or through the API:

| Call | Meaning |
|------|---------|
| `GET /api/lookups` | Tables with `kind`, `owner`, `columns`, `entries` and `referenced_by` (the `lookup` nodes using it). |
| `PUT /api/lookups/{name}` `{kind, description?, owner?, columns?}` | Create or update a table. Name: `[a-z0-9][a-z0-9_-]{0,62}`; the kind of an existing table cannot change. |
| `DELETE /api/lookups/{name}[?force=true]` | Delete; `409` with `referenced_by` while nodes use it, unless `force`. |
| `GET /api/lookups/{name}/entries?q=&limit=&offset=` | Entries (search matches key and values). |
| `POST /api/lookups/{name}/entries` `{entries:[{key, values}]}` | Upsert (at most 5000 per call); new columns are added to the table. |
| `POST /api/lookups/{name}/entries/delete` `{keys:[...]}` | Delete entries. |
| `POST /api/lookups/{name}/import?mode=merge\|replace&dry_run=true` | CSV body, see below. |
| `GET /api/lookups/{name}/csv` | CSV export. |
| `GET /api/lookups/{name}/resolve?key=` | One lookup: `{found, values?}`. |

### CSV import

A header row, then one row per entry. The column named `key` (else the first column) is the key, the
others are value columns; every cell is text (an empty cell is an empty string). The separator `,`, `;`
or tab is detected from the header (Excel in an Indonesian locale saves `;`); fields may be quoted,
with `""` for a quote and line breaks inside. `merge` (default) adds new keys and updates the columns
the file carries, keeping other keys and columns; `replace` makes the table equal to the file. The
import is all-or-nothing: with any error (empty or duplicate key, a row with the wrong number of
fields, a bad column name) nothing is applied and the response (`422`) lists `errors:[{line, message}]`
(`line` counts the header as row 1). `dry_run=true` reports `{rows, created, updated, deleted, columns}`
without changing anything.

## Delivering events

`filter` and `branch` conditions work on any JSON record: read a field as `metadata.<path>`
(`metadata.status == "SUCCESS"`, `metadata.order.total > 1000000`; on an envelope
`metadata.data.amount`). Machine records are unchanged.

A `sink` node that receives an envelope delivers it to the sinks that can carry events:

| Sink | Behaviour |
|------|-----------|
| `storage` | Event history `<data_dir>/events.db`, idempotent on `event_id`, kept 90 days. Query with `GET /api/events/log?type=&subject=&correlation_id=&source=&from=&to=&limit=` (newest first, at most 500). |
| `hub` | Live feed on the WebSocket `GET /api/ws/events?type=&token=`. |

`push`, `nats`, `mqtt`, `clickhouse` and `postgres` carry machine telemetry only: an envelope reaching
them is rejected with a reason (and a deploy warning names the sink when an `event` node feeds it).
To deliver an event to another system use an [http node](/docs/flows#http-node-authentication-and-posting-mode)
(NetSuite, an ERP, a partner API) with the envelope or its `data` as the body.

## Example: payment confirmed

```json
{
  "flows": [{ "id": "main", "name": "Payments" }],
  "nodes": [
{ "id": "src", "type": "source", "match": "webhook:payments",
  "webhook": { "signature": { "scheme": "hmac_sha256", "header": "X-Signature", "secret": "${PAYMENTS_SECRET}" } } },
{ "id": "check", "type": "validate", "schema": "payment.confirmed" },
{ "id": "cc", "type": "lookup", "table": "outlets", "key": "${outlet_code}",
  "set": { "cost_centre": "cost_centre" } },
{ "id": "wrap", "type": "event", "event_type": "payment.confirmed",
  "subject": "${order_id}", "occurred_at": "${paid_at}" },
{ "id": "hist", "type": "sink", "sink": "storage" }
  ],
  "edges": [
{ "from": "src", "to": "check" }, { "from": "check", "to": "cc" },
{ "from": "cc", "to": "wrap" }, { "from": "wrap", "to": "hist" }
  ]
}
```

A payment record with a negative amount is answered `422` and parked in the dead-letter queue with the
schema violations; a valid one gets `cost_centre` from the outlets table, is wrapped and stored. The
catalogue then lists `payment.confirmed` with consumer `Payments / check` fed by `webhook:payments` and
producer `Payments / wrap`.

## Limits

- Lookup values are flat (string, number, boolean); tables are read through one SQLite connection
  (fine for lookups by key, not meant for millions of rows).
- Event history retention is fixed at 90 days.
- Events are delivered to `storage` and `hub`; delivery to messaging sinks (NATS, MQTT) is not part of
  the event model, use an http node.
- JSON Schema `$ref` to remote or file URLs is not resolved (keep schemas self-contained, `$defs` is fine).