events

Imported from waclawek/open-logistic (packages/events/AGENTS.md). Install upstream with npx skills add waclawek/open-logistic --skill events. Copyright stays with the author.

Events Package — Agent Guidelines

Use @open-mercato/events for all event-driven communication between modules. MUST NOT use direct module-to-module function calls for side effects.

Always

1. MUST declare events in the emitting module's events.ts — use createModuleEvents() with as const for type safety
2. MUST run yarn generate after creating or modifying events.ts files
3. MUST export metadata from every subscriber with { event, persistent?, id? }
4. MUST keep subscribers focused — one side effect per subscriber file
5. MUST make persistent subscribers idempotent — they may be retried on failure

Ask First

• Ask before renaming event IDs, changing persistent delivery semantics, or altering SSE audience filtering.
• Ask before increasing SSE payload size limits or heartbeat/deduplication behavior.

Never

• Never use direct module-to-module function calls for side effects.
• Never emit undeclared events — undeclared events trigger TypeScript errors and runtime warnings.
• Never rely on payload-provided tenant or organization scope when trusted scope is available.

Validation Commands

yarn generate
yarn workspace @open-mercato/events test
yarn workspace @open-mercato/events build

Event Declaration

Declare events in the emitting module's events.ts. See packages/core/AGENTS.md → Events for the full declaration pattern, field reference (id, label, category, entity, excludeFromTriggers), and code example.

Quick reference:

import { createModuleEvents } from '@open-mercato/shared/modules/events'
const events = [
  { id: 'module.entity.created', label: 'Entity Created', entity: 'entity', category: 'crud' },
] as const
export const eventsConfig = createModuleEvents({ moduleId: 'module', events })
export default eventsConfig

Subscription Types

Adding an Event Subscriber

1. Create subscriber file in src/modules/<module>/subscribers/<event-name>.ts
2. Export metadata with { event: 'module.entity.created', persistent: true, id: 'my-subscriber' }
3. Export default async handler function
4. Keep the handler focused on one side effect
5. Make the handler idempotent if persistent: true — it may be retried
6. Run yarn generate to register the subscriber
7. Test that the subscriber fires correctly after the event is emitted

Subscriber Contract

export const metadata = { event: 'module.entity.created', persistent: true, id: 'entity-created-notify' }
export default async function handler(payload, ctx) { /* ... */ }

Event Bus Architecture

• Supports local (in-process) and async (Redis-backed) event dispatch
• Events are auto-discovered by generators → generated/events.generated.ts
• When QUEUE_STRATEGY=async, persistent events dispatch through the queue package (BullMQ)
• When QUEUE_STRATEGY=local, persistent events process from .mercato/queue/ (or QUEUE_BASE_DIR)
• Ephemeral subscribers always run in-process regardless of queue strategy

Persistent delivery: single-delivery (OM_EVENTS_SINGLE_DELIVERY, default ON)

Single-delivery is the default. A persistent emit is delivered on exactly one path:

• the bus skips inline delivery of persistent-marked subscribers on a persistent emit (ephemeral subscribers still run inline — read-your-writes paths like query_index.upsert_one are persistent: false and are unaffected);
• the events worker dispatches persistent subscribers via matchEventPattern, so wildcard (event: '*') persistent subscribers (workflow triggers, business-rules CRUD trigger, webhook outbound dispatch) are reached.

This avoids the legacy dual-dispatch, where persistent emits ran inline and in the worker - double-running exact-match persistent subscribers (duplicate notifications/emails) and never reaching wildcard persistent subscribers in the worker. Both halves ask the SAME bus (EventBus.dispatchQueued), so they cannot disagree within a process.

No worker yet is not the same as lost. With single-delivery on, persistent subscribers run ONLY in the worker. The bus does not second-guess that against "is a worker running": the queue is durable, so a persistent emit with no worker yet is delayed, not dropped, and a worker started later drains the backlog. Delivering inline instead would move the work back onto the caller's request path - exactly what a split app/worker deployment sets AUTO_SPAWN_WORKERS=false to avoid. The mercato server/start bootstrap still reconciles the flag for a process it knows runs no worker (reconcileSingleDelivery, mirrored in packages/cli/src/lib/events-single-delivery.ts); when it does, the bus stamps the queued job persistentDeliveredInline: true so a worker draining the queue skips it. Delivery is exactly-once on either path when handlers succeed; see the failure note below.

Failure on the inline path. Inline delivery logs each handler error and continues, so a persistent subscriber that throws inline has no retry of its own. The producer therefore stamps persistentDeliveredInline only when every persistent subscriber succeeded; if one threw, the job stays unstamped and the worker runs it with queue retry + dead-lettering. A retry re-runs the persistent subscribers that already succeeded, which is why persistent subscribers MUST be idempotent.

Enqueue-only emits. Pass { persistent: true, deliverInline: false } to hand a heavy persistent job (e.g. a full query-index rebuild) to the durable queue without ANY inline delivery, independent of the single-delivery flag. Only use it when every subscriber to the event is persistent: true. This is the "Ask First: changing persistent delivery semantics" surface — coordinate before altering these defaults.

Queue Integration

When QUEUE_STRATEGY=async, persistent event workers run as background processes. Start them with:

yarn mercato queue worker events --concurrency=5

Structure

packages/events/src/
├── modules/
│   └── events/
│       └── workers/    # Async event processing workers
└── __tests__/

Workers

Workers in modules/events/workers/ handle async event processing. Follow the standard worker contract: export default handler + metadata with { queue, id?, concurrency? }.

The events worker MUST NOT keep a subscriber registry of its own - it resolves eventBus from the per-job DI container and calls dispatchQueued. It previously built one from getCliModules(), which only the mercato bin populates, so a worker started any other way dispatched zero subscribers and completed the job silently. When no bus is resolvable it now throws, so the job retries and dead-letters with a visible cause.

Testing

• Tests inside packages/events SHOULD import the public @open-mercato/events/... API when validating package behavior
• tenantId and organizationId in subscriber context are trusted scope inputs from emit(..., options) or queued job options, not from arbitrary payload fields
• Add regression tests for both paths:
  • trusted scope is forwarded when explicitly provided
  • payload-provided scope is ignored when trusted scope is omitted

Cross-Reference

• Declaring events in a module: packages/core/AGENTS.md → Events
• Adding subscribers in a module: packages/core/AGENTS.md → Events → Event Subscribers
• Queue worker contract: packages/queue/AGENTS.md

DOM Event Bridge (SSE)

The DOM Event Bridge streams server-side events to the browser via Server-Sent Events (SSE).

How It Works

1. Module declares events with clientBroadcast: true in events.ts
2. SSE endpoint at /api/events/stream subscribes to the event bus
3. Client-side eventBridge.ts connects via EventSource with auto-reconnect
4. Events are dispatched as om:event CustomEvents on window
5. Widgets/components listen via useAppEvent(pattern, handler) hook

Enabling Broadcast on Events

In your module's events.ts:

const events = [
  { id: 'mymod.entity.created', label: 'Created', category: 'crud', clientBroadcast: true },
] as const

Consuming Events in Components

import { useAppEvent } from '@open-mercato/ui/backend/injection/useAppEvent'

// Wildcard: listen to all events from a module
useAppEvent('mymod.*', (event) => {
  console.log(event.id, event.payload)
}, [])

// Exact match
useAppEvent('mymod.entity.created', (event) => {
  // refresh data
}, [])

Browser Delivery Rules

• Events are server-filtered by audience before SSE send:
  • Tenant: tenantId must match
  • Organization: organizationId or organizationIds must match selected organization
  • Recipient user: recipientUserId or recipientUserIds must include connection user
  • Recipient role: recipientRoleId or recipientRoleIds must intersect connection roles
• Missing tenantId in event payload means no delivery
• SSE sends heartbeats every 30s; client auto-reconnects if no heartbeat within 45s
• Max payload size is 4096 bytes per event
• Client deduplicates events within a 500ms window
• isBroadcastEvent(eventId) checks if an event has clientBroadcast: true
• The useEventBridge() hook must be mounted once in the app shell to start receiving events

Private cross-process coordination

Use crossProcessBroadcast: true for server-to-server invalidation or coordination that must cross process boundaries but must not be exposed through browser SSE. The event bus publishes both clientBroadcast and crossProcessBroadcast events to the cross-process bridge, while the DOM Event Bridge delivers only clientBroadcast events. Do not use clientBroadcast merely to reach another server process when the payload contains record-scoped identifiers or activity that the organization-level SSE audience cannot authorize.