wip(db): async call-site conversion in progress (Phase 3) — DO NOT MERGE yet
On the db-migration branch only; master stays clean + deployable. Foundation (adapter, pg schema, smoke harness) is already on master and safe. Done: - repos.js fully async (Phase 2, validated: node --check clean, no missed transforms). - session.js currentUser/apiKeyFromReq async. - Mechanical `await` prefix applied across routes/static/calls/signaling/reminders/ webhooks/push. Remaining (does NOT compile yet — deterministic to finish): 1. Async cascade: helper fns that now contain `await` must be marked async and their callers awaited. node --check points to each (namesFor, authAttachmentRaw/ authAttachment in static, the WS handlers in calls/signaling, reminders/webhooks loops). 2. DTO builders are the real work: namesFor, avatarsFor, buildPollDTO, buildMsgDTO, recDTO all became async — every `.map(x => buildMsgDTO(...))` etc. must become `await Promise.all(arr.map(async x => ...))`. 3. Chained calls `R.x.y(...).map/.length/.includes` → `(await R.x.y(...)).method`. 4. Then: node --check all green → node test/db-smoke.js green → e2e → merge to master. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
+63
-61
@@ -5,28 +5,34 @@
|
||||
// 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; },
|
||||
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: (email) => !!db.prepare('SELECT 1 FROM users WHERE email=? COLLATE NOCASE').get(email),
|
||||
emailExists: async (email) => !!(await db.prepare('SELECT 1 FROM users WHERE email=? COLLATE NOCASE').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 +40,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 +63,43 @@ 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);
|
||||
await run('UPDATE OR IGNORE message_reactions SET user_id=? WHERE user_id=?', intoId, fromId);
|
||||
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('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('DELETE FROM poll_votes WHERE user_id=?', fromId);
|
||||
await run('UPDATE OR IGNORE favorites SET user_id=? WHERE user_id=?', intoId, fromId);
|
||||
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 OR IGNORE favorites SET target='dm:'||? WHERE target='dm:'||?", intoId, fromId);
|
||||
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);
|
||||
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);
|
||||
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 OR REPLACE INTO user_aliases (old_id,user_id,team_id,created_at) VALUES (?,?,?,?)', 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 +118,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;
|
||||
},
|
||||
@@ -239,18 +241,18 @@ const messages = {
|
||||
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),
|
||||
@@ -271,16 +273,16 @@ const conversations = {
|
||||
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),
|
||||
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 +290,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,7 +354,7 @@ 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),
|
||||
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()),
|
||||
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 +373,7 @@ 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('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 = {
|
||||
|
||||
Reference in New Issue
Block a user