Presence, Typing and Receipts

import type { Redis } from "ioredis";

export type Status = "online" | "away" | "offline";
export interface PresenceEvent { userId: string; status: Status; lastSeen?: number; room?: string }

export class PresenceStore {
  readonly redis: Redis;
  readonly heartbeatMs: number;
  readonly graceMs: number;
  readonly ttlMs: number;

  constructor(redis: Redis, opts: { heartbeatMs?: number; graceMs?: number } = {}) {
    this.redis = redis;
    this.heartbeatMs = opts.heartbeatMs ?? 20_000;
    this.graceMs = opts.graceMs ?? 15_000;
    this.ttlMs = 2 * this.heartbeatMs + 2_000; // the server closes a silent socket after 2 x heartbeatMs
  }

  // {user} is a hash tag: every key of one user lives in the same Redis Cluster slot.
  private key(userId: string, part: string): string { return `pres:{${userId}}:${part}`; }

  /** Called for hello, for every presence envelope (the heartbeat) and for any other frame. */
  async beat(userId: string, connId: string, status: "online" | "away"): Promise<void> {
    const now = Date.now();
    await this.redis.pipeline()
      .set(this.key(userId, `c:${connId}`), status, "PX", this.ttlMs)
      .sadd(this.key(userId, "conns"), connId)
      .pexpire(this.key(userId, "conns"), 600_000)
      .set(this.key(userId, "seen"), now)
      .zadd("pres:due", "GT", now + this.ttlMs + this.graceMs, userId)
      .exec();
    await this.announce(userId);
  }

  /** The aggregate over all devices: online if any is online, away if all are away, else offline. */
  async aggregate(userId: string): Promise<Status> {
    const ids = await this.redis.smembers(this.key(userId, "conns"));
    if (ids.length === 0) return "offline";
    const values = await this.redis.mget(ids.map((id) => this.key(userId, `c:${id}`)));
    const dead = ids.filter((_, i) => values[i] === null);
    if (dead.length > 0) await this.redis.srem(this.key(userId, "conns"), ...dead); // lazy clean-up
    const live = values.filter((v) => v !== null);
    return live.length === 0 ? "offline" : live.includes("online") ? "online" : "away";
  }

  /** Publishes online or away when the published status changes. Offline is announced only by sweep(). */
  private async announce(userId: string): Promise<void> {
    const status = await this.aggregate(userId);
    if (status === "offline") return;
    const previous = await this.redis.set(this.key(userId, "pub"), status, "GET");
    if (previous !== status) await this.publish({ userId, status });
  }

  async close(userId: string, connId: string): Promise<void> {
    await this.redis.pipeline().del(this.key(userId, `c:${connId}`)).srem(this.key(userId, "conns"), connId).exec();
    if ((await this.aggregate(userId)) === "offline") {
      await this.redis.zadd("pres:due", Date.now() + this.graceMs, userId); // re-check after the grace period
    } else {
      await this.announce(userId);
    }
  }

  /** Run every second on every node. Handles clean closes (after the grace) and silent drops (after TTL + grace). */
  async sweep(): Promise<void> {
    const due = await this.redis.zrangebyscore("pres:due", "-inf", Date.now(), "LIMIT", 0, 100);
    for (const userId of due) {
      if ((await this.redis.zrem("pres:due", userId)) === 0) continue; // another node took it
      if ((await this.aggregate(userId)) !== "offline") continue;       // came back during the grace period
      const previous = await this.redis.set(this.key(userId, "pub"), "offline", "GET");
      if (previous !== null && previous !== "offline") {
        const seen = await this.redis.get(this.key(userId, "seen"));
        await this.publish({ userId, status: "offline", lastSeen: seen ? Number(seen) : undefined });
      }
    }
  }

  /** Used by join: tell one room about the user's current status. */
  async announceTo(userId: string, room: string): Promise<void> {
    const status = (await this.published([userId]))[userId];
    await this.publish({ userId, status, room });
  }

  async published(userIds: string[]): Promise<Record<string, Status>> {
    const values = userIds.length ? await this.redis.mget(userIds.map((u) => this.key(u, "pub"))) : [];
    return Object.fromEntries(userIds.map((u, i) => [u, (values[i] as Status | null) ?? "offline"]));
  }

  private async publish(event: PresenceEvent): Promise<void> {
    await this.redis.publish("presence", JSON.stringify(event));
  }
}

import { Redis } from "ioredis";
import { PresenceStore } from "./presence.ts";

const redis = new Redis(process.env.REDIS_URL ?? "redis://127.0.0.1:6379");
const sub = new Redis(process.env.REDIS_URL ?? "redis://127.0.0.1:6379");
const store = new PresenceStore(redis, { heartbeatMs: 1000, graceMs: 1500 }); // short values for the demo
const t0 = Date.now();
const log = (text: string) => console.log(`${String(Date.now() - t0).padStart(5)} ms  ${text}`);
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

await redis.flushdb();
await sub.subscribe("presence");
sub.on("message", (_channel, raw) => log(`event ${raw}`));
const sweeper = setInterval(() => void store.sweep(), 250);

log("laptop: online");        await store.beat("alice", "laptop", "online");
log("phone: away");           await store.beat("alice", "phone", "away");        // no event, aggregate stays online
log("laptop: away");          await store.beat("alice", "laptop", "away");       // all devices away -> away
log("phone: online");         await store.beat("alice", "phone", "online");      // online again
log("laptop closes");         await store.close("alice", "laptop");              // no event, the phone is still there
log("phone closes");          await store.close("alice", "phone");               // offline only after the grace period
await sleep(2200);
log("bob: one beat, then silence"); await store.beat("bob", "tab", "online");
await sleep(4500);
log("carol: closes, reconnects within the grace");
await store.beat("carol", "tab", "online"); await store.close("carol", "tab"); await sleep(500);
await store.beat("carol", "tab2", "online"); await sleep(2500);
log(`published: ${JSON.stringify(await store.published(["alice", "bob", "carol"]))}`);
clearInterval(sweeper); redis.disconnect(); sub.disconnect();

    7 ms  laptop: online
   11 ms  event {"userId":"alice","status":"online"}
   11 ms  phone: away
   12 ms  laptop: away
   14 ms  phone: online
   14 ms  event {"userId":"alice","status":"away"}
   15 ms  laptop closes
   15 ms  event {"userId":"alice","status":"online"}
   17 ms  phone closes
 1772 ms  event {"userId":"alice","status":"offline","lastSeen":1792879942898}
 2218 ms  bob: one beat, then silence
 2233 ms  event {"userId":"bob","status":"online"}
 6733 ms  carol: closes, reconnects within the grace
 6741 ms  event {"userId":"carol","status":"online"}
 7799 ms  event {"userId":"bob","status":"offline","lastSeen":1792879945102}
 9757 ms  published: {"alice":"offline","bob":"offline","carol":"online"}

> SET pres:{alice}:c:laptop online PX 42000
OK
> SET pres:{alice}:pub online GET
(nil)
> SET pres:{alice}:pub away GET
"online"
> SET typing:{general}:bob 1 PX 2500 NX
OK
> SET typing:{general}:bob 1 PX 2500 NX
(nil)
> ZADD pres:due GT 1792879600000 alice
(integer) 0

export const SEND_EVERY_MS = 3_000; // protocol: at most one typing frame per 3 s
export const IDLE_STOP_MS = 4_000;  // stop after this long without a keystroke
export const EXPIRE_MS = 5_000;     // receivers drop a "start" after 5 s

/** Sending side: call keystroke() on every input event, stopped() on send, blur or an empty field. */
export class TypingSender {
  private lastStart = -Infinity;
  private idle: ReturnType<typeof setTimeout> | undefined;
  private active = false;
  readonly emit: (state: "start" | "stop") => void;

  constructor(emit: (state: "start" | "stop") => void) { this.emit = emit; }

  keystroke(): void {
    const now = Date.now();
    if (!this.active || now - this.lastStart >= SEND_EVERY_MS) {
      this.emit("start");
      this.lastStart = now;
      this.active = true;
    }
    clearTimeout(this.idle);
    this.idle = setTimeout(() => this.stopped(), IDLE_STOP_MS);
  }

  stopped(): void {
    clearTimeout(this.idle);
    if (this.active) this.emit("stop");
    this.active = false;
    this.lastStart = -Infinity;
  }
}

/** Receiving side: one expiry timer per user, so a lost "stop" never leaves the indicator stuck. */
export class TypingView {
  private timers = new Map<string, ReturnType<typeof setTimeout>>();
  readonly onChange: (typingUserIds: string[]) => void;

  constructor(onChange: (typingUserIds: string[]) => void) { this.onChange = onChange; }

  apply(userId: string, state: "start" | "stop"): void {
    clearTimeout(this.timers.get(userId));
    if (state === "start") this.timers.set(userId, setTimeout(() => this.apply(userId, "stop"), EXPIRE_MS));
    else this.timers.delete(userId);
    this.onChange([...this.timers.keys()]);
  }

  /** Call when a message from that user arrives. */
  messageFrom(userId: string): void { if (this.timers.has(userId)) this.apply(userId, "stop"); }
}

import type { Redis } from "ioredis";

export type Kind = "delivered" | "read";
export type Outcome = "advanced" | "stale" | "bad_request";

// KEYS[1] = rcpt:{room} (hash), KEYS[2] = seq:{room} (the room's last seq)
// ARGV = kind, upTo, userId. Returns 1 advanced, 0 stale, -1 beyond the last seq.
const ADVANCE = `
local last = tonumber(redis.call('GET', KEYS[2]) or '0')
local upTo = tonumber(ARGV[2])
if upTo > last then return -1 end
local field = ARGV[3] .. ':' .. ARGV[1]
local current = tonumber(redis.call('HGET', KEYS[1], field) or '0')
if upTo <= current then return 0 end
redis.call('HSET', KEYS[1], field, upTo)
if ARGV[1] == 'read' then
  local delivered = ARGV[3] .. ':delivered'
  if upTo > tonumber(redis.call('HGET', KEYS[1], delivered) or '0') then
    redis.call('HSET', KEYS[1], delivered, upTo)
  end
end
return 1`;

export class Receipts {
  readonly redis: Redis;
  constructor(redis: Redis) { this.redis = redis; }

  /** Cursors only move forward, so a duplicate or reordered receipt changes nothing. */
  async advance(room: string, userId: string, kind: Kind, upTo: number): Promise<Outcome> {
    const code = Number(await this.redis.eval(ADVANCE, 2, `rcpt:{${room}}`, `seq:{${room}}`, kind, upTo, userId));
    return code === 1 ? "advanced" : code === 0 ? "stale" : "bad_request";
  }

  /** Messages after the user's read cursor. msgs:{room} holds the seq of every message envelope (not edits or reactions). */
  async unread(room: string, userId: string): Promise<number> {
    const cursor = (await this.redis.hget(`rcpt:{${room}}`, `${userId}:read`)) ?? "0";
    return this.redis.zcount(`msgs:{${room}}`, `(${cursor}`, "+inf");
  }

  /** Who has read message `seq`: the data behind a "seen by" list. */
  async readers(room: string, seq: number): Promise<string[]> {
    const all = await this.redis.hgetall(`rcpt:{${room}}`);
    return Object.entries(all).filter(([f, v]) => f.endsWith(":read") && Number(v) >= seq).map(([f]) => f.slice(0, -5));
  }
}

delivered 41   advanced
read 43        advanced
read 42 (late) stale
read 99        bad_request
cursors        { 'bob:delivered': '43', 'bob:read': '43' }
bob unread     1 | carol unread 4

import type { Redis } from "ioredis";

export interface Privacy {
  presence: "everyone" | "contacts" | "nobody"; // who sees online/away; "nobody" is invisible mode
  lastSeen: "everyone" | "contacts" | "nobody";
  readReceipts: boolean;                         // reciprocal: off means you neither send nor see read receipts
  typing: boolean;
}
export const DEFAULTS: Privacy = { presence: "everyone", lastSeen: "contacts", readReceipts: true, typing: true };

export async function loadPrivacy(redis: Redis, userId: string): Promise<Privacy> {
  const h = await redis.hgetall(`priv:{${userId}}`);
  return {
    presence: (h.presence as Privacy["presence"]) ?? DEFAULTS.presence,
    lastSeen: (h.lastSeen as Privacy["lastSeen"]) ?? DEFAULTS.lastSeen,
    readReceipts: (h.readReceipts ?? String(DEFAULTS.readReceipts)) === "true",
    typing: (h.typing ?? String(DEFAULTS.typing)) === "true",
  };
}

const allowed = (rule: Privacy["presence"], isContact: boolean) => rule === "everyone" || (rule === "contacts" && isContact);

/** Applied for every recipient at delivery time, never trusted from the client. */
export function visibleStatus(from: Privacy, isContact: boolean, status: string): string {
  return allowed(from.presence, isContact) ? status : "offline";
}
export function visibleLastSeen(from: Privacy, isContact: boolean, lastSeen: number | undefined): number | undefined {
  return allowed(from.lastSeen, isContact) && allowed(from.presence, isContact) ? lastSeen : undefined;
}
export function mayDeliver(kind: "typing" | "delivered" | "read", from: Privacy, to: Privacy): boolean {
  if (kind === "typing") return from.typing;
  if (kind === "read") return from.readReceipts && to.readReceipts;
  return true;
}

// Excerpt of class ChatNode in handlers.ts: the relays. Conn, ProtocolError and the imports come from the same file.
  async onTyping(c: Conn, room: string, state: "start" | "stop"): Promise<void> {
    if (!c.rooms.has(room)) return;                          // fire-and-forget: no error frame
    if (state === "start") {
      const first = await this.redis.set(`typing:{${room}}:${c.userId}`, "1", "PX", 2500, "NX");
      if (first === null) return;                            // more than one start per 2.5 s: drop it
    }
    await this.fanout(room, "typing", { userId: c.userId, state }, c.userId);
  }

  async onReceipt(c: Conn, room: string, kind: Kind, upTo: number): Promise<void> {
    if (!c.rooms.has(room)) return;
    const outcome = await this.receipts.advance(room, c.userId, kind, upTo);
    if (outcome === "bad_request") throw new ProtocolError("bad_request", "upTo is beyond the last seq of the room");
    if (outcome === "advanced") await this.fanout(room, "receipt", { userId: c.userId, kind, upTo }, c.userId);
  }

  async fanout(room: string, type: string, payload: { userId: string; [k: string]: unknown }, except: string): Promise<void> {
    await this.redis.publish("fanout", JSON.stringify({ room, type, payload, except }));
  }

  private async deliverFanout(m: { room: string; type: string; payload: { userId: string; kind?: Kind; state?: string }; except: string }) {
    const sender = await loadPrivacy(this.redis, m.payload.userId);
    for (const c of this.local.get(m.room) ?? []) {
      if (c.userId === m.except) continue;
      const kind = m.type === "typing" ? "typing" : (m.payload.kind as "delivered" | "read");
      if (mayDeliver(kind, sender, await loadPrivacy(this.redis, c.userId))) c.send(m.type, m.room, m.payload);
    }
  }