Chat Architecture

-- Chatterbox storage schema (PostgreSQL 18)
CREATE TABLE users (
  id           text PRIMARY KEY,
  name         text NOT NULL,
  created_at   timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE conversations (
  name         text PRIMARY KEY CHECK (name ~ '^[a-z0-9-]{1,64}$'),
  kind         text NOT NULL CHECK (kind IN ('public', 'private', 'dm')),
  last_seq     bigint NOT NULL DEFAULT 0,          -- the per-room sequence counter
  created_at   timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE memberships (
  room            text   NOT NULL REFERENCES conversations (name),
  user_id         text   NOT NULL REFERENCES users (id),
  role            text   NOT NULL DEFAULT 'member' CHECK (role IN ('owner', 'moderator', 'member')),
  joined_at       timestamptz NOT NULL DEFAULT now(),
  delivered_seq   bigint NOT NULL DEFAULT 0,       -- cumulative delivered receipt
  read_seq        bigint NOT NULL DEFAULT 0,       -- cumulative read receipt
  PRIMARY KEY (room, user_id)
);
CREATE INDEX memberships_by_user ON memberships (user_id);

-- One row per message; edits update text and edited_at, deletes set deleted_at.
CREATE TABLE messages (
  room         text   NOT NULL REFERENCES conversations (name),
  seq          bigint NOT NULL,
  id           text   NOT NULL UNIQUE,             -- the client-generated envelope id
  sender_id    text   NOT NULL REFERENCES users (id),
  text         text   NOT NULL CHECK (char_length(text) BETWEEN 1 AND 4000),
  reply_to     text   REFERENCES messages (id),
  attachments  jsonb  NOT NULL DEFAULT '[]',
  created_at   timestamptz NOT NULL DEFAULT now(),
  edited_at    timestamptz,
  deleted_at   timestamptz,
  PRIMARY KEY (room, seq)
);

-- Edits and reactions share the room's counter, so replay can follow one ordering key.
CREATE TABLE room_events (
  room         text   NOT NULL REFERENCES conversations (name),
  seq          bigint NOT NULL,
  kind         text   NOT NULL CHECK (kind IN ('edit', 'react')),
  id           text   NOT NULL,
  user_id      text   NOT NULL REFERENCES users (id),
  body         jsonb  NOT NULL,
  created_at   timestamptz NOT NULL DEFAULT now(),
  PRIMARY KEY (room, seq)
);

-- Remembered acknowledgements: (user, envelope id) -> what the client was told.
CREATE TABLE acks (
  user_id      text   NOT NULL REFERENCES users (id),
  id           text   NOT NULL,
  room         text,
  seq          bigint,
  created_at   timestamptz NOT NULL DEFAULT now(),
  PRIMARY KEY (user_id, id)
);
CREATE INDEX acks_by_age ON acks (created_at);     -- for the 24 h retention sweep

// Chatterbox storage layer: idempotent persist, fan-out publish, history and unread counts.
import pg from 'pg';
import { createClient } from 'redis';

export const pool = new pg.Pool({
  connectionString: process.env.DATABASE_URL ?? 'postgres://chatterbox:change-me@127.0.0.1:47000/chatterbox',
  max: 10,
});

export class ChatError extends Error {
  constructor(code, message) {
    super(message);
    this.code = code;
  }
}

/**
 * Persist one "message" envelope. Returns the ack payload: { ok, seq, dup, result }.
 * The whole function is one transaction, so a retry with the same envelope id can never
 * create a second message or burn a second sequence number.
 */
export async function persistMessage(userId, env) {
  const { id, room, payload } = env;
  const client = await pool.connect();
  try {
    await client.query('BEGIN');

    // 1. Reserve the (user, id) pair. A concurrent duplicate waits here for the first
    //    transaction to finish, then sees the committed row and inserts nothing.
    //    Read the stored ack on the SAME client: asking the pool for a second connection while
    //    holding one deadlocks as soon as every pooled connection is a waiting duplicate.
    const reserve = await client.query(
      'INSERT INTO acks (user_id, id, room) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING RETURNING id',
      [userId, id, room],
    );
    if (reserve.rowCount === 0) {
      await client.query('ROLLBACK');
      const prior = await client.query('SELECT room, seq FROM acks WHERE user_id = $1 AND id = $2', [userId, id]);
      return { ok: true, seq: Number(prior.rows[0].seq), dup: true, result: {} };
    }

    // 2. Authorisation: only members may post.
    const member = await client.query('SELECT 1 FROM memberships WHERE room = $1 AND user_id = $2', [room, userId]);
    if (member.rowCount === 0) throw new ChatError('forbidden', 'not a member of ' + room);

    // 3. replyTo must be a message of the same room.
    if (payload.replyTo) {
      const parent = await client.query('SELECT 1 FROM messages WHERE id = $1 AND room = $2', [payload.replyTo, room]);
      if (parent.rowCount === 0) throw new ChatError('not_found', 'replyTo is not a message of this room');
    }

    // 4. Take the next sequence number. The row lock serialises writers of one room only.
    const next = await client.query(
      'UPDATE conversations SET last_seq = last_seq + 1 WHERE name = $1 RETURNING last_seq',
      [room],
    );
    if (next.rowCount === 0) throw new ChatError('not_found', 'unknown room');
    const seq = Number(next.rows[0].last_seq);

    // 5. Store the message and complete the remembered ack.
    await client.query(
      `INSERT INTO messages (room, seq, id, sender_id, text, reply_to, attachments)
       VALUES ($1, $2, $3, $4, $5, $6, $7)`,
      [room, seq, id, userId, payload.text, payload.replyTo ?? null, JSON.stringify(payload.attachments ?? [])],
    );
    await client.query('UPDATE acks SET seq = $3 WHERE user_id = $1 AND id = $2', [userId, id, seq]);

    await client.query('COMMIT');
    return { ok: true, seq, dup: false, result: {} };
  } catch (err) {
    await client.query('ROLLBACK').catch(() => {});
    if (err.code === '23505') throw new ChatError('conflict', 'message id already used by another sender'); // unique_violation on messages.id
    throw err;
  } finally {
    client.release();
  }
}

/** Fan-out: publish the committed message to every gateway. Called after COMMIT. */
export async function publishMessage(pub, env, seq, from) {
  const out = {
    v: 1, type: 'message', id: env.id, room: env.room, ts: Date.now(),
    payload: { from, text: env.payload.text, replyTo: env.payload.replyTo ?? null,
               attachments: env.payload.attachments ?? [], seq, edited: false },
  };
  await pub.publish('cb:room:' + env.room, JSON.stringify(out));
  return out;
}

/**
 * Gateway side of the fan-out: one pattern subscription per gateway process, not one per room.
 * deliver(room, envelope) writes to the sockets of that room that live in THIS process.
 */
export async function startFanout(sub, deliver) {
  await sub.pSubscribe('cb:room:*', (raw, channel) => {
    const room = channel.slice('cb:room:'.length);
    deliver(room, JSON.parse(raw));
  });
}

/** Cursor pagination on the sequence number, newest page first. */
export async function history(room, before = null, limit = 50) {
  const { rows } = await pool.query(
    `SELECT m.seq, m.id, m.sender_id, u.name, m.text, m.reply_to, m.edited_at IS NOT NULL AS edited
       FROM messages m JOIN users u ON u.id = m.sender_id
      WHERE m.room = $1 AND m.deleted_at IS NULL AND ($2::bigint IS NULL OR m.seq < $2)
      ORDER BY m.seq DESC LIMIT $3`,
    [room, before, limit + 1],
  );
  const hasMore = rows.length > limit;
  const page = rows.slice(0, limit).reverse();
  return { messages: page, hasMore, next: hasMore ? Number(page[0].seq) : null };
}

/** Unread count per room for a user: everything above the cumulative read marker. */
export async function unreadCounts(userId) {
  const { rows } = await pool.query(
    `SELECT mb.room, c.last_seq - mb.read_seq AS unread
       FROM memberships mb JOIN conversations c ON c.name = mb.room
      WHERE mb.user_id = $1 ORDER BY mb.room`,
    [userId],
  );
  return rows.map((r) => ({ room: r.room, unread: Number(r.unread) }));
}

/** Cumulative receipt: GREATEST keeps the marker monotonic when receipts arrive out of order. */
export async function markRead(userId, room, upTo) {
  const r = await pool.query(
    `UPDATE memberships mb SET read_seq = GREATEST(mb.read_seq, $3), delivered_seq = GREATEST(mb.delivered_seq, $3)
       FROM conversations c
      WHERE mb.room = $2 AND mb.user_id = $1 AND c.name = mb.room AND $3 <= c.last_seq
      RETURNING mb.read_seq`,
    [userId, room, upTo],
  );
  if (r.rowCount === 0) throw new ChatError('bad_request', 'upTo is above the last seq or not a member');
  return Number(r.rows[0].read_seq);
}

/**
 * Alternative sequence source: a Redis counter. Seed it from the database once (SET ... NX), then INCR.
 * Faster than a row lock, but values burnt by failed transactions leave gaps and a lost Redis needs a reseed.
 */
export async function nextSeqFromRedis(redis, room) {
  const key = 'seq:' + room;
  const seeded = await redis.set(key, '0', { NX: true });
  if (seeded === 'OK') {
    const { rows } = await pool.query('SELECT last_seq FROM conversations WHERE name = $1', [room]);
    await redis.set(key, String(rows[0]?.last_seq ?? 0), { XX: true });
  }
  return redis.incr(key);
}

export function redisClient() {
  return createClient({ url: process.env.REDIS_URL ?? 'redis://127.0.0.1:47001' });
}

export async function nextSeqFromRedis(redis, room) {
  const key = 'seq:' + room;
  const seeded = await redis.set(key, '0', { NX: true });
  if (seeded === 'OK') {
    const { rows } = await pool.query('SELECT last_seq FROM conversations WHERE name = $1', [room]);
    await redis.set(key, String(rows[0]?.last_seq ?? 0), { XX: true });
  }
  return redis.incr(key);
}

export async function startFanout(sub, deliver) {
  await sub.pSubscribe('cb:room:*', (raw, channel) => {
    const room = channel.slice('cb:room:'.length);
    deliver(room, JSON.parse(raw));
  });
}

name: g6-chat-architecture
services:
  postgres:
    image: postgres:18-alpine
    container_name: g6-chat-architecture-postgres
    environment:
      POSTGRES_USER: chatterbox
      POSTGRES_PASSWORD: change-me
      POSTGRES_DB: chatterbox
    ports:
      - "127.0.0.1:47000:5432"
    volumes:
      - ./sql/schema.sql:/docker-entrypoint-initdb.d/01-schema.sql:ro
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U chatterbox"]
      interval: 2s
      timeout: 3s
      retries: 30
  redis:
    image: redis:8.8.3-alpine
    container_name: g6-chat-architecture-redis
    ports:
      - "127.0.0.1:47001:6379"
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 2s
      timeout: 3s
      retries: 30

{ "name": "chatterbox-storage-demo", "version": "1.0.0", "private": true, "type": "module",
  "dependencies": { "pg": "8.23.1", "redis": "6.3.0" } }

import assert from 'node:assert/strict';
import { pool, persistMessage, publishMessage, startFanout, history, unreadCounts, markRead, nextSeqFromRedis, redisClient, ChatError } from './store.mjs';

const env = (id, room, text, extra = {}) => ({ v: 1, type: 'message', id, room, ts: 0, payload: { text, replyTo: null, attachments: [], ...extra } });
const ULID = (n) => 'MSG' + String(n).padStart(23, '0');

await pool.query("INSERT INTO users (id, name) VALUES ('u_ana','Ana'),('u_ben','Ben'),('u_eve','Eve') ON CONFLICT DO NOTHING");
await pool.query("INSERT INTO conversations (name, kind) VALUES ('general','public') ON CONFLICT DO NOTHING");
await pool.query("INSERT INTO memberships (room, user_id) VALUES ('general','u_ana'),('general','u_ben') ON CONFLICT DO NOTHING");

// 1. persist + fan-out through Redis with two "gateways"
const pub = redisClient(), subA = redisClient(), subB = redisClient();
await Promise.all([pub.connect(), subA.connect(), subB.connect()]);
const gotA = [], gotB = [];
await startFanout(subA, (room, e) => gotA.push({ room, ...e }));
await startFanout(subB, (room, e) => gotB.push({ room, ...e }));
const ack1 = await persistMessage('u_ana', env(ULID(1), 'general', 'Hello, Chatterbox'));
assert.deepEqual(ack1, { ok: true, seq: 1, dup: false, result: {} });
await publishMessage(pub, env(ULID(1), 'general', 'Hello, Chatterbox'), ack1.seq, { userId: 'u_ana', name: 'Ana' });
await new Promise((r) => setTimeout(r, 200));
assert.equal(gotA.length, 1); assert.equal(gotB.length, 1); assert.equal(gotA[0].payload.seq, 1); assert.equal(gotB[0].room, 'general');
console.log('1 persist+fanout ok', ack1);

// 2. retry with the same id -> dup:true, same seq, no new row
const ack2 = await persistMessage('u_ana', env(ULID(1), 'general', 'Hello, Chatterbox'));
assert.deepEqual(ack2, { ok: true, seq: 1, dup: true, result: {} });
console.log('2 retry ok', ack2);

// 3. 20 concurrent identical retries -> exactly one new row
const same = await Promise.all(Array.from({ length: 20 }, () => persistMessage('u_ana', env(ULID(2), 'general', 'racing'))));
assert.equal(same.filter((a) => !a.dup).length, 1);
assert.ok(same.every((a) => a.seq === 2));
console.log('3 concurrent duplicates ok: 1 new, 19 dup, seq', same[0].seq);

// 4. 50 concurrent distinct messages -> gapless seq 3..52
const many = await Promise.all(Array.from({ length: 50 }, (_, i) => persistMessage(i % 2 ? 'u_ana' : 'u_ben', env(ULID(100 + i), 'general', 'm' + i))));
const seqs = many.map((a) => a.seq).sort((a, b) => a - b);
assert.deepEqual(seqs, Array.from({ length: 50 }, (_, i) => i + 3));
console.log('4 concurrent distinct ok:', seqs[0], '..', seqs.at(-1));

// 5. errors
const expect = async (p, code) => { try { await p; } catch (e) { assert.ok(e instanceof ChatError, String(e)); assert.equal(e.code, code); return; } assert.fail('no error ' + code); };
await expect(persistMessage('u_eve', env(ULID(900), 'general', 'let me in')), 'forbidden');
await expect(persistMessage('u_ben', env(ULID(1), 'general', 'stolen id')), 'conflict');
await expect(persistMessage('u_ana', env(ULID(901), 'general', 'x', { replyTo: 'nope' })), 'not_found');
await expect(persistMessage('u_ana', env(ULID(902), 'nowhere', 'x')), 'forbidden');
const lost = await pool.query("SELECT count(*)::int AS n FROM acks WHERE id IN ($1,$2,$3)", [ULID(900), ULID(901), ULID(902)]);
assert.equal(lost.rows[0].n, 0); // failed transactions leave no ack behind
const last = await pool.query("SELECT last_seq FROM conversations WHERE name='general'");
assert.equal(Number(last.rows[0].last_seq), 52);
console.log('5 errors ok; failed attempts burnt no seq, last_seq =', last.rows[0].last_seq);

// 6. history, unread, receipts
const h1 = await history('general', null, 20);
assert.equal(h1.messages.length, 20); assert.equal(h1.hasMore, true); assert.equal(h1.messages.at(-1).seq, '52');
const h2 = await history('general', h1.next, 20);
assert.equal(h2.messages.at(-1).seq, String(h1.next - 1));
console.log('6a history ok next=', h1.next, 'then', h2.next);
console.log('6b unread', await unreadCounts('u_ben'));
assert.equal(await markRead('u_ben', 'general', 40), 40);
assert.equal(await markRead('u_ben', 'general', 10), 40); // older receipt cannot move marker back
console.log('6c unread after read 40', await unreadCounts('u_ben'));
await expect(markRead('u_ben', 'general', 999), 'bad_request');

// 7. redis counter
const r = redisClient(); await r.connect();
const a = await nextSeqFromRedis(r, 'general'); const b = await nextSeqFromRedis(r, 'general');
assert.equal(a, 53); assert.equal(b, 54);
console.log('7 redis seq ok', a, b);

await Promise.all([subA.quit(), subB.quit(), pub.quit(), r.quit()]); await pool.end();
console.log('ALL OK');

docker compose up -d --wait
npm install
node demo.mjs
docker compose down -v

1 persist+fanout ok { ok: true, seq: 1, dup: false, result: {} }
2 retry ok { ok: true, seq: 1, dup: true, result: {} }
3 concurrent duplicates ok: 1 new, 19 dup, seq 2
4 concurrent distinct ok: 3 .. 52
5 errors ok; failed attempts burnt no seq, last_seq = 52
6a history ok next= 33 then 13
6b unread [ { room: 'general', unread: 52 } ]
6c unread after read 40 [ { room: 'general', unread: 12 } ]
7 redis seq ok 53 54
ALL OK