feat(store): dual-write mirror + auth/billing specs to drop chat-docdb (0.9.15)
The Mongo→SQLite cutover, completed so chat-docdb can be DELETED with zero
data loss and an instant revert at every step.
- DualWriteModel: wraps a primary (served) + mirror model; reads/props fall
through to primary, the 10 write methods run on primary then replicate the
affected docs to the mirror KEYED BY THE PRIMARY'S _id (upsert if present,
delete if gone). Symmetric — same code mirrors mongoose->sqlite (pre-flip)
and sqlite->mongoose (post-flip escape hatch). Mirror failures are logged,
never thrown, so the served path can't break; gaps are caught by the
pre-flip count reconcile + the idempotent backfill.
- DocModel.upsertRaw: exact by-_id upsert (verbatim, no timestamp stamping) —
the shared primitive for the mirror AND the backfill, so live mirroring and
the one-shot copy converge on one keyspace without duplicating.
- Seam: applySqliteOverrides now honors TWO flags — CHAT_STORE_SQLITE (served)
+ CHAT_STORE_DUALWRITE (mirrored) — yielding the four cutover states. One
shared node:sqlite connection per process (was per-createModels-call).
- Route User/Session/Token through the handle (methods/index.ts) and add the
auth+billing CollectionSpecs (User/Session/Token/Balance/Transaction). These
are the collections actually populated in chat-docdb outside the 24 already
wired; User is the hot-path record every request loads and every conversation
references by _id, so it MUST move for a lossless Mongo delete.
- connectDb() is now Mongo-optional: unset MONGO_URI => SQLite-only mode, skip
the connection (final state, chat-docdb gone) with bufferCommands off so a
stray mongoose query fails fast instead of hanging.
- Dockerfile{,.multi,.static}: node:20/22-alpine -> ghcr.io/hanzoai/nodejs
:v24.18.0 (Node 24; node:sqlite built in, better-sqlite3 compiles).
- config/backfill-sqlite.js: one-shot Mongo->SQLite copy + count reconcile.
Tests: DualWriteModel.spec (10) green — mirror-by-primary-_id both directions,
read passthrough, bulkWrite, idempotent backfill, all 29 specs build.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
zeekay
+4
-5
@@ -1,10 +1,9 @@
|
||||
# v0.8.3-rc1
|
||||
|
||||
# Base node image — Node 22 (parity with Dockerfile.static). Node >= 22.5 ships
|
||||
# `node:sqlite` (DatabaseSync), required by the SQLite document store once
|
||||
# CHAT_STORE_SQLITE / CHAT_STORE_DUALWRITE is enabled. The store lazy-requires it
|
||||
# (see stores/sqlite/index.ts) so the runtime still boots with the flag unset.
|
||||
FROM node:22-alpine AS node
|
||||
# Base node image — Hanzo Node 24 (Alpine) base: node:sqlite (DatabaseSync) is
|
||||
# built in and native better-sqlite3 compiles (build-base + python3 + g++ baked
|
||||
# in), so the CHAT_STORE_SQLITE document store runs. Node 20 lacked node:sqlite.
|
||||
FROM ghcr.io/hanzoai/nodejs:v24.18.0 AS node
|
||||
|
||||
# Install jemalloc
|
||||
RUN apk add --no-cache jemalloc
|
||||
|
||||
+1
-1
@@ -5,7 +5,7 @@
|
||||
ARG NODE_MAX_OLD_SPACE_SIZE=6144
|
||||
|
||||
# Base for all builds
|
||||
FROM node:20-alpine AS base-min
|
||||
FROM ghcr.io/hanzoai/nodejs:v24.18.0 AS base-min
|
||||
# Install jemalloc
|
||||
RUN apk add --no-cache jemalloc
|
||||
# Set environment variable to use jemalloc
|
||||
|
||||
+1
-1
@@ -6,7 +6,7 @@
|
||||
# Build: docker build -f Dockerfile.static -t ghcr.io/hanzoai/chat:latest .
|
||||
# Run: docker run -p 3080:3080 ghcr.io/hanzoai/chat:latest
|
||||
|
||||
FROM node:22-alpine
|
||||
FROM ghcr.io/hanzoai/nodejs:v24.18.0
|
||||
|
||||
RUN npm install -g serve@14
|
||||
|
||||
|
||||
+15
-3
@@ -5,9 +5,11 @@ const { logger } = require('@librechat/data-schemas');
|
||||
const mongoose = require('mongoose');
|
||||
const MONGO_URI = process.env.MONGO_URI;
|
||||
|
||||
if (!MONGO_URI) {
|
||||
throw new Error('Please define the MONGO_URI environment variable');
|
||||
}
|
||||
// MONGO_URI is intentionally optional: in SQLite-only mode (every collection
|
||||
// served from CHAT_STORE_SQLITE, chat-docdb deleted) there is no Mongo to
|
||||
// connect to. connectDb() below handles the unset case by skipping the
|
||||
// connection rather than failing boot. When MONGO_URI IS set (default and the
|
||||
// dual-write migration window) the connection is required as before.
|
||||
/** The maximum number of connections in the connection pool. */
|
||||
const maxPoolSize = parseInt(process.env.MONGO_MAX_POOL_SIZE) || undefined;
|
||||
/** The minimum number of connections in the connection pool. */
|
||||
@@ -45,6 +47,16 @@ mongoose.connection.on('error', (err) => {
|
||||
});
|
||||
|
||||
async function connectDb() {
|
||||
if (!MONGO_URI) {
|
||||
// SQLite-only mode: skip Mongo entirely. Disable command buffering so any
|
||||
// stray mongoose query (e.g. Meili indexSync, which still references
|
||||
// mongoose.models directly) fails fast and is swallowed by its caller
|
||||
// instead of hanging forever on a pool that will never connect.
|
||||
mongoose.set('bufferCommands', false);
|
||||
logger.info('[connectDb] MONGO_URI unset — SQLite-only mode, skipping MongoDB connection');
|
||||
return null;
|
||||
}
|
||||
|
||||
if (cached.conn && cached.conn?._readyState === 1) {
|
||||
return cached.conn;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,121 @@
|
||||
/**
|
||||
* One-shot Mongo → SQLite backfill + reconcile for the chat-docdb drop.
|
||||
*
|
||||
* Copies every migrated collection from MongoDB (chat-docdb) into the embedded
|
||||
* SQLite document store at CHAT_SQLITE_PATH, keyed by each document's own `_id`
|
||||
* (BSON ObjectId → hex, preserved exactly). It uses the SAME `upsertRaw`
|
||||
* primitive the live dual-write mirror uses, so running it while dual-write is
|
||||
* enabled is safe and idempotent: a doc already mirrored by a live write and the
|
||||
* same doc copied here share the `_id` and collapse to one row (INSERT OR
|
||||
* REPLACE) — no duplicates. Re-run it as many times as you like.
|
||||
*
|
||||
* Then it reconciles: for every collection it prints `mongo == sqlite` and exits
|
||||
* non-zero if ANY collection's SQLite count is short of Mongo — the gate that
|
||||
* must be green before flipping CHAT_STORE_SQLITE.
|
||||
*
|
||||
* Run INSIDE the chat pod (has MONGO_URI + CHAT_SQLITE_PATH + the built package):
|
||||
* node config/backfill-sqlite.js
|
||||
* Optional: pass a comma list of collection (model) names to limit scope.
|
||||
*/
|
||||
const { MongoClient } = require('mongodb');
|
||||
const { createSqliteHandle, CHAT_COLLECTION_SPECS } = require('@librechat/data-schemas');
|
||||
|
||||
// model (spec) name -> mongo collection name. Explicit, not derived, so the
|
||||
// mapping is auditable and never depends on mongoose pluralization quirks.
|
||||
const MONGO_COLLECTION = {
|
||||
Conversation: 'conversations',
|
||||
Message: 'messages',
|
||||
Preset: 'presets',
|
||||
ConversationTag: 'conversationtags',
|
||||
SharedLink: 'sharedlinks',
|
||||
Project: 'projects',
|
||||
File: 'files',
|
||||
Key: 'keys',
|
||||
PluginAuth: 'pluginauths',
|
||||
Banner: 'banners',
|
||||
MCPServer: 'mcpservers',
|
||||
Prompt: 'prompts',
|
||||
PromptGroup: 'promptgroups',
|
||||
AgentApiKey: 'agentapikeys',
|
||||
Assistant: 'assistants',
|
||||
Action: 'actions',
|
||||
ToolCall: 'toolcalls',
|
||||
MemoryEntry: 'memoryentries',
|
||||
Role: 'roles',
|
||||
AccessRole: 'accessroles',
|
||||
Agent: 'agents',
|
||||
AgentCategory: 'agentcategories',
|
||||
AclEntry: 'aclentries',
|
||||
Group: 'groups',
|
||||
User: 'users',
|
||||
Session: 'sessions',
|
||||
Token: 'tokens',
|
||||
Balance: 'balances',
|
||||
Transaction: 'transactions',
|
||||
};
|
||||
|
||||
async function main() {
|
||||
const uri = process.env.MONGO_URI;
|
||||
const dbPath = process.env.CHAT_SQLITE_PATH;
|
||||
if (!uri) {
|
||||
throw new Error('MONGO_URI is required');
|
||||
}
|
||||
if (!dbPath) {
|
||||
throw new Error('CHAT_SQLITE_PATH is required');
|
||||
}
|
||||
|
||||
const only = (process.argv[2] || '')
|
||||
.split(',')
|
||||
.map((s) => s.trim())
|
||||
.filter(Boolean);
|
||||
const names = (only.length ? only : Object.keys(MONGO_COLLECTION)).filter(
|
||||
(n) => n in CHAT_COLLECTION_SPECS && n in MONGO_COLLECTION,
|
||||
);
|
||||
|
||||
console.log(`[backfill] sqlite=${dbPath} collections=${names.length}`);
|
||||
|
||||
const client = new MongoClient(uri);
|
||||
await client.connect();
|
||||
const db = client.db();
|
||||
const handle = createSqliteHandle(names, { dbPath });
|
||||
|
||||
const report = [];
|
||||
let failed = false;
|
||||
|
||||
for (const name of names) {
|
||||
const coll = db.collection(MONGO_COLLECTION[name]);
|
||||
const model = handle.models[name];
|
||||
const mongoCount = await coll.countDocuments();
|
||||
let copied = 0;
|
||||
const cursor = coll.find({});
|
||||
for await (const doc of cursor) {
|
||||
model.upsertRaw(doc);
|
||||
copied++;
|
||||
}
|
||||
const sqliteCount = await model.countDocuments({});
|
||||
const ok = sqliteCount >= mongoCount;
|
||||
if (!ok) {
|
||||
failed = true;
|
||||
}
|
||||
report.push({ name, mongo: mongoCount, copied, sqlite: sqliteCount, ok });
|
||||
console.log(
|
||||
`[backfill] ${ok ? 'OK ' : 'MISMATCH'} ${name.padEnd(16)} mongo=${String(mongoCount).padStart(6)} sqlite=${String(sqliteCount).padStart(6)}`,
|
||||
);
|
||||
}
|
||||
|
||||
handle.close();
|
||||
await client.close();
|
||||
|
||||
const matched = report.filter((r) => r.ok).length;
|
||||
console.log(`[backfill] reconcile: ${matched}/${report.length} collections mongo<=sqlite`);
|
||||
if (failed) {
|
||||
console.error('[backfill] FAILED — at least one collection is short in SQLite. Re-run.');
|
||||
process.exit(1);
|
||||
}
|
||||
console.log('[backfill] DONE — all collections reconciled. Safe to flip CHAT_STORE_SQLITE.');
|
||||
}
|
||||
|
||||
main().catch((err) => {
|
||||
console.error('[backfill] ERROR:', err);
|
||||
process.exit(1);
|
||||
});
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@hanzochat/chat",
|
||||
"version": "0.9.13",
|
||||
"version": "0.9.15",
|
||||
"description": "Chat - Unified interface to Agents, LLMs, MCPs and any AI model or OpenAPI documented API with full RAG stack",
|
||||
"packageManager": "pnpm@10.27.0",
|
||||
"workspaces": [
|
||||
|
||||
@@ -44,15 +44,18 @@ export type AllMethods = UserMethods &
|
||||
* @param mongoose - Mongoose instance
|
||||
*/
|
||||
export function createMethods(mongoose: typeof import('mongoose')): AllMethods {
|
||||
// Registry-aware handle so already-migrated domains (Share) resolve to the
|
||||
// backend selected by CHAT_STORE_SQLITE. Not-yet-migrated factories keep
|
||||
// reading the mongoose registry directly; they move to `dbHandle` as they
|
||||
// migrate. Unset flag => createModels returns pure mongoose => unchanged.
|
||||
// Registry-aware handle so migrated domains resolve to the backend selected by
|
||||
// CHAT_STORE_SQLITE / CHAT_STORE_DUALWRITE. Unset flags => createModels returns
|
||||
// pure mongoose => unchanged. User/Session/Token read only `.models`/`.Types`,
|
||||
// both of which this handle provides, so they route through the seam via the
|
||||
// cast below without any edit to their factory bodies (a mongoose instance is
|
||||
// itself structurally `{ models, Types }`, and so is this handle).
|
||||
const dbHandle = { models: createModels(mongoose), Types: mongoose.Types };
|
||||
const storeHandle = dbHandle as unknown as typeof import('mongoose');
|
||||
return {
|
||||
...createUserMethods(mongoose),
|
||||
...createSessionMethods(mongoose),
|
||||
...createTokenMethods(mongoose),
|
||||
...createUserMethods(storeHandle),
|
||||
...createSessionMethods(storeHandle),
|
||||
...createTokenMethods(storeHandle),
|
||||
...createRoleMethods(dbHandle),
|
||||
...createKeyMethods(dbHandle),
|
||||
...createFileMethods(dbHandle),
|
||||
|
||||
@@ -27,34 +27,81 @@ import { createMemoryModel } from './memory';
|
||||
import { createAccessRoleModel } from './accessRole';
|
||||
import { createAclEntryModel } from './aclEntry';
|
||||
import { createGroupModel } from './group';
|
||||
import { createSqliteHandle, CHAT_COLLECTION_SPECS } from '~/stores/sqlite';
|
||||
import {
|
||||
createSqliteHandle,
|
||||
createDualWriteModel,
|
||||
CHAT_COLLECTION_SPECS,
|
||||
type SqliteHandle,
|
||||
} from '~/stores/sqlite';
|
||||
|
||||
/**
|
||||
* Per-domain backend selection — the migration seam.
|
||||
* Per-domain backend selection — the migration seam. Two orthogonal CSV env
|
||||
* flags of collection names (only names with a CollectionSpec are honored;
|
||||
* unknown names are ignored, failing closed):
|
||||
*
|
||||
* `CHAT_STORE_SQLITE` is a CSV of collection names to serve from the SQLite
|
||||
* document store instead of mongoose (e.g. `Conversation,Message`). Unset (the
|
||||
* default) leaves every collection on mongoose, byte-for-byte unchanged — so the
|
||||
* live deployment is untouched until a domain is explicitly flipped. Only
|
||||
* collections with a CollectionSpec can be overridden; anything else throws,
|
||||
* failing closed rather than silently mis-storing.
|
||||
* CHAT_STORE_SQLITE — collections SERVED from the SQLite document store
|
||||
* (reads + the primary write target).
|
||||
* CHAT_STORE_DUALWRITE — collections written to BOTH stores, mirrored by the
|
||||
* primary store's `_id`.
|
||||
*
|
||||
* The four states this yields drive the whole Mongo→SQLite cutover, and any one
|
||||
* is a pure config change (no redeploy of logic):
|
||||
*
|
||||
* neither → mongoose only (untouched default).
|
||||
* dualwrite only → mongoose primary (served) + SQLite mirror [pre-flip].
|
||||
* sqlite + dualwrite → SQLite primary (served) + mongoose mirror [post-flip,
|
||||
* Mongo kept intact as the instant escape hatch].
|
||||
* sqlite only → SQLite only, no mirror [Mongo gone].
|
||||
*
|
||||
* Unsetting CHAT_STORE_SQLITE reverts serving to mongoose instantly (the mirror
|
||||
* kept it current), so the flip is reversible until Mongo is deleted.
|
||||
*/
|
||||
function applySqliteOverrides<T extends Record<string, unknown>>(models: T): T {
|
||||
const csv = process.env.CHAT_STORE_SQLITE?.trim();
|
||||
if (!csv) {
|
||||
return models;
|
||||
}
|
||||
const names = csv
|
||||
function parseStoreCsv(value?: string): string[] {
|
||||
return (value ?? '')
|
||||
.split(',')
|
||||
.map((s) => s.trim())
|
||||
.filter((s) => s in CHAT_COLLECTION_SPECS);
|
||||
if (names.length === 0) {
|
||||
}
|
||||
|
||||
/**
|
||||
* One SQLite handle (one `node:sqlite` connection) per process, shared across
|
||||
* every `createModels` call — the api requires the data-schemas index from three
|
||||
* entry points, and a per-call handle would open three connections to the same
|
||||
* file and race for the WAL write lock. Keyed by the collection set so a changed
|
||||
* flag set (only happens across a restart) rebuilds cleanly.
|
||||
*/
|
||||
let sharedHandle: SqliteHandle | undefined;
|
||||
let sharedHandleKey = '';
|
||||
|
||||
function sharedSqliteHandle(names: string[]): SqliteHandle {
|
||||
const key = [...names].sort().join(',');
|
||||
if (!sharedHandle || sharedHandleKey !== key) {
|
||||
sharedHandle = createSqliteHandle(names);
|
||||
sharedHandleKey = key;
|
||||
}
|
||||
return sharedHandle;
|
||||
}
|
||||
|
||||
function applySqliteOverrides<T extends Record<string, unknown>>(models: T): T {
|
||||
const sqliteNames = new Set(parseStoreCsv(process.env.CHAT_STORE_SQLITE));
|
||||
const dualNames = new Set(parseStoreCsv(process.env.CHAT_STORE_DUALWRITE));
|
||||
const union = [...new Set([...sqliteNames, ...dualNames])];
|
||||
if (union.length === 0) {
|
||||
return models;
|
||||
}
|
||||
const handle = createSqliteHandle(names);
|
||||
const handle = sharedSqliteHandle(union);
|
||||
const out = { ...models } as Record<string, unknown>;
|
||||
for (const name of names) {
|
||||
out[name] = handle.models[name];
|
||||
for (const name of union) {
|
||||
const mongooseModel = models[name];
|
||||
const sqliteModel = handle.models[name];
|
||||
const sqlitePrimary = sqliteNames.has(name);
|
||||
if (dualNames.has(name)) {
|
||||
const primary = sqlitePrimary ? sqliteModel : mongooseModel;
|
||||
const mirror = sqlitePrimary ? mongooseModel : sqliteModel;
|
||||
out[name] = createDualWriteModel(primary, mirror);
|
||||
} else {
|
||||
out[name] = sqliteModel;
|
||||
}
|
||||
}
|
||||
return out as T;
|
||||
}
|
||||
|
||||
@@ -686,6 +686,27 @@ export class DocModel {
|
||||
return hydrate(this.insertDoc(doc));
|
||||
}
|
||||
|
||||
/**
|
||||
* Exact by-`_id` upsert — the shared primitive for the dual-write mirror and
|
||||
* the Mongo→SQLite backfill. Writes the document verbatim: ObjectId-like
|
||||
* values coerced to hex (so a doc read from a mongoose `.lean()` serializes
|
||||
* correctly), Dates preserved, keyed by the document's own `_id`. NO timestamp
|
||||
* stamping, NO schema defaults, NO tenant scoping — the source store already
|
||||
* owns those. Idempotent: re-running with the same `_id` replaces the row, so
|
||||
* the backfill and live mirroring converge on the same keyspace and never
|
||||
* duplicate (both sides key on the primary store's `_id`).
|
||||
*/
|
||||
upsertRaw(input: Doc): void {
|
||||
const doc = deepCoerceIds(input) as Doc;
|
||||
const id = doc._id;
|
||||
if (id == null) {
|
||||
throw new Error(`[DocModel:${this.modelName}] upsertRaw requires _id`);
|
||||
}
|
||||
this.db
|
||||
.prepare(`INSERT OR REPLACE INTO ${this.table} (_id, doc) VALUES (?, ?)`)
|
||||
.run(String(id), this.serialize(doc));
|
||||
}
|
||||
|
||||
async bulkWrite(
|
||||
ops: Array<Record<string, unknown>>,
|
||||
): Promise<BulkResult> {
|
||||
|
||||
@@ -0,0 +1,173 @@
|
||||
import { createSqliteHandle, type SqliteHandle } from './index';
|
||||
import { createDualWriteModel } from './DualWriteModel';
|
||||
import { CHAT_COLLECTION_SPECS } from './collections';
|
||||
import type { DocModel } from './DocModel';
|
||||
|
||||
/**
|
||||
* A minimal mongoose-Model stand-in for the mongo-mirror branch: it is NOT a
|
||||
* DocModel, exposes `.base.Types.ObjectId` and a native-driver-shaped
|
||||
* `.collection` with replaceOne/deleteOne, and records writes in a Map so the
|
||||
* test can assert the mirror is keyed by the primary's `_id` (as an ObjectId).
|
||||
*/
|
||||
class FakeObjectId {
|
||||
constructor(private readonly hex: string) {}
|
||||
toString() {
|
||||
return this.hex;
|
||||
}
|
||||
toHexString() {
|
||||
return this.hex;
|
||||
}
|
||||
}
|
||||
function makeFakeMongoModel() {
|
||||
const store = new Map<string, Record<string, unknown>>();
|
||||
return {
|
||||
store,
|
||||
base: { Types: { ObjectId: FakeObjectId } },
|
||||
collection: {
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
replaceOne: async (filter: any, doc: any) => {
|
||||
store.set(String(filter._id), doc);
|
||||
return { acknowledged: true };
|
||||
},
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
deleteOne: async (filter: any) => {
|
||||
store.delete(String(filter._id));
|
||||
return { acknowledged: true };
|
||||
},
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/* eslint-disable @typescript-eslint/no-explicit-any */
|
||||
describe('DualWriteModel — SQLite primary, SQLite mirror', () => {
|
||||
let primaryH: SqliteHandle;
|
||||
let mirrorH: SqliteHandle;
|
||||
let primary: DocModel;
|
||||
let mirror: DocModel;
|
||||
let dual: any;
|
||||
|
||||
beforeEach(() => {
|
||||
primaryH = createSqliteHandle(['Conversation']);
|
||||
mirrorH = createSqliteHandle(['Conversation']);
|
||||
primary = primaryH.models.Conversation;
|
||||
mirror = mirrorH.models.Conversation;
|
||||
dual = createDualWriteModel(primary, mirror);
|
||||
});
|
||||
afterEach(() => {
|
||||
primaryH.close();
|
||||
mirrorH.close();
|
||||
});
|
||||
|
||||
it('create mirrors the doc by the primary _id', async () => {
|
||||
const created: any = await dual.create({ conversationId: 'c1', user: 'u1', title: 'A' });
|
||||
expect(created._id).toBeTruthy();
|
||||
const inMirror = await mirror.findOne({ _id: created._id }).lean();
|
||||
expect(inMirror).toBeTruthy();
|
||||
expect((inMirror as any).conversationId).toBe('c1');
|
||||
// Same _id in both stores (mirror keyed by primary _id, not by value).
|
||||
expect(String((inMirror as any)._id)).toBe(String(created._id));
|
||||
expect(await primary.countDocuments({})).toBe(1);
|
||||
expect(await mirror.countDocuments({})).toBe(1);
|
||||
});
|
||||
|
||||
it('reads pass through to primary only', async () => {
|
||||
await primary.create({ conversationId: 'only-primary', user: 'u1' });
|
||||
const viaDual = await dual.findOne({ conversationId: 'only-primary' }).lean();
|
||||
expect(viaDual).toBeTruthy();
|
||||
// mirror never saw a direct primary.create — proves reads do not touch it
|
||||
expect(await mirror.countDocuments({})).toBe(0);
|
||||
});
|
||||
|
||||
it('updateOne mirrors the mutation', async () => {
|
||||
const c: any = await dual.create({ conversationId: 'c2', user: 'u1', title: 'old' });
|
||||
await dual.updateOne({ conversationId: 'c2' }, { $set: { title: 'new' } });
|
||||
const m = await mirror.findOne({ _id: c._id }).lean();
|
||||
expect((m as any).title).toBe('new');
|
||||
});
|
||||
|
||||
it('findOneAndUpdate upsert mirrors the inserted doc', async () => {
|
||||
const created: any = await dual.findOneAndUpdate(
|
||||
{ conversationId: 'c3', user: 'u1' },
|
||||
{ $set: { title: 'up' } },
|
||||
{ upsert: true, new: true },
|
||||
);
|
||||
expect(created.conversationId).toBe('c3');
|
||||
const m = await mirror.findOne({ conversationId: 'c3' }).lean();
|
||||
expect(m).toBeTruthy();
|
||||
expect(String((m as any)._id)).toBe(String(created._id));
|
||||
});
|
||||
|
||||
it('deleteOne removes the doc from the mirror', async () => {
|
||||
const c: any = await dual.create({ conversationId: 'c4', user: 'u1' });
|
||||
expect(await mirror.countDocuments({})).toBe(1);
|
||||
await dual.deleteOne({ conversationId: 'c4' });
|
||||
expect(await mirror.findOne({ _id: c._id }).lean()).toBeNull();
|
||||
expect(await mirror.countDocuments({})).toBe(0);
|
||||
});
|
||||
|
||||
it('bulkWrite mirrors inserts, updates and deletes', async () => {
|
||||
const seed: any = await dual.create({ conversationId: 'keep', user: 'u1', title: 't' });
|
||||
await dual.bulkWrite([
|
||||
{ insertOne: { document: { conversationId: 'bulk-ins', user: 'u1' } } },
|
||||
{ updateOne: { filter: { conversationId: 'keep' }, update: { $set: { title: 't2' } } } },
|
||||
]);
|
||||
expect(await mirror.findOne({ conversationId: 'bulk-ins' }).lean()).toBeTruthy();
|
||||
const kept = await mirror.findOne({ _id: seed._id }).lean();
|
||||
expect((kept as any).title).toBe('t2');
|
||||
});
|
||||
|
||||
it('idempotent: a re-mirror of the same _id does not duplicate (backfill safety)', async () => {
|
||||
const c: any = await dual.create({ conversationId: 'c5', user: 'u1' });
|
||||
// simulate the backfill upserting the same primary doc again by _id
|
||||
const raw = await primary.findById(String(c._id)).lean();
|
||||
mirror.upsertRaw(raw as any);
|
||||
mirror.upsertRaw(raw as any);
|
||||
expect(await mirror.countDocuments({ conversationId: 'c5' })).toBe(1);
|
||||
});
|
||||
});
|
||||
|
||||
describe('DualWriteModel — SQLite primary, mongo mirror (escape-hatch direction)', () => {
|
||||
let primaryH: SqliteHandle;
|
||||
let primary: DocModel;
|
||||
let fake: ReturnType<typeof makeFakeMongoModel>;
|
||||
let dual: any;
|
||||
|
||||
beforeEach(() => {
|
||||
primaryH = createSqliteHandle(['Conversation']);
|
||||
primary = primaryH.models.Conversation;
|
||||
fake = makeFakeMongoModel();
|
||||
dual = createDualWriteModel(primary, fake);
|
||||
});
|
||||
afterEach(() => primaryH.close());
|
||||
|
||||
it('mirrors creates to the mongo collection keyed by an ObjectId _id', async () => {
|
||||
const created: any = await dual.create({ conversationId: 'm1', user: 'u1' });
|
||||
const hex = String(created._id);
|
||||
expect(fake.store.has(hex)).toBe(true);
|
||||
expect((fake.store.get(hex) as any)._id).toBeInstanceOf(FakeObjectId);
|
||||
expect((fake.store.get(hex) as any).conversationId).toBe('m1');
|
||||
});
|
||||
|
||||
it('mirrors deletes to the mongo collection', async () => {
|
||||
const created: any = await dual.create({ conversationId: 'm2', user: 'u1' });
|
||||
const hex = String(created._id);
|
||||
expect(fake.store.has(hex)).toBe(true);
|
||||
await dual.deleteOne({ conversationId: 'm2' });
|
||||
expect(fake.store.has(hex)).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe('CHAT_COLLECTION_SPECS — all specs build a valid store', () => {
|
||||
it('opens a handle over every spec without throwing (validates the 5 new specs)', () => {
|
||||
const names = Object.keys(CHAT_COLLECTION_SPECS);
|
||||
const handle = createSqliteHandle(names);
|
||||
for (const name of names) {
|
||||
expect(handle.models[name]).toBeTruthy();
|
||||
}
|
||||
// the auth+billing batch must be present and tableable
|
||||
for (const n of ['User', 'Session', 'Token', 'Balance', 'Transaction']) {
|
||||
expect(names).toContain(n);
|
||||
}
|
||||
handle.close();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,338 @@
|
||||
/**
|
||||
* Dual-write mirror — the migration bridge between the mongoose store and the
|
||||
* SQLite document store.
|
||||
*
|
||||
* A `DualWriteModel` wraps a `primary` model (the one that SERVES reads) and a
|
||||
* `mirror` model (kept convergent for the escape hatch / the flip). Every read
|
||||
* and every non-write property falls through to `primary` unchanged; the ten
|
||||
* write methods run on `primary` first, then replicate the resulting document
|
||||
* state to `mirror` KEYED BY THE PRIMARY'S `_id` (upsert if the doc still
|
||||
* exists, delete if it was removed). Because the mirror is addressed by the
|
||||
* primary's `_id` — never by value — the live mirror and the one-shot backfill
|
||||
* write the same keyspace and converge without duplicating.
|
||||
*
|
||||
* It is symmetric, so it powers both migration directions with the same code:
|
||||
* - pre-flip (CHAT_STORE_DUALWRITE only): primary = mongoose, mirror = SQLite
|
||||
* - post-flip (CHAT_STORE_SQLITE + DUALWRITE): primary = SQLite, mirror = mongoose
|
||||
*
|
||||
* `primary` / `mirror` are each either a mongoose `Model` or a `DocModel`; both
|
||||
* expose the same bounded Model API the chat data methods use (the pure-SQLite
|
||||
* seam already proves the app runs against that subset), so the wrapper only has
|
||||
* to special-case the mirror WRITE primitives per store kind.
|
||||
*/
|
||||
import { DocModel } from './DocModel';
|
||||
import { deepCoerceIds, type Doc, type Filter } from './engine';
|
||||
|
||||
/** Minimal shape the wrapper depends on — satisfied by mongoose Model and DocModel. */
|
||||
interface ModelLike {
|
||||
findById(id: unknown, projection?: unknown): { select(p: string): { lean(): Promise<Doc | null> }; lean(): Promise<Doc | null> };
|
||||
findOne(filter: Filter, projection?: unknown): { select(p: string): { lean(): Promise<Doc | null> } };
|
||||
find(filter: Filter, projection?: unknown): { select(p: string): { lean(): Promise<Doc[]> } };
|
||||
create(input: unknown): Promise<unknown>;
|
||||
insertMany(docs: unknown[], options?: unknown): Promise<unknown>;
|
||||
updateOne(filter: Filter, update: unknown, options?: unknown): Promise<{ upsertedId?: unknown }>;
|
||||
updateMany(filter: Filter, update: unknown, options?: unknown): Promise<{ upsertedId?: unknown }>;
|
||||
deleteOne(filter: Filter): Promise<unknown>;
|
||||
deleteMany(filter: Filter): Promise<unknown>;
|
||||
bulkWrite(ops: Array<Record<string, unknown>>, options?: unknown): Promise<Record<string, unknown>>;
|
||||
findOneAndUpdate(filter: Filter, update: unknown, options?: unknown): unknown;
|
||||
findByIdAndUpdate(id: unknown, update: unknown, options?: unknown): unknown;
|
||||
findOneAndDelete(filter: Filter, projection?: unknown): unknown;
|
||||
[key: string]: unknown;
|
||||
}
|
||||
|
||||
function idOf(doc: unknown): string | null {
|
||||
if (doc == null) {
|
||||
return null;
|
||||
}
|
||||
const id = (doc as Doc)._id ?? doc;
|
||||
return id == null ? null : String((id as { toString(): string }).toString());
|
||||
}
|
||||
|
||||
/** Extracts the id list from a bulkWrite result's insertedIds/upsertedIds map. */
|
||||
function idsFromResultMap(map: unknown): string[] {
|
||||
if (map == null || typeof map !== 'object') {
|
||||
return [];
|
||||
}
|
||||
return Object.values(map as Record<string, unknown>)
|
||||
.map((v) => idOf(v))
|
||||
.filter((v): v is string => v != null);
|
||||
}
|
||||
|
||||
class DualWriteModel {
|
||||
readonly primary: ModelLike;
|
||||
private readonly mirror: ModelLike;
|
||||
private readonly mirrorIsSqlite: boolean;
|
||||
/** mongoose `Types.ObjectId` ctor, needed to key the native mongo mirror. */
|
||||
private readonly ObjectId?: new (hex: string) => unknown;
|
||||
|
||||
constructor(primary: ModelLike, mirror: ModelLike) {
|
||||
this.primary = primary;
|
||||
this.mirror = mirror;
|
||||
this.mirrorIsSqlite = mirror instanceof DocModel;
|
||||
if (!this.mirrorIsSqlite) {
|
||||
// mongoose Model: `.base` is the mongoose instance carrying Types.ObjectId.
|
||||
const base = (mirror as { base?: { Types?: { ObjectId?: unknown } } }).base;
|
||||
this.ObjectId = base?.Types?.ObjectId as new (hex: string) => unknown;
|
||||
}
|
||||
}
|
||||
|
||||
/* ---------------------------------------------------------- mirror sync */
|
||||
|
||||
/**
|
||||
* Re-reads each id from `primary` and makes `mirror` match (upsert or delete).
|
||||
* A mirror failure is logged, never thrown: the primary (served) write already
|
||||
* succeeded, so the request must not fail because the insurance copy hiccuped.
|
||||
* Any gap is caught by the pre-flip count reconcile + the idempotent backfill.
|
||||
*/
|
||||
private async syncIds(ids: Array<string | null | undefined>): Promise<void> {
|
||||
const uniq = [...new Set(ids.map((i) => (i == null ? null : String(i))))].filter(
|
||||
(i): i is string => !!i,
|
||||
);
|
||||
for (const id of uniq) {
|
||||
try {
|
||||
const doc = await this.primary.findById(id).lean();
|
||||
if (doc) {
|
||||
await this.mirrorUpsert(doc);
|
||||
} else {
|
||||
await this.mirrorDelete(id);
|
||||
}
|
||||
} catch (err) {
|
||||
// eslint-disable-next-line no-console
|
||||
console.error(`[dual-write] mirror sync failed for _id=${id}:`, (err as Error)?.message);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async mirrorUpsert(doc: Doc): Promise<void> {
|
||||
if (this.mirrorIsSqlite) {
|
||||
(this.mirror as unknown as DocModel).upsertRaw(doc);
|
||||
return;
|
||||
}
|
||||
// mongoose mirror: write verbatim through the native driver (bypasses schema
|
||||
// casting + timestamp stamping) so the copy is exact. Top-level `_id` must be
|
||||
// a real ObjectId for mongo; nested id refs stay as stored (acceptable — the
|
||||
// mongo mirror is the escape hatch, never the served store post-flip).
|
||||
const oid = this.toObjectId(doc._id);
|
||||
const raw = { ...doc, _id: oid };
|
||||
await (this.mirror as { collection: { replaceOne: Function } }).collection.replaceOne(
|
||||
{ _id: oid },
|
||||
raw,
|
||||
{ upsert: true },
|
||||
);
|
||||
}
|
||||
|
||||
private async mirrorDelete(id: string): Promise<void> {
|
||||
if (this.mirrorIsSqlite) {
|
||||
await (this.mirror as unknown as DocModel).deleteOne({ _id: id });
|
||||
return;
|
||||
}
|
||||
await (this.mirror as { collection: { deleteOne: Function } }).collection.deleteOne({
|
||||
_id: this.toObjectId(id),
|
||||
});
|
||||
}
|
||||
|
||||
private toObjectId(id: unknown): unknown {
|
||||
const hex = String((id as { toString(): string }).toString());
|
||||
if (this.ObjectId && /^[0-9a-fA-F]{24}$/.test(hex)) {
|
||||
return new this.ObjectId(hex);
|
||||
}
|
||||
return id;
|
||||
}
|
||||
|
||||
/* ------------------------------------------------------------ writes */
|
||||
|
||||
async create(...args: unknown[]): Promise<unknown> {
|
||||
const res = await (this.primary.create as (...a: unknown[]) => Promise<unknown>)(...args);
|
||||
const docs = Array.isArray(res) ? res : [res];
|
||||
await this.syncIds(docs.map(idOf));
|
||||
return res;
|
||||
}
|
||||
|
||||
async insertMany(docs: unknown[], options?: unknown): Promise<unknown> {
|
||||
const res = await this.primary.insertMany(docs, options);
|
||||
const arr = Array.isArray(res) ? res : [res];
|
||||
await this.syncIds(arr.map(idOf));
|
||||
return res;
|
||||
}
|
||||
|
||||
async updateOne(filter: Filter, update: unknown, options?: unknown): Promise<unknown> {
|
||||
const pre = await this.primary.findOne(filter).select('_id').lean();
|
||||
const res = await this.primary.updateOne(filter, update, options);
|
||||
await this.syncIds([idOf(pre), idOf(res?.upsertedId)]);
|
||||
return res;
|
||||
}
|
||||
|
||||
async updateMany(filter: Filter, update: unknown, options?: unknown): Promise<unknown> {
|
||||
const pre = await this.primary.find(filter).select('_id').lean();
|
||||
const res = await this.primary.updateMany(filter, update, options);
|
||||
await this.syncIds([...pre.map(idOf), idOf(res?.upsertedId)]);
|
||||
return res;
|
||||
}
|
||||
|
||||
async deleteOne(filter: Filter): Promise<unknown> {
|
||||
const pre = await this.primary.findOne(filter).select('_id').lean();
|
||||
const res = await this.primary.deleteOne(filter);
|
||||
await this.syncIds([idOf(pre)]);
|
||||
return res;
|
||||
}
|
||||
|
||||
async deleteMany(filter: Filter): Promise<unknown> {
|
||||
const pre = await this.primary.find(filter).select('_id').lean();
|
||||
const res = await this.primary.deleteMany(filter);
|
||||
await this.syncIds(pre.map(idOf));
|
||||
return res;
|
||||
}
|
||||
|
||||
async bulkWrite(ops: Array<Record<string, unknown>>, options?: unknown): Promise<unknown> {
|
||||
const preIds: Array<string | null> = [];
|
||||
for (const op of ops) {
|
||||
const one = (op.updateOne ?? op.deleteOne) as { filter?: Filter } | undefined;
|
||||
const many = (op.updateMany ?? op.deleteMany) as { filter?: Filter } | undefined;
|
||||
if (many?.filter) {
|
||||
const rows = await this.primary.find(many.filter).select('_id').lean();
|
||||
preIds.push(...rows.map(idOf));
|
||||
} else if (one?.filter) {
|
||||
const row = await this.primary.findOne(one.filter).select('_id').lean();
|
||||
preIds.push(idOf(row));
|
||||
}
|
||||
}
|
||||
const res = await this.primary.bulkWrite(ops, options);
|
||||
await this.syncIds([
|
||||
...preIds,
|
||||
...idsFromResultMap(res?.insertedIds),
|
||||
...idsFromResultMap(res?.upsertedIds),
|
||||
]);
|
||||
return res;
|
||||
}
|
||||
|
||||
findOneAndUpdate(filter: Filter, update: unknown, options: { upsert?: boolean } = {}): DualWriteQuery {
|
||||
return new DualWriteQuery(
|
||||
this,
|
||||
() => this.primary.findOneAndUpdate(filter, update, options),
|
||||
filter,
|
||||
!!options.upsert,
|
||||
);
|
||||
}
|
||||
|
||||
findByIdAndUpdate(id: unknown, update: unknown, options: { upsert?: boolean } = {}): DualWriteQuery {
|
||||
return new DualWriteQuery(
|
||||
this,
|
||||
() => this.primary.findByIdAndUpdate(id, update, options),
|
||||
{ _id: id } as Filter,
|
||||
!!options.upsert,
|
||||
);
|
||||
}
|
||||
|
||||
findOneAndDelete(filter: Filter, projection?: unknown): DualWriteQuery {
|
||||
return new DualWriteQuery(
|
||||
this,
|
||||
() => this.primary.findOneAndDelete(filter, projection),
|
||||
filter,
|
||||
false,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Lazy, chainable result for the `findOneAnd*` methods. Forwards the mongoose /
|
||||
* DocModel query-builder chain (`select/lean/populate/sort/session/…`) to the
|
||||
* primary query so the caller's return shape is preserved, and mirrors the
|
||||
* affected document once the write resolves. Order matters: the pre-image id is
|
||||
* captured BEFORE the (lazy) primary query executes its write.
|
||||
*/
|
||||
class DualWriteQuery implements PromiseLike<unknown> {
|
||||
private pq: { [k: string]: Function } | null = null;
|
||||
|
||||
constructor(
|
||||
private readonly dwm: DualWriteModel,
|
||||
private readonly makePrimaryQuery: () => unknown,
|
||||
private readonly filter: Filter,
|
||||
private readonly upsert: boolean,
|
||||
) {}
|
||||
|
||||
private query(): { [k: string]: Function } {
|
||||
if (!this.pq) {
|
||||
this.pq = this.makePrimaryQuery() as { [k: string]: Function };
|
||||
}
|
||||
return this.pq;
|
||||
}
|
||||
|
||||
private chain(method: string, ...args: unknown[]): this {
|
||||
const q = this.query();
|
||||
if (typeof q[method] === 'function') {
|
||||
q[method](...args);
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
select(p: unknown): this {
|
||||
return this.chain('select', p);
|
||||
}
|
||||
lean(): this {
|
||||
return this.chain('lean');
|
||||
}
|
||||
populate(p: unknown): this {
|
||||
return this.chain('populate', p);
|
||||
}
|
||||
sort(p: unknown): this {
|
||||
return this.chain('sort', p);
|
||||
}
|
||||
skip(n: unknown): this {
|
||||
return this.chain('skip', n);
|
||||
}
|
||||
limit(n: unknown): this {
|
||||
return this.chain('limit', n);
|
||||
}
|
||||
session(s: unknown): this {
|
||||
return this.chain('session', s);
|
||||
}
|
||||
|
||||
private async run(): Promise<unknown> {
|
||||
const primary = (this.dwm as unknown as { primary: ModelLike }).primary;
|
||||
const pre = await primary.findOne(this.filter).select('_id').lean();
|
||||
const result = await (this.query() as unknown as Promise<Doc | null>);
|
||||
const ids = [idOf(pre), idOf(result)];
|
||||
if (!pre && !result && this.upsert) {
|
||||
const seeded = await primary.findOne(this.filter).select('_id').lean();
|
||||
ids.push(idOf(seeded));
|
||||
}
|
||||
await (this.dwm as unknown as { syncIds(ids: unknown[]): Promise<void> })['syncIds'](ids);
|
||||
return result;
|
||||
}
|
||||
|
||||
then<TResult1 = unknown, TResult2 = never>(
|
||||
onfulfilled?: ((value: unknown) => TResult1 | PromiseLike<TResult1>) | null,
|
||||
onrejected?: ((reason: unknown) => TResult2 | PromiseLike<TResult2>) | null,
|
||||
): Promise<TResult1 | TResult2> {
|
||||
return this.run().then(onfulfilled, onrejected);
|
||||
}
|
||||
catch<T = never>(onrejected?: ((reason: unknown) => T | PromiseLike<T>) | null): Promise<unknown> {
|
||||
return this.run().catch(onrejected);
|
||||
}
|
||||
exec(): Promise<unknown> {
|
||||
return this.run();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Builds a dual-write model as a Proxy: the write methods above take precedence,
|
||||
* every other access (find/findOne/aggregate/countDocuments/modelName/…) falls
|
||||
* through to `primary`, so the wrapper's surface is exactly the primary's plus
|
||||
* mirrored writes. `syncIds` is private but reached by DualWriteQuery via the
|
||||
* bracket accessor, so it must resolve on the target, not the primary.
|
||||
*/
|
||||
export function createDualWriteModel(primary: unknown, mirror: unknown): unknown {
|
||||
const dwm = new DualWriteModel(primary as ModelLike, mirror as ModelLike);
|
||||
return new Proxy(dwm, {
|
||||
get(target, prop, receiver) {
|
||||
if (prop in target) {
|
||||
return Reflect.get(target, prop, receiver);
|
||||
}
|
||||
const value = (primary as Record<string | symbol, unknown>)[prop];
|
||||
return typeof value === 'function' ? (value as Function).bind(primary) : value;
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
export { DualWriteModel };
|
||||
@@ -208,4 +208,41 @@ export const CHAT_COLLECTION_SPECS: Record<string, CollectionSpec> = {
|
||||
index: ['source', 'idOnTheSource', 'name', 'memberIds'],
|
||||
dateFields: ['createdAt', 'updatedAt'],
|
||||
},
|
||||
|
||||
// ---- Batch 9: auth + billing — the collections that ACTUALLY hold live data
|
||||
// in chat-docdb outside the domains above (users/sessions/transactions/balances
|
||||
// are populated; tokens is on the same auth path). These must move to SQLite
|
||||
// for chat-docdb to be deletable without data loss — User is the hot-path record
|
||||
// that every authenticated request loads and that every conversation references
|
||||
// by `_id`, so its `_id` MUST survive the migration (backfill preserves it).
|
||||
// Their data methods read `.models`/`.Types` off the seam handle, so they route
|
||||
// here once flagged (see methods/index.ts, api/models/Transaction.js). Uniques
|
||||
// on the sparse social/openid ids are safe: an absent id is NULL in json_extract
|
||||
// and SQLite treats NULLs as distinct, so it mirrors the mongoose `sparse`.
|
||||
User: {
|
||||
name: 'User',
|
||||
unique: ['email', 'openidId', 'googleId', 'githubId'],
|
||||
index: ['organization', 'provider', 'username', 'idOnTheSource', 'role'],
|
||||
dateFields: ['createdAt', 'updatedAt', 'expiresAt'],
|
||||
},
|
||||
Session: {
|
||||
name: 'Session',
|
||||
index: ['refreshToken', 'user', 'expiration'],
|
||||
dateFields: ['expiration', 'createdAt', 'updatedAt'],
|
||||
},
|
||||
Token: {
|
||||
name: 'Token',
|
||||
index: ['userId', 'token', 'email', 'type', 'identifier'],
|
||||
dateFields: ['createdAt', 'expiresAt'],
|
||||
},
|
||||
Balance: {
|
||||
name: 'Balance',
|
||||
index: ['user', 'commerceUserId'],
|
||||
dateFields: ['lastRefill', 'expiresAt', 'creditsGrantedAt', 'createdAt', 'updatedAt'],
|
||||
},
|
||||
Transaction: {
|
||||
name: 'Transaction',
|
||||
index: ['user', 'conversationId', 'model', 'createdAt'],
|
||||
dateFields: ['createdAt', 'updatedAt'],
|
||||
},
|
||||
};
|
||||
|
||||
@@ -18,6 +18,7 @@ import { ObjectId } from './engine';
|
||||
export { DocModel, type CollectionSpec } from './DocModel';
|
||||
export { CHAT_COLLECTION_SPECS } from './collections';
|
||||
export { ObjectId } from './engine';
|
||||
export { createDualWriteModel, DualWriteModel } from './DualWriteModel';
|
||||
|
||||
export interface SqliteHandle {
|
||||
models: Record<string, DocModel>;
|
||||
@@ -42,6 +43,10 @@ export function openDatabase(dbPath?: string): DatabaseSync {
|
||||
if (path !== ':memory:') {
|
||||
db.exec('PRAGMA journal_mode = WAL');
|
||||
db.exec('PRAGMA synchronous = NORMAL');
|
||||
// The one-shot Mongo→SQLite backfill runs as a SEPARATE process against this
|
||||
// same file; WAL permits one writer at a time, so wait for the lock rather
|
||||
// than erroring when the live pod and the backfill overlap.
|
||||
db.exec('PRAGMA busy_timeout = 5000');
|
||||
}
|
||||
db.exec('PRAGMA foreign_keys = ON');
|
||||
return db;
|
||||
|
||||
Reference in New Issue
Block a user