AgentHub/server/lib/chat.js

/**
 * @file chat.js
 * @description Chat between the users of a server-mode install, shown in the
 * pet's window beside the chat with the pet: a direct chat (`dm`, at most one
 * per pair of users) or a group anyone may start and name. Only a room's
 * MEMBERS read or write it — an admin or analyst included, and there is no
 * route that lists another user's rooms. A group's creator adds, removes and
 * renames; any member may leave; anyone deletes their own messages (only). A message is text (≤ 8000 chars) and up to
 * `MAX_FILES` files of ≤ `MAX_FILE_BYTES` each, stored once under
 * `<data dir>/chat/<room id>/<random id>` — a name the server makes, never one
 * from the request — and served back only to the room's members (the route
 * adds a `sandbox` CSP and `nosniff`; only raster images are shown inline).
 * Every change is told to exactly the users concerned
 * (`chat_message` / `chat_room` frames, `to` = their ids, see
 * `scope.exactAudience`).
 * @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
 */

"use strict";

const fs = require("node:fs");
const path = require("node:path");
const crypto = require("node:crypto");
const { getDataDir } = require("./claude-home");

const db = () => require("../db").db;
const sql = (q) => db().prepare(q);

const MAX_BODY = 8000;
const MAX_NAME = 80;
const MAX_FILES = 10;
const MAX_FILE_BYTES = 25 * 1024 * 1024;
const MAX_GROUP = 200;
/** Raster types shown inline; anything else is a download. */
const INLINE_TYPES = new Set(["image/png", "image/jpeg", "image/gif", "image/webp"]);

function err(code, message, status = 400) {
  return Object.assign(new Error(message), { code, status });
}

const chatDir = () => path.join(getDataDir(), "chat");
const fileDir = (roomId) => path.join(chatDir(), String(Number(roomId)));

/** A user who may be chatted with: exists and is not disabled. */
function activeUser(id) {
  const n = Number(id);
  if (!Number.isSafeInteger(n) || n <= 0) return null;
  return sql("SELECT id FROM users WHERE id = ? AND COALESCE(disabled, 0) = 0").get(n) || null;
}

/** Every enabled account but the caller — who a chat can be started with. */
function people(me) {
  return sql(
    `SELECT id, username, display_name, avatar_at FROM users
     WHERE COALESCE(disabled, 0) = 0 AND id != ? ORDER BY COALESCE(display_name, username)`
  )
    .all(me)
    .map((u) => ({
      id: u.id,
      username: u.username,
      displayName: u.display_name ?? null,
      avatarAt: u.avatar_at ?? null,
    }));
}

function memberIds(roomId) {
  return sql("SELECT user_id FROM chat_members WHERE room_id = ? ORDER BY user_id")
    .all(roomId)
    .map((r) => r.user_id);
}

/** The room, when `me` is one of its members; else ENOTFOUND (404) — never "forbidden". */
function roomFor(me, roomId) {
  const id = Number(roomId);
  const row =
    Number.isSafeInteger(id) && id > 0
      ? sql(
          `SELECT r.* FROM chat_rooms r JOIN chat_members m ON m.room_id = r.id
           WHERE r.id = ? AND m.user_id = ?`
        ).get(id, me)
      : null;
  if (!row) throw err("ENOTFOUND", "no such chat", 404);
  return row;
}

function filesOf(messageIds) {
  if (!messageIds.length) return new Map();
  const rows = sql(
    `SELECT id, message_id, name, type, size FROM chat_files
     WHERE message_id IN (${messageIds.map(() => "?").join(",")}) ORDER BY rowid`
  ).all(...messageIds);
  const out = new Map();
  for (const f of rows) {
    if (!out.has(f.message_id)) out.set(f.message_id, []);
    out.get(f.message_id).push({ id: f.id, name: f.name, type: f.type, size: f.size });
  }
  return out;
}

function shapeMessages(rows) {
  const files = filesOf(rows.map((r) => r.id));
  return rows.map((r) => ({
    id: r.id,
    roomId: r.room_id,
    userId: r.user_id,
    kind: r.kind,
    body: r.body,
    createdAt: r.created_at,
    files: files.get(r.id) || [],
  }));
}

/** One room as `me` sees it: members, newest message, unread count. */
function shapeRoom(me, r) {
  const mine = sql("SELECT last_read_id FROM chat_members WHERE room_id = ? AND user_id = ?").get(
    r.id,
    me
  );
  const last = sql("SELECT * FROM chat_messages WHERE room_id = ? ORDER BY id DESC LIMIT 1").get(
    r.id
  );
  const unread = sql(
    "SELECT COUNT(*) AS n FROM chat_messages WHERE room_id = ? AND id > ? AND COALESCE(user_id, -1) != ?"
  ).get(r.id, mine ? mine.last_read_id : 0, me).n;
  return {
    id: r.id,
    kind: r.kind,
    name: r.name,
    createdBy: r.created_by,
    createdAt: r.created_at,
    lastMessageAt: r.last_message_at,
    members: memberIds(r.id),
    lastMessage: last ? shapeMessages([last])[0] : null,
    lastReadId: mine ? mine.last_read_id : 0,
    unread,
  };
}

/** `me`'s rooms, the most recently active first. */
function listRooms(me) {
  return sql(
    `SELECT r.* FROM chat_rooms r JOIN chat_members m ON m.room_id = r.id
     WHERE m.user_id = ? ORDER BY COALESCE(r.last_message_at, r.created_at) DESC`
  )
    .all(me)
    .map((r) => shapeRoom(me, r));
}

function cleanName(name) {
  if (name === undefined || name === null) return "";
  if (typeof name !== "string") throw err("EBADFIELD", "name: text");
  return name.trim().slice(0, MAX_NAME);
}

function userList(value, field) {
  if (value === undefined) return [];
  if (!Array.isArray(value) || value.length > MAX_GROUP)
    throw err("EBADFIELD", `${field}: a list of user ids`);
  const ids = [...new Set(value.map(Number))];
  for (const id of ids) if (!activeUser(id)) throw err("ENOUSER", `no such user: ${id}`);
  return ids;
}

/** The direct chat between `me` and `other`, made on first use. `created` says which. */
function openDm(me, other) {
  const them = activeUser(other);
  if (!them || them.id === me) throw err("ENOUSER", "pick another user");
  const key = [me, them.id].sort((a, b) => a - b).join(":");
  const found = sql("SELECT * FROM chat_rooms WHERE dm_key = ?").get(key);
  if (found) return { room: shapeRoom(me, found), created: false };
  const id = db().transaction(() => {
    const { lastInsertRowid } = sql(
      "INSERT INTO chat_rooms (kind, dm_key, created_by) VALUES ('dm', ?, ?)"
    ).run(key, me);
    const add = sql("INSERT INTO chat_members (room_id, user_id) VALUES (?, ?)");
    add.run(lastInsertRowid, me);
    add.run(lastInsertRowid, them.id);
    return lastInsertRowid;
  })();
  return {
    room: shapeRoom(me, sql("SELECT * FROM chat_rooms WHERE id = ?").get(id)),
    created: true,
  };
}

function systemLine(roomId, me, body) {
  sql("INSERT INTO chat_messages (room_id, user_id, kind, body) VALUES (?, ?, 'system', ?)").run(
    roomId,
    me,
    body
  );
  sql(
    "UPDATE chat_rooms SET last_message_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') WHERE id = ?"
  ).run(roomId);
}

/** A new group of `me` plus `userIds`. */
function createGroup(me, { name, userIds }) {
  const title = cleanName(name);
  const ids = userList(userIds, "userIds").filter((id) => id !== me);
  if (!ids.length) throw err("EBADFIELD", "a group needs at least one other member");
  const id = db().transaction(() => {
    const { lastInsertRowid } = sql(
      "INSERT INTO chat_rooms (kind, name, created_by) VALUES ('group', ?, ?)"
    ).run(title, me);
    const add = sql("INSERT INTO chat_members (room_id, user_id) VALUES (?, ?)");
    for (const uid of [me, ...ids]) add.run(lastInsertRowid, uid);
    systemLine(lastInsertRowid, me, "created");
    return lastInsertRowid;
  })();
  return shapeRoom(me, sql("SELECT * FROM chat_rooms WHERE id = ?").get(id));
}

/**
 * A group's creator renames it, adds or removes members. Answers the room and
 * `notify` = every user to tell (removed ones included, so they drop it).
 */
function updateGroup(me, roomId, { name, add, remove }) {
  const room = roomFor(me, roomId);
  if (room.kind !== "group") throw err("EBADFIELD", "a direct chat has no members to change");
  if (room.created_by !== me) throw err("EFORBIDDEN", "only the group's creator changes it", 403);
  const before = memberIds(room.id);
  const adds = userList(add, "add").filter((id) => !before.includes(id));
  const drops = userList(remove, "remove").filter((id) => id !== me && before.includes(id));
  db().transaction(() => {
    if (name !== undefined) {
      const title = cleanName(name);
      if (title !== room.name) {
        sql("UPDATE chat_rooms SET name = ? WHERE id = ?").run(title, room.id);
        systemLine(room.id, me, "renamed");
      }
    }
    const ins = sql("INSERT OR IGNORE INTO chat_members (room_id, user_id) VALUES (?, ?)");
    for (const uid of adds) ins.run(room.id, uid);
    const del = sql("DELETE FROM chat_members WHERE room_id = ? AND user_id = ?");
    for (const uid of drops) del.run(room.id, uid);
    if (adds.length) systemLine(room.id, me, `added:${adds.join(",")}`);
    if (drops.length) systemLine(room.id, me, `removed:${drops.join(",")}`);
  })();
  return {
    room: shapeRoom(me, sql("SELECT * FROM chat_rooms WHERE id = ?").get(room.id)),
    notify: [...new Set([...before, ...adds])],
  };
}

/** `me` leaves a group; the last one out deletes it and its files. Answers who to tell. */
function leaveGroup(me, roomId) {
  const room = roomFor(me, roomId);
  if (room.kind !== "group") throw err("EBADFIELD", "a direct chat cannot be left");
  const before = memberIds(room.id);
  db().transaction(() => {
    sql("DELETE FROM chat_members WHERE room_id = ? AND user_id = ?").run(room.id, me);
    if (before.length > 1) systemLine(room.id, me, "left");
    else sql("DELETE FROM chat_rooms WHERE id = ?").run(room.id);
  })();
  if (before.length <= 1) fs.rmSync(fileDir(room.id), { recursive: true, force: true });
  return { notify: before };
}

/** A page of messages, oldest first: the newest `limit`, or those before `before`. */
function messages(me, roomId, { before, limit } = {}) {
  const room = roomFor(me, roomId);
  const n = Math.min(Math.max(parseInt(limit, 10) || 50, 1), 200);
  const b = parseInt(before, 10);
  const rows = Number.isSafeInteger(b)
    ? sql("SELECT * FROM chat_messages WHERE room_id = ? AND id < ? ORDER BY id DESC LIMIT ?").all(
        room.id,
        b,
        n
      )
    : sql("SELECT * FROM chat_messages WHERE room_id = ? ORDER BY id DESC LIMIT ?").all(room.id, n);
  return { messages: shapeMessages(rows.reverse()), hasMore: rows.length === n };
}

/** A file's media type, kept only when it is a plain `type/subtype`. */
function cleanType(t) {
  return typeof t === "string" && /^[a-z0-9.+-]+\/[a-z0-9.+-]+$/i.test(t)
    ? t.toLowerCase()
    : "application/octet-stream";
}

function cleanFileName(n) {
  const base = path.basename(String(n || "file")).replace(/[\u0000-\u001f\u007f]/g, "");
  return base.slice(0, 200) || "file";
}

/**
 * `me` posts to a room: `body` text and `files` ([{name, type, buffer}]).
 * Answers the message and who to tell (every member).
 */
function post(me, roomId, { body, files = [] }) {
  const room = roomFor(me, roomId);
  const text = typeof body === "string" ? body.replace(/\s+$/, "") : "";
  if (text.length > MAX_BODY)
    throw err("ETOOBIG", `a message is at most ${MAX_BODY} characters`, 413);
  if (files.length > MAX_FILES) throw err("ETOOBIG", `at most ${MAX_FILES} files`, 413);
  for (const f of files)
    if (f.buffer.length > MAX_FILE_BYTES) throw err("ETOOBIG", "a file is at most 25 MB", 413);
  if (!text.trim() && !files.length) throw err("EBADFIELD", "an empty message");
  const dir = fileDir(room.id);
  const stored = files.map((f) => ({
    id: crypto.randomBytes(16).toString("hex"),
    name: cleanFileName(f.name),
    type: cleanType(f.type),
    buffer: f.buffer,
  }));
  if (stored.length) {
    fs.mkdirSync(dir, { recursive: true, mode: 0o700 });
    for (const f of stored) fs.writeFileSync(path.join(dir, f.id), f.buffer, { mode: 0o600 });
  }
  let id;
  try {
    id = db().transaction(() => {
      const { lastInsertRowid } = sql(
        "INSERT INTO chat_messages (room_id, user_id, body) VALUES (?, ?, ?)"
      ).run(room.id, me, text);
      const ins = sql(
        "INSERT INTO chat_files (id, message_id, room_id, name, type, size) VALUES (?, ?, ?, ?, ?, ?)"
      );
      for (const f of stored)
        ins.run(f.id, lastInsertRowid, room.id, f.name, f.type, f.buffer.length);
      sql(
        "UPDATE chat_rooms SET last_message_at = (SELECT created_at FROM chat_messages WHERE id = ?) WHERE id = ?"
      ).run(lastInsertRowid, room.id);
      // Your own message is read.
      sql("UPDATE chat_members SET last_read_id = ? WHERE room_id = ? AND user_id = ?").run(
        lastInsertRowid,
        room.id,
        me
      );
      return lastInsertRowid;
    })();
  } catch (e) {
    for (const f of stored) fs.rmSync(path.join(dir, f.id), { force: true });
    throw e;
  }
  const [message] = shapeMessages([sql("SELECT * FROM chat_messages WHERE id = ?").get(id)]);
  return { message, notify: memberIds(room.id) };
}

/** `me` has read the room up to message `upTo` (never moves back). */
function markRead(me, roomId, upTo) {
  const room = roomFor(me, roomId);
  const n = Number(upTo);
  if (!Number.isSafeInteger(n) || n < 0) throw err("EBADFIELD", "upTo: a message id");
  sql(
    "UPDATE chat_members SET last_read_id = MAX(last_read_id, ?) WHERE room_id = ? AND user_id = ?"
  ).run(n, room.id, me);
  return { lastReadId: Math.max(n, 0) };
}

/**
 * `me` deletes their OWN text message (and its files) — never anyone else's,
 * an admin included. Answers who to tell (every member).
 */
function removeMessage(me, roomId, messageId) {
  const room = roomFor(me, roomId);
  const id = Number(messageId);
  const m = Number.isSafeInteger(id)
    ? sql("SELECT * FROM chat_messages WHERE id = ? AND room_id = ?").get(id, room.id)
    : null;
  if (!m) throw err("ENOTFOUND", "no such message", 404);
  if (m.user_id !== me || m.kind !== "text")
    throw err("EFORBIDDEN", "only your own messages can be deleted", 403);
  const files = sql("SELECT id FROM chat_files WHERE message_id = ?").all(id);
  db().transaction(() => {
    sql("DELETE FROM chat_messages WHERE id = ?").run(id);
    const last = sql("SELECT MAX(created_at) AS at FROM chat_messages WHERE room_id = ?").get(
      room.id
    );
    sql("UPDATE chat_rooms SET last_message_at = ? WHERE id = ?").run(last.at ?? null, room.id);
  })();
  for (const f of files) fs.rmSync(path.join(fileDir(room.id), f.id), { force: true });
  return { notify: memberIds(room.id) };
}

/** A file of a room `me` is in: `{path, name, type, inline}`. */
function file(me, roomId, fileId) {
  const room = roomFor(me, roomId);
  if (typeof fileId !== "string" || !/^[0-9a-f]{32}$/.test(fileId))
    throw err("ENOTFOUND", "no such file", 404);
  const f = sql("SELECT * FROM chat_files WHERE id = ? AND room_id = ?").get(fileId, room.id);
  const p = f && path.join(fileDir(room.id), f.id);
  if (!f || !fs.existsSync(p)) throw err("ENOTFOUND", "no such file", 404);
  return { path: p, name: f.name, type: f.type, inline: INLINE_TYPES.has(f.type) };
}

module.exports = {
  MAX_FILES,
  MAX_FILE_BYTES,
  people,
  listRooms,
  openDm,
  createGroup,
  updateGroup,
  leaveGroup,
  messages,
  post,
  markRead,
  removeMessage,
  file,
  roomFor,
  shapeRoom,
};