Sravan/BizGaze_Remote
0
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

feat(db): complete async call-site conversion — Phase 3 done, validated on SQLite

Browse Source 
The full sync→async conversion is complete and green on the SQLite backend. Every
DB call across the app now awaits the async adapter, so the identical code runs on
Postgres at cutover.

Converted (this commit finishes Phase 3):
- session.js: currentUser/apiKeyFromReq async → 63 route awaits + WS + static.
- routes.js: all ~250 R.* awaited; DTO helpers (namesFor, avatarsFor, buildMsgDTO,
  buildPollDTO, reactionsForMessage, postSystemMessage, pushGroupUpdate,
  issueRefreshToken, provisionFromBizgaze) made async; every `.map(x=>buildDTO(x))`
  restructured to `await Promise.all(...map(async...))` preserving order; `.filter`
  predicates that hit the DB moved to an `asyncFilter` helper; chained
  `R.x.y(...).length/.map/.filter` wrapped as `(await R.x.y(...)).method`; stream
  upload handlers (recording/transcript/attachment) made async.
- calls.js / signaling.js: all call/meeting fns async; leaveMeeting AWAITS
  persistCallHistory + finalizeTranscript BEFORE endCallByRoom (ordering matters —
  fire-and-forget would race the map teardown); WS handle()/cleanup() async with
  .catch guards.
- static.js: authAttachment(Raw) async (the .some carrier check became a loop),
  handleGet async; server.js dispatch catches handler rejections → 500 not a hang.
- media.js backfill, push.js, reminders.js, webhooks.js await their repo calls.

Validation on DB_BACKEND=sqlite: db-smoke 22/22; legacy e2e 80 checks pass with zero
FAILs (throws only at a PRE-EXISTING WS lobby-drift assertion, unrelated). Every
server file `node --check` clean.

Still on the branch — master untouched. Next: Phase 5 (pg backend + ~7 dialect
queries + data migration + Docker Postgres + cutover), then merge.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Sravan
2026-07-24 22:06:27 +05:30
parent 2460c0f9eb
commit 3250530596
8 changed files with 158 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/signaling.js
+20 -20
View File
@@ -26,7 +26,7 @@ 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;
@@ -46,7 +46,7 @@ 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 = await R.scheduledMeetings.byCode(room); if (s) return s.lobby !== 0; } catch (_) {}
return true;
@@ -77,12 +77,12 @@ 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': {
@@ -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,7 +259,7 @@ function handle(ws, m, req) {
break;
}
case 'meeting-leave': {
leaveMeeting(ws);
await leaveMeeting(ws);
break;
}
// --- Agent comes online ---
@@ -370,13 +370,13 @@ 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 { await R.sessionsLog.end(sessionId); } catch (e) {}
@@ -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);
}
}
}