hub/chat.js

// Family Chat: a messenger built into the hub, for the people who have an account here. Direct chats and groups, text, attachments, replies, reactions, edit and delete,
// unread counts, search, mute and archive, live delivery (server-sent events) and phone push when someone is not looking. Everything is kept on this server, in
// DATA_DIR/chat: conversations and messages as JSON, attachments as files. (Messages are stored readable on the server; end-to-end encryption is a later decision.)
const fs = require('fs'), path = require('path'), crypto = require('crypto');
const { pipeline } = require('stream/promises');
const atomicWrite = require('./atomic');
const users = require('./users');
const push = require('./push');
const { readBody } = require('./body');

const DATA_DIR = process.env.DATA_DIR || path.join(__dirname, 'data'), ROOT = () => path.join(DATA_DIR, 'chat');
const MAX_TEXT = 8000, MAX_FILE = 100 * 1024 * 1024, MAX_FILES = 10, MAX_MEMBERS = 40, MAX_MSGS = 50000, PAGE = 60;
const EMOJI = /^[\p{Extended_Pictographic}‍️⃣\p{Emoji_Modifier}]{1,12}$/u;
const json = (res, code, o) => { res.writeHead(code, { 'content-type': 'application/json', 'cache-control': 'no-store' }); res.end(JSON.stringify(o)); };
const fail = (code, message) => Object.assign(new Error(message), { code });
const str = (v, n) => String(v == null ? '' : v).replace(/[\u0000-\u0008\u000b\u000c\u000e-\u001f]/g, '').slice(0, n);
const newId = () => crypto.randomBytes(8).toString('hex');
const idOk = s => /^[0-9a-f]{16}$/.test(s || '') || /^d:[0-9a-z]+:[0-9a-z]+$/.test(s || '');

// ---- storage (small enough to hold in memory and write whole)
const cvFile = () => path.join(ROOT(), 'convs.json'), msFile = id => path.join(ROOT(), 'msgs', id.replace(/:/g, '_') + '.json'), flFile = () => path.join(ROOT(), 'files.json'), blob = id => path.join(ROOT(), 'files', id);
const read = (f, d) => { try { return JSON.parse(fs.readFileSync(f, 'utf8')); } catch { return d; } };
const write = (f, v) => { fs.mkdirSync(path.dirname(f), { recursive: true, mode: 0o700 }); atomicWrite(f, JSON.stringify(v), { mode: 0o600 }); };
let convsC = null, filesC = null; const msgsC = new Map();
const convs = () => convsC || (convsC = read(cvFile(), []));
const saveConvs = () => write(cvFile(), convs());
const files = () => filesC || (filesC = read(flFile(), {}));
const saveFiles = () => write(flFile(), files());
const msgs = id => { let m = msgsC.get(id); if (!m) { m = read(msFile(id), []); msgsC.set(id, m); if (msgsC.size > 60) msgsC.delete(msgsC.keys().next().value); } return m; };
const saveMsgs = id => write(msFile(id), msgs(id));
// each person's own marks: how far they have read, and which chats they muted or archived
const stFile = uid => path.join(users.dirOf(uid), 'chat-state.json');
const stateOf = uid => { const s = read(stFile(uid), {}); return { read: s.read || {}, muted: s.muted || [], archived: s.archived || [] }; };
const saveState = (uid, s) => write(stFile(uid), s);

const nameOf = uid => { const u = users.get(uid); return u ? u.name : 'Someone'; };
const live = uid => { const u = users.get(uid); return !!u && !u.disabled && users.can(u, 'chat'); };
const people = () => users.all().filter(u => !u.disabled && users.can(u, 'chat')).map(u => ({ id: u.id, name: u.name }));
const convOf = (id, me) => { const c = convs().find(x => x.id === id); if (!c || !c.members.includes(me.id)) throw fail(404, 'Conversation not found'); return c; };
const unreadIn = (c, uid, st) => { const r = st.read[c.id] || 0; return msgs(c.id).filter(m => m.ts > r && m.from !== uid && !m.deleted).length; };
const totalUnread = uid => { const st = stateOf(uid); return convs().filter(c => c.members.includes(uid) && !st.muted.includes(c.id) && !st.archived.includes(c.id)).reduce((n, c) => n + unreadIn(c, uid, st), 0); };
const preview = m => m ? (m.deleted ? 'This message was deleted' : m.text ? str(m.text, 120) : m.att && m.att.length ? '📎 ' + (m.att[0].name || 'Attachment') : '') : '';

// ---- live updates (server-sent events), one stream per open app
const clients = new Map(); // user id -> Set of responses
const online = uid => (clients.get(uid) || new Set()).size > 0;
function emit(uids, ev, per) { for (const uid of uids) for (const res of clients.get(uid) || []) { try { res.write('data: ' + JSON.stringify(per ? per(uid, ev) : ev) + '\n\n'); } catch { /* closed */ } } }

const view = (c, uid, st) => { const last = msgs(c.id).slice(-1)[0], others = c.members.filter(x => x !== uid);
  return { id: c.id, type: c.type, name: c.type === 'group' ? c.name : nameOf(others[0] || uid), members: c.members.map(x => ({ id: x, name: nameOf(x) })), by: c.by, created: c.created, ts: last ? last.ts : c.created, preview: last ? (c.type === 'group' && last.from !== uid ? nameOf(last.from) + ': ' : last.from === uid ? 'You: ' : '') + preview(last) : '', unread: unreadIn(c, uid, st), muted: st.muted.includes(c.id), archived: st.archived.includes(c.id) }; };
const msgView = (m, uid) => ({ id: m.id, ts: m.ts, from: m.from, name: nameOf(m.from), out: m.from === uid, text: m.text, att: m.att, quote: m.quote, react: m.react, edited: m.edited, deleted: m.deleted });

function send(me, b) {
  const c = convOf(String(b.conv || ''), me), text = str(b.text, MAX_TEXT).trim(), ids = (Array.isArray(b.files) ? b.files : []).map(String).slice(0, MAX_FILES), att = [];
  for (const id of ids) { const f = files()[id]; if (!f || f.conv !== c.id || f.by !== me.id || f.attached) throw fail(400, 'An attachment is not available (upload it again)'); att.push({ id, name: f.name, type: f.type, size: f.size }); }
  if (!text && !att.length) throw fail(400, 'Write something or attach a file');
  const list = msgs(c.id); let quote;
  if (b.quote) { const q = list.find(x => x.id === String(b.quote)); if (q) quote = { id: q.id, from: q.from, name: nameOf(q.from), text: q.deleted ? '' : str(q.text || (q.att && q.att[0] ? q.att[0].name : ''), 200) }; }
  const last = list[list.length - 1], m = { id: newId(), ts: Math.max(Date.now(), last ? last.ts + 1 : 0), from: me.id, text, att: att.length ? att : undefined, quote, react: {} };
  list.push(m); if (list.length > MAX_MSGS) list.splice(0, list.length - MAX_MSGS); att.forEach(a => { files()[a.id].attached = true; }); saveMsgs(c.id); if (att.length) saveFiles();
  const body = (c.type === 'group' ? me.name + ': ' : '') + (text ? str(text, 140) : '📎 ' + att[0].name), title = c.type === 'group' ? c.name : me.name;
  emit(c.members, { t: 'msg', conv: c.id, id: m.id }, uid => ({ t: 'msg', conv: c.id, id: m.id, out: uid === me.id, name: title, preview: body, unread: totalUnread(uid) }));
  for (const uid of c.members) if (uid !== me.id && !online(uid) && !stateOf(uid).muted.includes(c.id) && live(uid)) push.notify(uid, { title, body, url: '#/chat/' + encodeURIComponent(c.id), tag: 'chat:' + c.id }).catch(() => {});
  return msgView(m, me.id);
}

async function handle(req, res) {
  const u = new URL(req.url, 'http://x'), p = (() => { try { return decodeURIComponent(u.pathname.slice('/api/chat'.length)) || '/'; } catch { return '/'; } })(), me = users.current(), q = u.searchParams; // (a chat id has colons, which a browser writes as %3A)
  try {
    let m;
    if (p === '/status') return json(res, 200, { ok: true, me: { id: me.id, name: me.name }, unread: totalUnread(me.id) });
    if (p === '/people') return json(res, 200, people().filter(x => x.id !== me.id));
    if (p === '/events') {
      res.writeHead(200, { 'content-type': 'text/event-stream', 'cache-control': 'no-store', connection: 'keep-alive', 'x-accel-buffering': 'no' }); res.write(': hello\n\n');
      const set = clients.get(me.id) || new Set(); set.add(res); clients.set(me.id, set);
      const ka = setInterval(() => { try { res.write(': \n\n'); } catch { /* closed */ } }, 25000);
      res.on('close', () => { clearInterval(ka); set.delete(res); if (!set.size) clients.delete(me.id); }); return;
    }
    if (p === '/convs' && req.method === 'GET') { const st = stateOf(me.id); return json(res, 200, { convs: convs().filter(c => c.members.includes(me.id)).map(c => view(c, me.id, st)).sort((a, b) => b.ts - a.ts), unread: totalUnread(me.id) }); }
    if (p === '/new' && req.method === 'POST') {
      const b = JSON.parse(await readBody(req, 1e4) || '{}'); let c;
      if (b.type === 'group') {
        const name = str(b.name, 60).trim(); if (!name) throw fail(400, 'Give the group a name'); const mem = [...new Set([me.id, ...(Array.isArray(b.members) ? b.members : [])].map(String))].filter(live); if (mem.length < 2) throw fail(400, 'Add at least one other person'); if (mem.length > MAX_MEMBERS) throw fail(400, 'That is a lot of people for one group');
        c = { id: newId(), type: 'group', name, members: mem, by: me.id, created: Date.now() }; convs().push(c);
      } else {
        const other = String(b.with || ''); if (other === me.id || !live(other)) throw fail(400, 'Choose someone from the family'); const id = 'd:' + [me.id, other].sort().join(':'); c = convs().find(x => x.id === id);
        if (!c) { c = { id, type: 'direct', members: [me.id, other].sort(), by: me.id, created: Date.now() }; convs().push(c); }
      }
      saveConvs(); emit(c.members, { t: 'conv', conv: c.id }); return json(res, 200, { conv: view(c, me.id, stateOf(me.id)) });
    }
    if (p === '/conv/update' && req.method === 'POST') { // rename, add or remove people, leave (groups)
      const b = JSON.parse(await readBody(req, 1e4) || '{}'), c = convOf(String(b.id || ''), me); if (c.type !== 'group') throw fail(400, 'Only groups can be changed');
      const isOwner = c.by === me.id || me.role === 'admin'; const before = [...c.members];
      if (b.name !== undefined) { const n = str(b.name, 60).trim(); if (!n) throw fail(400, 'Give the group a name'); c.name = n; }
      for (const id of Array.isArray(b.add) ? b.add : []) if (live(String(id)) && !c.members.includes(String(id)) && c.members.length < MAX_MEMBERS) c.members.push(String(id));
      for (const id of Array.isArray(b.remove) ? b.remove : []) { if (!isOwner && String(id) !== me.id) throw fail(403, 'Only the person who made the group, or an admin, can remove others'); c.members = c.members.filter(x => x !== String(id)); }
      if (b.leave) c.members = c.members.filter(x => x !== me.id);
      if (!c.members.length) convsC = convs().filter(x => x !== c); else if (c.by && !c.members.includes(c.by)) c.by = c.members[0];
      saveConvs(); emit([...new Set([...before, ...c.members])], { t: 'conv', conv: c.id }); return json(res, 200, c.members.includes(me.id) ? { conv: view(c, me.id, stateOf(me.id)) } : { left: true });
    }
    if ((m = /^\/conv\/(\w[\w:]*)$/.exec(p)) && req.method === 'GET') {
      const c = convOf(m[1], me), all = msgs(c.id), before = +q.get('before') || Infinity, eligible = all.filter(x => x.ts < before), page = eligible.slice(-PAGE);
      return json(res, 200, { conv: view(c, me.id, stateOf(me.id)), msgs: page.map(x => msgView(x, me.id)), more: eligible.length > page.length, names: Object.fromEntries(c.members.map(x => [x, nameOf(x)])) });
    }
    if (p === '/send' && req.method === 'POST') return json(res, 200, { msg: send(me, JSON.parse(await readBody(req, 5e4) || '{}')) });
    if (p === '/upload' && req.method === 'POST') { // the body is the file; ?conv= says which chat it is for
      const c = convOf(String(q.get('conv') || ''), me); if ((+req.headers['content-length'] || 0) > MAX_FILE) throw fail(413, 'That file is too large (100 MB at most)');
      const id = newId(), type = /^[\w.+-]+\/[\w.+-]+$/.test(String(req.headers['content-type'] || '').split(';')[0].trim()) ? String(req.headers['content-type']).split(';')[0].trim().toLowerCase() : 'application/octet-stream', name = path.basename(str(q.get('name'), 160)).replace(/[\\/:*?"<>|]+/g, '_') || 'file';
      fs.mkdirSync(path.dirname(blob(id)), { recursive: true, mode: 0o700 }); let n = 0; req.on('data', d => { n += d.length; if (n > MAX_FILE) req.destroy(fail(413, 'That file is too large')); });
      try { await pipeline(req, fs.createWriteStream(blob(id) + '.part', { mode: 0o600 })); } catch (e) { fs.rm(blob(id) + '.part', { force: true }, () => {}); throw e.code === 413 ? e : fail(400, 'The upload did not finish'); }
      fs.renameSync(blob(id) + '.part', blob(id)); files()[id] = { conv: c.id, by: me.id, name, type, size: n, ts: Date.now() }; saveFiles(); return json(res, 200, { id, name, type, size: n });
    }
    if ((m = /^\/file\/([0-9a-f]{16})$/.exec(p)) && req.method === 'GET') {
      const f = files()[m[1]]; if (!f) throw fail(404, 'Not found'); convOf(f.conv, me); let st; try { st = fs.statSync(blob(m[1])); } catch { throw fail(404, 'Not found'); }
      const inline = /^(image\/(png|jpeg|gif|webp|heic|heif|avif)|audio\/|video\/|application\/pdf$)/.test(f.type), h = { 'content-type': inline ? f.type : 'application/octet-stream', 'x-content-type-options': 'nosniff', 'cache-control': 'private, max-age=3600', 'accept-ranges': 'bytes', 'content-security-policy': "default-src 'none'; sandbox" };
      if (!inline || q.get('dl') === '1') h['content-disposition'] = `attachment; filename="${f.name.replace(/[^\x20-\x7e]/g, '_').replace(/"/g, '')}"; filename*=UTF-8''${encodeURIComponent(f.name)}`;
      const rg = /^bytes=(\d*)-(\d*)$/.exec(req.headers.range || '');
      if (rg && (rg[1] || rg[2])) { const a = rg[1] ? +rg[1] : Math.max(0, st.size - +rg[2]), b = rg[1] && rg[2] ? Math.min(+rg[2], st.size - 1) : st.size - 1; if (a > b || a >= st.size) { res.writeHead(416, { 'content-range': 'bytes */' + st.size }); return res.end(); } res.writeHead(206, { ...h, 'content-range': `bytes ${a}-${b}/${st.size}`, 'content-length': b - a + 1 }); return fs.createReadStream(blob(m[1]), { start: a, end: b }).pipe(res); }
      res.writeHead(200, { ...h, 'content-length': st.size }); return fs.createReadStream(blob(m[1])).pipe(res);
    }
    if (p === '/react' && req.method === 'POST') {
      const b = JSON.parse(await readBody(req, 1e4) || '{}'), c = convOf(String(b.conv || ''), me), x = msgs(c.id).find(y => y.id === String(b.msg)); if (!x || x.deleted) throw fail(404, 'Message not found');
      const e = String(b.emoji || ''); if (e && !EMOJI.test(e)) throw fail(400, 'That is not an emoji'); x.react = x.react || {}; if (e) x.react[me.id] = e; else delete x.react[me.id]; saveMsgs(c.id);
      emit(c.members, { t: 'upd', conv: c.id, id: x.id }); return json(res, 200, { ok: true });
    }
    if ((p === '/edit' || p === '/delete') && req.method === 'POST') {
      const b = JSON.parse(await readBody(req, 2e4) || '{}'), c = convOf(String(b.conv || ''), me), x = msgs(c.id).find(y => y.id === String(b.msg)); if (!x) throw fail(404, 'Message not found'); if (x.from !== me.id) throw fail(403, 'You can only change your own messages'); if (x.deleted) throw fail(400, 'That message was deleted');
      if (p === '/edit') { const t = str(b.text, MAX_TEXT).trim(); if (!t) throw fail(400, 'Write something'); if (Date.now() - x.ts > 48 * 3600e3) throw fail(400, 'Messages can be edited for two days'); x.text = t; x.edited = true; }
      else { x.deleted = true; x.text = ''; for (const a of x.att || []) { fs.rm(blob(a.id), { force: true }, () => {}); delete files()[a.id]; } x.att = undefined; x.react = {}; saveFiles(); }
      saveMsgs(c.id); emit(c.members, { t: 'upd', conv: c.id, id: x.id }); return json(res, 200, { ok: true });
    }
    if (p === '/read' && req.method === 'POST') { const b = JSON.parse(await readBody(req, 1e4) || '{}'), c = convOf(String(b.id || ''), me), st = stateOf(me.id), last = msgs(c.id).slice(-1)[0]; st.read[c.id] = last ? last.ts : Date.now(); saveState(me.id, st); emit([me.id], { t: 'read', conv: c.id, unread: totalUnread(me.id) }); return json(res, 200, { ok: true, unread: totalUnread(me.id) }); }
    if ((p === '/conv/mute' || p === '/conv/archive') && req.method === 'POST') { const b = JSON.parse(await readBody(req, 1e4) || '{}'), c = convOf(String(b.id || ''), me), st = stateOf(me.id), k = p.endsWith('mute') ? 'muted' : 'archived'; st[k] = st[k].filter(x => x !== c.id); if (b.on) st[k].push(c.id); saveState(me.id, st); emit([me.id], { t: 'conv', conv: c.id, unread: totalUnread(me.id) }); return json(res, 200, { ok: true }); }
    if (p === '/search' && req.method === 'GET') {
      const needle = str(q.get('q'), 100).trim().toLowerCase(); if (needle.length < 2) return json(res, 200, { results: [] }); const out = [], st = stateOf(me.id);
      for (const c of convs().filter(x => x.members.includes(me.id))) { const cv = view(c, me.id, st); for (const x of msgs(c.id)) if (!x.deleted && x.text && x.text.toLowerCase().includes(needle)) out.push({ conv: c.id, convName: cv.name, msg: msgView(x, me.id) }); }
      return json(res, 200, { results: out.sort((a, b) => b.msg.ts - a.msg.ts).slice(0, 40) });
    }
    return json(res, 404, { error: 'Not found' });
  } catch (e) { if (res.headersSent) return res.destroy(); return json(res, e.code >= 400 && e.code < 600 ? e.code : e.status === 413 ? 413 : 400, { error: e instanceof SyntaxError ? 'Bad request' : e.message }); }
}

module.exports = { handle, totalUnread };