fix(db): portable thread queries for Postgres (conditional cursor + subquery alias)

The DM/group thread queries used `(? IS NULL OR created_at < ?)` — an all-NULL
param Postgres can't type ('could not determine data type of parameter') — and an
unaliased FROM-subquery (Postgres requires an alias). Both rewritten to add the
`created_at < ?` clause only when a cursor is given, and alias the subquery `t`.
Portable; sqlite db-smoke still 22/22.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

server/calls.js

@@ -11,14 +11,14 @@ const now = () => Date.now();
 const pairKey = (a, b) => [a, b].sort().join('|');
 
 // Resolve a room's meeting context (group / scheduled meeting / title) for labelling recordings.
-function meetingContext(room) {
+async function meetingContext(room) {
   const ctx = { groupId: null, meetingId: null, title: 'Meeting' };
   try {
-    const sched = R.scheduledMeetings.byCode(room);
+    const sched = await R.scheduledMeetings.byCode(room);
     if (sched) { ctx.meetingId = sched.id; ctx.groupId = sched.group_id || null; ctx.title = sched.title || 'Meeting'; }
   } catch (_) {}
   if (!ctx.groupId) { const gid = roomToGroupCall.get(room); if (gid) ctx.groupId = gid; }
-  if (ctx.groupId && ctx.title === 'Meeting') { try { const g = R.conversations.byId(ctx.groupId); if (g) ctx.title = g.name || 'Group'; } catch (_) {} }
+  if (ctx.groupId && ctx.title === 'Meeting') { try { const g = await R.conversations.byId(ctx.groupId); if (g) ctx.title = g.name || 'Group'; } catch (_) {} }
   if (!ctx.groupId && !ctx.meetingId && roomToDmCall.has(room)) ctx.title = 'Direct Call';
   return ctx;
 }
@@ -26,21 +26,21 @@ function meetingContext(room) {
 // Save the FULL shared conversation transcript as a PRIVATE copy for each subscriber. onlyUserId
 // finalizes just that subscriber (on their leave / opt-out); omit to flush all remaining (room end).
 // Must run BEFORE endCallByRoom (which clears the room→meeting maps meetingContext relies on).
-function finalizeTranscript(room, onlyUserId) {
+async function finalizeTranscript(room, onlyUserId) {
   const subs = transcriptSubs.get(room); if (!subs || !subs.size) { if (!onlyUserId) { transcriptBuffers.delete(room); transcriptSubs.delete(room); } return; }
   const buf = transcriptBuffers.get(room) || [];
   const ids = onlyUserId ? (subs.has(onlyUserId) ? [onlyUserId] : []) : [...subs];
   if (ids.length && buf.length) {
-    const ctx = meetingContext(room);
+    const ctx = await meetingContext(room);
     const lines = buf.map((s) => { const ts = new Date(s.t); const hh = String(ts.getHours()).padStart(2, '0'), mm = String(ts.getMinutes()).padStart(2, '0'); return '[' + hh + ':' + mm + '] ' + s.speaker + ': ' + s.text; });
     const body = ctx.title + ' — transcript\n' + new Date(buf[0].t).toLocaleString() + '\n\n' + lines.join('\n') + '\n';
     for (const uid of ids) {
-      let user = null; try { user = R.users.byId(uid); } catch (_) {}
+      let user = null; try { user = await R.users.byId(uid); } catch (_) {}
       if (!user) { subs.delete(uid); continue; }
       const id = A.id(); const file = 'm_' + id + '.txt';
       try { fs.writeFileSync(path.join(TRANS_DIR, file), body); } catch (e) { continue; }
       // groupId null → private to its creator (see canSeeRec / /mrec auth).
-      R.recordings.create({ id, teamId: user.team_id, room, groupId: null, meetingId: ctx.meetingId, title: ctx.title, kind: 'transcript', file, mime: 'text/plain', size: null, durationMs: null, createdBy: uid, createdByName: user.name || user.email });
+      await R.recordings.create({ id, teamId: user.team_id, room, groupId: null, meetingId: ctx.meetingId, title: ctx.title, kind: 'transcript', file, mime: 'text/plain', size: null, durationMs: null, createdBy: uid, createdByName: user.name || user.email });
       subs.delete(uid);
     }
   } else { ids.forEach((uid) => subs.delete(uid)); }
@@ -49,46 +49,46 @@ function finalizeTranscript(room, onlyUserId) {
 
 function fmtDur(ms) { const s = Math.max(0, Math.round(ms / 1000)); const m = Math.floor(s / 60); return m ? (m + 'm ' + (s % 60) + 's') : (s + 's'); }
 
-function broadcast(group, evt) { try { for (const mid of R.conversations.members(group)) CHAT.pushToUser(mid, evt); } catch (_) {} }
+async function broadcast(group, evt) { try { for (const mid of await R.conversations.members(group)) CHAT.pushToUser(mid, evt); } catch (_) {} }
 
 // Post a centered activity line into the group (system sender → no ping on clients).
-function postSystem(group, teamId, text) {
+async function postSystem(group, teamId, text) {
   const id = A.id();
-  R.messages.send({ id, teamId, senderId: '__system__', recipientId: '', body: text, conversationId: group });
-  const m = R.messages.byId(id);
+  await R.messages.send({ id, teamId, senderId: '__system__', recipientId: '', body: text, conversationId: group });
+  const m = await R.messages.byId(id);
   broadcast(group, { type: 'chat-message', message: { id: m.id, from: '__system__', conversation_id: group, body: m.body, created_at: m.created_at, system: true } });
 }
 
-function startGroupCall(group, teamId, user) {
+async function startGroupCall(group, teamId, user) {
   const existing = groupCalls.get(group);
   if (existing) return { room: existing.room, active: true, already: true };
   let room; do { room = A.numericCode(6); } while (meetingRooms.has(room));
   meetingRooms.set(room, new Map());
   const call = { room, startedAt: now(), startedBy: user.id, startedByName: user.name || user.email };
   // Log the call as a meeting so it appears under Past meetings (history) with the group name.
-  try { const hid = A.id(); R.scheduledMeetings.create({ id: hid, teamId, groupId: group, roomCode: room, title: 'Group call', description: null, scheduledAt: now(), createdBy: user.id }); call.historyId = hid; call.teamId = teamId; } catch (_) {}
+  try { const hid = A.id(); await R.scheduledMeetings.create({ id: hid, teamId, groupId: group, roomCode: room, title: 'Group call', description: null, scheduledAt: now(), createdBy: user.id }); call.historyId = hid; call.teamId = teamId; } catch (_) {}
   groupCalls.set(group, call); roomToGroupCall.set(room, group); roomHost.set(room, user.id); // creator = host
-  postSystem(group, teamId, '📞 ' + call.startedByName + ' started a group call');
-  let gName = 'Group'; try { const g = R.conversations.byId(group); if (g) gName = g.name || 'Group'; } catch (_) {}
+  postSystem(group, teamId, '📞 ' + call.startedByName + ' started a group call').catch(() => {});
+  let gName = 'Group'; try { const g = await R.conversations.byId(group); if (g) gName = g.name || 'Group'; } catch (_) {}
   broadcast(group, { type: 'group-call', group, active: true, room, by: user.id, startedByName: call.startedByName, groupName: gName });
   return { room, active: true };
 }
 
 // Called from signaling when a mesh room empties — ends the group call if this room was one.
-function endGroupCallByRoom(room) {
+async function endGroupCallByRoom(room) {
   const group = roomToGroupCall.get(room);
   if (!group) return;
   const call = groupCalls.get(group);
   roomToGroupCall.delete(room); groupCalls.delete(group); roomHost.delete(room);
   if (call) {
-    let teamId = call.teamId; try { const g = R.conversations.byId(group); if (g) { teamId = g.team_id; postSystem(group, g.team_id, '📞 Group call ended · ' + fmtDur(now() - call.startedAt)); } } catch (_) {}
-    if (call.historyId && teamId) { try { R.scheduledMeetings.end(call.historyId, teamId); } catch (_) {} } // mark the history row past
+    let teamId = call.teamId; try { const g = await R.conversations.byId(group); if (g) { teamId = g.team_id; postSystem(group, g.team_id, '📞 Group call ended · ' + fmtDur(now() - call.startedAt)).catch(() => {}); } } catch (_) {}
+    if (call.historyId && teamId) { try { await R.scheduledMeetings.end(call.historyId, teamId); } catch (_) {} } // mark the history row past
     broadcast(group, { type: 'group-call', group, active: false, room });
   }
 }
 
 // 1:1 (DM) call. Notifies both parties (state + a chat line) so the callee sees "Join".
-function startDmCall(me, otherId, teamId) {
+async function startDmCall(me, otherId, teamId) {
   const key = pairKey(me.id, otherId);
   const existing = dmCalls.get(key);
   if (existing) return { room: existing.room, active: true, already: true };
@@ -97,7 +97,7 @@ function startDmCall(me, otherId, teamId) {
   const byName = me.name || me.email;
   const call = { room, startedAt: now(), startedBy: me.id, startedByName: byName, users: [me.id, otherId], teamId, answered: false };
   // Log to history (both participants) so the call shows under Past meetings with its transcript.
-  try { const hid = A.id(); R.scheduledMeetings.create({ id: hid, teamId, groupId: null, roomCode: room, title: 'Direct Call', description: null, scheduledAt: now(), createdBy: me.id, participants: [me.id, otherId] }); call.historyId = hid; } catch (_) {}
+  try { const hid = A.id(); await R.scheduledMeetings.create({ id: hid, teamId, groupId: null, roomCode: room, title: 'Direct Call', description: null, scheduledAt: now(), createdBy: me.id, participants: [me.id, otherId] }); call.historyId = hid; } catch (_) {}
   dmCalls.set(key, call); roomToDmCall.set(room, key); roomHost.set(room, me.id); // caller = host
   // #9 (unanswered): if the callee never joins within the ring window, auto-end and mark it missed —
   // so the caller isn't stuck "ringing" forever.
@@ -109,8 +109,8 @@ function startDmCall(me, otherId, teamId) {
   }, 40000);
   // A viewer-relative activity line: the caller sees "You started a call", the callee sees the name.
   const mid = A.id();
-  R.messages.send({ id: mid, teamId, senderId: me.id, recipientId: otherId, body: '📞 Started a call', msgType: 'call-start' });
-  const m = R.messages.byId(mid); const dto = { id: m.id, from: me.id, to: otherId, conversation_id: null, body: m.body, created_at: m.created_at, system: true, evt: 'call-start', byName };
+  await R.messages.send({ id: mid, teamId, senderId: me.id, recipientId: otherId, body: '📞 Started a call', msgType: 'call-start' });
+  const m = await R.messages.byId(mid); const dto = { id: m.id, from: me.id, to: otherId, conversation_id: null, body: m.body, created_at: m.created_at, system: true, evt: 'call-start', byName };
   try { CHAT.pushToUser(otherId, { type: 'chat-message', message: dto }); } catch (_) {}
   try { CHAT.pushToUser(me.id, { type: 'chat-message', message: dto }); } catch (_) {}
   try { CHAT.pushToUser(otherId, { type: 'dm-call', active: true, room, with: me.id, by: me.id, byName }); } catch (_) {}
@@ -118,18 +118,18 @@ function startDmCall(me, otherId, teamId) {
   return { room, active: true };
 }
 
-function endDmCallByRoom(room, silent) {
+async function endDmCallByRoom(room, silent) {
   const key = roomToDmCall.get(room); if (!key) return;
   const call = dmCalls.get(key);
   roomToDmCall.delete(room); dmCalls.delete(key); roomHost.delete(room);
   if (!call) return;
   if (call.ringTimer) { try { clearTimeout(call.ringTimer); } catch (_) {} }
-  if (call.historyId && call.teamId) { try { R.scheduledMeetings.end(call.historyId, call.teamId); } catch (_) {} } // mark history past
+  if (call.historyId && call.teamId) { try { await R.scheduledMeetings.end(call.historyId, call.teamId); } catch (_) {} } // mark history past
   // Activity line: a duration ONLY if the call was answered; otherwise "Missed call" (#7/#9).
   if (!silent) try {
     const mid = A.id(); const body = call.answered ? ('📞 Call ended · ' + fmtDur(now() - (call.answeredAt || call.startedAt))) : '📞 Missed call';
-    R.messages.send({ id: mid, teamId: call.teamId, senderId: call.startedBy, recipientId: call.users.find((u) => u !== call.startedBy) || '', body, msgType: 'call-end' });
-    const m = R.messages.byId(mid); const dto = { id: m.id, from: call.startedBy, to: m.recipient_id, conversation_id: null, body, created_at: m.created_at, system: true, evt: 'call-end' };
+    await R.messages.send({ id: mid, teamId: call.teamId, senderId: call.startedBy, recipientId: call.users.find((u) => u !== call.startedBy) || '', body, msgType: 'call-end' });
+    const m = await R.messages.byId(mid); const dto = { id: m.id, from: call.startedBy, to: m.recipient_id, conversation_id: null, body, created_at: m.created_at, system: true, evt: 'call-end' };
     call.users.forEach((uid) => { try { CHAT.pushToUser(uid, { type: 'chat-message', message: dto }); } catch (_) {} });
   } catch (_) {}
   call.users.forEach((uid, i) => { try { CHAT.pushToUser(uid, { type: 'dm-call', active: false, with: call.users[1 - i], room }); } catch (_) {} });
@@ -143,10 +143,10 @@ function markDmAnswered(room, userId) {
   if (userId && userId !== call.startedBy && !call.answered) { call.answered = true; call.answeredAt = now(); if (call.ringTimer) { try { clearTimeout(call.ringTimer); } catch (_) {} call.ringTimer = null; } }
 }
 // Called from signaling when any mesh room empties.
-function endCallByRoom(room) { endGroupCallByRoom(room); endDmCallByRoom(room); }
+async function endCallByRoom(room) { await endGroupCallByRoom(room); await endDmCallByRoom(room); }
 
 // Callee declines a 1:1 call: post "Call declined" into the DM, drop the waiting caller, end it.
-function declineDmCall(room, byUser) {
+async function declineDmCall(room, byUser) {
   const key = roomToDmCall.get(room); if (!key) return { ok: false };
   const call = dmCalls.get(key); if (!call) return { ok: false };
   // #8: if this user has ALREADY accepted on another device (they're in the room), this is just the
@@ -156,8 +156,8 @@ function declineDmCall(room, byUser) {
   const callerId = call.users.find((id) => id !== byUser.id) || call.startedBy;
   try {
     const mid = A.id();
-    R.messages.send({ id: mid, teamId: byUser.team_id, senderId: byUser.id, recipientId: callerId, body: '📞 Call declined', msgType: 'call-end' });
-    const mm = R.messages.byId(mid); const dto = { id: mm.id, from: byUser.id, to: callerId, conversation_id: null, body: mm.body, created_at: mm.created_at, system: true, evt: 'call-end' };
+    await R.messages.send({ id: mid, teamId: byUser.team_id, senderId: byUser.id, recipientId: callerId, body: '📞 Call declined', msgType: 'call-end' });
+    const mm = await R.messages.byId(mid); const dto = { id: mm.id, from: byUser.id, to: callerId, conversation_id: null, body: mm.body, created_at: mm.created_at, system: true, evt: 'call-end' };
     CHAT.pushToUser(callerId, { type: 'chat-message', message: dto });
     CHAT.pushToUser(byUser.id, { type: 'chat-message', message: dto });
   } catch (_) {}

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); });

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 };

server/media.js

@@ -193,9 +193,9 @@ function dropDerived(id) {
 // persist on the data volume, so on a normal restart this finds nothing to do and costs one query.
 // Deliberately delayed and rate-limited by the same 2-at-a-time queue — boot must not stall on it.
 function backfill() {
-  setTimeout(() => {
+  setTimeout(async () => {
     let rows = [];
-    try { rows = require('./repos').attachments.allVideos(); } catch (e) { return; }
+    try { rows = await require('./repos').attachments.allVideos(); } catch (e) { return; }
     let queued = 0;
     for (const r of rows) {
       if (hasWebRendition(r.id)) continue;

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"
   },

server/push.js

@@ -114,21 +114,21 @@ async function sendToUser(userId, payload) {
   const data = JSON.stringify(payload || {});
   if (webReady) {
     let subs = [];
-    try { subs = R.pushSubs.byUser(userId); } catch (_) { subs = []; }
+    try { subs = await R.pushSubs.byUser(userId); } catch (_) { subs = []; }
     for (const s of subs) {
       try { await webpush.sendNotification({ endpoint: s.endpoint, keys: { p256dh: s.p256dh, auth: s.auth } }, data, { TTL: 600 }); }
-      catch (err) { const c = err && err.statusCode; if (c === 404 || c === 410) { try { R.pushSubs.removeByEndpoint(s.endpoint); } catch (_) {} } }
+      catch (err) { const c = err && err.statusCode; if (c === 404 || c === 410) { try { await R.pushSubs.removeByEndpoint(s.endpoint); } catch (_) {} } }
     }
   }
   if (nativeReady) {
     let toks = [];
-    try { toks = R.deviceTokens.byUser(userId); } catch (_) { toks = []; }
+    try { toks = await R.deviceTokens.byUser(userId); } catch (_) { toks = []; }
     for (const t of toks) {
       try {
         let r = null;
         if (t.platform === 'android' && fcmSA) r = await sendFcm(t.token, payload);
         else if (t.platform === 'ios' && apnsCfg) r = await sendApns(t.token, payload);
-        if (r && r.dead) { try { R.deviceTokens.removeByToken(t.token); } catch (_) {} }
+        if (r && r.dead) { try { await R.deviceTokens.removeByToken(t.token); } catch (_) {} }
       } catch (_) { /* best-effort */ }
     }
   }

server/reminders.js

@@ -3,18 +3,18 @@
 const R = require('./repos');
 const CHAT = require('./chat');
 
-function tick() {
+async function tick() {
   try {
     const now = Date.now();
-    const due = R.scheduledMeetings.dueForReminder(now, now + 10 * 60 * 1000); // starting within 10 min
+    const due = await R.scheduledMeetings.dueForReminder(now, now + 10 * 60 * 1000); // starting within 10 min
     for (const s of due) {
       const recipients = new Set([s.created_by]);
       let invited = []; try { invited = JSON.parse(s.participants || '[]'); } catch (_) {}
       invited.forEach((id) => recipients.add(id));
-      if (s.group_id) { try { R.conversations.members(s.group_id).forEach((m) => recipients.add(m)); } catch (_) {} }
+      if (s.group_id) { try { (await R.conversations.members(s.group_id)).forEach((m) => recipients.add(m)); } catch (_) {} }
       const evt = { type: 'meeting-reminder', meeting: { id: s.id, title: s.title, scheduledAt: s.scheduled_at, room: s.room_code } };
       recipients.forEach((uid) => { try { CHAT.pushToUser(uid, evt); } catch (_) {} });
-      R.scheduledMeetings.markReminded(s.id);
+      await R.scheduledMeetings.markReminded(s.id);
     }
   } catch (_) { /* never let the timer die */ }
 }

server/repos.js

@@ -5,28 +5,35 @@
 // TENANT ABSTRACTION: a "tenant" currently maps 1:1 to a team (column `team_id`).
 // Repo signatures take `tenantId` so that when the tenant is later elevated to a
 // first-class Organization (Phase 3), callers and the API/auth built on top stay unchanged.
-const db = require('./db');
+//
+// ASYNC: queries go through the db adapter (dbx.js), so every method returns a Promise — the same repo
+// code runs on synchronous SQLite and asynchronous Postgres. Callers MUST await. Methods that only
+// `return db.prepare(...).get/all/run(...)` already yield the adapter's Promise; methods that transform a
+// result (.map, !!, .c, or run multiple statements) are written async/await so they don't operate on a
+// bare Promise.
+const db = require('./dbx');
 const A = require('./auth');
 const now = () => Date.now();
 
 const teams = {
   first: () => db.prepare('SELECT * FROM teams LIMIT 1').get(),
   byId: (id) => db.prepare('SELECT * FROM teams WHERE id=?').get(id),
-  create: (name) => {
+  create: async (name) => {
     const id = A.id();
-    db.prepare('INSERT INTO teams (id,name,created_at) VALUES (?,?,?)').run(id, name, now());
+    await db.prepare('INSERT INTO teams (id,name,created_at) VALUES (?,?,?)').run(id, name, now());
     return db.prepare('SELECT * FROM teams WHERE id=?').get(id);
   },
 };
 
 const users = {
-  anyExists: () => !!db.prepare('SELECT 1 FROM users LIMIT 1').get(),
+  anyExists: async () => !!(await db.prepare('SELECT 1 FROM users LIMIT 1').get()),
   byId: (id) => db.prepare('SELECT * FROM users WHERE id=?').get(id),
   // 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: (id) => { if (!id) return id; const a = 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: (email) => !!db.prepare('SELECT 1 FROM users WHERE email=? COLLATE NOCASE').get(email),
+  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; },
+  // 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),
@@ -34,9 +41,9 @@ const users = {
     db.prepare('SELECT id,email,name,role,active,avatar_url,created_at FROM users WHERE team_id=?').all(tenantId),
   inTenant: (id, tenantId) =>
     db.prepare('SELECT * FROM users WHERE id=? AND team_id=?').get(id, tenantId),
-  create: ({ tenantId, email, hash, salt, role, name, mfaSecret }) => {
+  create: async ({ tenantId, email, hash, salt, role, name, mfaSecret }) => {
     const id = A.id();
-    db.prepare(`INSERT INTO users (id,team_id,email,pw_hash,pw_salt,role,name,mfa_secret,mfa_enabled,created_at)
+    await db.prepare(`INSERT INTO users (id,team_id,email,pw_hash,pw_salt,role,name,mfa_secret,mfa_enabled,created_at)
                 VALUES (?,?,?,?,?,?,?,?,0,?)`)
       .run(id, tenantId, email, hash, salt, role, name || null, mfaSecret, now());
     return id;
@@ -57,47 +64,46 @@ const users = {
   // BizGaze person (#2). Runs in a single transaction so a failure leaves the data untouched.
   // For composite-key tables we UPDATE OR IGNORE (move what won't collide) then DELETE the rest
   // (the survivor already has that membership/reaction/vote).
-  mergeInto: (fromId, intoId) => {
+  mergeInto: async (fromId, intoId) => {
     if (!fromId || !intoId || fromId === intoId) return;
-    const run = (sql, ...a) => db.prepare(sql).run(...a);
-    db.exec('BEGIN');
-    try {
-      run('UPDATE messages SET sender_id=? WHERE sender_id=?', intoId, fromId);
-      run('UPDATE messages SET recipient_id=? WHERE recipient_id=?', intoId, fromId);
-      run('UPDATE OR IGNORE message_reactions SET user_id=? WHERE user_id=?', intoId, fromId);
-      run('DELETE FROM message_reactions WHERE user_id=?', fromId);
-      run('UPDATE OR IGNORE conversation_members SET user_id=? WHERE user_id=?', intoId, fromId);
-      run('DELETE FROM conversation_members WHERE user_id=?', fromId);
-      run('UPDATE OR IGNORE poll_votes SET user_id=? WHERE user_id=?', intoId, fromId);
-      run('DELETE FROM poll_votes WHERE user_id=?', fromId);
-      run('UPDATE OR IGNORE favorites SET user_id=? WHERE user_id=?', intoId, fromId);
-      run('DELETE FROM favorites WHERE user_id=?', fromId);
+    // db.tx() gives one atomic unit on both engines (sqlite single-connection; pg one pooled client).
+    return db.tx(async (t) => {
+      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);
+      // 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 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 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 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.
-      run("UPDATE OR IGNORE favorites SET target='dm:'||? WHERE target='dm:'||?", intoId, fromId);
-      run("DELETE FROM favorites WHERE target='dm:'||?", fromId);
-      run('UPDATE conversations SET created_by=? WHERE created_by=?', intoId, fromId);
-      run('UPDATE polls SET created_by=? WHERE created_by=?', intoId, fromId);
-      run('UPDATE scheduled_meetings SET created_by=? WHERE created_by=?', intoId, fromId);
-      run('UPDATE recordings SET created_by=? WHERE created_by=?', intoId, fromId);
-      run('UPDATE attachments SET uploader_id=? WHERE uploader_id=?', intoId, fromId);
-      run('UPDATE push_subscriptions SET user_id=? WHERE user_id=?', intoId, fromId);
-      run('UPDATE OR IGNORE device_tokens SET user_id=? WHERE user_id=?', intoId, fromId);
-      run('DELETE FROM device_tokens WHERE user_id=?', fromId);
-      run('UPDATE app_installs SET user_id=? WHERE user_id=?', 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);
+      await run('UPDATE scheduled_meetings SET created_by=? WHERE created_by=?', intoId, fromId);
+      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);
+      // 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.
-      run('DELETE FROM sessions_auth WHERE user_id=?', fromId);
-      run('DELETE FROM refresh_tokens WHERE user_id=?', fromId);
+      await run('DELETE FROM sessions_auth WHERE user_id=?', fromId);
+      await run('DELETE FROM refresh_tokens WHERE user_id=?', fromId);
       // 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 = db.prepare('SELECT team_id FROM users WHERE id=?').get(intoId);
-      run('INSERT OR REPLACE INTO user_aliases (old_id,user_id,team_id,created_at) VALUES (?,?,?,?)', fromId, intoId, (_iu && _iu.team_id) || null, now());
-      run('UPDATE user_aliases SET user_id=? WHERE user_id=?', intoId, fromId); // re-chain earlier aliases to the new survivor
-      run('DELETE FROM users WHERE id=?', fromId);
-      db.exec('COMMIT');
-    } catch (e) {
-      db.exec('ROLLBACK');
-      throw e;
-    }
+      const _iu = await t.get('SELECT team_id FROM users WHERE id=?', intoId);
+      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);
+    });
   },
 };
 
@@ -116,9 +122,9 @@ const machines = {
   inTenant: (id, tenantId) => db.prepare('SELECT * FROM machines WHERE id=? AND team_id=?').get(id, tenantId),
   listByTenant: (tenantId) =>
     db.prepare('SELECT id,name,unattended,last_seen FROM machines WHERE team_id=?').all(tenantId),
-  create: ({ tenantId, name, enrollToken, unattended }) => {
+  create: async ({ tenantId, name, enrollToken, unattended }) => {
     const id = A.id();
-    db.prepare('INSERT INTO machines (id,team_id,name,enroll_token,unattended,created_at) VALUES (?,?,?,?,?,?)')
+    await db.prepare('INSERT INTO machines (id,team_id,name,enroll_token,unattended,created_at) VALUES (?,?,?,?,?,?)')
       .run(id, tenantId, name, enrollToken, unattended ? 1 : 0, now());
     return id;
   },
@@ -126,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),
 };
@@ -210,13 +213,17 @@ const messages = {
   // Full 1:1 (DM) thread between two users (both directions). Take the NEWEST `limit` messages
   // (inner DESC), then present them oldest-first. The old plain "ASC LIMIT" returned the OLDEST 300
   // and silently dropped everything newer once a thread passed 300 — so new messages "disappeared".
-  thread: (teamId, a, b, limit = 500, before = null) =>
-    db.prepare(`SELECT * FROM (
+  // The `before` cursor is added CONDITIONALLY (not as `? IS NULL OR …`): an all-NULL param has no type
+  // for Postgres to infer. The subquery also needs an alias (`t`) — Postgres requires it. Both portable.
+  thread: (teamId, a, b, limit = 500, before = null) => {
+    const cond = before != null ? ' AND created_at < ?' : '';
+    const args = before != null ? [teamId, a, b, b, a, before, limit] : [teamId, a, b, b, a, limit];
+    return db.prepare(`SELECT * FROM (
         SELECT * FROM messages WHERE team_id=? AND conversation_id IS NULL
-          AND ((sender_id=? AND recipient_id=?) OR (sender_id=? AND recipient_id=?))
-          AND (? IS NULL OR created_at < ?)
+          AND ((sender_id=? AND recipient_id=?) OR (sender_id=? AND recipient_id=?))${cond}
           ORDER BY created_at DESC LIMIT ?
-      ) ORDER BY created_at ASC`).all(teamId, a, b, b, a, before, before, limit),
+      ) t ORDER BY created_at ASC`).all(...args);
+  },
   // Full-thread search (ALL messages, not just the loaded window). `like` is the escaped LIKE pattern.
   searchThread: (teamId, a, b, like, limit = 300) =>
     db.prepare(`SELECT id, created_at, sender_id FROM messages WHERE team_id=? AND conversation_id IS NULL
@@ -230,27 +237,30 @@ const messages = {
     db.prepare('SELECT * FROM messages WHERE team_id=? AND conversation_id IS NULL AND (sender_id=? OR recipient_id=?) ORDER BY created_at DESC LIMIT ?')
       .all(teamId, userId, userId, limit),
   // Group conversation helpers.
-  threadByConversation: (conversationId, limit = 500, before = null) =>
-    db.prepare(`SELECT * FROM (
-        SELECT * FROM messages WHERE conversation_id=? AND (? IS NULL OR created_at < ?) ORDER BY created_at DESC LIMIT ?
-      ) ORDER BY created_at ASC`).all(conversationId, before, before, limit),
+  threadByConversation: (conversationId, limit = 500, before = null) => {
+    const cond = before != null ? ' AND created_at < ?' : '';
+    const args = before != null ? [conversationId, before, limit] : [conversationId, limit];
+    return db.prepare(`SELECT * FROM (
+        SELECT * FROM messages WHERE conversation_id=?${cond} ORDER BY created_at DESC LIMIT ?
+      ) t ORDER BY created_at ASC`).all(...args);
+  },
   searchConversation: (conversationId, like, limit = 300) =>
     db.prepare(`SELECT id, created_at, sender_id FROM messages WHERE conversation_id=? AND deleted=0
         AND body LIKE ? ESCAPE '\\' ORDER BY created_at ASC LIMIT ?`).all(conversationId, like, limit),
   lastInConversation: (conversationId) =>
     db.prepare('SELECT * FROM messages WHERE conversation_id=? ORDER BY created_at DESC LIMIT 1').get(conversationId),
-  unreadInConversation: (conversationId, userId, since) =>
-    db.prepare('SELECT COUNT(*) AS c FROM messages WHERE conversation_id=? AND sender_id<>? AND created_at>?').get(conversationId, userId, since).c,
+  unreadInConversation: async (conversationId, userId, since) =>
+    (await db.prepare('SELECT COUNT(*) AS c FROM messages WHERE conversation_id=? AND sender_id<>? AND created_at>?').get(conversationId, userId, since)).c,
 };
 
 const reactions = {
   // Toggle with ONE reaction per user per message: picking an emoji replaces any prior
   // reaction by that user; picking the same one again removes it. Returns true if added.
-  toggle: (messageId, userId, emoji) => {
-    const had = db.prepare('SELECT 1 FROM message_reactions WHERE message_id=? AND user_id=? AND emoji=?').get(messageId, userId, emoji);
-    db.prepare('DELETE FROM message_reactions WHERE message_id=? AND user_id=?').run(messageId, userId);
+  toggle: async (messageId, userId, emoji) => {
+    const had = await db.prepare('SELECT 1 FROM message_reactions WHERE message_id=? AND user_id=? AND emoji=?').get(messageId, userId, emoji);
+    await db.prepare('DELETE FROM message_reactions WHERE message_id=? AND user_id=?').run(messageId, userId);
     if (had) return false;
-    db.prepare('INSERT INTO message_reactions (message_id,user_id,emoji,created_at) VALUES (?,?,?,?)').run(messageId, userId, emoji, now());
+    await db.prepare('INSERT INTO message_reactions (message_id,user_id,emoji,created_at) VALUES (?,?,?,?)').run(messageId, userId, emoji, now());
     return true;
   },
   forMessage: (messageId) => db.prepare('SELECT user_id, emoji FROM message_reactions WHERE message_id=? ORDER BY created_at ASC').all(messageId),
@@ -270,17 +280,17 @@ 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),
-  members: (conversationId) => db.prepare('SELECT user_id FROM conversation_members WHERE conversation_id=? ORDER BY joined_at ASC').all(conversationId).map((r) => r.user_id),
-  isMember: (conversationId, userId) => !!db.prepare('SELECT 1 FROM conversation_members WHERE conversation_id=? AND user_id=?').get(conversationId, userId),
+    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) ----
-  isAdmin: (conversationId, userId) => !!db.prepare('SELECT 1 FROM conversation_members WHERE conversation_id=? AND user_id=? AND admin=1').get(conversationId, userId),
-  admins: (conversationId) => db.prepare('SELECT user_id FROM conversation_members WHERE conversation_id=? AND admin=1').all(conversationId).map((r) => r.user_id),
+  isAdmin: async (conversationId, userId) => !!(await db.prepare('SELECT 1 FROM conversation_members WHERE conversation_id=? AND user_id=? AND admin=1').get(conversationId, userId)),
+  admins: async (conversationId) => (await db.prepare('SELECT user_id FROM conversation_members WHERE conversation_id=? AND admin=1').all(conversationId)).map((r) => r.user_id),
   setMemberAdmin: (conversationId, userId, v) => db.prepare('UPDATE conversation_members SET admin=? WHERE conversation_id=? AND user_id=?').run(v ? 1 : 0, conversationId, userId),
-  oldestMember: (conversationId) => { const r = db.prepare('SELECT user_id FROM conversation_members WHERE conversation_id=? ORDER BY joined_at ASC LIMIT 1').get(conversationId); return r ? r.user_id : null; },
+  oldestMember: async (conversationId) => { const r = await db.prepare('SELECT user_id FROM conversation_members WHERE conversation_id=? ORDER BY joined_at ASC LIMIT 1').get(conversationId); return r ? r.user_id : null; },
   listForUser: (teamId, userId) =>
     db.prepare('SELECT c.* FROM conversations c JOIN conversation_members m ON m.conversation_id=c.id WHERE c.team_id=? AND m.user_id=?').all(teamId, userId),
-  lastReadAt: (conversationId, userId) => { const r = db.prepare('SELECT last_read_at FROM conversation_members WHERE conversation_id=? AND user_id=?').get(conversationId, userId); return r ? r.last_read_at : 0; },
+  lastReadAt: async (conversationId, userId) => { const r = await db.prepare('SELECT last_read_at FROM conversation_members WHERE conversation_id=? AND user_id=?').get(conversationId, userId); return r ? r.last_read_at : 0; },
   memberReads: (conversationId) => db.prepare('SELECT user_id, last_read_at FROM conversation_members WHERE conversation_id=?').all(conversationId),
   setAdminOnly: (id, v) => db.prepare('UPDATE conversations SET admin_only=? WHERE id=?').run(v ? 1 : 0, id),
   markRead: (conversationId, userId) => db.prepare('UPDATE conversation_members SET last_read_at=? WHERE conversation_id=? AND user_id=?').run(now(), conversationId, userId),
@@ -288,7 +298,7 @@ const conversations = {
   setAvatar: (id, attachmentId) => db.prepare('UPDATE conversations SET avatar_id=? WHERE id=?').run(attachmentId || null, id),
   byAvatar: (attachmentId) => db.prepare('SELECT * FROM conversations WHERE avatar_id=? LIMIT 1').get(attachmentId),
   removeMember: (conversationId, userId) => db.prepare('DELETE FROM conversation_members WHERE conversation_id=? AND user_id=?').run(conversationId, userId),
-  remove: (id) => { db.prepare('DELETE FROM conversation_members WHERE conversation_id=?').run(id); db.prepare('DELETE FROM conversations WHERE id=?').run(id); },
+  remove: async (id) => { await db.prepare('DELETE FROM conversation_members WHERE conversation_id=?').run(id); await db.prepare('DELETE FROM conversations WHERE id=?').run(id); },
 };
 
 const attachments = {
@@ -352,8 +362,8 @@ const polls = {
 
 const pollVotes = {
   forPoll: (pollId) => db.prepare('SELECT user_id, option_idx FROM poll_votes WHERE poll_id=?').all(pollId),
-  hasVoted: (pollId, userId, idx) => !!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()),
+  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 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),
 };
@@ -369,9 +379,9 @@ 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: (userId) => db.prepare('SELECT target FROM favorites WHERE user_id=?').all(userId).map((r) => r.target),
+  forUser: async (userId) => (await db.prepare('SELECT target FROM favorites WHERE user_id=?').all(userId)).map((r) => r.target),
 };
 
 const deviceTokens = {

server/server.js

@@ -23,43 +23,55 @@ const { onConnection } = require('./signaling');
 // ---------- HTTP request dispatch ----------
 const server = http.createServer((req, res) => {
   const key = `${req.method} ${req.url.split('?')[0]}`;
-  if (routes[key]) return routes[key](req, res);
-  if (req.method === 'GET') return handleGet(req, res); // downloads + static
-  json(res, 404, { error: 'not found' });
+  // Route/static handlers are async now (DB adapter). Catch any rejection so a handler error becomes a
+  // 500 instead of a hung request + unhandled promise rejection.
+  let p;
+  if (routes[key]) p = routes[key](req, res);
+  else if (req.method === 'GET') p = handleGet(req, res); // downloads + static
+  else return json(res, 404, { error: 'not found' });
+  if (p && typeof p.catch === 'function') p.catch((e) => { try { console.error('handler error', key, e && e.message); json(res, 500, { error: 'server error' }); } catch (_) {} });
 });
 
 // ---------- WebSocket signaling ----------
 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 };

server/session.js

@@ -25,27 +25,28 @@ function tokenFromReq(req) {
 }
 
 // Resolve the logged-in user from the request. Returns user row (with mfa state) or null.
-function currentUser(req, { requireMfa = true } = {}) {
+// Async: the repos it reads go through the DB adapter, so every caller must `await currentUser(...)`.
+async function currentUser(req, { requireMfa = true } = {}) {
   const tok = tokenFromReq(req);
   if (!tok) return null;
-  const s = R.authSessions.byToken(tok);
+  const s = await R.authSessions.byToken(tok);
   if (!s || s.expires_at < now()) return null;
   if (requireMfa && !s.mfa_passed) return null;
-  const u = R.users.byId(s.user_id);
+  const u = await R.users.byId(s.user_id);
   if (!u || u.active === 0) return null;
   return { ...u, _session: s };
 }
 
 // Resolve a third-party API key from `X-API-Key` or `Authorization: Bearer bzc_...`.
 // Returns { id, teamId, scopes:[], name } or null. Keys are prefixed `bzc_` and stored hashed.
-function apiKeyFromReq(req) {
+async function apiKeyFromReq(req) {
   let raw = req.headers && req.headers['x-api-key'];
   if (!raw) {
     const h = req.headers && (req.headers.authorization || req.headers.Authorization);
     if (h && /^Bearer\s+bzc_/i.test(h)) raw = h.replace(/^Bearer\s+/i, '').trim();
   }
   if (!raw || !/^bzc_/.test(raw)) return null;
-  const row = R.apiKeys.byHash(A.hashToken(raw));
+  const row = await R.apiKeys.byHash(A.hashToken(raw));
   if (!row || row.revoked) return null;
   return { id: row.id, teamId: row.team_id, scopes: String(row.scopes || '').split(',').map((s) => s.trim()).filter(Boolean), name: row.name };
 }

server/signaling.js

@@ -26,15 +26,15 @@ function noteRoomStat(room, ws, size) {
   if (ws._meetingUserId) st.uids.add(ws._meetingUserId);
 }
 // Called as a room is torn down. Scheduled meetings already have their own row, so they're skipped.
-function persistCallHistory(room) {
+async function persistCallHistory(room) {
   const st = roomStats.get(room);
   roomStats.delete(room);
   if (!st || !st.teamId || st.peak < 1) return;
   try {
-    if (R.scheduledMeetings.byCode(room)) return;      // the scheduled meeting row already represents this
+    if (await R.scheduledMeetings.byCode(room)) return;      // the scheduled meeting row already represents this
     const gid = roomToGroupCall.get(room) || null;
     const isDm = roomToDmCall.has(room);
-    R.callHistory.create({
+    await R.callHistory.create({
       id: A.id(), teamId: st.teamId, room, groupId: gid,
       kind: isDm ? 'dm' : (gid ? 'group' : 'adhoc'),
       title: isDm ? 'Direct call' : (gid ? null : 'Meeting'),
@@ -46,9 +46,9 @@ function persistCallHistory(room) {
 // A room requires host approval for GUESTS when explicitly set (ad-hoc) or by the scheduled meeting's
 // `lobby` flag. Default: require approval (the "people with the link join directly" concern) unless the
 // organizer chose "join directly". Logged-in tenant users are never held — only guests.
-function meetingRoomRequiresApproval(room) {
+async function meetingRoomRequiresApproval(room) {
   if (roomLobby.has(room)) return !!roomLobby.get(room);
-  try { const s = R.scheduledMeetings.byCode(room); if (s) return s.lobby !== 0; } catch (_) {}
+  try { const s = await R.scheduledMeetings.byCode(room); if (s) return s.lobby !== 0; } catch (_) {}
   return true;
 }
 // Actually add a newcomer to the room: tell them who's here, tell others they arrived, and (if they're
@@ -77,16 +77,16 @@ function onConnection(ws, req) {
   }, 25000);
   ws.on('message', (raw) => {
     let m; try { m = JSON.parse(raw); } catch { return; }
-    handle(ws, m, req);
+    handle(ws, m, req).catch(() => {}); // handle is async now; never let a rejection go unhandled
   });
-  ws.on('close', () => { clearInterval(hb); cleanup(ws); });
+  ws.on('close', () => { clearInterval(hb); cleanup(ws).catch(() => {}); });
 }
 
-function handle(ws, m, req) {
+async function handle(ws, m, req) {
   switch (m.type) {
     // --- Logged-in user registers this socket for live chat delivery ---
     case 'chat-hello': {
-      const u = currentUser(req); // identity from the cookie/Bearer on the WS upgrade
+      const u = await currentUser(req); // identity from the cookie/Bearer on the WS upgrade
       if (!u) return ws.send(JSON.stringify({ type: 'error', message: 'unauthorized' }));
       ws._chatUserId = u.id; ws._chatTeamId = u.team_id;
       CHAT.register(u.id, ws);
@@ -97,10 +97,10 @@ function handle(ws, m, req) {
     // Recipient's client acknowledges a DM was delivered → mark it + tell the sender.
     case 'chat-delivered': {
       if (!ws._chatUserId || !m.id) break;
-      const msg = R.messages.byId(m.id);
+      const msg = await R.messages.byId(m.id);
       if (!msg || msg.conversation_id || msg.team_id !== ws._chatTeamId) break; // DMs only
       if (msg.recipient_id !== ws._chatUserId) break; // only the recipient can ack
-      if (!msg.delivered_at) { R.messages.markDelivered(m.id); try { CHAT.pushToUser(msg.sender_id, { type: 'chat-delivered', id: m.id, with: msg.recipient_id }); } catch (_) {} }
+      if (!msg.delivered_at) { await R.messages.markDelivered(m.id); try { CHAT.pushToUser(msg.sender_id, { type: 'chat-delivered', id: m.id, with: msg.recipient_id }); } catch (_) {} }
       break;
     }
     // Live "is typing…" — ephemeral, never persisted. Relay to the DM peer, or fan out to
@@ -108,9 +108,9 @@ function handle(ws, m, req) {
     case 'chat-typing': {
       const uid = ws._chatUserId; if (!uid) break;
       const on = !!m.on;
-      let name = ''; try { const u = R.users.byId(uid); name = (u && u.name) || ''; } catch (_) {}
+      let name = ''; try { const u = await R.users.byId(uid); name = (u && u.name) || ''; } catch (_) {}
       if (m.group) {
-        let members; try { if (!R.conversations.isMember(m.group, uid)) break; members = R.conversations.members(m.group); } catch (_) { break; }
+        let members; try { if (!await R.conversations.isMember(m.group, uid)) break; members = await R.conversations.members(m.group); } catch (_) { break; }
         for (const mid of members) { if (mid !== uid) { try { CHAT.pushToUser(mid, { type: 'chat-typing', group: m.group, from: uid, name, on }); } catch (_) {} } }
       } else if (m.to) {
         try { CHAT.pushToUser(m.to, { type: 'chat-typing', from: uid, name, on }); } catch (_) {}
@@ -121,7 +121,7 @@ function handle(ws, m, req) {
     case 'meeting-create': {
       let code; do { code = A.numericCode(6); } while (meetingRooms.has(code));
       meetingRooms.set(code, new Map());
-      const cu = currentUser(req); if (cu) roomHost.set(code, cu.id); // ad-hoc meeting: creator = host
+      const cu = await currentUser(req); if (cu) roomHost.set(code, cu.id); // ad-hoc meeting: creator = host
       // Lobby preference for guests joining this ad-hoc room by link (default: require approval).
       roomLobby.set(code, m.lobby === false ? false : true);
       ws.send(JSON.stringify({ type: 'meeting-created', room: code }));
@@ -132,7 +132,7 @@ function handle(ws, m, req) {
       let peers = meetingRooms.get(room);
       // A scheduled meeting's room is created lazily on first join (its code lives in the DB).
       if (!peers) {
-        const sched = R.scheduledMeetings.byCode(room);
+        const sched = await R.scheduledMeetings.byCode(room);
         if (sched && !sched.ended_at) { peers = new Map(); meetingRooms.set(room, peers); }
       }
       if (!peers) return ws.send(JSON.stringify({ type: 'error', message: 'Meeting not found' }));
@@ -141,8 +141,8 @@ function handle(ws, m, req) {
       ws.kind = 'meeting'; ws._meetingRoom = room; ws._peerId = peerId; ws._peerName = name;
       // Host = the meeting's creator. roomHost is set on call/meeting creation; scheduled meetings fall back to created_by.
       let hostUserId = roomHost.get(room);
-      if (hostUserId === undefined) { try { const s = R.scheduledMeetings.byCode(room); if (s) { hostUserId = s.created_by; roomHost.set(room, hostUserId); } } catch (_) {} }
-      const ju = currentUser(req);
+      if (hostUserId === undefined) { try { const s = await R.scheduledMeetings.byCode(room); if (s) { hostUserId = s.created_by; roomHost.set(room, hostUserId); } } catch (_) {} }
+      const ju = await currentUser(req);
       // Identity used to map LiveKit media → this tile (peerIdForUid). Logged-in users use their user id
       // (their LiveKit token identity is the same). GUESTS have no session, so they pass a stable client
       // guest id here that ALSO becomes their LiveKit token identity — otherwise their media never maps
@@ -154,7 +154,7 @@ function handle(ws, m, req) {
       if (ju) ws._meetingTeamId = ju.team_id; // #7: which tenant owns this call log
       // LOBBY (#4): a GUEST (no session) waits for the host to admit them when the room requires approval.
       // Logged-in tenant members always join directly.
-      if (!ju && meetingRoomRequiresApproval(room)) {
+      if (!ju && await meetingRoomRequiresApproval(room)) {
         let pend = lobbyPending.get(room); if (!pend) { pend = new Map(); lobbyPending.set(room, pend); }
         pend.set(peerId, ws); ws._lobbyRoom = room;
         ws.send(JSON.stringify({ type: 'meeting-lobby-wait' }));
@@ -259,25 +259,25 @@ function handle(ws, m, req) {
       break;
     }
     case 'meeting-leave': {
-      leaveMeeting(ws);
+      await leaveMeeting(ws);
       break;
     }
     // --- Agent comes online ---
     case 'agent-hello': {
-      const machine = R.machines.byEnrollToken(m.enrollToken);
+      const machine = await R.machines.byEnrollToken(m.enrollToken);
       if (!machine) return ws.send(JSON.stringify({ type: 'error', message: 'invalid enroll token' }));
       ws.kind = 'agent'; ws.machineId = machine.id;
       onlineAgents.set(machine.id, { ws, machine });
-      R.machines.touch(machine.id);
+      await R.machines.touch(machine.id);
       ws.send(JSON.stringify({ type: 'agent-registered', machineId: machine.id, name: machine.name }));
       break;
     }
     // --- Technician requests control of a machine ---
     case 'viewer-connect': {
-      const u = currentUser(req); // cookie sent on WS upgrade
+      const u = await currentUser(req); // cookie sent on WS upgrade
       if (!u) return ws.send(JSON.stringify({ type: 'error', message: 'unauthorized' }));
       const agent = onlineAgents.get(m.machineId);
-      const machine = R.machines.inTenant(m.machineId, u.team_id);
+      const machine = await R.machines.inTenant(m.machineId, u.team_id);
       if (!machine) return ws.send(JSON.stringify({ type: 'error', message: 'no such machine' }));
       if (!agent) return ws.send(JSON.stringify({ type: 'error', message: 'machine offline' }));
       if (u.role === 'viewer' && false) {} // view-only still allowed to watch; control gated agent-side
@@ -301,7 +301,7 @@ function handle(ws, m, req) {
       if (m.granted) {
         audit({ team_id: sess.machine.team_id, user_id: sess.user.id, user_email: sess.user.email, machine_id: sess.machine.id, machine_name: sess.machine.name, action: 'consent_granted', detail: sess.ticket ? 'Ticket ' + sess.ticket : (sess.machine.id ? null : 'Direct session') });
         try {
-          R.sessionsLog.create({ id: m.sessionId, tenantId: sess.machine.team_id, agentEmail: sess.user.email, agentName: sess.agentName || sess.user.email, ticket: sess.ticket || null });
+          await R.sessionsLog.create({ id: m.sessionId, tenantId: sess.machine.team_id, agentEmail: sess.user.email, agentName: sess.agentName || sess.user.email, ticket: sess.ticket || null });
         } catch (e) { /* duplicate consent */ }
         try { W.emit('session.started', sess.machine.team_id, { sessionId: m.sessionId, agent_email: sess.user.email, agent_name: sess.agentName || sess.user.email, ticket: sess.ticket || null, started_at: Date.now() }); } catch (_) {}
         sess.viewerWs.send(JSON.stringify({ type: 'session-ready', sessionId: m.sessionId }));
@@ -325,7 +325,7 @@ function handle(ws, m, req) {
     }
     // --- Logged-in agent enters the code (+ ticket) to connect ---
     case 'code-connect': {
-      const agent = currentUser(req); // identity from the agent's authenticated session
+      const agent = await currentUser(req); // identity from the agent's authenticated session
       if (!agent) {
         return ws.send(JSON.stringify({ type: 'error', message: 'Please sign in as an agent first' }));
       }
@@ -370,18 +370,18 @@ function handle(ws, m, req) {
       break;
     }
     case 'end-session': {
-      endSession(ws.sessionId, m.reason || null);
+      await endSession(ws.sessionId, m.reason || null);
       break;
     }
   }
 }
 
-function endSession(sessionId, reason) {
+async function endSession(sessionId, reason) {
   const sess = liveSessions.get(sessionId);
   if (!sess) return;
-  try { R.sessionsLog.end(sessionId); } catch (e) {}
+  try { await R.sessionsLog.end(sessionId); } catch (e) {}
   try {
-    const row = R.sessionsLog.byId(sessionId);
+    const row = await R.sessionsLog.byId(sessionId);
     if (row) W.emit('session.ended', sess.machine.team_id, { sessionId: row.id, agent_email: row.agent_email,
       agent_name: row.agent_name, ticket: row.ticket, started_at: row.started_at, ended_at: row.ended_at,
       duration_ms: row.ended_at ? row.ended_at - row.started_at : null });
@@ -393,7 +393,7 @@ function endSession(sessionId, reason) {
   liveSessions.delete(sessionId);
 }
 
-function leaveMeeting(ws) {
+async function leaveMeeting(ws) {
   const room = ws._meetingRoom;
   if (!room) return;
   const peers = meetingRooms.get(room);
@@ -401,34 +401,34 @@ function leaveMeeting(ws) {
   const pid = ws._peerId;
   const leaverId = ws._meetingUserId;
   if (!peers) { if (leaverId) CHAT.broadcastPresence(leaverId); return; }
-  try { require('./calls').finalizeTranscript(room, ws._meetingUserId); } catch (_) {} // save THIS user's transcript
+  try { await require('./calls').finalizeTranscript(room, ws._meetingUserId); } catch (_) {} // save THIS user's transcript
   peers.delete(pid);
   // 1:1 call: when either party leaves, end it for everyone (a DM call has no "remaining" call).
   if (roomToDmCall.has(room)) {
     const others = [...peers.values()].map((p) => p.ws && p.ws._meetingUserId).filter(Boolean);
     for (const [, p] of peers) { if (p.ws.readyState === 1) { try { p.ws.send(JSON.stringify({ type: 'meeting-ended' })); } catch (_) {} p.ws._meetingRoom = null; } }
-    persistCallHistory(room); // #7: log the call BEFORE endCallByRoom clears roomToDmCall/roomToGroupCall
+    await persistCallHistory(room); // #7: log the call BEFORE endCallByRoom clears roomToDmCall/roomToGroupCall
     meetingRooms.delete(room);
-    try { require('./calls').finalizeTranscript(room); } catch (_) {} // any remaining buffers (safety)
+    try { await require('./calls').finalizeTranscript(room); } catch (_) {} // any remaining buffers (safety)
     roomHost.delete(room);
-    try { require('./calls').endCallByRoom(room); } catch (_) {}
+    try { await require('./calls').endCallByRoom(room); } catch (_) {}
     if (leaverId) CHAT.broadcastPresence(leaverId);
     others.forEach((uid) => CHAT.broadcastPresence(uid)); // both parties are now out of the call → live update
     return;
   }
   for (const [, p] of peers) { if (p.ws.readyState === 1) p.ws.send(JSON.stringify({ type: 'meeting-peer-left', peerId: pid })); }
   if (peers.size === 0) {
-    persistCallHistory(room); // #7: log the finished call (before the room maps are cleared)
+    await persistCallHistory(room); // #7: log the finished call (before the room maps are cleared)
     meetingRooms.delete(room);
     lobbyPending.delete(room); roomLobby.delete(room); // room gone → drop its lobby state
-    try { require('./calls').finalizeTranscript(room); } catch (_) {} // before endCallByRoom clears the maps
+    try { await require('./calls').finalizeTranscript(room); } catch (_) {} // before endCallByRoom clears the maps
     roomHost.delete(room);
-    try { require('./calls').endCallByRoom(room); } catch (_) {}
+    try { await require('./calls').endCallByRoom(room); } catch (_) {}
   }
   if (leaverId) CHAT.broadcastPresence(leaverId); // this user left the call → update contacts live
 }
 
-function cleanup(ws) {
+async function cleanup(ws) {
   const goneUserId = ws._chatUserId; // capture before unregister so we can announce the change
   // A guest waiting in the lobby dropped → remove their pending request and tell the host to clear it.
   if (ws._lobbyRoom) {
@@ -438,14 +438,14 @@ function cleanup(ws) {
     ws._lobbyRoom = null;
   }
   CHAT.unregister(ws);
-  leaveMeeting(ws);
+  await leaveMeeting(ws);
   if (goneUserId) CHAT.broadcastPresence(goneUserId); // now reflects offline / no-longer-in-call
 
   if (ws.kind === 'agent' && ws.machineId) onlineAgents.delete(ws.machineId);
   if (ws.kind === 'sharer' && ws.shareCode) pendingShares.delete(ws.shareCode);
   if (ws.sessionId) {
     for (const [sid, sess] of liveSessions) {
-      if (sess.agentWs === ws || sess.viewerWs === ws) endSession(sid);
+      if (sess.agentWs === ws || sess.viewerWs === ws) await endSession(sid);
     }
   }
 }

server/static.js

@@ -12,27 +12,32 @@ const MIME = { '.html': 'text/html', '.js': 'text/javascript', '.css': 'text/css
 
 // Authorize an attachment id: the uploader, a member of the group using it as an avatar, or a participant of
 // ANY message carrying it (the "any" covers forwarded attachments, which reuse the same id). Returns the row.
-function authAttachmentRaw(id, u) {
-  const a = R.attachments.byId(id);
+async function authAttachmentRaw(id, u) {
+  const a = await R.attachments.byId(id);
   if (!a || a.team_id !== u.team_id) return null;
-  const avatarGroup = R.conversations.byAvatar(id);
-  const carriers = R.messages.allByAttachment(id);
-  const ok = a.uploader_id === u.id
-    || (avatarGroup && R.conversations.isMember(avatarGroup.id, u.id))
-    || carriers.some((msg) => msg.conversation_id
-      ? R.conversations.isMember(msg.conversation_id, u.id)
-      : (msg.sender_id === u.id || msg.recipient_id === u.id));
+  const avatarGroup = await R.conversations.byAvatar(id);
+  const carriers = await R.messages.allByAttachment(id);
+  let ok = a.uploader_id === u.id || (avatarGroup && await R.conversations.isMember(avatarGroup.id, u.id));
+  if (!ok) {
+    // A .some() predicate can't await, so walk the carriers explicitly.
+    for (const msg of carriers) {
+      const carried = msg.conversation_id
+        ? await R.conversations.isMember(msg.conversation_id, u.id)
+        : (msg.sender_id === u.id || msg.recipient_id === u.id);
+      if (carried) { ok = true; break; }
+    }
+  }
   return ok ? a : null;
 }
 // Video playback fires MANY /files Range requests, and authAttachmentRaw scans messages by attachment_id
-// (un-indexed) each time — that per-chunk scan is what made playback stutter ("buffers and plays…"). Cache the
+// each time — that per-chunk scan is what made playback stutter ("buffers and plays…"). Cache the resolved
 // decision per user+attachment for 60s so range requests after the first are ~free. Bounded to keep memory flat.
 const _attAuth = new Map();
-function authAttachment(id, u) {
+async function authAttachment(id, u) {
   const key = u.id + ':' + id, now = Date.now();
   const hit = _attAuth.get(key);
   if (hit && hit.exp > now) return hit.a;
-  const a = authAttachmentRaw(id, u);
+  const a = await authAttachmentRaw(id, u);
   if (_attAuth.size > 4000) _attAuth.clear();
   _attAuth.set(key, { a, exp: now + 60000 });
   return a;
@@ -100,7 +105,7 @@ function serveStatic(req, res) {
 }
 
 // GET fallback: authenticated transcript/recording downloads, else static files.
-function handleGet(req, res) {
+async function handleGet(req, res) {
   const pathOnly = req.url.split('?')[0];
   // Stable "latest Windows installer" link (used by the site's Download button). Reads the
   // electron-updater manifest and redirects to the current versioned .exe.
@@ -141,11 +146,11 @@ function handleGet(req, res) {
     });
   }
   if (pathOnly.startsWith('/transcripts/')) {
-    const u = currentUser(req);
+    const u = await currentUser(req);
     if (!u) return json(res, 401, { error: 'unauthorized' });
     const name = path.basename(decodeURIComponent(pathOnly));
     const sid = name.replace(/\.txt$/i, '');
-    const row = R.sessionsLog.byIdInTenant(sid, u.team_id);
+    const row = await R.sessionsLog.byIdInTenant(sid, u.team_id);
     if (!row || !row.transcript) return json(res, 404, { error: 'not found' });
     const fp = path.join(TRANS_DIR, row.transcript);
     if (!fp.startsWith(TRANS_DIR)) return json(res, 403, { error: 'forbidden' });
@@ -158,11 +163,11 @@ function handleGet(req, res) {
     });
   }
   if (pathOnly.startsWith('/recordings/')) {
-    const u = currentUser(req);
+    const u = await currentUser(req);
     if (!u) return json(res, 401, { error: 'unauthorized' });
     const name = path.basename(decodeURIComponent(pathOnly));
     const sid = name.replace(/\.(webm|mp4)$/i, '');
-    const row = R.sessionsLog.byIdInTenant(sid, u.team_id);
+    const row = await R.sessionsLog.byIdInTenant(sid, u.team_id);
     if (!row || !row.recording) return json(res, 404, { error: 'not found' });
     const fp = path.join(REC_DIR, row.recording);
     if (!fp.startsWith(REC_DIR)) return json(res, 403, { error: 'forbidden' });
@@ -179,16 +184,16 @@ function handleGet(req, res) {
   // Meeting recordings & transcripts (/mrec/<id>). Visible to the creator, group members, or those
   // who can see the scheduled meeting it belongs to.
   if (pathOnly.startsWith('/mrec/')) {
-    const u = currentUser(req);
+    const u = await currentUser(req);
     if (!u) return json(res, 401, { error: 'unauthorized' });
     const id = path.basename(decodeURIComponent(pathOnly));
-    const r = R.recordings.byId(id);
+    const r = await R.recordings.byId(id);
     if (!r || r.team_id !== u.team_id || !r.file) return json(res, 404, { error: 'not found' });
     let allowed = r.created_by === u.id;
     if (r.kind === 'transcript') allowed = r.created_by === u.id; // transcripts are private to their owner
     else {
-      if (!allowed && r.group_id) allowed = R.conversations.isMember(r.group_id, u.id);
-      if (!allowed && r.meeting_id) { const s = R.scheduledMeetings.byId(r.meeting_id); if (s) allowed = s.created_by === u.id || (s.participants && s.participants.includes('"' + u.id + '"')); }
+      if (!allowed && r.group_id) allowed = await R.conversations.isMember(r.group_id, u.id);
+      if (!allowed && r.meeting_id) { const s = await R.scheduledMeetings.byId(r.meeting_id); if (s) allowed = s.created_by === u.id || (s.participants && s.participants.includes('"' + u.id + '"')); }
     }
     if (!allowed) return json(res, 403, { error: 'forbidden' });
     const isVideo = r.kind === 'video';
@@ -208,10 +213,10 @@ function handleGet(req, res) {
   // Video POSTER thumbnail — first frame extracted with ffmpeg, cached next to the file. Cosmetic: if ffmpeg
   // is missing or fails we 404 and the <video> just falls back to its own (black) poster.
   if (pathOnly.startsWith('/thumbs/')) {
-    const u = currentUser(req);
+    const u = await currentUser(req);
     if (!u) return json(res, 401, { error: 'unauthorized' });
     const id = path.basename(decodeURIComponent(pathOnly));
-    const a = authAttachment(id, u);
+    const a = await authAttachment(id, u);
     if (!a) return json(res, 404, { error: 'not found' });
     if (!/^video\//.test(a.mime || '')) return json(res, 404, { error: 'not a video' });
     const src = path.join(UPLOADS_DIR, id);
@@ -236,10 +241,10 @@ function handleGet(req, res) {
   // back to the original bytes while that is still transcoding, so a video is never unplayable. The
   // download button keeps pointing at /files, which always serves the untouched original.
   if (pathOnly.startsWith('/stream/')) {
-    const u = currentUser(req);
+    const u = await currentUser(req);
     if (!u) return json(res, 401, { error: 'unauthorized' });
     const id = path.basename(decodeURIComponent(pathOnly));
-    const a = authAttachment(id, u);
+    const a = await authAttachment(id, u);
     if (!a) return json(res, 404, { error: 'not found' });
     media.ensureWebRendition(id, a.mime);        // idempotent — also backfills pre-existing uploads
     const ready = media.hasWebRendition(id);
@@ -253,10 +258,10 @@ function handleGet(req, res) {
     });
   }
   if (pathOnly.startsWith('/files/')) {
-    const u = currentUser(req);
+    const u = await currentUser(req);
     if (!u) return json(res, 401, { error: 'unauthorized' });
     const id = path.basename(decodeURIComponent(pathOnly));
-    const a = authAttachment(id, u);
+    const a = await authAttachment(id, u);
     if (!a) return json(res, 404, { error: 'not found' });
     const fp = path.join(UPLOADS_DIR, id);
     if (!fp.startsWith(UPLOADS_DIR)) return json(res, 403, { error: 'forbidden' });

server/webhooks.js

@@ -34,14 +34,14 @@ function deliver(url, secret, body, onDone) {
   go();
 }
 
-function emit(event, tenantId, payload) {
+async function emit(event, tenantId, payload) {
   const body = JSON.stringify({ event, ...payload });
   // Per-tenant subscriptions
   try {
-    for (const h of R.webhooks.activeForTenant(tenantId)) {
+    for (const h of await R.webhooks.activeForTenant(tenantId)) {
       const subs = String(h.events || '').split(',').map((s) => s.trim());
       if (subs.includes('*') || subs.includes(event)) {
-        deliver(h.url, h.secret, body, (r) => { try { R.webhooks.setStatus(h.id, r.ok ? 1 : 0, r.err || ('HTTP ' + r.status)); } catch (_) {} });
+        deliver(h.url, h.secret, body, async (r) => { try { await R.webhooks.setStatus(h.id, r.ok ? 1 : 0, r.err || ('HTTP ' + r.status)); } catch (_) {} });
       }
     }
   } catch (_) {}