@eco-foundation/eco-messaging

A hardened NestJS AMQP transport for Eco's services: connection lifecycle, topology as code, a confirm-based publisher, a transactional outbox with relay, and an idempotent consumer host. Extracted so that correct publish/consume behaviour is the default rather than something each service re-derives and re-reviews on its own.

pnpm add @eco-foundation/eco-messaging @eco-foundation/eco-message-contracts

@eco-foundation/eco-message-contracts is a peer, not a dependency, so installing this package alone leaves an unmet peer — install both explicitly. The remaining peers are things a NestJS service already has, and are peers so this library never pulls a second copy of your framework or AMQP client into the tree: @nestjs/common (^10 || ^11), @nestjs/core (^10.2 || ^11), amqplib (^0.10), amqp-connection-manager (^5), and zod (^3.25 || ^4). @opentelemetry/api and testcontainers are optional peers, needed only if you wire OTel or import the /testing subpath.

Note the zod floor: ^3.25.0. A service declaring an older range needs to raise the declared range even if its lockfile already resolves something newer.

The guarantee

At-least-once delivery plus idempotent consumption. This library does not provide exactly-once delivery — that primitive does not exist here — so your handler must be safe to run twice. What the library provides is that a message is not acknowledged until your handler has resolved, that a duplicate delivery of the same message ID is skipped rather than reapplied, and that a failure retries through broker topology rather than an in-process sleep.

Wiring it up

import { Module } from '@nestjs/common';
import {
  EcoMessagingModule,
  MongoOutbox,
  MongoIdempotencyLedger,
} from '@eco-foundation/eco-messaging';

@Module({
  imports: [
    EcoMessagingModule.forRoot({
      url: process.env.RABBITMQ_URL!,
      exchange: 'my-service.events',
      sourceService: 'my-service',
      // Host-supplied, already-connected. The library never opens its own datastore
      // connection — it piggybacks on yours and owns only its own collections.
      outbox: MongoOutbox.forConnection(mongooseConnection),
      ledger: MongoIdempotencyLedger.forConnection(mongooseConnection),
    }),
  ],
})
export class AppModule {}

forRoot/forRootAsync (not register) on purpose: there is exactly one messaging connection per service, and the method name says so. The module is @Global — import it once at the app root. Importing it per feature module is how a service ends up with two connections, and heartbeats are per-connection, so a publisher on one can look healthy while a consumer on the other is dead.

A publisher-only service can omit outbox/ledger entirely; an in-memory outbox is wired automatically. Registering a consumer with no ledger is a refused configuration — consuming without dedup is not allowed to boot silently degraded.

Persistence is a port with adapters, and the host chooses: MemoryOutbox/MongoOutbox and MemoryLedger/MongoIdempotencyLedger/RedisIdempotencyLedger. The library never opens its own datastore connection, never calls mongoose.connect(), and has no mongoose peer dependency — which is what lets services on different Mongoose majors share it.

Publishing

A producer publishes through the outbox by default, and for a required contract that is not merely recommended — it is the only admitted path:

import { obligationFromContract } from '@eco-foundation/eco-messaging';

await outbox.record(
  obligationFromContract(InventorySnapshot, payload, {
    now: new Date(),
    sourceService: 'my-service',
  }),
  dbSession, // your own transaction handle, so the record commits atomically with the event
);

publisher.publishNowOrThrow(contract, payload) is the fail-closed direct path: it awaits the confirm and throws on nack, return, timeout, or disconnect rather than queuing anything. It admits a required contract only when that contract declares durability: 'caller-ledger'.

On failure, EcoPublishError answers two orthogonal questions. isRetryablePublicationFailure asks "can retrying this exact message ever succeed?"; wireOutcome asks "did the broker ever see it?" — which is what decides whether a failure should spend a finite attempt budget. A caller reducing the three wire outcomes to two must cut not-attempted vs. everything-else, never broker-verdict vs. everything-else; the latter lets an ambiguous publish retry without bound.

If you lease an obligation before publishing, size the lease TTL above maxPublishWallTimeMs(options). Past that bound a slow-but-live publish has its lease stolen mid-flight and the lease starts producing duplicates.

Consuming

import { Injectable } from '@nestjs/common';
import type { HandlerContext } from '@eco-foundation/eco-messaging';
import {
  EcoConsumer,
  EcoMessageHandlerProvider,
  TransientError,
} from '@eco-foundation/eco-messaging';

@Injectable()
@EcoConsumer({ queue: 'my-service.snapshots', contracts: [InventorySnapshot] })
export class SnapshotHandler implements EcoMessageHandlerProvider {
  async handle(payload: unknown, context: HandlerContext): Promise<void> {
    // Your resolve is your durability attestation: the ack happens only after this resolves.
    // Throw TransientError to retry via topology-based backoff, or PermanentError to park
    // immediately in the DLQ.
    if (somethingRecoverable) throw new TransientError('downstream unavailable, retry me');
  }
}

Registering a consumer materializes a trio of queues at boot and on every reconnect: X.v2 (the quorum work queue), X.v2.retry (TTL, dead-lettering back to work), and X.v2.dlq (the parking lot, no further dead-lettering). Subscribing without that trio asserted is a refused configuration.

For a producer-owned, un-enveloped, latest-value feed you do not control, EcoExternalSnapshotConsumer reads it on the shared connection — a bounded x-max-length: 1 queue where a failure requeues once and then drops, because the message being retried is guaranteed stale.

Readiness

Wire EcoMessagingHealth.isReady() into your probe, not EcoConnectionManager.isConnected(): a connected client whose queues do not exist is not ready to consume, and reporting ready in that state is how a deploy silently black-holes traffic.

Readiness is the conjunction of named gates. The library owns topology; a service that rebuilds in-memory state on boot registers its own, so "running" and "rehydrated" stay distinguishable:

this.rehydration = health.registerGate('rehydration'); // registered closed
// ...later
this.rehydration.open(`rebuilt ${restored} obligations`);

A closed gate withholds readiness only — never consumption, and never the outbox relay's drain. Both are deliberate: a service whose rehydration source is the bus would deadlock otherwise, and an obligation committed before a crash must publish regardless of how long the rebuild takes.

EcoMessagingStatus is the separate, evented view of connection and publisher state for host lifecycle adapters (watchdogs, readiness latches, worker gates). It gates nothing, and credentials are redacted before any detail reaches a listener.

Observability

ALERT exports five structured log-line literals. Eco alerting is log-based — this library's custom OTLP metrics do not reach Datadog — so those five strings are the only thing that can page an on-call engineer about a messaging failure. They are pinned byte-for-byte by tests; renaming one is a breaking change.

METRIC and METRIC_KINDS name the instruments, covering both halves of durability: the transport (publish.*, consume.*, outbox.*) and the caller-ledger obligation lifecycle (obligation.state, obligation.pending, obligation.oldest_age_ms). A gauge with no measurement reports nothing rather than zero.

Subpath exports

If you wire OTel, read the two ordering traps in the repository README first — registering the MeterProvider after the metrics sink is built silently produces no metrics.

Upgrading from 1.x

2.0.0 is a major for topology-surface reasons; nothing existing changes behaviour and the new bounds default to what hosts already had. There is one breaking source change: RawPublishInput.envelope is now Omit<Envelope, 'correlationId'>, because publishRawOrThrow has always minted a fresh correlation ID per attempt and discarded whatever was passed. A caller passing a full Envelope object literal must drop that field. See CHANGELOG.md for the full entry.

Documentation

These packages are the reference implementation of the standard, not its definition. The definition is two language-agnostic documents, both machine-checked against this code:

• Messaging conformance profile — envelope, topology, settlement, dedup, ordering, observability.
• Persist-before-publish and rehydration — producer durability and readiness.
• Alerting — what each ALERT means and its monitor status.
• Repository README — the full guide, including the .v2 classic→quorum cutover rule, what is deliberately out of scope, and the accepted residual risks worth reading before you touch connection or consumer lifecycle.