From e67a783bdc2c50f022420b3a29341c273ba09cb2 Mon Sep 17 00:00:00 2001 From: sravan Date: Fri, 24 Jul 2026 22:28:04 +0530 Subject: [PATCH] feat(db): Postgres backend + dialect-portable queries + data migration (Phase 5) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - db/pg.js: the pg backend (prepare/exec/tx/init) — ?→$N translation, BIGINT parsed as Number (matches sqlite; else expires_at --- server/db/migrate-sqlite-to-pg.js | 56 +++++++++++++++++++++++++++++ server/db/pg.js | 59 ++++++++++++++++++++++++++++++ server/package.json | 1 + server/repos.js | 39 ++++++++++---------- server/server.js | 60 +++++++++++++++++-------------- 5 files changed, 170 insertions(+), 45 deletions(-) create mode 100644 server/db/migrate-sqlite-to-pg.js create mode 100644 server/db/pg.js diff --git a/server/db/migrate-sqlite-to-pg.js b/server/db/migrate-sqlite-to-pg.js new file mode 100644 index 0000000..2a3bcbe --- /dev/null +++ b/server/db/migrate-sqlite-to-pg.js @@ -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); }); diff --git a/server/db/pg.js b/server/db/pg.js new file mode 100644 index 0000000..a8a799b --- /dev/null +++ b/server/db/pg.js @@ -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 }; diff --git a/server/package.json b/server/package.json index 67a0b7e..5436820 100644 --- a/server/package.json +++ b/server/package.json @@ -10,6 +10,7 @@ "node": ">=22.5.0" }, "dependencies": { + "pg": "^8.13.1", "web-push": "^3.6.7", "ws": "^8.18.0" }, diff --git a/server/repos.js b/server/repos.js index aae4d29..baf2195 100644 --- a/server/repos.js +++ b/server/repos.js @@ -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), }; diff --git a/server/server.js b/server/server.js index aa2e3c6..187aa67 100644 --- a/server/server.js +++ b/server/server.js @@ -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://:${HTTPS_PORT}/share`); - console.log(` Technician connects: https://:${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://:${HTTPS_PORT}/share`); + console.log(` Technician connects: https://:${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 };