@classytic/mongokit

Production-grade MongoDB repository pattern. Implements the StandardRepo<TDoc> contract from @classytic/repo-core — the same contract sqlitekit + future pgkit / prismakit implement. Controller code written against the contract runs unchanged on any kit. 2009 integration tests + cross-kit conformance suite.

Requires: Mongoose >=9.4.1 | @classytic/repo-core >=0.3.0 | Node.js >=22

Install

npm install @classytic/mongokit @classytic/repo-core mongoose

Both @classytic/repo-core and mongoose are peer deps — kit never bundles them.

Core pattern

import { Repository } from "@classytic/mongokit";

const repo = new Repository(UserModel);

const user = await repo.create({ name: "Alice", email: "a@example.com" });
const page = await repo.getAll({ page: 1, limit: 20 });
const found = await repo.getById(id);        // null on miss
const updated = await repo.update(id, {...}); // null on miss
const result = await repo.delete(id);         // { success: false, ... } on miss

Miss semantics (MinimalRepo contract)

getById / update / delete return null / { success: false } on miss by default — not a thrown 404. Invalid-shape ids ('not-a-valid-id' on an ObjectId _id) short-circuit to the same miss result rather than raising mongoose CastError.

Legacy throw behavior is one opt-in away:

await repo.getById(id, { throwOnNotFound: true });   // throws { status: 404 }
await repo.update(id, data, { throwOnNotFound: true });
await repo.delete(id, { throwOnNotFound: true });

Full API

| Method | -------------------------------------- | create(data, opts) | createMany(items[], opts) | getById(id, opts) | getByQuery(query, opts) | getOne(filter, opts) | getAll(params, opts) | findAll(filter?, opts) | getOrCreate(query, data, opts) | update(id, data, opts) | findOneAndUpdate(filter, | delete(id, opts) | count(filter, opts) | exists(filter, opts) | distinct(field, filter, opts) | aggregate(req: AggRequest) | aggregatePipeline(stages[]) | aggregatePaginate(req) | aggregatePipelinePaginate(opts) | bulkWrite(operations[]) | lookupPopulate(params) | purgeByField(field, value, | withTransaction(async | isDuplicateKeyError(err) | Returns on miss | | ---------------------------------------- | | — | | — | | null | | null | | null | | envelope with empty docs | | [] | | inserts + returns new doc | | null | update, opts) | null | | { success: false, ... } | | 0 | | null | | [] | | portable cross-kit IR | | kit-native mongo pipeline | | portable IR + pagination envelope | | pipeline + pagination envelope | | heterogeneous insert/update/delete batch | | { docs, total, pages, hasNext, ... } | strategy) | compliance cleanup (see below) | txRepo => ...) | tx-bound repo (see Transactions) | | true for E11000 / wrapped 409 |

All filter arguments accept either plain mongo queries ({ status: 'active' }) or Filter IR from @classytic/repo-core/filter (eq('status', 'active')). Same code works on sqlitekit.

Single-doc reads with `sort`

getOne / getByQuery accept sort to disambiguate when the filter matches multiple docs — FIFO claim, "latest pending", "oldest unprocessed":

const oldest = await repo.getOne({ status: 'pending' }, { sort: { createdAt: 1 } });
const latest = await repo.getByQuery({ userId }, { sort: { createdAt: -1 } });

For atomic claim-and-mutate, prefer findOneAndUpdate(filter, update, { sort }) — single round-trip vs read-then-write.

`purgeByField` — compliance-grade tenant cleanup

Process every row matching field = value under a declared strategy. Chunked, plugin-composed, abort-aware. Powers arc's cascadeDeleteForOrganization.

// GDPR right-to-be-forgotten — hard delete:
await repo.purgeByField('organizationId', orgId, { type: 'hard' }, { batchSize: 1000 });

// SOX-retained ledger — keep row, clear PII:
await repo.purgeByField('organizationId', orgId, {
  type: 'anonymize',
  fields: { customerName: '[REDACTED]', customerEmail: null },
});

// Soft delete with TTL eventual cleanup:
await repo.purgeByField('organizationId', orgId, { type: 'soft' });

// Returns { strategy, processed, ok, durationMs, error? }

Function-form anonymize replacers (fields: { hash: (doc) => sha256(doc.email) }) compile to a single bulkWrite per chunk — N rows in one round-trip, not N+1.

Index requirement: the purge field MUST be indexed (Schema.index({ organizationId: 1 }) or a compound leading with it) or chunked SELECTs run O(n²).

Options: batchSize (default 1000), onProgress, signal (AbortSignal), retry: { maxAttempts, baseDelayMs, shouldRetry }. See contract types in @classytic/repo-core/repository.

Pagination (auto-detected)

// Offset — pass `page`
await repo.getAll({ page: 1, limit: 20, filters: { status: "active" }, sort: { createdAt: -1 } });
// → { method: 'offset', docs, total, pages, hasNext, hasPrev }

// Keyset — pass `sort` without `page` (or `after` for next page)
const first = await repo.getAll({ sort: { createdAt: -1 }, limit: 20 });
// → { method: 'keyset', docs, hasMore, next: 'eyJ2...' }
const second = await repo.getAll({ after: first.next, sort: { createdAt: -1 }, limit: 20 });

Detection: page → offset | after → keyset | sort only → keyset | default → offset.

Keyset indexes: create a compound index on the sort keys + _id:

Schema.index({ createdAt: -1, _id: -1 });
Schema.index({ organizationId: 1, createdAt: -1, _id: -1 }); // multi-tenant

Aggregation — two surfaces

aggregate(req: AggRequest) — portable IR, same input + output on mongokit and sqlitekit:

const { rows } = await repo.aggregate({
  filter: { active: true },
  groupBy: 'category',
  measures: { total: { op: 'sum', field: 'amount' }, n: { op: 'count' } },
  having: gt('total', 1000),
  sort: { total: -1 },
});
// rows: [{ category: 'admin', total: 1200, n: 5 }, ...]

aggregatePipeline(stages) — kit-native mongo pipeline. Use for $lookup, $unwind, $facet, $graphLookup, $bucket, window fields — anything that doesn't translate across backends:

const stats = await repo.aggregatePipeline([
  { $match: { active: true } },
  { $lookup: { from: 'orders', localField: '_id', foreignField: 'userId', as: 'orders' } },
  { $addFields: { orderCount: { $size: '$orders' } } },
]);

Rule of thumb: reach for portable aggregate first. Drop to aggregatePipeline only when the query needs MongoDB-specific stages.

Plugins (17)

Order matters — plugins run at declared priorities (POLICY → CACHE → OBSERVABILITY → DEFAULT):

const repo = new Repository(UserModel, [
  timestampPlugin(),
  multiTenantPlugin({ tenantField: 'organizationId' }),
  softDeletePlugin(),
  cachePlugin({ adapter: createMemoryCache(), ttl: 60 }),
]);

| Plugin | Adds | | ----------------------------------- | -------------------------------------------- | | timestampPlugin() | createdAt / updatedAt | | softDeletePlugin(opts) | deletedAt mark + auto read-filter | | auditLogPlugin(logger) | external CUD log | | auditTrailPlugin(opts) | DB-persisted audit trail + field-diffs. mode: 'transactional' = awaited, session-joined (atomic ONLY inside withTransaction; compliance-grade); default 'best-effort' = fire-and-forget (observability only). Call ensureAuditTrailReady() at boot (pre-creates collection+indexes so the first transaction doesn't). Pass connection for createConnection() apps; the writing plugin's conflicting ttlDays throws — readers (AuditTrailQuery) omit ttlDays to reuse it. | | cachePlugin(opts) | Redis/memory read cache + auto-invalidation | | validationChainPlugin(validators) | custom validation rules | | fieldFilterPlugin(preset) | role-based field visibility (RBAC) | | cascadePlugin(opts) | auto-delete related docs | | multiTenantPlugin(opts) | tenant scope injection (fieldType casting) | | customIdPlugin(opts) | sequential / random ID generation | | observabilityPlugin(opts) | timing + metrics + slow-op callback | | methodRegistryPlugin() | base for mongoOperations / batchOps / … | | mongoOperationsPlugin() | increment, pushToArray, upsert | | batchOperationsPlugin() | updateMany, deleteMany, bulkWrite | | aggregateHelpersPlugin() | groupBy, sum, average | | subdocumentPlugin() | array-subdoc CRUD methods | | elasticSearchPlugin(opts) | delegate ?search= to ES / OpenSearch |

Soft delete

const repo = new Repository(UserModel, [
  methodRegistryPlugin(),
  batchOperationsPlugin(),
  softDeletePlugin({ deletedField: 'deletedAt' }),
]);

await repo.delete(id);                         // sets deletedAt
await repo.getAll();                            // excludes soft-deleted
await repo.getAll({ includeDeleted: true });   // includes them
await repo.delete(id, { mode: 'hard' });       // actually removes
await repo.deleteMany({ status: 'draft' });    // soft-deletes in batch

Use partial unique indexes so soft-deleted rows don't block new inserts:

Schema.index({ email: 1 }, { unique: true, partialFilterExpression: { deletedAt: null } });

Cache

cachePlugin({ adapter, ttl: 60, byIdTtl: 300, queryTtl: 30 });

Adapter shape:

const redisAdapter: CacheAdapter = {
  async get(key)          { return JSON.parse((await redis.get(key)) || 'null'); },
  async set(key, v, ttl)  { await redis.setex(key, ttl, JSON.stringify(v)); },
  async delete(key)       { await redis.del(key); },       // 3.10: renamed from del()
  async clear(pattern)    { /* bulk invalidation */ },
};

Multi-tenant

MultiTenantOptions extends Pick<TenantConfig, ...> from @classytic/repo-core/tenant — TenantConfig, TenantStrategy, TenantFieldType, resolveTenantConfig, DEFAULT_TENANT_CONFIG, ResolvedTenantConfig all live in repo-core; mongokit only contributes Mongoose-specific extras (e.g. fieldType: 'objectId').

multiTenantPlugin({
  tenantField: 'organizationId',
  contextKey: 'organizationId',
  required: true,
  fieldType: 'objectId',  // cast to ObjectId for $lookup / .populate() to work
});

await repo.getAll({ organizationId: 'org_123' }); // auto-scoped
await repo.update(id, data, { organizationId: 'org_attacker' }); // → null (cross-tenant miss)

Use createTenantContext() with AsyncLocalStorage to avoid passing organizationId on every call.

`@classytic/mongokit/tenant` — schema-side injection + index scope (3.39)

The plugin scopes QUERIES; this scopes the SCHEMA. Kernels import it instead of each carrying an inject-tenant.ts copy (PACKAGE_RULES P11).

import { injectTenantField, type IndexDeclaration } from '@classytic/mongokit/tenant';

injectTenantField(schema, resolveTenantConfig(shape.tenant), {
  skipIndexes: ['order_event_id_unique'],  // a kernel's OWN inline global index, by name
  indexes: shape.indexes,                  // a host's declarations, applied AFTER the prepend
});

It adds the tenant field, leads every declared compound with it (skipping one that already names it, a TTL index, or one in skipIndexes), applies indexes honouring each declaration's scope, then adds a bare { tenant: 1 } only when nothing was prepended.

scope is the thing to get right. Default 'tenant' is led by the tenant field. 'global' is kept as written, for an IDENTITY read that spans tenants by design — a buyer's history across every seller, a kernel-generated event id unique everywhere, a webhook looking up a provider session id it was never told the tenant of. A global index must be NAMED and must not name the tenant field (both refused at describe time), and must be paired with a bypassTenant: true read; one nothing reads across tenants is pure write amplification.

{ fields: { customerId: 1, createdAt: -1 }, scope: 'global', options: { name: 'order_buyer_history_global' } }

Expose an indexes option on your kernel's shape and thread it here: a host declaring an index on the blueprint gets it built and verified with everything else, where a host-side collection.createIndex() after bind is invisible to the deploy step and dropped by the next syncIndexes.

Custom IDs

import { customIdPlugin, sequentialId, prefixedId, dateSequentialId } from '@classytic/mongokit';

customIdPlugin({ generate: sequentialId({ counterModel, field: 'sku' }) });      // 1, 2, 3
customIdPlugin({ generate: prefixedId({ prefix: 'USR_', length: 8 }) });          // USR_a1b2c3d4
customIdPlugin({ generate: dateSequentialId({ counterModel, pattern: 'YYYYMM' }) }); // 2026040001

QueryParser (HTTP → mongo)

Turns URL query strings into mongo filters + pagination params:

import { QueryParser } from '@classytic/mongokit';

const parser = new QueryParser({ schema: UserSchema });
// Fail-closed by DEFAULT (invalidInput: 'throw'): invalid/blocked query input
// → HttpError 400 (code INVALID_QUERY_INPUT). Opt out with
// invalidInput: 'drop' (legacy warn-and-drop; can broaden results) only for
// trusted migration/compat tooling. search/like/contains stay literal (escape,
// never reject — ?search=c++ never 400s).
const parsed = parser.parse(req.query); // { filters, sort, page, limit, lookups, select, populate }

const result = await repo.getAll(parsed);

URL syntax:

GET /users?status=active&age[gte]=18&sort=-createdAt&page=1&limit=20
GET /users?populate=orders,profile
GET /users?populate[orders][select]=id,total&populate[orders][match][status]=paid
GET /products?lookup[category][from]=categories&lookup[category][localField]=categorySlug&lookup[category][foreignField]=slug&lookup[category][single]=true

Ref-less $lookup via lookup[...]: join by any field (slug, code, SKU) without declaring a Mongoose ref. See docs/LOOKUP_GUIDE.md for the full grammar.

Transactions

withTransaction receives a tx-bound repo (NOT a raw mongoose session — that's the standalone helper). Every method on txRepo auto-threads the session:

await ordersRepo.withTransaction(async (txRepo) => {
  const order = await txRepo.create({ total: 100 });
  await txRepo.update(order._id, { confirmed: true });
  return order;
});
// Both writes commit, or neither does.

Cross-repo transactions need the standalone export (raw session, exported as withTransaction from @classytic/mongokit):

import { withTransaction } from '@classytic/mongokit';

await withTransaction(mongoose.connection, async (session) => {
  await ordersRepo.create({ ... }, { session });
  await outboxRepo.create({ ... }, { session });
});

Standalone MongoDB doesn't support transactions — need a replica set (even single-node). Tests use mongodb-memory-server with MongoMemoryReplSet.

Events

import { HOOK_PRIORITY } from '@classytic/mongokit';

repo.on('after:create', async ({ context, result }) => {
  await kafka.publish('users.created', { doc: result, tenant: context.organizationId });
}, { priority: HOOK_PRIORITY.OBSERVABILITY });

repo.on('before:update', async ({ context }) => { /* mutate context.data */ });
repo.on('error:delete', async ({ context, error }) => { /* metric / alert */ });

Events per op: before:<op>, after:<op>, error:<op>. Priority ordering: POLICY → CACHE → DEFAULT → OBSERVABILITY.

Swap-ability with sqlitekit

Both kits implement StandardRepo<TDoc> from @classytic/repo-core/repository. Cross-kit dashboard / admin / test code can target the contract and run on either backend unchanged:

import type { StandardRepo } from '@classytic/repo-core/repository';

function listActive<T>(repo: StandardRepo<T>): Promise<T[]> {
  return repo.findAll!({ status: 'active' });
}

Conformance: tests/integration/conformance.test.ts runs the shared suite from @classytic/repo-core/testing. When both kits pass, swap-ability is provable.