@resilientmq/mongoose-connector

Durable MongoDB inbox and outbox persistence for @resilientmq/core, with Mongoose models, idempotent insertion, bounded backlog reads, and reusable RabbitMQ runtimes.

const connector = new MongooseConnector(config);
await connector.startConsumer();
await connector.publish(event);

Table of contents

• Features
• Compatibility
• Installation
• Quick start
• Delivery model
• Connector lifecycle
• Event store
• Custom models and serialization
• Configuration
• Architecture
• Version lines
• Development
• Documentation
• Contributors
• License

Features

• Durable inbox and outbox — persists ResilientMQ event state in MongoDB.
• Idempotent insertion — converts MongoDB duplicate-key conflicts into a deterministic false result.
• Atomic inbox leases — one replica owns a delivery until completion or lease expiration.
• Fenced transitions — stale process generations cannot overwrite a newer claim.
• Distributed outbox — replicas split pending work through conditional MongoDB claims instead of publishing the same row independently.
• Efficient backlog processing — bounded, oldest-first pending queries and one-call bulkWrite status transitions.
• Reusable connections — one object-oriented runtime shares MongoDB and long-lived RabbitMQ consumer and publisher resources.
• Independent stores — consumer and publisher models and serializers never leak into each other.
• Custom persistence — applications may provide their own Mongoose model, identity filter, status field, and serialization format.
• Safe shutdown — RabbitMQ runtimes stop before the shared MongoDB connection is closed.
• Typed API — ESM package with generated TypeScript declarations.

Compatibility

Install the connector major matching the @resilientmq/core major:

| Connector | ResilientMQ core | Mongoose | Node.js | Ownership model | | --- | --- | --- | --- | --- | | 1.x | ^1.2.12 | 8.x–9.x | 20.19, 22, 24 | Durable CRUD and deduplication | | 2.x | ^2.3.1 | 8.x–9.x | 20.19, 22, 24 | Idempotent batch operations | | 3.x | ^3.0.0 | 8.x–9.x | 20.19, 22, 24 | Atomic leases and fencing |

This branch builds connector 3.x for the Core 3 atomic persistence contract. Inbox ownership is partitioned by stable serviceId and messageId; outbox ownership uses expiring leases and a fresh fencing token on every claim.

Every supported Node.js line, the declared core peer dependency, Mongoose, and a real MongoDB 8 service are exercised by CI.

Installation

npm install @resilientmq/core@^3.0.0 mongoose @resilientmq/mongoose-connector@^3

Quick start

Prefer one MongooseConnector per application process:

import {randomUUID} from 'node:crypto';
import {MongooseConnector} from '@resilientmq/mongoose-connector';

const connector = new MongooseConnector({
  mongo: {
    uri: process.env.MONGODB_URL!
  },
  rabbit: {
    consumer: {
      connection: process.env.AMQP_URL!,
      serviceId: 'orders-consumer',
      processingTimeoutMs: 60_000,
      processingLeaseMs: 90_000,
      consumeQueue: {
        queue: 'orders.events',
        options: {durable: true}
      },
      eventsToProcess: [{
        type: 'order.created',
        handler: async event => processOrder(event.payload)
      }]
    },
    publisher: {
      connection: process.env.AMQP_URL!,
      serviceId: 'orders-publisher',
      exchange: {
        name: 'domain.events',
        type: 'topic',
        options: {durable: true}
      }
    }
  }
});

await connector.startConsumer();
await connector.publish({
  messageId: randomUUID(),
  type: 'order.accepted',
  routingKey: 'order.accepted',
  payload: {orderId: 'order-42'},
  status: 'PENDING_PUBLICATION'
});

process.once('SIGTERM', () => {
  void connector.disconnect();
});

The connector opens MongoDB lazily, creates each RabbitMQ runtime once, and reuses the publisher across calls. It does not connect and disconnect for every event.

Delivery model

• MongoDB provides durable event state and indexed message identities.
• Duplicate inserts are rejected by the database rather than by an unsafe read-before-write check.
• Inbox claims use the stable core service hash plus the message ID, so distinct logical consumers can process the same RabbitMQ event independently.
• Active ownership records the ephemeral process ID, a unique fencing token, and an expiration time.
• Pending publication claims are ordered, bounded, and atomically partitioned across replicas.
• A process that resumes after its lease was recovered cannot apply its stale completion or retry transition.
• RabbitMQ and MongoDB do not participate in one distributed transaction.
• Message handlers and external domain effects must therefore remain idempotent.

These guarantees provide at-least-once delivery with exclusive database ownership; they do not provide exactly-once external side effects. The compatibility guide documents the contract in detail.

Connector lifecycle

MongooseConnector owns the runtimes it creates while retaining one shared MongoDB connection:

| Method | Behavior | | --- | --- | | connect() | Opens MongoDB if it is not already connected. | | createConsumer() | Lazily creates and reuses the configured consumer. | | startConsumer() | Connects MongoDB and starts the consumer. | | createPublisher() | Lazily creates and reuses the configured publisher. | | publish(event, options?) | Connects MongoDB and publishes through the reusable publisher. | | disconnect() | Stops consumer and publisher, then closes MongoDB. |

Existing applications can retain the functional compatibility facade:

import {
  consume,
  disconnect,
  publish,
  setEnvironment
} from '@resilientmq/mongoose-connector';

await setEnvironment(config);
await consume();
await publish(event);
await disconnect();

The facade delegates to the same reusable connector lifecycle.

Event store

GenericMongooseStore implements the full Core 3 consumer and distributed publisher contracts:

| Operation | MongoDB behavior | | --- | --- | | saveEvent | Creates one event document. | | saveEventIfNotExists | Uses the unique index and handles error 11000. | | getEvent / deleteEvent | Uses the serializer-defined identity filter. | | updateEventStatus | Updates the serializer-defined status path. | | getPendingEvents | Returns a bounded, oldest-first batch. | | getEventsByStatus | Returns events matching an exact status. | | batchUpdateEventStatus | Applies transitions through one bulkWrite. | | claimConsumeEvent | Atomically inserts or recovers a PROCESSING lease. | | transitionConsumeEvent | Transitions only for the current service, instance, and token. | | claimPublishEvent | Claims one ready or expired outbox document. | | claimPendingEvents | Atomically claims a bounded, ordered batch across replicas. | | completePublishedEvent | Completes only the active confirmed publication claim. | | releasePublishEvent | Releases the active claim with a delayed retry deadline. |

The default model persists messageId, type, payload, status, routingKey, AMQP properties, service and instance identity, fencing token, lease and retry deadlines, attempts, completion timestamps, and bounded error details. Default indexes cover compound inbox identity, unique outbox identity, lease recovery, and pending status scans.

Custom models and serialization

Consumer and publisher stores accept independent model, modelName, and serializer settings. A custom serializer must match its model and define the identity filter whenever the event ID is not stored at messageId:

import type {EventSerializer} from '@resilientmq/mongoose-connector';

const serializer: EventSerializer = {
  toStorageFormat: event => ({
    _id: event.messageId,
    body: event.payload,
    lifecycle: event.status
  }),
  fromStorageFormat: document => ({
    messageId: String(document._id),
    payload: document.body,
    status: String(document.lifecycle)
  }),
  getIdentityFilter: event => ({_id: event.messageId}),
  getStatusField: () => 'lifecycle'
};

Serializer objects must be stateless or safe for concurrent calls.

Custom Core 3 models must also declare serviceId, instanceId, fencingToken, leaseExpiresAt, attempt, lastAttemptAt, completedAt, nextAttemptAt, publishedAt, errorName, errorMessage, and errorStack. Consumer models require a unique {serviceId, messageId} index; publisher models require a unique {messageId} index. The application owns these indexes when it supplies rabbit.*.model.

Configuration

new MongooseConnector({mongo, rabbit, logLevel});

| Option | Required | Description | | --- | :---: | --- | | mongo.uri | Yes | MongoDB connection URI. | | mongo.options | No | Application-specific Mongoose connection options. | | rabbit.consumer | No | Core consumer configuration plus optional store customization. | | rabbit.publisher | No | Core publisher configuration plus optional store customization. | | rabbit.*.model | No | Existing application-owned Mongoose model. | | rabbit.*.modelName | No | Name used when creating the default model. | | rabbit.*.serializer | No | Serializer matching the selected model. | | logLevel | No | none, warn, info, or error. |

The application supplies at least the RabbitMQ runtime it intends to use. Calling a consumer or publisher method without its matching configuration fails immediately with a descriptive error.

Architecture

Application
  └─ MongooseConnector
      ├─ ResilientConsumer ── fenced inbox store ───── consumer model
      ├─ ResilientPublisher ─ distributed outbox ───── publisher model
      └─ MongoConnection ───────────────────────────── MongoDB

The connector separates RabbitMQ runtime ownership from persistence mapping. Both stores share the MongoDB connection, while their models and serializers remain isolated.

Version lines

The connector majors are maintained as sequential compatibility branches:

| Branch | Package line | Purpose | | --- | --- | --- | | release/core-1.x | 1.x | Latest core 1.x CRUD contract. | | release/core-2.x | 2.x | Core 2 batch and idempotent store contract. | | release/core-3.x | 3.x | Atomic inbox/outbox leases and fencing. |

Upgrade one major at a time. The peer dependency deliberately rejects unsupported core/connector combinations instead of allowing a resilience contract to degrade silently.

Development

npm ci
npm run typecheck
npm run test:coverage
npm run build
npm audit --audit-level=high
npm pack --dry-run

Run the real MongoDB integration suite with:

MONGODB_URL=mongodb://localhost:27017/resilientmq npm run test:integration

CI validates Node.js 20.19, 22, and 24, enforces at least 90% statement, line, and function coverage plus 80% branch coverage, and exercises concurrent claim, lease recovery, stale fencing, delayed retry, and MongoDB 8 behavior.

Documentation

• Compatibility and release guarantees
• Changelog
• Contributing guide
• Security policy
• CI and trusted publishing

Contributors

Thanks to everyone who has contributed to this project:

Want to help? Read CONTRIBUTING.md.

License

MIT © ResilientMQ