প্রজেক্ট: চ্যাট ও প্রেজেন্স সিস্টেম

import { WebSocketServer, WebSocket } from 'ws';
import { createClient, RedisClientType } from 'redis';
import crypto from 'node:crypto';

// ===========================================
// 1. CONFIG & TYPES
// ===========================================
const NODE_ID = process.env.NODE_ID || `gw-baghdad-${process.pid}`;
const PORT = parseInt(process.env.PORT || '8080', 10);
const HEARTBEAT_INTERVAL_MS = 30_000;
const HEARTBEAT_TIMEOUT_MS = 90_000;
const REGISTRY_TTL_SECONDS = 120;
const MAX_OUTBOUND_QUEUE = 512;

type Frame =
	| { t: 'hello'; userId: string; deviceId: string; token: string }
	| { t: 'ping'; ts: number }
	| { t: 'send'; convId: string; clientMsgId: string; body: string }
	| { t: 'ack'; convId: string; seq: number; kind: 'delivered' | 'read' }
	| { t: 'resume'; convId: string; fromSeq: number }
	| { t: 'sub_presence'; userIds: string[] };

interface Conn {
	id: string;
	userId: string;
	deviceId: string;
	socket: WebSocket;
	lastSeenAt: number;
	// Outbound frames buffered while the socket drains. Bounded — see back-pressure.
	pending: string[];
	closed: boolean;
}

// ===========================================
// 2. CONNECTION REGISTRY (Redis-backed)
// ===========================================
// Maps userId -> set of "nodeId:deviceId". A user may hold several devices
// on several nodes at once, so the value is a hash, not a scalar.
class ConnectionRegistry {
	constructor(private redis: RedisClientType) {}

	private key(userId: string): string {
		return `conn:user:${userId}`;
	}

	async register(userId: string, deviceId: string, nodeId: string): Promise<void> {
		const k = this.key(userId);
		await this.redis
			.multi()
			.hSet(k, deviceId, `${nodeId}:${Date.now()}`)
			.expire(k, REGISTRY_TTL_SECONDS)
			.exec();
	}

	// Called on every heartbeat. Refreshes the TTL so a crashed node's entries
	// expire on their own without any cleanup job.
	async refresh(userId: string, deviceId: string, nodeId: string): Promise<void> {
		await this.register(userId, deviceId, nodeId);
	}

	async unregister(userId: string, deviceId: string): Promise<void> {
		const k = this.key(userId);
		await this.redis.hDel(k, deviceId);
	}

	// Returns the node that owns each of the user's live devices.
	async lookup(userId: string): Promise<Array<{ deviceId: string; nodeId: string }>> {
		const entries = await this.redis.hGetAll(this.key(userId));
		return Object.entries(entries).map(([deviceId, value]) => ({
			deviceId,
			nodeId: value.split(':')[0]
		}));
	}
}

// ===========================================
// 3. GATEWAY
// ===========================================
class Gateway {
	private conns = new Map<string, Conn>();
	// userId -> connection ids (multi-device)
	private byUser = new Map<string, Set<string>>();
	private sweeper: ReturnType<typeof setInterval>;

	constructor(
		private registry: ConnectionRegistry,
		private bus: NodeBus,
		private msgClient: MessageServiceClient,
		private presence: PresenceClient
	) {
		this.sweeper = setInterval(() => this.sweepDeadConnections(), HEARTBEAT_INTERVAL_MS);
		// Frames routed to this node by fan-out workers arrive here.
		this.bus.onInbound(NODE_ID, (userId, payload) => this.deliverLocal(userId, payload));
	}

	async onConnection(socket: WebSocket): Promise<void> {
		// The socket is anonymous until the hello frame authenticates it.
		const authTimer = setTimeout(() => socket.close(4401, 'auth timeout'), 10_000);
		let conn: Conn | null = null;

		socket.on('message', async (raw) => {
			let frame: Frame;
			try {
				frame = JSON.parse(raw.toString()) as Frame;
			} catch {
				socket.close(4400, 'malformed frame');
				return;
			}

			if (!conn) {
				if (frame.t !== 'hello') {
					socket.close(4401, 'hello expected');
					return;
				}
				const userId = await verifyToken(frame.token, frame.userId);
				if (!userId) {
					socket.close(4403, 'bad token');
					return;
				}
				clearTimeout(authTimer);
				conn = this.attach(userId, frame.deviceId, socket);
				await this.registry.register(userId, frame.deviceId, NODE_ID);
				await this.presence.heartbeat(userId);
				this.send(conn, { t: 'hello_ok', nodeId: NODE_ID, serverTime: Date.now() });
				return;
			}

			conn.lastSeenAt = Date.now();
			await this.handleFrame(conn, frame);
		});

		socket.on('close', async () => {
			clearTimeout(authTimer);
			if (conn) await this.detach(conn);
		});

		socket.on('error', () => {
			if (conn) void this.detach(conn);
		});
	}

	private attach(userId: string, deviceId: string, socket: WebSocket): Conn {
		const conn: Conn = {
			id: crypto.randomUUID(),
			userId,
			deviceId,
			socket,
			lastSeenAt: Date.now(),
			pending: [],
			closed: false
		};
		this.conns.set(conn.id, conn);
		let set = this.byUser.get(userId);
		if (!set) {
			set = new Set();
			this.byUser.set(userId, set);
		}
		set.add(conn.id);
		return conn;
	}

	private async detach(conn: Conn): Promise<void> {
		if (conn.closed) return;
		conn.closed = true;
		this.conns.delete(conn.id);
		const set = this.byUser.get(conn.userId);
		if (set) {
			set.delete(conn.id);
			if (set.size === 0) this.byUser.delete(conn.userId);
		}
		await this.registry.unregister(conn.userId, conn.deviceId);
		// Do NOT mark offline here. The TTL in the presence service decides that,
		// so a two-second network blip does not flap the green dot.
	}

	private async handleFrame(conn: Conn, frame: Frame): Promise<void> {
		switch (frame.t) {
			case 'ping': {
				await this.registry.refresh(conn.userId, conn.deviceId, NODE_ID);
				await this.presence.heartbeat(conn.userId);
				this.send(conn, { t: 'pong', ts: frame.ts, serverTime: Date.now() });
				return;
			}
			case 'send': {
				const result = await this.msgClient.send({
					convId: frame.convId,
					senderId: conn.userId,
					clientMsgId: frame.clientMsgId,
					body: frame.body
				});
				// Ack carries the assigned seq so the sender can order its own view.
				this.send(conn, {
					t: 'send_ok',
					clientMsgId: frame.clientMsgId,
					convId: frame.convId,
					seq: result.seq,
					messageId: result.messageId
				});
				return;
			}
			case 'ack': {
				await this.msgClient.recordReceipt({
					convId: frame.convId,
					userId: conn.userId,
					seq: frame.seq,
					kind: frame.kind
				});
				return;
			}
			case 'resume': {
				const batch = await this.msgClient.backfill(frame.convId, conn.userId, frame.fromSeq, 200);
				for (const m of batch.messages) {
					this.send(conn, { t: 'msg', ...m });
				}
				this.send(conn, {
					t: 'resume_done',
					convId: frame.convId,
					upToSeq: batch.upToSeq,
					hasMore: batch.hasMore
				});
				return;
			}
			case 'sub_presence': {
				// Scope guard: a client may not subscribe to unbounded user sets.
				const scoped = frame.userIds.slice(0, 500);
				await this.presence.subscribe(conn.userId, NODE_ID, scoped);
				const snapshot = await this.presence.snapshot(scoped);
				this.send(conn, { t: 'presence_snapshot', states: snapshot });
				return;
			}
		}
	}

	// Called by the fan-out worker via the node bus.
	private deliverLocal(userId: string, payload: unknown): void {
		const ids = this.byUser.get(userId);
		if (!ids) return; // user moved or disconnected; worker falls back to push
		for (const id of ids) {
			const conn = this.conns.get(id);
			if (conn) this.send(conn, payload);
		}
	}

	// Bounded write with back-pressure: a client that cannot keep up is
	// disconnected rather than allowed to consume the node's heap.
	private send(conn: Conn, payload: unknown): void {
		if (conn.closed) return;
		if (conn.socket.bufferedAmount > 1_000_000 || conn.pending.length > MAX_OUTBOUND_QUEUE) {
			console.warn(`[BACKPRESSURE] dropping slow consumer user=${conn.userId}`);
			conn.socket.close(4408, 'too slow');
			void this.detach(conn);
			return;
		}
		conn.socket.send(JSON.stringify(payload));
	}

	private sweepDeadConnections(): void {
		const cutoff = Date.now() - HEARTBEAT_TIMEOUT_MS;
		for (const conn of this.conns.values()) {
			if (conn.lastSeenAt < cutoff) {
				console.log(`[SWEEP] stale conn user=${conn.userId} device=${conn.deviceId}`);
				conn.socket.close(4409, 'heartbeat timeout');
				void this.detach(conn);
			}
		}
	}

	async shutdown(): Promise<void> {
		clearInterval(this.sweeper);
		// Staggered close so 10k clients do not reconnect in the same millisecond.
		const all = [...this.conns.values()];
		for (let i = 0; i < all.length; i++) {
			const delay = Math.floor((i / all.length) * 15_000);
			setTimeout(() => all[i].socket.close(1001, 'server restart'), delay);
		}
	}
}

// ===========================================
// 4. WIRING
// ===========================================
async function main(): Promise<void> {
	const redis = createClient({ url: process.env.REDIS_URL }) as RedisClientType;
	await redis.connect();

	const registry = new ConnectionRegistry(redis);
	const gateway = new Gateway(
		registry,
		new NodeBus(redis),
		new MessageServiceClient(process.env.MESSAGE_SERVICE_URL!),
		new PresenceClient(redis)
	);

	const wss = new WebSocketServer({ port: PORT, maxPayload: 64 * 1024 });
	wss.on('connection', (socket) => void gateway.onConnection(socket));
	console.log(`gateway ${NODE_ID} listening on :${PORT}`);

	process.on('SIGTERM', async () => {
		await gateway.shutdown();
		wss.close();
		await redis.quit();
	});
}

void main();

import { Pool } from 'pg';
import { RedisClientType } from 'redis';
import { Kafka, Producer } from 'kafkajs';
import crypto from 'node:crypto';

// ===========================================
// 1. TYPES
// ===========================================
type FanoutMode = 'write' | 'hybrid' | 'read';

const SMALL_GROUP_MAX = 200;
const HYBRID_GROUP_MAX = 2000;

interface SendRequest {
	convId: string;
	senderId: string;
	clientMsgId: string;
	body: string;
}

interface SendResult {
	messageId: string;
	seq: number;
	deduped: boolean;
}

interface StoredMessage {
	convId: string;
	seq: number;
	messageId: string;
	senderId: string;
	body: string;
	createdAt: string;
}

// ===========================================
// 2. SEQUENCE ALLOCATOR
// ===========================================
// Redis is the fast path; the database is the authority. If the Redis counter
// is cold or was lost, it is re-seeded from the persisted maximum. A unique
// index on (conv_id, seq) is the final guard against a duplicate allocation.
class SeqAllocator {
	constructor(
		private redis: RedisClientType,
		private db: Pool
	) {}

	async next(convId: string): Promise<number> {
		const key = `seq:conv:${convId}`;
		const exists = await this.redis.exists(key);

		if (!exists) {
			const { rows } = await this.db.query<{ max_seq: string | null }>(
				'SELECT MAX(seq)::text AS max_seq FROM messages WHERE conv_id = $1',
				[convId]
			);
			const seed = rows[0]?.max_seq ? parseInt(rows[0].max_seq, 10) : 0;
			// NX so a concurrent seeder does not clobber a counter already in use.
			await this.redis.set(key, String(seed), { NX: true });
		}

		const next = await this.redis.incr(key);
		await this.redis.expire(key, 86_400); // idle conversations release memory
		return next;
	}

	// Called when a write loses the (conv_id, seq) uniqueness race.
	async repair(convId: string): Promise<void> {
		const { rows } = await this.db.query<{ max_seq: string | null }>(
			'SELECT MAX(seq)::text AS max_seq FROM messages WHERE conv_id = $1',
			[convId]
		);
		const seed = rows[0]?.max_seq ? parseInt(rows[0].max_seq, 10) : 0;
		await this.redis.set(`seq:conv:${convId}`, String(seed));
	}
}

// ===========================================
// 3. MESSAGE SERVICE
// ===========================================
class MessageService {
	constructor(
		private db: Pool,
		private redis: RedisClientType,
		private seq: SeqAllocator,
		private producer: Producer
	) {}

	async send(req: SendRequest): Promise<SendResult> {
		// --- Sender idempotency: the same clientMsgId never creates a second row.
		const existing = await this.db.query<{ message_id: string; seq: string }>(
			`SELECT message_id, seq::text AS seq FROM messages
       WHERE conv_id = $1 AND sender_id = $2 AND client_msg_id = $3`,
			[req.convId, req.senderId, req.clientMsgId]
		);
		if (existing.rowCount) {
			return {
				messageId: existing.rows[0].message_id,
				seq: parseInt(existing.rows[0].seq, 10),
				deduped: true
			};
		}

		const messageId = crypto.randomUUID();

		// --- Allocate and persist, retrying once if two writers collided.
		let seq = 0;
		for (let attempt = 0; attempt < 3; attempt++) {
			seq = await this.seq.next(req.convId);
			try {
				await this.db.query(
					`INSERT INTO messages (conv_id, seq, message_id, sender_id, body, client_msg_id)
           VALUES ($1, $2, $3, $4, $5, $6)`,
					[req.convId, seq, messageId, req.senderId, req.body, req.clientMsgId]
				);
				break;
			} catch (err: unknown) {
				const code = (err as { code?: string }).code;
				if (code === '23505' && attempt < 2) {
					// Unique violation: either a seq collision (repair and retry)
					// or a concurrent duplicate send (return the winner).
					await this.seq.repair(req.convId);
					continue;
				}
				throw err;
			}
		}

		await this.db.query(
			'UPDATE conversations SET next_seq = GREATEST(next_seq, $2 + 1) WHERE conv_id = $1',
			[req.convId, seq]
		);

		// --- Sender is acked here. Fan-out happens asynchronously downstream.
		await this.producer.send({
			topic: 'chat.messages',
			messages: [
				{
					key: req.convId, // partition by conversation: preserves per-conv order
					value: JSON.stringify({
						convId: req.convId,
						seq,
						messageId,
						senderId: req.senderId,
						body: req.body,
						createdAt: new Date().toISOString()
					})
				}
			]
		});

		return { messageId, seq, deduped: false };
	}

	// --- Receipts. Monotonic, therefore idempotent by construction.
	async recordReceipt(input: {
		convId: string;
		userId: string;
		seq: number;
		kind: 'delivered' | 'read';
	}): Promise<{ applied: boolean; cursor: number }> {
		const column = input.kind === 'read' ? 'read_up_to' : 'delivered_up_to';

		const { rows } = await this.db.query<{ cursor: string; applied: boolean }>(
			`INSERT INTO read_cursors (conv_id, user_id, ${column}, updated_at)
       VALUES ($1, $2, $3, now())
       ON CONFLICT (conv_id, user_id) DO UPDATE
         SET ${column} = GREATEST(read_cursors.${column}, EXCLUDED.${column}),
             updated_at = now()
       RETURNING ${column}::text AS cursor,
                 (read_cursors.${column} = EXCLUDED.${column}) AS applied`,
			[input.convId, input.userId, input.seq]
		);

		const cursor = parseInt(rows[0].cursor, 10);

		// Only broadcast when the cursor actually moved — a repeated ack is silent.
		if (rows[0].applied) {
			await this.producer.send({
				topic: 'chat.receipts',
				messages: [
					{
						key: input.convId,
						value: JSON.stringify({ ...input, cursor })
					}
				]
			});
		}

		return { applied: rows[0].applied, cursor };
	}

	// --- Cursor pagination on (conv_id, seq). No OFFSET anywhere.
	async history(
		convId: string,
		userId: string,
		beforeSeq: number | null,
		limit = 50
	): Promise<{ messages: StoredMessage[]; nextCursor: number | null }> {
		const member = await this.db.query<{ joined_seq: string }>(
			'SELECT joined_seq::text AS joined_seq FROM conversation_members WHERE conv_id = $1 AND user_id = $2',
			[convId, userId]
		);
		if (!member.rowCount) throw new Error('not a member');
		const floor = parseInt(member.rows[0].joined_seq, 10);

		const cap = Math.min(limit, 200);
		const { rows } = await this.db.query(
			`SELECT conv_id, seq::text AS seq, message_id, sender_id, body, created_at
       FROM messages
       WHERE conv_id = $1
         AND seq >= $2
         AND ($3::bigint IS NULL OR seq < $3)
         AND deleted = false
       ORDER BY seq DESC
       LIMIT $4`,
			[convId, floor, beforeSeq, cap]
		);

		const messages: StoredMessage[] = rows.map((r) => ({
			convId: r.conv_id,
			seq: parseInt(r.seq, 10),
			messageId: r.message_id,
			senderId: r.sender_id,
			body: r.body,
			createdAt: r.created_at.toISOString()
		}));

		const nextCursor = messages.length === cap ? messages[messages.length - 1].seq : null;
		return { messages, nextCursor };
	}

	// --- Resume: everything the client missed, oldest first, capped.
	async backfill(
		convId: string,
		userId: string,
		fromSeq: number,
		limit = 200
	): Promise<{ messages: StoredMessage[]; upToSeq: number; hasMore: boolean }> {
		const cap = Math.min(limit, 500);
		const { rows } = await this.db.query(
			`SELECT conv_id, seq::text AS seq, message_id, sender_id, body, created_at
       FROM messages
       WHERE conv_id = $1 AND seq > $2 AND deleted = false
       ORDER BY seq ASC
       LIMIT $3`,
			[convId, fromSeq, cap + 1]
		);

		const hasMore = rows.length > cap;
		const page = hasMore ? rows.slice(0, cap) : rows;
		const messages: StoredMessage[] = page.map((r) => ({
			convId: r.conv_id,
			seq: parseInt(r.seq, 10),
			messageId: r.message_id,
			senderId: r.sender_id,
			body: r.body,
			createdAt: r.created_at.toISOString()
		}));

		return {
			messages,
			upToSeq: messages.length ? messages[messages.length - 1].seq : fromSeq,
			hasMore
		};
	}
}

// ===========================================
// 4. FAN-OUT WORKER
// ===========================================
class FanoutWorker {
	constructor(
		private db: Pool,
		private redis: RedisClientType,
		private nodeBus: NodeBus,
		private push: PushClient
	) {}

	private modeFor(memberCount: number): FanoutMode {
		if (memberCount <= SMALL_GROUP_MAX) return 'write';
		if (memberCount <= HYBRID_GROUP_MAX) return 'hybrid';
		return 'read';
	}

	async handle(msg: StoredMessage): Promise<void> {
		const conv = await this.db.query<{ member_count: number }>(
			'SELECT member_count FROM conversations WHERE conv_id = $1',
			[msg.convId]
		);
		const mode = this.modeFor(conv.rows[0]?.member_count ?? 2);

		if (mode === 'read') {
			// Large channel: publish a lightweight tail pointer only. Readers pull.
			await this.redis.set(`tail:conv:${msg.convId}`, String(msg.seq));
			await this.nodeBus.broadcastConversationTail(msg.convId, msg.seq);
			return;
		}

		const members = await this.db.query<{ user_id: string; muted: boolean }>(
			'SELECT user_id, muted FROM conversation_members WHERE conv_id = $1',
			[msg.convId]
		);

		for (const member of members.rows) {
			if (member.user_id === msg.senderId) continue;

			const locations = await this.lookupRegistry(member.user_id);

			if (locations.length > 0) {
				for (const loc of locations) {
					await this.nodeBus.route(loc.nodeId, member.user_id, { t: 'msg', ...msg });
				}
				if (mode === 'write') {
					await this.db.query(
						'INSERT INTO inbox (user_id, conv_id, seq) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING',
						[member.user_id, msg.convId, msg.seq]
					);
				}
				continue;
			}

			// Offline. In hybrid mode we write nothing — the client pulls on open.
			if (mode === 'write') {
				await this.db.query(
					'INSERT INTO inbox (user_id, conv_id, seq) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING',
					[member.user_id, msg.convId, msg.seq]
				);
			}
			if (!member.muted) {
				await this.push.enqueue(member.user_id, msg);
			}
		}
	}

	private async lookupRegistry(userId: string): Promise<Array<{ nodeId: string }>> {
		const entries = await this.redis.hGetAll(`conn:user:${userId}`);
		return Object.values(entries).map((v) => ({ nodeId: v.split(':')[0] }));
	}
}

export { MessageService, SeqAllocator, FanoutWorker };

import { RedisClientType } from 'redis';

// ===========================================
// 1. CONFIG
// ===========================================
const PRESENCE_TTL_SECONDS = 45; // grace period: outlives one missed heartbeat
const COALESCE_WINDOW_MS = 5_000;
const MAX_SUBSCRIPTIONS_PER_USER = 500;

type PresenceState = 'online' | 'offline';

interface PresenceChange {
	userId: string;
	state: PresenceState;
	lastSeenAt: number;
}

// ===========================================
// 2. PRESENCE SERVICE
// ===========================================
// Storage is trivial: one TTL key per user. The expensive part is deciding
// who hears about a change, and how often.
class PresenceService {
	// Pending changes waiting for the next coalesce flush.
	private pending = new Map<string, PresenceChange>();
	private flusher: ReturnType<typeof setInterval>;

	constructor(
		private redis: RedisClientType,
		private nodeBus: NodeBus
	) {
		this.flusher = setInterval(() => void this.flush(), COALESCE_WINDOW_MS);
	}

	private stateKey(userId: string): string {
		return `presence:${userId}`;
	}

	// Subscribers of a user, stored as a set of "nodeId|watcherId".
	private watchersKey(userId: string): string {
		return `presence:watchers:${userId}`;
	}

	// --- Called on every gateway heartbeat. Cheap: one SETEX.
	async heartbeat(userId: string): Promise<void> {
		const key = this.stateKey(userId);
		const wasOnline = await this.redis.exists(key);
		await this.redis.setEx(key, PRESENCE_TTL_SECONDS, String(Date.now()));

		// Only an actual transition is worth broadcasting.
		if (!wasOnline) {
			this.enqueueChange({ userId, state: 'online', lastSeenAt: Date.now() });
		}
	}

	// --- Explicit offline (app backgrounded, user logged out).
	async markOffline(userId: string): Promise<void> {
		const removed = await this.redis.del(this.stateKey(userId));
		if (removed) {
			this.enqueueChange({ userId, state: 'offline', lastSeenAt: Date.now() });
		}
	}

	// --- A watcher subscribes only to what is on screen right now.
	async subscribe(watcherId: string, nodeId: string, targetIds: string[]): Promise<void> {
		const scoped = targetIds.slice(0, MAX_SUBSCRIPTIONS_PER_USER);
		const member = `${nodeId}|${watcherId}`;

		const multi = this.redis.multi();
		// Replace, do not append: scrolling changes the visible window.
		multi.del(`presence:subs:${watcherId}`);
		for (const target of scoped) {
			multi.sAdd(this.watchersKey(target), member);
			multi.expire(this.watchersKey(target), 3600);
			multi.sAdd(`presence:subs:${watcherId}`, target);
		}
		multi.expire(`presence:subs:${watcherId}`, 3600);
		await multi.exec();
	}

	async unsubscribeAll(watcherId: string, nodeId: string): Promise<void> {
		const targets = await this.redis.sMembers(`presence:subs:${watcherId}`);
		const member = `${nodeId}|${watcherId}`;
		const multi = this.redis.multi();
		for (const target of targets) {
			multi.sRem(this.watchersKey(target), member);
		}
		multi.del(`presence:subs:${watcherId}`);
		await multi.exec();
	}

	// --- Point-in-time read, used when a client first opens a list.
	async snapshot(userIds: string[]): Promise<Record<string, PresenceState>> {
		if (userIds.length === 0) return {};
		const keys = userIds.map((id) => this.stateKey(id));
		const values = await this.redis.mGet(keys);
		const out: Record<string, PresenceState> = {};
		userIds.forEach((id, i) => {
			out[id] = values[i] ? 'online' : 'offline';
		});
		return out;
	}

	// --- Coalescing. Repeated flaps inside one window collapse to one update.
	private enqueueChange(change: PresenceChange): void {
		this.pending.set(change.userId, change);
	}

	private async flush(): Promise<void> {
		if (this.pending.size === 0) return;
		const batch = [...this.pending.values()];
		this.pending.clear();

		// Group deliveries by gateway node so each node receives one frame
		// carrying many changes, instead of many frames carrying one each.
		const perNode = new Map<string, Map<string, PresenceChange[]>>();

		for (const change of batch) {
			const watchers = await this.redis.sMembers(this.watchersKey(change.userId));
			for (const w of watchers) {
				const [nodeId, watcherId] = w.split('|');
				let nodeMap = perNode.get(nodeId);
				if (!nodeMap) {
					nodeMap = new Map();
					perNode.set(nodeId, nodeMap);
				}
				const list = nodeMap.get(watcherId) ?? [];
				list.push(change);
				nodeMap.set(watcherId, list);
			}
		}

		for (const [nodeId, watcherMap] of perNode) {
			for (const [watcherId, changes] of watcherMap) {
				await this.nodeBus.route(nodeId, watcherId, {
					t: 'presence_update',
					changes
				});
			}
		}

		console.log(`[PRESENCE] flushed ${batch.length} changes to ${perNode.size} nodes`);
	}

	// --- Typing: best effort, never persisted, never retried.
	async typing(convId: string, userId: string, memberNodes: string[]): Promise<void> {
		for (const nodeId of memberNodes) {
			void this.nodeBus.broadcast(nodeId, {
				t: 'typing',
				convId,
				userId,
				expiresInMs: 3000
			});
		}
	}

	stop(): void {
		clearInterval(this.flusher);
	}
}

// ===========================================
// 3. RESUME COORDINATOR
// ===========================================
// After a reconnect the client says, per conversation, "I have up to seq N".
// The coordinator returns the gap, capped, oldest first, plus a flag telling
// the client to fall back to a full history load if it drifted too far.
interface ResumeCursor {
	convId: string;
	haveUpToSeq: number;
}

const MAX_BACKFILL_PER_CONV = 200;
const RESYNC_THRESHOLD = 1000;

class ResumeCoordinator {
	constructor(
		private messages: MessageService,
		private redis: RedisClientType
	) {}

	async resume(
		userId: string,
		cursors: ResumeCursor[]
	): Promise<{
		conversations: Array<{
			convId: string;
			messages: unknown[];
			upToSeq: number;
			needsFullResync: boolean;
		}>;
	}> {
		const results = [];

		for (const cursor of cursors.slice(0, 200)) {
			// The conversation tail is cached, so a conversation with no gap
			// costs a single Redis read instead of a database query.
			const tailRaw = await this.redis.get(`tail:conv:${cursor.convId}`);
			const tail = tailRaw ? parseInt(tailRaw, 10) : cursor.haveUpToSeq;

			if (tail <= cursor.haveUpToSeq) {
				results.push({
					convId: cursor.convId,
					messages: [],
					upToSeq: cursor.haveUpToSeq,
					needsFullResync: false
				});
				continue;
			}

			// Drifted too far: replaying thousands of messages over a mobile
			// link is worse than telling the client to reload the tail.
			if (tail - cursor.haveUpToSeq > RESYNC_THRESHOLD) {
				results.push({
					convId: cursor.convId,
					messages: [],
					upToSeq: cursor.haveUpToSeq,
					needsFullResync: true
				});
				continue;
			}

			const batch = await this.messages.backfill(
				cursor.convId,
				userId,
				cursor.haveUpToSeq,
				MAX_BACKFILL_PER_CONV
			);
			results.push({
				convId: cursor.convId,
				messages: batch.messages,
				upToSeq: batch.upToSeq,
				needsFullResync: false
			});
		}

		return { conversations: results };
	}
}

export { PresenceService, ResumeCoordinator };