Case studiesReference integration: event history and work order record

Business events into Postgres as the operational record

How OpexFlow turns work order and alarm events from JetStream into rows people can query in Postgres: one row per event, one row per work order, safe against a resend, a redelivery and an arrival in the wrong order.

Why keep the events in a table

JetStream is the log: ordered, durable, replayable. It is not where a planner asks which work orders are open on machine cnc-02, or where an auditor asks who produced the event behind a ticket. Those questions belong to SQL. The job of the flow is to write what the bus carries into tables that answer them, without ever writing the same fact twice.

Three tables

bus_events keeps every accepted event whole, with its envelope fields and its payload as jsonb. It is the history and the audit trail. work_orders holds the current state of each order, assembled from the ERP release, the MES creation and the answers of OpexMX and the ERP. alarms holds the alarms and the maintenance ticket each one opened.

bus_events: the primary key is the eventId of the envelope

CREATE TABLE bus_events (
  event_id       uuid PRIMARY KEY,
  schema_id      text        NOT NULL,
  nats_subject   text        NOT NULL,
  producer       text        NOT NULL,
  correlation_id text,
  occurred_at    timestamptz NOT NULL,
  received_at    timestamptz NOT NULL DEFAULT now(),
  payload        jsonb       NOT NULL
);

The flows write with the sql node. Every value is a bound parameter, never text in the statement. The statements are upserts, so running one a second time changes nothing.

The release adds the ERP reference and never moves a created order backwards

INSERT INTO work_orders (work_order_id, machine_id, erp_ref, wo_type, title, priority, status, released_at)
VALUES (${metadata.value.payload.workOrderId}, ${metadata.value.payload.machineId}, ${metadata.value.payload.erpRef?},
        ${metadata.value.payload.type?}, ${metadata.value.payload.title?}, ${metadata.value.payload.priority?}, 'released',
        ${metadata.value.payload.releasedAt})
ON CONFLICT (work_order_id) DO UPDATE SET
  erp_ref = COALESCE(EXCLUDED.erp_ref, work_orders.erp_ref),
  released_at = EXCLUDED.released_at,
  machine_id = COALESCE(work_orders.machine_id, EXCLUDED.machine_id),
  updated_at = now()

Three layers against a repeat

A message can come twice for three different reasons, and each has its own guard.

Why a message repeats and what stops it
Cause	Guard	Where it acts
The producer resends and sets the same Nats-Msg-Id	JetStream duplicate window (24 hours for OPEX_EVENTS)	Before OpexFlow sees the message
The producer resends without the header, or after the window	The dedupe node on the eventId, for 30 days	In the flow, before any write or call
OpexFlow redelivers after a failure further down	The primary key of bus_events and the upserts	In Postgres

The dedupe node remembers a key only after the whole path has finished, so a message that failed halfway passes it again on redelivery and is made safe by the third guard. This is the difference between at-least-once delivery and a result that is correct when read.

Tested cases

Test cases and results
Case	What happened	Result
Resent with and without the header	The same created event three times: with Nats-Msg-Id twice, then without it	The second was rejected by JetStream (duplicate: true), the third reached the flow and the dedupe node dropped it. One row in bus_events, one call to OpexMX
OpexMX answers 503	Three created events, each attempted four times while the call was refused	bus_events held one row per event the whole time. The work order rows existed without a ticket until OpexMX returned, then all three had one
Created before released	MES created the order, the ERP release arrived six seconds later	One work order row with both timestamps, the ERP reference and the acknowledgement. The release did not reset the status
Contract violation	A created event with a wrong type and no createdAt	Nothing written; parked in Dead letters at the payload check
An alarm	alarm.raised.v1 for machine cnc-02	One row in alarms with the OpexMX ticket MXT-AL-2001

The released flow does the second half of the ERP hand-over. When the release arrives after the order was created, it finds the order in Postgres with an OpexMX ticket and sends the status to the ERP itself. When it arrives before, the created flow does it. Either order ends with one acknowledged work order.

The work order released flow in the editor: a JetStream source, a filter, two validate nodes, a dedupe node, a Postgres event node and a Postgres release node in the first row; then a created-by-MES check, an OpexMX has it check, a POST of the status to the ERP and a Postgres ERP acknowledgement node in the second row. — The released flow. After recording the release it asks Postgres whether MES already created the order, and sends the status to the ERP if so.

The alarm flow in the editor: the same intake, then a Postgres alarm node, a map to a ticket, a POST to OpexMX and a Postgres node that stores the ticket number. — The alarm flow: the same intake, the alarm row, a ticket in OpexMX and its number stored on the alarm.

What the tables answer

Two work orders were taken through the system: WO-1001 in the normal order and WO-OOO-1 in the reverse order. The first query returns both with their ERP reference, OpexMX ticket and ERP acknowledgement. The second lists their events in the order they arrived, whichever system produced them. The third is the alarm and its ticket.

A terminal with three Postgres queries. work_orders: WO-1001 on cnc-01 and WO-OOO-1 on cnc-02, both created, with MXWO tickets and ERPACK acknowledgements. bus_events: four rows, released and created events of both orders with their producers erp:sap and mes. alarms: AL-2001 on cnc-02, code 2004, severity error, ticket MXT-AL-2001. — The work orders, their events and the alarm, as Postgres returns them.

A terminal showing a resent event. Three publishes: the first accepted, the second rejected by JetStream as a duplicate, the third without the header accepted at a new sequence. A query on the eventId returns count 1, and the OpexMX call log shows 1 call for the work order. — The resend: JetStream rejects the one with the header, the dedupe node drops the one without. One row, one call.

Scope

Tested against Postgres 16 with plain tables. The TimescaleDB image of the repository compose file was not used, because these tables hold events, not telemetry.
The producers and the OpexMX and ERP endpoints are simulated. The messages were checked against the opex-contracts schemas before they were published.
The statements shown are the ones that ran. Table design, indexes, retention of bus_events and the status model of a work order are set per installation.
The ERP status call can go twice if the release and the created event are processed at the same moment. The status call is idempotent in the stand-in service.
A sql or transform node that adds a top-level field to the record makes a later filter or branch node refuse it. The flows therefore keep their filters before the first SQL step and stop a record with a transform that returns nothing.