Eventing Pipeline

Every entity change in ORE Studio flows through a four-stage pipeline:

1. PostgreSQL fires a pg_notify on a named channel.
2. The service's postgres_event_source deserialises the notification and publishes the entity's typed event, and an entity_change_event that carries the tenant, to the in-process event_bus.
3. A subscriber in the service publishes the typed event to NATS on the subject its entity_event_traits state.
4. A client (or any NATS subscriber) receives the NATS message and reacts.

The diagram below shows these stages for a workunit change:

┌─────────────────────────────────────────────────────────────────────┐
│ PostgreSQL                                                          │
│  ores_compute_workunits_tbl → ores_compute_workunits_notify_trg     │
│  → pg_notify('ores_compute_workunits', {event_id, entity, key,      │
│      action, version, occurred_at, correlation_id, tenant_id})      │
└───────────────────────────┬─────────────────────────────────────────┘
                            │  LISTEN/NOTIFY (libpq)
┌───────────────────────────▼─────────────────────────────────────────┐
│ ores.compute.service (server process)                               │
│  postgres_event_source → event_bus                                  │
│    workunit_event, entity_change_event                              │
│  → subscriber → nats.publish("compute.v1.events.workunits.updated") │
└───────────────────────────┬─────────────────────────────────────────┘
                            │  NATS publish/subscribe
┌───────────────────────────▼─────────────────────────────────────────┐
│ NATS subscriber (another service, or the web server's event stream) │
│  subscribe("compute.v1.workunits_events.>")                         │
└─────────────────────────────────────────────────────────────────────┘

Every table that participates in eventing has a dedicated AFTER trigger that calls pg_notify. The trigger is generated into ores.sql; the workunit example is in projects/ores.sql/create/compute/compute_workunits_notify_trigger_create.sql.

The notification payload is a JSON object, the specification's event fields plus the tenant:

The channel name (first argument to pg_notify) follows the convention ores_<table_base_name>, e.g. ores_compute_workunits for ores_compute_workunits_tbl. A versioned table's update closes the current row and the insert that follows carries the change, so the close announces nothing.

postgres_event_source (in ores.eventing) maintains a persistent PostgreSQL LISTEN connection. On each incoming notification it:

1. Deserialises the JSON payload into an entity_event_notification.
2. Finds the mapping registered for the channel and converts the notification into the entity's typed event, through the event's own entity_event_traits::from_notification, which knows the entity's key.
3. Publishes the typed event on the in-process event_bus.
4. Publishes an entity_change_event – the entity's name, the changed ids and the tenant – on the same bus.
5. Publishes a published_entity_event<Event> – the typed event with the tenant and the party the store reported – on the same bus. The generated registrar subscribes to this one and publishes to NATS.

Mappings are registered during application startup. The generated per-entity registrar does it:

event_source.register_entity_event_mapping<compute::messaging::workunit_event>(
    "ores_compute_workunits");

The argument is the PostgreSQL channel name; it must match pg_notify.

The generated registrar also subscribes to the typed event and publishes it to NATS on the subject the action names:

return event_bus.subscribe<compute::messaging::workunit_event>(
    [&nats](const compute::messaging::workunit_event& e) {
        ev::service::publish_entity_event(
            nats, ev::domain::event_subject<compute::messaging::workunit_event>(e.action), e);
    });

A NATS subscriber registers interest in a prefix's wildcard and receives the typed event as the payload. The tenant and the party are in the envelope, as X-Tenant-Id and X-Party-Id; the protocol specification defines them (Events, Tenancy of an event). The web server holds one subscription per entity for all the sessions watching it and relays the events to each browser on GET /api/events; see Change notification. The registry listens on the wildcard of each collection's prefix, such as iam.v1.accounts_events.>, and relays the time of the change and the key of the record, which the typed event carries in its key field. It joins close events into one message and names up to 20 records; see Change notification. It tells a session only of an event whose tenant is the session's own or the system tenant's, and, for a member, whose party the member works in. An event with no tenant header is told to nobody. The broker does not enforce this, so the rule is kept by the subscriber.

The events commands of ores.shell print "Event subscriptions are not available in NATS mode", so they cannot show an event. To watch one, subscribe to the wildcard with the NATS command line client, under the environment's subject prefix (the services log it at startup), and write to the entity by any path:

nats sub 'ores.dev.<environment>.compute.v1.workunits_events.>'

If nothing arrives within a second, check:

1. The SQL trigger exists: \df ores_compute_workunits_notify_fn in psql.
2. The service started its event source, and the log names the channel.
3. The subject matches exactly; subjects are case-sensitive.

Eventing is generated. The nats-eventing and nats-event-registrar facets of the entity's model produce:

1. SQL trigger — <table>_notify_trigger_create.sql, calling pg_notify after insert, update and delete.
2. Event traits — <domain>.api/include/<domain>.api/eventing/<entity>_event.hpp with the entity_event_traits specialisation that states the subject prefix. The event struct sits beside the entity's key in its protocol header.
3. Registrar — register_<entity>_event_mapping(), which maps the channel to the event and publishes it to NATS.

The service's application.cpp calls the registrar before event_source.start(), and a registrar that is not called publishes nothing. Add ores.eventing.core to the service CMakeLists.

• ores.eventing — component overview
• ores.sql — database schema and triggers
• Change notification — how the web interface consumes these events