BizGaze_Remote/server/repos.js

// Data-access layer (Phase 1).
// All SQL lives here, never in route/signaling handlers. This decouples the rest of
// the app from SQLite so the store can later move to Postgres without touching callers.
//
// 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.
//
// 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: async (name) => {
    const id = A.id();
    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: 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: 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),
  listByTenant: (tenantId) =>
    // last_seen + status MUST be selected: the contacts/conversations DTOs read x.last_seen / x.status. Omitting
    // them made every list payload carry lastSeen:null (and status:'active'), so a fresh load showed a bare
    // "Offline" with no time — last-seen only appeared via live presence events (which read the full row).
    db.prepare('SELECT id,email,name,role,active,avatar_url,created_at,last_seen,status FROM users WHERE team_id=?').all(tenantId),
  inTenant: (id, tenantId) =>
    db.prepare('SELECT * FROM users WHERE id=? AND team_id=?').get(id, tenantId),
  create: async ({ tenantId, email, hash, salt, role, name, mfaSecret }) => {
    const id = A.id();
    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;
  },
  enableMfa: (id) => db.prepare('UPDATE users SET mfa_enabled=1 WHERE id=?').run(id),
  setName: (id, name) => db.prepare('UPDATE users SET name=? WHERE id=?').run(name, id),
  setRole: (id, role) => db.prepare('UPDATE users SET role=? WHERE id=?').run(role, id),
  setPassword: (id, hash, salt) => db.prepare('UPDATE users SET pw_hash=?, pw_salt=? WHERE id=?').run(hash, salt, id),
  setActive: (id, active) => db.prepare('UPDATE users SET active=? WHERE id=?').run(active ? 1 : 0, id),
  setAvatar: (id, url) => db.prepare('UPDATE users SET avatar_url=? WHERE id=?').run(url || null, id),
  setStatus: (id, status) => db.prepare('UPDATE users SET status=? WHERE id=?').run(status, id),
  // #2 "last seen": stamped when a user connects and when their last socket drops.
  touchSeen: (id) => db.prepare('UPDATE users SET last_seen=? WHERE id=?').run(now(), id),
  remove: (id) => db.prepare('DELETE FROM users WHERE id=?').run(id),
  // Fold one account (fromId) into another (intoId): reassign everything the merged-away user
  // owns/authored to the survivor, then delete the empty row. Used when a person turns out to
  // have two Biz Connect accounts — one per login identifier (email vs mobile) — for the same
  // 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: async (fromId, intoId) => {
    if (!fromId || !intoId || fromId === intoId) return;
    // 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.
      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.
      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 = 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);
    });
  },
};

const authSessions = {
  byToken: (token) => db.prepare('SELECT * FROM sessions_auth WHERE token=?').get(token),
  create: ({ token, userId, mfaPassed, ttl }) =>
    db.prepare('INSERT INTO sessions_auth (token,user_id,mfa_passed,created_at,expires_at) VALUES (?,?,?,?,?)')
      .run(token, userId, mfaPassed ? 1 : 0, now(), now() + ttl),
  markMfaPassed: (token) => db.prepare('UPDATE sessions_auth SET mfa_passed=1 WHERE token=?').run(token),
  touch: (token, ttl) => db.prepare('UPDATE sessions_auth SET expires_at=? WHERE token=?').run(now() + ttl, token), // slide the expiry forward on activity
  deleteByToken: (token) => db.prepare('DELETE FROM sessions_auth WHERE token=?').run(token),
  deleteByUser: (userId) => db.prepare('DELETE FROM sessions_auth WHERE user_id=?').run(userId),
};

const machines = {
  byEnrollToken: (t) => db.prepare('SELECT * FROM machines WHERE enroll_token=?').get(t),
  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: async ({ tenantId, name, enrollToken, unattended }) => {
    const id = A.id();
    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;
  },
  touch: (id) => db.prepare('UPDATE machines SET last_seen=? WHERE id=?').run(now(), id),
};

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 (?,?,?,?,?,?,?,?)`)
      .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),
};

const sessionsLog = {
  byId: (id) => db.prepare('SELECT * FROM sessions_log WHERE id=?').get(id),
  byIdInTenant: (id, tenantId) => db.prepare('SELECT * FROM sessions_log WHERE id=? AND team_id=?').get(id, tenantId),
  create: ({ id, tenantId, agentEmail, agentName, ticket }) =>
    db.prepare('INSERT INTO sessions_log (id,team_id,agent_email,agent_name,ticket,started_at) VALUES (?,?,?,?,?,?)')
      .run(id, tenantId, agentEmail, agentName, ticket || null, now()),
  end: (id) => db.prepare('UPDATE sessions_log SET ended_at=? WHERE id=? AND ended_at IS NULL').run(now(), id),
  setRecording: (id, fname) => db.prepare('UPDATE sessions_log SET recording=? WHERE id=?').run(fname, id),
  setTranscript: (id, fname) => db.prepare('UPDATE sessions_log SET transcript=? WHERE id=?').run(fname, id),
  // Role-scoping is the caller's job: pass agentEmail to restrict to one agent (non-admins).
  report: ({ tenantId, agentEmail, from, to }) => {
    let sql = 'SELECT * FROM sessions_log WHERE team_id=?';
    const args = [tenantId];
    if (agentEmail) { sql += ' AND agent_email=?'; args.push(agentEmail); }
    if (from) { sql += ' AND started_at>=?'; args.push(from); }
    if (to) { sql += ' AND started_at<=?'; args.push(to); }
    sql += ' ORDER BY started_at DESC LIMIT 500';
    return db.prepare(sql).all(...args);
  },
};

const refreshTokens = {
  create: ({ userId, tokenHash, ttl }) =>
    db.prepare('INSERT INTO refresh_tokens (token_hash,user_id,created_at,expires_at,revoked) VALUES (?,?,?,?,0)')
      .run(tokenHash, userId, now(), now() + ttl),
  byHash: (h) => db.prepare('SELECT * FROM refresh_tokens WHERE token_hash=?').get(h),
  revoke: (h) => db.prepare('UPDATE refresh_tokens SET revoked=1 WHERE token_hash=?').run(h),
  revokeByUser: (userId) => db.prepare('UPDATE refresh_tokens SET revoked=1 WHERE user_id=?').run(userId),
};

const apiKeys = {
  create: ({ id, tenantId, name, keyHash, scopes, createdBy }) =>
    db.prepare('INSERT INTO api_keys (id,team_id,name,key_hash,scopes,created_by,created_at,revoked) VALUES (?,?,?,?,?,?,?,0)')
      .run(id, tenantId, name || null, keyHash, scopes || '', createdBy || null, now()),
  byHash: (h) => db.prepare('SELECT * FROM api_keys WHERE key_hash=?').get(h),
  listByTenant: (tenantId) =>
    db.prepare('SELECT id,name,scopes,created_by,created_at,last_used_at,revoked FROM api_keys WHERE team_id=? ORDER BY created_at DESC').all(tenantId),
  revoke: (id, tenantId) => db.prepare('UPDATE api_keys SET revoked=1 WHERE id=? AND team_id=?').run(id, tenantId),
  touch: (id) => db.prepare('UPDATE api_keys SET last_used_at=? WHERE id=?').run(now(), id),
};

const webhooks = {
  create: ({ id, tenantId, url, secret, events, createdBy }) =>
    db.prepare('INSERT INTO webhooks (id,team_id,url,secret,events,active,created_by,created_at) VALUES (?,?,?,?,?,1,?,?)')
      .run(id, tenantId, url, secret, events || '', createdBy || null, now()),
  activeForTenant: (tenantId) => db.prepare('SELECT * FROM webhooks WHERE team_id=? AND active=1').all(tenantId),
  listByTenant: (tenantId) =>
    db.prepare('SELECT id,url,events,active,created_by,created_at,last_status,last_error,last_at FROM webhooks WHERE team_id=? ORDER BY created_at DESC').all(tenantId),
  remove: (id, tenantId) => db.prepare('DELETE FROM webhooks WHERE id=? AND team_id=?').run(id, tenantId),
  setStatus: (id, status, err) => db.prepare('UPDATE webhooks SET last_status=?, last_error=?, last_at=? WHERE id=?').run(status, err || null, now(), id),
};

const messages = {
  send: ({ id, teamId, senderId, recipientId, body, replyTo, attachmentId, conversationId, mentions, msgType, fwdFrom }) =>
    db.prepare('INSERT INTO messages (id,team_id,sender_id,recipient_id,body,created_at,reply_to,attachment_id,conversation_id,mentions,msg_type,fwd_from) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)')
      .run(id, teamId, senderId, recipientId || '', body, now(), replyTo || null, attachmentId || null, conversationId || null, (mentions && mentions.length) ? JSON.stringify(mentions) : null, msgType || null, fwdFrom || null),
  byId: (id) => db.prepare('SELECT * FROM messages WHERE id=?').get(id),
  byAttachment: (attachmentId) => db.prepare('SELECT * FROM messages WHERE attachment_id=? LIMIT 1').get(attachmentId),
  allByAttachment: (attachmentId) => db.prepare('SELECT sender_id, recipient_id, conversation_id FROM messages WHERE attachment_id=? AND deleted=0').all(attachmentId),
  setPoll: (messageId, pollId) => db.prepare('UPDATE messages SET poll_id=? WHERE id=?').run(pollId, messageId),
  markDelivered: (id) => db.prepare('UPDATE messages SET delivered_at=? WHERE id=? AND delivered_at IS NULL').run(now(), id),
  editBody: (id, body) => db.prepare('UPDATE messages SET body=?, edited_at=? WHERE id=?').run(body, now(), id),
  // Delete-for-everyone: clear the content but keep the row (renders as a placeholder).
  markDeleted: (id) => db.prepare("UPDATE messages SET deleted=1, body='', attachment_id=NULL, poll_id=NULL WHERE id=?").run(id),
  // #18 Delete-for-me: hide a message from ONE user's view (the row + everyone else are untouched).
  hideForUser: (messageId, userId) => db.prepare('INSERT INTO message_hidden (message_id,user_id,hidden_at) VALUES (?,?,?) ON CONFLICT(message_id,user_id) DO NOTHING').run(messageId, userId, now()),
  hiddenForUser: async (userId) => (await db.prepare('SELECT message_id FROM message_hidden WHERE user_id=?').all(userId)).map((r) => r.message_id),
  // Last message in a group that THIS user hasn't hidden (so a "delete for me" on the last message
  // rolls the sidebar preview back to the previous one, instead of showing what they just removed).
  lastInConversationForUser: (conversationId, userId) => db.prepare('SELECT * FROM messages WHERE conversation_id=? AND id NOT IN (SELECT message_id FROM message_hidden WHERE user_id=?) ORDER BY created_at DESC LIMIT 1').get(conversationId, userId),
  // #13 Pin a message: set/clear pinned_at + pinned_by.
  setPinned: (id, pinnedAt, pinnedBy) => db.prepare('UPDATE messages SET pinned_at=?, pinned_by=? WHERE id=?').run(pinnedAt, pinnedBy, id),
  // Pinned messages in a group / DM (newest pin first, deleted excluded).
  pinnedInConversation: (conversationId) => db.prepare('SELECT * FROM messages WHERE conversation_id=? AND pinned_at IS NOT NULL AND deleted=0 ORDER BY pinned_at DESC').all(conversationId),
  pinnedInDm: (teamId, a, b) => db.prepare('SELECT * FROM messages WHERE team_id=? AND conversation_id IS NULL AND ((sender_id=? AND recipient_id=?) OR (sender_id=? AND recipient_id=?)) AND pinned_at IS NOT NULL AND deleted=0 ORDER BY pinned_at DESC').all(teamId, a, b, b, a),
  // Shared media/files in a conversation (group) or DM — newest first.
  attachmentsForConversation: (teamId, conversationId) => db.prepare(`SELECT a.id, a.name, a.mime, a.size, m.created_at FROM messages m JOIN attachments a ON a.id=m.attachment_id WHERE m.team_id=? AND m.conversation_id=? AND m.deleted=0 ORDER BY m.created_at DESC`).all(teamId, conversationId),
  attachmentsForDm: (teamId, a, b) => db.prepare(`SELECT at.id, at.name, at.mime, at.size, m.created_at FROM messages m JOIN attachments at ON at.id=m.attachment_id WHERE m.team_id=? AND m.conversation_id IS NULL AND ((m.sender_id=? AND m.recipient_id=?) OR (m.sender_id=? AND m.recipient_id=?)) AND m.deleted=0 ORDER BY m.created_at DESC`).all(teamId, a, b, b, a),
  linksForConversation: (teamId, conversationId) => db.prepare("SELECT body, created_at FROM messages WHERE team_id=? AND conversation_id=? AND deleted=0 AND body LIKE '%http%' ORDER BY created_at DESC").all(teamId, conversationId),
  linksForDm: (teamId, a, b) => db.prepare("SELECT body, created_at FROM messages WHERE team_id=? AND conversation_id IS NULL AND ((sender_id=? AND recipient_id=?) OR (sender_id=? AND recipient_id=?)) AND deleted=0 AND body LIKE '%http%' ORDER BY created_at DESC").all(teamId, a, b, b, a),
  // 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".
  // 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.
  // `a` is the VIEWER (u.id). Exclude the viewer's "deleted for me" (message_hidden) rows in SQL — not in
  // JS afterwards — so the LIMIT counts only VISIBLE messages. Filtering after the LIMIT returned < PAGE rows
  // whenever a recent message had been hidden, and the client read that as "no older history" and stopped
  // paginating (a chat with a deleted recent message wouldn't scroll back).
  thread: (teamId, a, b, limit = 500, before = null) => {
    const cond = before != null ? ' AND created_at < ?' : '';
    const args = before != null ? [teamId, a, b, b, a, a, before, limit] : [teamId, a, b, b, a, 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 id NOT IN (SELECT message_id FROM message_hidden WHERE user_id=?)${cond}
          ORDER BY created_at DESC 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
        AND ((sender_id=? AND recipient_id=?) OR (sender_id=? AND recipient_id=?))
        AND deleted=0 AND body LIKE ? ESCAPE '\\' ORDER BY created_at ASC LIMIT ?`).all(teamId, a, b, b, a, like, limit),
  markRead: (teamId, recipientId, senderId) =>
    db.prepare('UPDATE messages SET read_at=? WHERE team_id=? AND conversation_id IS NULL AND recipient_id=? AND sender_id=? AND read_at IS NULL')
      .run(now(), teamId, recipientId, senderId),
  // Recent DM messages involving a user (newest first) — reduced into per-contact conversations.
  recentFor: (teamId, userId, limit = 1000) =>
    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, userId, limit = 500, before = null) => {
    const cond = before != null ? ' AND created_at < ?' : '';
    const args = before != null ? [conversationId, userId, before, limit] : [conversationId, userId, limit];
    return db.prepare(`SELECT * FROM (
        SELECT * FROM messages WHERE conversation_id=?
          AND id NOT IN (SELECT message_id FROM message_hidden WHERE user_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: 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: 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;
    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),
  // All reactions on the messages in a 1:1 thread.
  forPair: (teamId, a, b) =>
    db.prepare(`SELECT r.message_id, r.user_id, r.emoji FROM message_reactions r
      JOIN messages m ON r.message_id = m.id
      WHERE m.team_id=? AND m.conversation_id IS NULL AND ((m.sender_id=? AND m.recipient_id=?) OR (m.sender_id=? AND m.recipient_id=?))`).all(teamId, a, b, b, a),
  // All reactions on the messages in a group conversation.
  forConversation: (conversationId) =>
    db.prepare(`SELECT r.message_id, r.user_id, r.emoji FROM message_reactions r
      JOIN messages m ON r.message_id = m.id WHERE m.conversation_id=?`).all(conversationId),
};

const conversations = {
  create: ({ id, teamId, name, createdBy }) =>
    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 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: 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: 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: 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),
  rename: (id, name) => db.prepare('UPDATE conversations SET name=? WHERE id=?').run(name, id),
  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: 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 = {
  create: ({ id, teamId, uploaderId, name, mime, size }) =>
    db.prepare('INSERT INTO attachments (id,team_id,uploader_id,name,mime,size,created_at) VALUES (?,?,?,?,?,?,?)')
      .run(id, teamId, uploaderId, name, mime || null, size || 0, now()),
  byId: (id) => db.prepare('SELECT * FROM attachments WHERE id=?').get(id),
  // Newest first: media.js backfills streaming renditions for uploads that predate it.
  allVideos: () => db.prepare("SELECT id, mime FROM attachments WHERE mime LIKE 'video/%' ORDER BY created_at DESC").all(),
};

const scheduledMeetings = {
  create: ({ id, teamId, groupId, roomCode, title, description, scheduledAt, createdBy, participants, durationMins, recurrence, guestEmails, lobby }) =>
    db.prepare('INSERT INTO scheduled_meetings (id,team_id,group_id,room_code,title,description,scheduled_at,created_by,created_at,participants,duration_mins,recurrence,guest_emails,lobby) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)')
      .run(id, teamId, groupId || null, roomCode, title, description || null, scheduledAt, createdBy, now(), (participants && participants.length) ? JSON.stringify(participants) : null, durationMins || null, (recurrence && recurrence.length) ? JSON.stringify(recurrence) : null, (guestEmails && guestEmails.length) ? JSON.stringify(guestEmails) : null, (lobby === false ? 0 : 1)),
  byId: (id) => db.prepare('SELECT * FROM scheduled_meetings WHERE id=?').get(id),
  byCode: (code) => db.prepare('SELECT * FROM scheduled_meetings WHERE room_code=? ORDER BY created_at DESC LIMIT 1').get(code),
  // Meetings a user can see: created by them, a member of the group, or an invited participant.
  listForUser: (teamId, userId) =>
    db.prepare(`SELECT s.* FROM scheduled_meetings s
      WHERE s.team_id=? AND (
        s.created_by=? OR
        (s.group_id IS NOT NULL AND EXISTS (SELECT 1 FROM conversation_members cm WHERE cm.conversation_id=s.group_id AND cm.user_id=?)) OR
        (s.participants IS NOT NULL AND s.participants LIKE '%'||?||'%'))
      ORDER BY s.scheduled_at ASC`).all(teamId, userId, userId, '"' + userId + '"'),
  dueForReminder: (fromTs, toTs) => db.prepare('SELECT * FROM scheduled_meetings WHERE reminded=0 AND ended_at IS NULL AND scheduled_at>=? AND scheduled_at<=?').all(fromTs, toTs),
  markReminded: (id) => db.prepare('UPDATE scheduled_meetings SET reminded=1 WHERE id=?').run(id),
  end: (id, teamId) => db.prepare('UPDATE scheduled_meetings SET ended_at=? WHERE id=? AND team_id=?').run(now(), id, teamId),
  cancel: (id, teamId) => db.prepare('UPDATE scheduled_meetings SET cancelled=1, ended_at=? WHERE id=? AND team_id=?').run(now(), id, teamId),
  reschedule: (id, teamId, ts) => db.prepare('UPDATE scheduled_meetings SET scheduled_at=?, reminded=0 WHERE id=? AND team_id=?').run(ts, id, teamId), // recurrence: roll to next occurrence
  update: (id, teamId, { title, description, scheduledAt, durationMins, participants, recurrence, guestEmails, lobby }) =>
    db.prepare('UPDATE scheduled_meetings SET title=?, description=?, scheduled_at=?, duration_mins=?, participants=?, recurrence=?, guest_emails=?, lobby=?, reminded=0 WHERE id=? AND team_id=?')
      .run(title, description || null, scheduledAt, durationMins || null, (participants && participants.length) ? JSON.stringify(participants) : null, (recurrence && recurrence.length) ? JSON.stringify(recurrence) : null, (guestEmails && guestEmails.length) ? JSON.stringify(guestEmails) : null, (lobby === false ? 0 : 1), id, teamId),
  remove: (id, teamId) => db.prepare('DELETE FROM scheduled_meetings WHERE id=? AND team_id=?').run(id, teamId),
};

// #7: finished calls (not scheduled meetings — those have their own table).
const callHistory = {
  create: ({ id, teamId, room, groupId, kind, title, peak, participants, uids, startedAt, endedAt }) =>
    db.prepare('INSERT INTO call_history (id,team_id,room,group_id,kind,title,peak,participants,uids,started_at,ended_at) VALUES (?,?,?,?,?,?,?,?,?,?,?)')
      .run(id, teamId, room, groupId || null, kind || null, title || null, peak || 0,
        JSON.stringify(participants || []), JSON.stringify(uids || []), startedAt, endedAt),
  forTeam: (teamId) => db.prepare('SELECT * FROM call_history WHERE team_id=? ORDER BY ended_at DESC').all(teamId),
};

const recordings = {
  create: ({ id, teamId, room, groupId, meetingId, title, kind, file, mime, size, durationMs, createdBy, createdByName }) =>
    db.prepare('INSERT INTO recordings (id,team_id,room,group_id,meeting_id,title,kind,file,mime,size,duration_ms,created_by,created_by_name,created_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)')
      .run(id, teamId, room || null, groupId || null, meetingId || null, title || null, kind, file || null, mime || null, size || null, durationMs || null, createdBy || null, createdByName || null, now()),
  byId: (id) => db.prepare('SELECT * FROM recordings WHERE id=?').get(id),
  forTeam: (teamId) => db.prepare('SELECT * FROM recordings WHERE team_id=? ORDER BY created_at DESC').all(teamId),
};

const polls = {
  create: ({ id, teamId, conversationId, messageId, question, options, multi, createdBy }) =>
    db.prepare('INSERT INTO polls (id,team_id,conversation_id,message_id,question,options,multi,closed,created_by,created_at) VALUES (?,?,?,?,?,?,?,0,?,?)')
      .run(id, teamId, conversationId, messageId || null, question, JSON.stringify(options), multi ? 1 : 0, createdBy, now()),
  byId: (id) => db.prepare('SELECT * FROM polls WHERE id=?').get(id),
  close: (id) => db.prepare('UPDATE polls SET closed=1 WHERE id=?').run(id),
};

const pollVotes = {
  forPoll: (pollId) => db.prepare('SELECT user_id, option_idx FROM poll_votes WHERE poll_id=?').all(pollId),
  hasVoted: async (pollId, userId, idx) => !!(await db.prepare('SELECT 1 FROM poll_votes WHERE poll_id=? AND user_id=? AND option_idx=?').get(pollId, userId, idx)),
  add: (pollId, userId, idx) => db.prepare('INSERT 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),
};

const pushSubs = {
  // Upsert by endpoint: re-subscribing the same browser updates its keys/owner.
  add: ({ id, userId, endpoint, p256dh, auth }) =>
    db.prepare('INSERT INTO push_subscriptions (id,user_id,endpoint,p256dh,auth,created_at) VALUES (?,?,?,?,?,?) ON CONFLICT(endpoint) DO UPDATE SET user_id=excluded.user_id, p256dh=excluded.p256dh, auth=excluded.auth')
      .run(id, userId, endpoint, p256dh, auth, now()),
  byUser: (userId) => db.prepare('SELECT * FROM push_subscriptions WHERE user_id=?').all(userId),
  removeByEndpoint: (endpoint) => db.prepare('DELETE FROM push_subscriptions WHERE endpoint=?').run(endpoint),
};

const favorites = {
  set: (userId, target, on) => on
    ? db.prepare('INSERT INTO favorites (user_id,target,created_at) VALUES (?,?,?) ON CONFLICT(user_id,target) DO NOTHING').run(userId, target, now())
    : db.prepare('DELETE FROM favorites WHERE user_id=? AND target=?').run(userId, target),
  forUser: async (userId) => (await db.prepare('SELECT target FROM favorites WHERE user_id=?').all(userId)).map((r) => r.target),
};

const deviceTokens = {
  // Upsert by token: re-registering the same device refreshes its owner/platform/last_seen.
  register: ({ id, userId, tenantId, platform, token }) =>
    db.prepare('INSERT INTO device_tokens (id,user_id,tenant_id,platform,token,created_at,last_seen) VALUES (?,?,?,?,?,?,?) ON CONFLICT(token) DO UPDATE SET user_id=excluded.user_id, tenant_id=excluded.tenant_id, platform=excluded.platform, last_seen=excluded.last_seen')
      .run(id, userId, tenantId || null, platform, token, now(), now()),
  byUser: (userId) => db.prepare('SELECT * FROM device_tokens WHERE user_id=?').all(userId),
  removeByToken: (token) => db.prepare('DELETE FROM device_tokens WHERE token=?').run(token),
};

const appInstalls = {
  // Upsert by install_id: each launch refreshes version/os/last_seen and fills in the user once
  // they sign in (COALESCE keeps a known user if a later anonymous ping arrives).
  record: ({ id, installId, userId, userEmail, tenantId, platform, appVersion, os }) =>
    db.prepare(`INSERT INTO app_installs (id,install_id,user_id,user_email,tenant_id,platform,app_version,os,first_seen,last_seen)
      VALUES (?,?,?,?,?,?,?,?,?,?)
      ON CONFLICT(install_id) DO UPDATE SET
        user_id=COALESCE(excluded.user_id, app_installs.user_id),
        user_email=COALESCE(excluded.user_email, app_installs.user_email),
        tenant_id=COALESCE(excluded.tenant_id, app_installs.tenant_id),
        platform=excluded.platform, app_version=excluded.app_version, os=excluded.os,
        last_seen=excluded.last_seen`)
      .run(id, installId, userId || null, userEmail || null, tenantId || null, platform || null, appVersion || null, os || null, now(), now()),
  listForTenant: (tenantId) => db.prepare('SELECT * FROM app_installs WHERE tenant_id=? ORDER BY last_seen DESC').all(tenantId),
};

module.exports = { teams, users, authSessions, machines, audit, sessionsLog, refreshTokens, apiKeys, webhooks, messages, reactions, attachments, conversations, scheduledMeetings, callHistory, recordings, polls, pollVotes, pushSubs, favorites, deviceTokens, appInstalls };