Business events, catalogue and lookups

Machine telemetry keeps its own canonical shape (CanonicalMachineData, see the 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, the ingest API or a schedule source. Three flow nodes and two stores make those records first class:

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

The envelope

{
  "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" }
}

event node

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

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

validate node

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

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

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 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

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

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:

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.

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:

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:

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:

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 (NetSuite, an ERP, a partner API) with the envelope or its data as the body.

Example: payment confirmed

{
  "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).