feat(db): Postgres backend + dialect-portable queries + data migration (Phase 5)

- db/pg.js: the pg backend (prepare/exec/tx/init) — ?→$N translation, BIGINT parsed
  as Number (matches sqlite; else expires_at<Date.now() compares string<number),
  transactions on one pooled client, init() applies schema.pg.sql. Same interface as
  db/sqlite.js, so repos are unchanged.
- repos.js: the ~7 SQLite-only queries rewritten to run on BOTH engines —
  audit.add @named→positional; email lookups COLLATE NOCASE→LOWER()=LOWER();
  INSERT OR IGNORE→ON CONFLICT DO NOTHING (addMember/poll vote/favorite);
  mergeInto's UPDATE OR IGNORE→UPDATE…WHERE NOT EXISTS/NOT IN and INSERT OR
  REPLACE→ON CONFLICT DO UPDATE. Re-validated on sqlite: db-smoke still 22/22.
- server.js: boot now `await db.init()` before listening (pg creates tables; sqlite
  no-op), so the first request can't hit a missing table.
- db/migrate-sqlite-to-pg.js: one-shot row copy in FK order (bulk insert, TRUNCATE
  first so re-runnable). audit_log id left to PG's identity.
- package.json: add pg ^8.13.1.

Next: validate DB_BACKEND=pg smoke against a real Postgres on the server, then merge.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-07-24 22:28:04 +05:30
parent 3250530596
commit e67a783bdc
5 changed files with 170 additions and 45 deletions
+56
View File
@@ -0,0 +1,56 @@
// One-shot data migration: copy every row from the SQLite data.db into Postgres. Run ONCE at cutover,
// with the app stopped, BEFORE switching DB_BACKEND to pg.
//
// DB_PATH=/data/data.db DATABASE_URL=postgres://user:pass@host/db node db/migrate-sqlite-to-pg.js
//
// It applies the Postgres schema first, TRUNCATEs the target tables (so a re-run re-copies cleanly), then
// bulk-inserts in FK-dependency order. audit_log.id is a GENERATED identity, so its id is not copied (PG
// assigns fresh ones — nothing references audit_log.id). Timestamps/flags are plain integers on both sides.
const fs = require('fs');
const path = require('path');
const { DatabaseSync } = require('node:sqlite');
const { Pool } = require('pg');
const SQLITE = process.env.DB_PATH || path.join(__dirname, '..', 'data.db');
if (!process.env.DATABASE_URL) { console.error('DATABASE_URL is required'); process.exit(1); }
const src = new DatabaseSync(SQLITE);
const pool = new Pool({ connectionString: process.env.DATABASE_URL, max: 4 });
// Parents before children (users→teams, sessions_auth→users, machines→teams); the rest have no FKs.
const ORDER = [
'teams', 'users', 'machines', 'sessions_auth', 'audit_log', 'sessions_log', 'refresh_tokens',
'api_keys', 'webhooks', 'messages', 'message_reactions', 'attachments', 'conversations',
'conversation_members', 'call_history', 'user_aliases', 'polls', 'poll_votes', 'scheduled_meetings',
'recordings', 'push_subscriptions', 'device_tokens', 'app_installs', 'favorites',
];
async function main() {
await pool.query(fs.readFileSync(path.join(__dirname, 'schema.pg.sql'), 'utf8')); // ensure schema exists
await pool.query('TRUNCATE ' + ORDER.map((t) => '"' + t + '"').join(', ') + ' RESTART IDENTITY CASCADE');
const totals = {};
for (const table of ORDER) {
let rows = [];
try { rows = src.prepare('SELECT * FROM ' + table).all(); } catch (e) { totals[table] = 'skip(' + e.message + ')'; continue; }
if (!rows.length) { totals[table] = 0; continue; }
let cols = Object.keys(rows[0]);
if (table === 'audit_log') cols = cols.filter((c) => c !== 'id'); // GENERATED — let PG assign
const colList = cols.map((c) => '"' + c + '"').join(',');
const CHUNK = 400; // keep param count well under Postgres' 65535 limit even for wide tables
for (let i = 0; i < rows.length; i += CHUNK) {
const batch = rows.slice(i, i + CHUNK);
const values = []; const params = [];
batch.forEach((r, ri) => {
values.push('(' + cols.map((c, ci) => '$' + (ri * cols.length + ci + 1)).join(',') + ')');
cols.forEach((c) => params.push(r[c] === undefined ? null : r[c]));
});
await pool.query('INSERT INTO "' + table + '" (' + colList + ') VALUES ' + values.join(','), params);
}
totals[table] = rows.length;
}
console.log('MIGRATED rows:', JSON.stringify(totals, null, 0));
await pool.end();
}
main().catch((e) => { console.error('MIGRATION FAILED:', e && e.message); process.exit(1); });
+59
View File
@@ -0,0 +1,59 @@
// PostgreSQL backend for the async DB adapter. Same interface as db/sqlite.js — prepare(sql).{get,all,run},
// exec(sql), tx(fn), init() — so repos and app code are engine-agnostic. Selected by DB_BACKEND=pg;
// connection string from DATABASE_URL.
const { Pool, types } = require('pg');
const fs = require('fs');
const path = require('path');
// BIGINT (int8, OID 20) defaults to STRING in node-postgres to avoid precision loss. Every BIGINT here is
// an epoch-ms timestamp or a byte size — all far below Number.MAX_SAFE_INTEGER — so parse them as numbers
// to match the SQLite backend. Otherwise `expires_at < Date.now()` would compare a string to a number.
types.setTypeParser(20, (v) => (v === null ? null : parseInt(v, 10)));
const pool = new Pool({ connectionString: process.env.DATABASE_URL, max: 10 });
// Repos use '?' placeholders (SQLite style); Postgres wants $1,$2,… — replace positionally. Safe because
// no literal '?' appears inside any SQL string literal in this codebase.
function toPg(sql) { let i = 0; return sql.replace(/\?/g, () => '$' + (++i)); }
function prepare(sql) {
const q = toPg(sql);
return {
get: (...p) => pool.query(q, p).then((r) => r.rows[0]),
all: (...p) => pool.query(q, p).then((r) => r.rows),
run: (...p) => pool.query(q, p).then((r) => ({ changes: r.rowCount, lastInsertRowid: undefined })),
};
}
function exec(sql) { return pool.query(sql).then(() => {}); }
// Transaction on ONE pooled client (a pool would scatter BEGIN/COMMIT across connections). Same runner
// shape the sqlite backend's tx() exposes, so repos.mergeInto is identical on both engines.
async function tx(fn) {
const client = await pool.connect();
try {
await client.query('BEGIN');
const t = {
run: (sql, ...p) => client.query(toPg(sql), p).then((r) => ({ changes: r.rowCount })),
get: (sql, ...p) => client.query(toPg(sql), p).then((r) => r.rows[0]),
all: (sql, ...p) => client.query(toPg(sql), p).then((r) => r.rows),
};
const out = await fn(t);
await client.query('COMMIT');
return out;
} catch (e) {
try { await client.query('ROLLBACK'); } catch (_) {}
throw e;
} finally {
client.release();
}
}
// Apply the schema (all CREATE ... IF NOT EXISTS — idempotent). Multi-statement, no params, so it runs via
// the simple-query protocol in one call. MUST be awaited before serving (server.js boot).
async function init() {
const sql = fs.readFileSync(path.join(__dirname, 'schema.pg.sql'), 'utf8');
await pool.query(sql);
}
module.exports = { prepare, exec, tx, init, name: 'pg', _pool: pool };
+1
View File
@@ -10,6 +10,7 @@
"node": ">=22.5.0"
},
"dependencies": {
"pg": "^8.13.1",
"web-push": "^3.6.7",
"ws": "^8.18.0"
},
+20 -19
View File
@@ -31,8 +31,9 @@ const users = {
// Follow a merge redirect: a merged-away id resolves to the surviving account, else returns id
// unchanged. Use for any user id that arrived from the client (DM recipient, thread peer).
resolve: async (id) => { if (!id) return id; const a = await db.prepare('SELECT user_id FROM user_aliases WHERE old_id=?').get(id); return a ? a.user_id : id; },
byEmail: (email) => db.prepare('SELECT * FROM users WHERE email=? COLLATE NOCASE').get(email),
emailExists: async (email) => !!(await db.prepare('SELECT 1 FROM users WHERE email=? COLLATE NOCASE').get(email)),
// LOWER()=LOWER() is case-insensitive on both engines (SQLite has no portable COLLATE NOCASE in Postgres).
byEmail: (email) => db.prepare('SELECT * FROM users WHERE LOWER(email)=LOWER(?)').get(email),
emailExists: async (email) => !!(await db.prepare('SELECT 1 FROM users WHERE LOWER(email)=LOWER(?)').get(email)),
// Match on the stable BizGaze person-id so email- and mobile-logins resolve to one account (#2).
byBizgazeId: (bizId) => (bizId ? db.prepare('SELECT * FROM users WHERE bizgaze_user_id=?').get(String(bizId)) : undefined),
setBizgazeId: (id, bizId) => db.prepare('UPDATE users SET bizgaze_user_id=? WHERE id=?').run(bizId != null ? String(bizId) : null, id),
@@ -70,16 +71,19 @@ const users = {
const run = (sql, ...a) => t.run(sql, ...a);
await run('UPDATE messages SET sender_id=? WHERE sender_id=?', intoId, fromId);
await run('UPDATE messages SET recipient_id=? WHERE recipient_id=?', intoId, fromId);
await run('UPDATE OR IGNORE message_reactions SET user_id=? WHERE user_id=?', intoId, fromId);
// Composite-key tables: MOVE only the rows that won't collide with the survivor's existing rows
// (NOT EXISTS / NOT IN — portable across SQLite & Postgres, replacing SQLite-only UPDATE OR IGNORE),
// then DELETE whatever remains (the survivor already had that membership/reaction/vote).
await run('UPDATE message_reactions SET user_id=? WHERE user_id=? AND NOT EXISTS (SELECT 1 FROM message_reactions x WHERE x.message_id=message_reactions.message_id AND x.emoji=message_reactions.emoji AND x.user_id=?)', intoId, fromId, intoId);
await run('DELETE FROM message_reactions WHERE user_id=?', fromId);
await run('UPDATE OR IGNORE conversation_members SET user_id=? WHERE user_id=?', intoId, fromId);
await run('UPDATE conversation_members SET user_id=? WHERE user_id=? AND conversation_id NOT IN (SELECT conversation_id FROM conversation_members WHERE user_id=?)', intoId, fromId, intoId);
await run('DELETE FROM conversation_members WHERE user_id=?', fromId);
await run('UPDATE OR IGNORE poll_votes SET user_id=? WHERE user_id=?', intoId, fromId);
await run('UPDATE poll_votes SET user_id=? WHERE user_id=? AND NOT EXISTS (SELECT 1 FROM poll_votes x WHERE x.poll_id=poll_votes.poll_id AND x.option_idx=poll_votes.option_idx AND x.user_id=?)', intoId, fromId, intoId);
await run('DELETE FROM poll_votes WHERE user_id=?', fromId);
await run('UPDATE OR IGNORE favorites SET user_id=? WHERE user_id=?', intoId, fromId);
await run('UPDATE favorites SET user_id=? WHERE user_id=? AND target NOT IN (SELECT target FROM favorites WHERE user_id=?)', intoId, fromId, intoId);
await run('DELETE FROM favorites WHERE user_id=?', fromId);
// A DM favourite pointing AT the merged-away user should now point at the survivor.
await run("UPDATE OR IGNORE favorites SET target='dm:'||? WHERE target='dm:'||?", intoId, fromId);
await run("UPDATE favorites SET target='dm:'||? WHERE target='dm:'||? AND NOT EXISTS (SELECT 1 FROM favorites x WHERE x.user_id=favorites.user_id AND x.target='dm:'||?)", intoId, fromId, intoId);
await run("DELETE FROM favorites WHERE target='dm:'||?", fromId);
await run('UPDATE conversations SET created_by=? WHERE created_by=?', intoId, fromId);
await run('UPDATE polls SET created_by=? WHERE created_by=?', intoId, fromId);
@@ -87,8 +91,8 @@ const users = {
await run('UPDATE recordings SET created_by=? WHERE created_by=?', intoId, fromId);
await run('UPDATE attachments SET uploader_id=? WHERE uploader_id=?', intoId, fromId);
await run('UPDATE push_subscriptions SET user_id=? WHERE user_id=?', intoId, fromId);
await run('UPDATE OR IGNORE device_tokens SET user_id=? WHERE user_id=?', intoId, fromId);
await run('DELETE FROM device_tokens WHERE user_id=?', fromId);
// device_tokens: changing user_id can't violate its PK (id) or UNIQUE (token), so a plain UPDATE.
await run('UPDATE device_tokens SET user_id=? WHERE user_id=?', intoId, fromId);
await run('UPDATE app_installs SET user_id=? WHERE user_id=?', intoId, fromId);
// Drop the merged-away account's auth so a stale token can't resurrect the empty row.
await run('DELETE FROM sessions_auth WHERE user_id=?', fromId);
@@ -96,7 +100,7 @@ const users = {
// Record old_id -> survivor so lingering references (cached contacts, in-flight DMs) resolve
// instead of hitting the deleted row (which made messages to merged contacts vanish).
const _iu = await t.get('SELECT team_id FROM users WHERE id=?', intoId);
await run('INSERT OR REPLACE INTO user_aliases (old_id,user_id,team_id,created_at) VALUES (?,?,?,?)', fromId, intoId, (_iu && _iu.team_id) || null, now());
await run('INSERT INTO user_aliases (old_id,user_id,team_id,created_at) VALUES (?,?,?,?) ON CONFLICT(old_id) DO UPDATE SET user_id=excluded.user_id, team_id=excluded.team_id, created_at=excluded.created_at', fromId, intoId, (_iu && _iu.team_id) || null, now());
await run('UPDATE user_aliases SET user_id=? WHERE user_id=?', intoId, fromId); // re-chain earlier aliases to the new survivor
await run('DELETE FROM users WHERE id=?', fromId);
});
@@ -128,14 +132,11 @@ const machines = {
};
const audit = {
// Positional params (not @named) so the same SQL runs on SQLite and Postgres.
add: (e) =>
db.prepare(`INSERT INTO audit_log (team_id,user_id,user_email,machine_id,machine_name,action,detail,at)
VALUES (@team_id,@user_id,@user_email,@machine_id,@machine_name,@action,@detail,@at)`)
.run({
team_id: e.team_id, user_id: e.user_id || null, user_email: e.user_email || null,
machine_id: e.machine_id || null, machine_name: e.machine_name || null,
action: e.action, detail: e.detail || null, at: now(),
}),
VALUES (?,?,?,?,?,?,?,?)`)
.run(e.team_id, e.user_id || null, e.user_email || null, e.machine_id || null, e.machine_name || null, e.action, e.detail || null, now()),
listByTenant: (tenantId) =>
db.prepare("SELECT * FROM audit_log WHERE team_id=? OR team_id='adhoc' ORDER BY at DESC LIMIT 200").all(tenantId),
};
@@ -272,7 +273,7 @@ const conversations = {
db.prepare('INSERT INTO conversations (id,team_id,type,name,created_by,created_at) VALUES (?,?,?,?,?,?)').run(id, teamId, 'group', name || null, createdBy || null, now()),
byId: (id) => db.prepare('SELECT * FROM conversations WHERE id=?').get(id),
addMember: (conversationId, userId, admin) =>
db.prepare('INSERT OR IGNORE INTO conversation_members (conversation_id,user_id,last_read_at,joined_at,admin) VALUES (?,?,?,?,?)').run(conversationId, userId, 0, now(), admin ? 1 : 0),
db.prepare('INSERT INTO conversation_members (conversation_id,user_id,last_read_at,joined_at,admin) VALUES (?,?,?,?,?) ON CONFLICT(conversation_id,user_id) DO NOTHING').run(conversationId, userId, 0, now(), admin ? 1 : 0),
members: async (conversationId) => (await db.prepare('SELECT user_id FROM conversation_members WHERE conversation_id=? ORDER BY joined_at ASC').all(conversationId)).map((r) => r.user_id),
isMember: async (conversationId, userId) => !!(await db.prepare('SELECT 1 FROM conversation_members WHERE conversation_id=? AND user_id=?').get(conversationId, userId)),
// ---- Group admins (multiple allowed) ----
@@ -355,7 +356,7 @@ const polls = {
const pollVotes = {
forPoll: (pollId) => db.prepare('SELECT user_id, option_idx FROM poll_votes WHERE poll_id=?').all(pollId),
hasVoted: async (pollId, userId, idx) => !!(await db.prepare('SELECT 1 FROM poll_votes WHERE poll_id=? AND user_id=? AND option_idx=?').get(pollId, userId, idx)),
add: (pollId, userId, idx) => db.prepare('INSERT OR IGNORE INTO poll_votes (poll_id,user_id,option_idx,created_at) VALUES (?,?,?,?)').run(pollId, userId, idx, now()),
add: (pollId, userId, idx) => db.prepare('INSERT INTO poll_votes (poll_id,user_id,option_idx,created_at) VALUES (?,?,?,?) ON CONFLICT(poll_id,user_id,option_idx) DO NOTHING').run(pollId, userId, idx, now()),
remove: (pollId, userId, idx) => db.prepare('DELETE FROM poll_votes WHERE poll_id=? AND user_id=? AND option_idx=?').run(pollId, userId, idx),
clearUser: (pollId, userId) => db.prepare('DELETE FROM poll_votes WHERE poll_id=? AND user_id=?').run(pollId, userId),
};
@@ -371,7 +372,7 @@ const pushSubs = {
const favorites = {
set: (userId, target, on) => on
? db.prepare('INSERT OR IGNORE INTO favorites (user_id,target,created_at) VALUES (?,?,?)').run(userId, target, now())
? db.prepare('INSERT INTO favorites (user_id,target,created_at) VALUES (?,?,?) ON CONFLICT(user_id,target) DO NOTHING').run(userId, target, now())
: db.prepare('DELETE FROM favorites WHERE user_id=? AND target=?').run(userId, target),
forUser: async (userId) => (await db.prepare('SELECT target FROM favorites WHERE user_id=?').all(userId)).map((r) => r.target),
};
+34 -26
View File
@@ -36,34 +36,42 @@ const server = http.createServer((req, res) => {
const wss = new WebSocketServer({ server, path: '/ws' });
wss.on('connection', onConnection);
server.listen(PORT, () => {
console.log(`HTTP on http://localhost:${PORT}`);
try { require('./media').backfill(); } catch (e) {} // streaming renditions for older video uploads
});
// Apply the DB schema BEFORE serving. For Postgres this creates the tables (async); for SQLite it's a
// no-op (the schema is already applied synchronously when db.js is required). Serving only starts once the
// store is ready, so the first request can never hit a missing table.
const db = require('./dbx');
// HTTPS — required so other devices can share their screen (browsers block
// screen capture on non-secure origins). Uses cert.pem/key.pem if present.
let httpsServer = null;
try {
const certPath = path.join(__dirname, 'cert.pem');
const keyPath = path.join(__dirname, 'key.pem');
if (fs.existsSync(certPath) && fs.existsSync(keyPath)) {
httpsServer = https.createServer(
{ cert: fs.readFileSync(certPath), key: fs.readFileSync(keyPath) },
(req, res) => server.emit('request', req, res)
);
const wssSecure = new WebSocketServer({ server: httpsServer, path: '/ws' });
wssSecure.on('connection', onConnection);
httpsServer.listen(HTTPS_PORT, () => {
console.log(`HTTPS on https://localhost:${HTTPS_PORT} (use this address from other devices)`);
console.log(` End user shares screen: https://<this-pc-ip>:${HTTPS_PORT}/share`);
console.log(` Technician connects: https://<this-pc-ip>:${HTTPS_PORT}/connect`);
});
} else {
console.log('(No cert.pem/key.pem found — HTTPS disabled. Other devices can view but not share their screen.)');
function startListening() {
server.listen(PORT, () => {
console.log(`HTTP on http://localhost:${PORT} (db=${db.name})`);
try { require('./media').backfill(); } catch (e) {} // streaming renditions for older video uploads
});
// HTTPS — required so other devices can share their screen (browsers block
// screen capture on non-secure origins). Uses cert.pem/key.pem if present.
try {
const certPath = path.join(__dirname, 'cert.pem');
const keyPath = path.join(__dirname, 'key.pem');
if (fs.existsSync(certPath) && fs.existsSync(keyPath)) {
const httpsServer = https.createServer(
{ cert: fs.readFileSync(certPath), key: fs.readFileSync(keyPath) },
(req, res) => server.emit('request', req, res)
);
const wssSecure = new WebSocketServer({ server: httpsServer, path: '/ws' });
wssSecure.on('connection', onConnection);
httpsServer.listen(HTTPS_PORT, () => {
console.log(`HTTPS on https://localhost:${HTTPS_PORT} (use this address from other devices)`);
console.log(` End user shares screen: https://<this-pc-ip>:${HTTPS_PORT}/share`);
console.log(` Technician connects: https://<this-pc-ip>:${HTTPS_PORT}/connect`);
});
} else {
console.log('(No cert.pem/key.pem found — HTTPS disabled. Other devices can view but not share their screen.)');
}
} catch (e) {
console.log('HTTPS failed to start:', e.message);
}
} catch (e) {
console.log('HTTPS failed to start:', e.message);
}
db.init().then(startListening).catch((e) => { console.error('DB init failed:', (e && e.message) || e); process.exit(1); });
module.exports = { server };