Tutorial: a chat with presence

import express from 'express';
import { ThuumClient } from '@thuum/client';

const app = express();
const thuum = new ThuumClient({ baseUrl: process.env.THUUM_API_URL! });
const ROOMS = new Set(['general', 'random', 'help']);

app.get('/api/thuum-token', requireUser, async (req, res) => {
	const { token } = await thuum.exchangeToken('acme', process.env.CHAT_BROWSER_KEY!);
	res.set('cache-control', 'no-store').json({
		baseUrl: process.env.THUUM_API_URL,
		domain: 'acme',
		token
	});
});

app.get('/api/rooms/:room/messages', requireUser, async (req, res) => {
	if (!ROOMS.has(req.params.room)) return res.status(404).end();
	const { token } = await thuum.exchangeToken('acme', process.env.CHAT_SERVER_KEY!);
	const events = await thuum
		.domain('acme', token)
		.eventBroker.data.list(`chat:${req.params.room}`, {
			type: 'message',
			since: req.query.since ? Number(req.query.since) : undefined,
			limit: 100
		});
	res.set('cache-control', 'no-store').json(events);
});

import type { BrokerEvent } from '@thuum/client';

export interface ChatMessage {
	id: string;
	ts: number;
	clientId: string;
	name: string;
	text: string;
}

export function toChatMessage(event: BrokerEvent): ChatMessage | null {
	if (event.type !== 'message') return null;
	const data = event.data as Record<string, unknown> | null;
	if (!data || typeof data !== 'object') return null;
	const { clientId, name, text } = data;
	if (typeof clientId !== 'string' || typeof name !== 'string' || typeof text !== 'string')
		return null;
	if (!text.trim() || text.length > 1000 || name.length > 32) return null;
	return { id: event.id, ts: event.ts, clientId, name, text };
}

import { EventBrokerSocket, StateBrokerSocket, type BrokerEvent } from '@thuum/client';

const HEARTBEAT_MS = 20_000;
const OFFLINE_AFTER_MS = 45_000;

export class Room {
	messages: ChatMessage[] = [];
	peers = new Map<string, { name: string; typing: boolean; seenAt: number }>();

	#eb!: EventBrokerSocket;
	#sb!: StateBrokerSocket;
	#heartbeat?: ReturnType<typeof setInterval>;
	#typing = false;

	constructor(
		private readonly room: string,
		private readonly clientId: string,
		private name: string,
		private readonly onUpdate: () => void
	) {}

	get #chatTopic() {
		return `chat:${this.room}`;
	}
	get #presenceTopic() {
		return `presence:${this.room}`;
	}

	async join(): Promise<void> {
		const grant = async () => (await fetch('/api/thuum-token', { cache: 'no-store' })).json();
		this.#eb = new EventBrokerSocket(await grant());
		this.#sb = new StateBrokerSocket(await grant());

		this.#eb.onMessage((event) => {
			const message = event.topic === this.#chatTopic ? toChatMessage(event) : null;
			if (message) this.#add([message]);
		});
		this.#sb.onChange(({ topic, key, value }) => {
			if (topic !== this.#presenceTopic) return;
			const peer = value as { name?: unknown; typing?: unknown } | null;
			if (peer && typeof peer.name === 'string') {
				this.peers.set(key, { name: peer.name, typing: peer.typing === true, seenAt: Date.now() });
			} else {
				this.peers.delete(key); // deleted, or not a presence value
			}
			this.onUpdate();
		});

		await Promise.all([this.#eb.ready, this.#sb.ready]);
		await Promise.all([this.#eb.sub(this.#chatTopic), this.#sb.sub(this.#presenceTopic)]);

		const history: BrokerEvent[] = await (await fetch(`/api/rooms/${this.room}/messages`)).json();
		this.#add(history.map(toChatMessage).filter((m): m is ChatMessage => m !== null));

		await this.#announce();
		this.#heartbeat = setInterval(() => void this.#announce(), HEARTBEAT_MS);
	}

	/** Shown when it comes back from the broker, so every screen has the same order. */
	async send(text: string): Promise<void> {
		await this.#eb.pub(this.#chatTopic, 'message', {
			clientId: this.clientId,
			name: this.name,
			text
		});
		this.setTyping(false);
	}

	setTyping(typing: boolean): void {
		if (typing === this.#typing) return;
		this.#typing = typing;
		void this.#announce();
	}

	/** Peers seen recently: a vanished tab disappears without waiting for the TTL. */
	online(now = Date.now()) {
		return [...this.peers.entries()].filter(([, p]) => now - p.seenAt < OFFLINE_AFTER_MS);
	}

	async leave(): Promise<void> {
		clearInterval(this.#heartbeat);
		await this.#sb.del(this.#presenceTopic, this.clientId).catch(() => {});
		this.#eb.close();
		this.#sb.close();
	}

	async #announce(): Promise<void> {
		await this.#sb
			.set(this.#presenceTopic, this.clientId, { name: this.name, typing: this.#typing })
			.catch(() => {});
	}

	#add(incoming: ChatMessage[]): void {
		const known = new Set(this.messages.map((m) => m.id));
		const fresh = incoming.filter((m) => !known.has(m.id));
		if (!fresh.length) return;
		this.messages = [...this.messages, ...fresh].sort(
			(a, b) => a.ts - b.ts || (a.id < b.id ? -1 : 1)
		);
		this.onUpdate();
	}
}

const room = new Room('general', crypto.randomUUID(), currentUser.name, render);
await room.join();

input.addEventListener('input', () => room.setTyping(input.value.length > 0));
form.addEventListener('submit', async (e) => {
	e.preventDefault();
	await room.send(input.value);
	input.value = '';
});
addEventListener('pagehide', () => void room.leave());
setInterval(render, 5_000); // ages presence entries