@ahrowe/mongo

A typed MongoDB CRUD service factory: connect() once, then createService<T>(...) gives you a collection wrapper with built-in transactions, an event bus (created/updated/removed), and optional document validation on writes.

Install

pnpm add @ahrowe/mongo

Usage

import db from '@ahrowe/mongo';

const { createService } = db.connect(process.env.MONGO_DB_URI!, 'myDb');

type User = { _id: string; email: string; createdOn?: Date; updatedOn?: Date };
const users = createService<User>('users');

const user = await users.insert({ email: '[email protected]' });
const found = await users.findOne({ _id: user._id });
await users.updateOne({ _id: user._id }, { $set: { email: '[email protected]' } });

Call createService once per collection and reuse that instance everywhere. Calling it again for the same collection name is supported (it reuses the first call's event bus and logs a warning, so listeners on either instance still fire) but other per-instance options are not shared: a mismatched emitOutboxEvents means writes through one instance never fire events at all (not a bus issue — no outbox row is written), a mismatched validate means the two instances enforce different schemas on the same collection, and a mismatched addCreatedOnField/addUpdatedOnField means only some documents get timestamped. Also, eventBus.removeAllListeners() on either instance clears listeners registered via both. Prefer sharing one instance unless one of these divergences is what you actually want (e.g. a backfill instance with emitOutboxEvents: false).

Validation on writes

Pass a validator function as the second argument to createService. It's called on every insert and on the document resulting from an update; a non-passing result throws. The validator contract is (entity: T) => { error?: unknown } — plug in whatever validation library you like, or none at all. On a non-empty error, that value is thrown as-is.

With Zod:

import { z } from 'zod';

const userSchema = z.object({ _id: z.string(), email: z.string().email() });
const users = createService<User>('users', (entity) => userSchema.safeParse(entity));

Or a plain TS guard with no dependency at all:

const users = createService<User>('users', (entity) => (
  entity.email.includes('@') ? {} : { error: new Error('email must contain @') }
));

Pass { skipValidation: true } on an individual call to bypass it.

Events (transactional outbox)

users.on('created', ({ doc, meta }) => { /* ... */ });
users.on('updated', ({ prevDoc, doc, meta }) => { /* ... */ });
users.on('removed', ({ doc, meta }) => { /* ... */ });

Writes don't dispatch events directly. Every insert/updateOne/updateMany/ removeOne/removeMany that actually changes a document writes one row per affected document to an internal outbox collection, in the same transaction as the data write. A separate relay process claims and dispatches those rows to your .on(...) listeners. This makes delivery durable: events survive a crash between the write and the listener running, and are never lost or duplicated across multiple pods racing the same row (claims are leased atomically). updateOne/updateMany only enqueue 'updated' for documents that actually changed.

updateMany/removeMany always process every matched document individually — one atomic operation per document, all inside one transaction, so the whole batch still rolls back together on failure — so each outbox row's { prevDoc, doc } pair is accurate and no document is missed even if the matched set changes mid-operation. Run the relay to drain the outbox:

import { startOutboxRelay } from '@ahrowe/mongo';

const relay = startOutboxRelay({ instanceId: process.env.HOSTNAME ?? 'local' });
// relay.stop() on shutdown

The reliable per-document path has no size cap by default — but a single transaction processing a very large matched set risks hitting MongoDB's transaction lifetime limit. Pass { maxBatchSize } (per-call, or as a createService option for a service-wide default) to enforce one and fail fast instead:

await users.updateMany({}, { $set: { plan: 'pro' } }, { maxBatchSize: 500 });

Past that limit it throws, pointing you at bulkWrite for larger jobs (no transaction/validation/event guarantees, but no cap either).

Pass { meta: {...} } on a call to thread arbitrary data through to listeners, so a specific handler can decide to skip its own side effect without anyone else missing the event:

await users.updateOne({ _id }, { $set: { email: '[email protected]' } }, { meta: { dontRegenPdf: true } });

users.on('updated', ({ doc, meta }) => {
  if (meta?.dontRegenPdf) return;
  regenerateInvoicePdf(doc);
});

Infra collections with no listeners (sequences, the outbox itself, etc.) can skip outbox writes entirely by passing { emitOutboxEvents: false } to createService — writes take a leaner fast path with no pre-reads or transaction-wrapped outbox insert.

If a listener needs request/actor context (e.g. an audit log reading from AsyncLocalStorage), register a provider once at startup. It's called synchronously at write time and the captured value is stored on the outbox row, so it's still available to the listener even after a process restart:

import { setOutboxContextProvider } from '@ahrowe/mongo';

setOutboxContextProvider(() => myAsyncLocalStorage.getStore());

Transactions

import db from '@ahrowe/mongo';

const session = await db.startSession();
await session.withTransaction(async () => {
  await users.insert({ email: '[email protected]' }, { session });
  await otherService.updateOne({ _id }, { $set: { count: 1 } }, { session });
});

insert/updateOne/updateMany automatically wrap themselves in a transaction when no session is passed.

Connection health

const { createService, on } = db.connect(process.env.MONGO_DB_URI!, 'myDb');

on('error', ({ source, error }) => { /* 'primary' | 'read' */ });
on('close', ({ source, error }) => { /* ... */ });

connect() wires up error/close handlers on both the primary and secondary-preferred read clients internally (so a connection issue can't crash the process), and re-emits them on a connection-level bus you can subscribe to for your own alerting.

Shutdown

import db from '@ahrowe/mongo';

await db.disconnect();

disconnect() closes both MongoDB clients. Outbox rows already committed are durable in MongoDB regardless — call this on graceful shutdown (e.g. on SIGTERM); if you also run startOutboxRelay in-process, call relay.stop() first so it releases any leases it's holding.

API

• connect(connectionString, databaseName?) → { createService, on }
• createService<T>(collectionName, validate?, { addCreatedOnField?, addUpdatedOnField?, maxLimit?, maxBatchSize?, emitOutboxEvents? }) → DbService<T>
• withSession(cb), startSession(options?), disconnect()
• setOutboxContextProvider(fn) — capture request/actor context at write time for outbox rows
• startOutboxRelay({ instanceId, pollIntervalMs?, batchSize?, leaseMs?, maxAttempts?, autoStart? }) → { stop, tick, drain }
• getServiceBus(collectionName), getRegisteredCollections(), getOutboxCollection() — relay/advanced internals
• OutboxOp, OutboxStatus, OutboxRow, OUTBOX_COLLECTION — outbox row shape, for tooling/inspection
• DbService<T> methods: find, findOne, findCursor, insert, updateOne, updateMany, removeOne, removeMany, count, exists, aggregate, distinct, createIndex, dropIndex, bulkWrite, on, onPropertiesUpdated, eventBus, generateId, name

See docs/CLAUDE.md for gotchas (upserts are rejected, findOne strictness, the opt-in maxLimit/maxBatchSize caps, bulkWrite safety gate).