Design a Chat System

-- One row per (user, channel). A 10,000-member channel does not get 10,000
-- read rows per message - it gets 10,000 rows in total, updated in place.
CREATE TABLE read_state (
  user_id              BIGINT NOT NULL,
  channel_id           BIGINT NOT NULL,
  last_read_message_id BIGINT NOT NULL,
  mention_count        INT    NOT NULL DEFAULT 0,
  PRIMARY KEY (user_id, channel_id)
);

-- Unread count is a bounded range count above the cursor, capped so a channel
-- nobody has opened for a year does not scan a year of messages.
SELECT count(*) FROM (
  SELECT 1 FROM messages
  WHERE channel_id = $1
    AND id > (SELECT last_read_message_id FROM read_state
              WHERE user_id = $2 AND channel_id = $1)
  LIMIT 100
) capped;

-- Marking read is one idempotent upsert that never moves backwards, so a
-- retrying or out-of-order client cannot un-read a channel.
INSERT INTO read_state (user_id, channel_id, last_read_message_id)
VALUES ($2, $1, $3)
ON CONFLICT (user_id, channel_id) DO UPDATE
SET last_read_message_id = GREATEST(read_state.last_read_message_id, excluded.last_read_message_id);

// Called after the message is durably written. Delivery here is best-effort:
// the stored channel is what a client reads on reconnect, so a lost frame is
// a slower render, not a lost message.
async function fanout(channelId, message) {
  const memberIds = await members.list(channelId);      // page this for huge channels
  const routes = await sessions.lookup(memberIds);      // Map<userId, [{node, sessionId}]>

  const byNode = new Map();
  const offline = [];
  for (const userId of memberIds) {
    const live = routes.get(userId) ?? [];
    if (live.length === 0) { offline.push(userId); continue; }
    for (const s of live) {
      if (!byNode.has(s.node)) byNode.set(s.node, []);
      byNode.get(s.node).push(s.sessionId);
    }
  }

  // One RPC per gateway node, not per session: a 50,000-member channel becomes
  // tens of sends instead of 50,000. allSettled because one dead node must not
  // stop delivery to the others.
  await Promise.allSettled(
    [...byNode].map(([node, sessionIds]) =>
      gatewayRpc(node, 'DISPATCH', { sessionIds, t: 'MESSAGE_CREATE', d: message })),
  );

  // Push is a separate queue with its own retries, rate limits and provider
  // failures - it must never be in line with the socket delivery.
  if (offline.length) {
    await pushQueue.enqueue(offline.map((userId) => ({
      userId, channelId, messageId: message.id, dedupeKey: `${message.id}:${userId}`,
    })));
  }
}

// Presence is the highest-volume signal in a chat product: it changes on every
// tab focus and every network blip, and it fans out to everyone who can see you.
// So it is memory-only, expiring, and never written to the message store.
const PRESENCE_TTL = 45; // seconds; longer than the client heartbeat interval

async function onHeartbeat(userId, sessionId, status) {
  // A hash per user holds one field per device, and the whole key expires on its
  // own - so a user whose last device hard-crashes goes offline with nobody
  // cleaning up after it.
  await redis.hSet(`presence:${userId}`, sessionId, status);
  await redis.expire(`presence:${userId}`, PRESENCE_TTL);

  // Fan out only on a transition, not on every heartbeat. Without this check,
  // one user with five devices generates a broadcast every few seconds.
  const effective = status === 'online' ? 'online' : await rollUp(userId);
  if (effective !== lastBroadcast.get(userId)) {
    lastBroadcast.set(userId, effective);
    await fanoutPresence(userId, effective);   // to that user's visible peers only
  }
}