Function onIntegrationEvent

Produces an inbox subscription for an integration-event contract (M108): each event's database writes are applied ONCE per consumer, although the outbox and every broker deliver at least once.

The handler receives a fourth argument — the unit of work of the transaction in which the inbox marker (consumer, topic, envelope id) is created FIRST — and every write through it commits with the marker or not at all. A duplicate delivery is acknowledged without running the handler; a handler that fails leaves no marker, so the broker's redelivery runs it again; after any rejection the marker is re-read, and only a present marker means another delivery already handled the event. The broker queue defaults to inbox. plus a 16-character hash of the consumer and the topic; set queue for a readable name.

The promise. For one consumer name, the handler's writes through the supplied unit of work are committed at most once per topic and envelope id while the marker is retained; a delivery after the marker is purged is processed again. Nothing is promised about effects outside that unit of work, about two processes using one consumer name with different handlers, or about a database other than the store's.

Returns a RegistryFactory rather than a definition: it is resolved against the registry — in MessagingPlugin({ subscriptions }) at onInit, after the inbox store is verified; imperatively, by calling it with ctx.services at or after onInit.

Produces a SubscriptionDefinition for an integration-event contract — the declarative form, plugging straight into MessagingPlugin({ subscriptions }), or spread by hand into an imperative broker.subscribe after start().

The returned wrapper validates the delivered message against the definition, runs definition.parse, rebuilds the envelope with its data set to the parsed value, and only then invokes the application handler. A malformed envelope, a mismatched type/version, or a rejecting parser throws IntegrationEventRejectedError — the application handler is never called with an unvalidated value — and the rejection follows the broker's OWN failure path, which differs per arm and is not a retry guarantee: RabbitMQ nacks with requeue DISABLED (dead-lettered when a DLX is configured, discarded otherwise) and logs; NATS naks, which redelivers while the stream retains the message; the in-memory broker reports to onDispatchError and drops. Since a rejection here is deterministic — the same envelope fails the same way every time — redelivery cannot resolve it, so a dead-letter queue, not a retry, is the place to inspect one. When MessagingPlugin({ behaviors }) is configured, the behaviour chain runs BEFORE this wrapper, so IngressContext.payload is the raw envelope and never the parsed payload.

A handler needing a resolved capability uses the existing RegistryFactory<SubscriptionDefinition> arm of SubscriptionEntry: (services) => onIntegrationEvent(definition, handlerFor(services)) — no factory variant of this function exists, and none is needed.