almost into a room

apps/notify/src/context.ts

@@ -0,0 +1,16 @@
+import type { NotificationsHub } from './notifications-hub'
+import type { HonoApp } from '@repo/hono-helpers'
+import type { SharedHonoEnv, SharedHonoVariables } from '@repo/hono-helpers/src/types'
+
+export type Env = SharedHonoEnv & {
+	/** Durable Object hosting the SignalR notifications hub. */
+	NOTIFICATIONS_HUB: DurableObjectNamespace<NotificationsHub>
+}
+
+/** Variables can be extended */
+export type Variables = SharedHonoVariables
+
+export interface App extends HonoApp {
+	Bindings: Env
+	Variables: Variables
+}

apps/notify/src/notifications-hub.ts

@@ -0,0 +1,291 @@
+import { DurableObject } from 'cloudflare:workers'
+
+import type { Env } from './context'
+
+/**
+ * Durable Object hosting the SignalR notifications hub, ported from the C#
+ * `NotificationsHub` + `NotificationService`. A single global instance plays the
+ * role of the C# static dictionaries (one process, shared across connections).
+ *
+ * It speaks the SignalR JSON Hub Protocol over a hibernatable WebSocket:
+ *   1. negotiate happens in the worker; the client then opens a WS to `/hub/v1`.
+ *   2. handshake: client sends `{"protocol":"json","version":1}␞`, we reply `{}␞`.
+ *   3. framed messages are `␞` (0x1e) delimited JSON; type 1 = invocation,
+ *      3 = completion, 6 = ping, 7 = close.
+ *
+ * Connection/subscription state lives in SQLite so it survives hibernation:
+ *   - `subscriptions(connectionId, playerId)` — both the connection→players and
+ *     (queried the other way) the player→connections maps from the C#.
+ *   - `pending(id, playerId, payload)` — the per-player queue delivered once a
+ *     player is subscribed.
+ */
+
+/** SignalR record separator (0x1e) that terminates every protocol message. */
+const RS = '\u001e'
+
+interface SocketState {
+	connectionId: string
+	handshakeDone: boolean
+}
+
+interface HubMessage {
+	type: number
+	target?: string
+	invocationId?: string
+	arguments?: unknown[]
+}
+
+export class NotificationsHub extends DurableObject<Env> {
+	constructor(ctx: DurableObjectState, env: Env) {
+		super(ctx, env)
+		void ctx.blockConcurrencyWhile(async () => {
+			this.ctx.storage.sql.exec(`
+				CREATE TABLE IF NOT EXISTS subscriptions (
+					connectionId TEXT NOT NULL,
+					playerId INTEGER NOT NULL
+				);
+				CREATE INDEX IF NOT EXISTS idx_sub_conn ON subscriptions(connectionId);
+				CREATE INDEX IF NOT EXISTS idx_sub_player ON subscriptions(playerId);
+				CREATE TABLE IF NOT EXISTS pending (
+					id INTEGER PRIMARY KEY AUTOINCREMENT,
+					playerId INTEGER NOT NULL,
+					payload TEXT NOT NULL
+				);
+				CREATE INDEX IF NOT EXISTS idx_pending_player ON pending(playerId);
+			`)
+		})
+	}
+
+	/** WebSocket upgrade entrypoint — the worker forwards `/hub/v1` here. */
+	override async fetch(request: Request): Promise<Response> {
+		if ((request.headers.get('Upgrade') ?? '').toLowerCase() !== 'websocket') {
+			return new Response('Expected a WebSocket upgrade request', { status: 426 })
+		}
+
+		const url = new URL(request.url)
+		// negotiate handed the client this id as `connectionToken`/`connectionId`.
+		const connectionId = url.searchParams.get('id') || crypto.randomUUID()
+
+		const pair = new WebSocketPair()
+		const server = pair[1]
+		// Tag by connectionId so we can find this socket via getWebSockets(id).
+		this.ctx.acceptWebSocket(server, [connectionId])
+		server.serializeAttachment({ connectionId, handshakeDone: false } satisfies SocketState)
+
+		return new Response(null, { status: 101, webSocket: pair[0] })
+	}
+
+	override async webSocketMessage(ws: WebSocket, message: string | ArrayBuffer): Promise<void> {
+		const text = typeof message === 'string' ? message : new TextDecoder().decode(message)
+		const state = ws.deserializeAttachment() as SocketState | null
+		if (!state) return
+
+		for (const record of text.split(RS)) {
+			if (record.length === 0) continue
+
+			if (!state.handshakeDone) {
+				// First frame is the SignalR handshake request.
+				this.completeHandshake(ws, record, state)
+				continue
+			}
+
+			let msg: HubMessage
+			try {
+				msg = JSON.parse(record) as HubMessage
+			} catch {
+				continue
+			}
+			this.handleMessage(ws, state.connectionId, msg)
+		}
+	}
+
+	override async webSocketClose(ws: WebSocket): Promise<void> {
+		const state = ws.deserializeAttachment() as SocketState | null
+		if (state) {
+			// Mirrors OnDisconnected: drop this connection's subscriptions, which
+			// also removes it from every player's connection set.
+			this.ctx.storage.sql.exec('DELETE FROM subscriptions WHERE connectionId = ?', state.connectionId)
+		}
+		try {
+			ws.close()
+		} catch {
+			// already closed
+		}
+	}
+
+	// ---- SignalR protocol ----------------------------------------------------
+
+	private completeHandshake(ws: WebSocket, record: string, state: SocketState): void {
+		let protocol = 'json'
+		try {
+			protocol = (JSON.parse(record) as { protocol?: string }).protocol ?? 'json'
+		} catch {
+			// fall through to the json default
+		}
+		if (protocol !== 'json') {
+			ws.send(JSON.stringify({ error: `Unsupported protocol '${protocol}'` }) + RS)
+			ws.close(1002, 'unsupported protocol')
+			return
+		}
+
+		// Empty object = handshake success.
+		ws.send('{}' + RS)
+		state.handshakeDone = true
+		ws.serializeAttachment(state)
+
+		// C# OnConnectedAsync sends "OnConnect" to the caller after connecting.
+		ws.send(this.invocation('OnConnect', []))
+	}
+
+	private handleMessage(ws: WebSocket, connectionId: string, msg: HubMessage): void {
+		switch (msg.type) {
+			case 1: // Invocation
+				this.handleInvocation(ws, connectionId, msg)
+				break
+			case 6: // Ping — echo to keep the connection alive.
+				ws.send(JSON.stringify({ type: 6 }) + RS)
+				break
+			case 7: // Close
+				ws.close()
+				break
+			default:
+				break
+		}
+	}
+
+	private handleInvocation(ws: WebSocket, connectionId: string, msg: HubMessage): void {
+		switch (msg.target) {
+			case 'SubscribeToPlayers': {
+				const arg = msg.arguments?.[0] as { playerIds?: number[] } | undefined
+				const playerIds = (arg?.playerIds ?? []).filter((n) => typeof n === 'number')
+				this.subscribeToPlayers(ws, connectionId, playerIds)
+				if (msg.invocationId) ws.send(this.completion(msg.invocationId, null))
+				break
+			}
+			case 'GetSubscriptions': {
+				const players = this.getSubscribedPlayers(connectionId)
+				if (msg.invocationId) ws.send(this.completion(msg.invocationId, players))
+				break
+			}
+			default:
+				if (msg.invocationId) {
+					ws.send(this.completionError(msg.invocationId, `Unknown method '${msg.target}'`))
+				}
+				break
+		}
+	}
+
+	private subscribeToPlayers(ws: WebSocket, connectionId: string, playerIds: number[]): void {
+		const unique = [...new Set(playerIds)]
+
+		// Replace this connection's subscription set.
+		this.ctx.storage.sql.exec('DELETE FROM subscriptions WHERE connectionId = ?', connectionId)
+		for (const playerId of unique) {
+			this.ctx.storage.sql.exec(
+				'INSERT INTO subscriptions (connectionId, playerId) VALUES (?, ?)',
+				connectionId,
+				playerId
+			)
+		}
+
+		// Flush any notifications queued while these players were offline.
+		for (const playerId of unique) {
+			const pending = this.ctx.storage.sql
+				.exec<{ payload: string }>('SELECT payload FROM pending WHERE playerId = ? ORDER BY id', playerId)
+				.toArray()
+			if (pending.length === 0) continue
+			for (const row of pending) {
+				ws.send(this.invocation('Notification', [row.payload]))
+			}
+			this.ctx.storage.sql.exec('DELETE FROM pending WHERE playerId = ?', playerId)
+		}
+	}
+
+	private getSubscribedPlayers(connectionId: string): number[] {
+		return this.ctx.storage.sql
+			.exec<{ playerId: number }>(
+				'SELECT DISTINCT playerId FROM subscriptions WHERE connectionId = ?',
+				connectionId
+			)
+			.toArray()
+			.map((r) => r.playerId)
+	}
+
+	// ---- Server → client RPC (callable from other workers) -------------------
+
+	/**
+	 * Send a notification to a player's connections, queueing it if the player
+	 * isn't currently connected (mirrors `SendNotificationToPlayer`).
+	 */
+	async notifyPlayer(
+		playerId: number,
+		notificationType: number,
+		data?: Record<string, unknown>
+	): Promise<{ delivered: number; queued: boolean }> {
+		const payload = this.buildNotificationPayload(notificationType, data)
+
+		const connectionIds = this.ctx.storage.sql
+			.exec<{ connectionId: string }>(
+				'SELECT DISTINCT connectionId FROM subscriptions WHERE playerId = ?',
+				playerId
+			)
+			.toArray()
+			.map((r) => r.connectionId)
+
+		let delivered = 0
+		for (const connectionId of connectionIds) {
+			for (const ws of this.ctx.getWebSockets(connectionId)) {
+				ws.send(this.invocation('Notification', [payload]))
+				delivered++
+			}
+		}
+
+		if (delivered === 0) {
+			this.ctx.storage.sql.exec('INSERT INTO pending (playerId, payload) VALUES (?, ?)', playerId, payload)
+			return { delivered: 0, queued: true }
+		}
+		return { delivered, queued: false }
+	}
+
+	/** Broadcast a notification to every connected (handshaken) client. */
+	async broadcast(notificationType: number, data?: Record<string, unknown>): Promise<{ delivered: number }> {
+		const payload = this.buildNotificationPayload(notificationType, data)
+		let delivered = 0
+		for (const ws of this.ctx.getWebSockets()) {
+			const state = ws.deserializeAttachment() as SocketState | null
+			if (!state?.handshakeDone) continue
+			ws.send(this.invocation('Notification', [payload]))
+			delivered++
+		}
+		return { delivered }
+	}
+
+	// ---- Helpers -------------------------------------------------------------
+
+	/**
+	 * Build the `Notification` argument: a JSON string `{ Id, Msg }`, matching the
+	 * C# `SendToConnection` (null values are dropped from `Msg`).
+	 */
+	private buildNotificationPayload(notificationType: number, data?: Record<string, unknown>): string {
+		const msg: Record<string, unknown> = {}
+		if (data) {
+			for (const [key, value] of Object.entries(data)) {
+				if (value === null || value === undefined) continue
+				msg[key] = value
+			}
+		}
+		return JSON.stringify({ Id: notificationType ?? 0, Msg: msg })
+	}
+
+	private invocation(target: string, args: unknown[]): string {
+		return JSON.stringify({ type: 1, target, arguments: args }) + RS
+	}
+
+	private completion(invocationId: string, result: unknown): string {
+		return JSON.stringify({ type: 3, invocationId, result }) + RS
+	}
+
+	private completionError(invocationId: string, error: string): string {
+		return JSON.stringify({ type: 3, invocationId, error }) + RS
+	}
+}

apps/notify/src/notify.app.ts

@@ -0,0 +1,91 @@
+import { Hono } from 'hono'
+import { useWorkersLogger } from 'workers-tagged-logger'
+
+import { logger, withNotFound, withOnError } from '@repo/hono-helpers'
+
+import type { App } from './context'
+import { NotificationsHub } from './notifications-hub'
+
+/**
+ * Ported from the C# `NotifyController`, which maps a SignalR hub at `/hub/v1`
+ * (see `NotificationsHub` / `NotificationService`). The hub itself — WebSocket
+ * transport, the SignalR JSON Hub Protocol, and the shared connection state —
+ * lives in the `NotificationsHub` Durable Object; this worker handles the
+ * SignalR negotiate handshake, forwards the WebSocket upgrade to the DO, and
+ * exposes internal send/broadcast endpoints for other workers.
+ */
+
+/** The hub state is global in the C# (static dictionaries) → one DO instance. */
+const HUB_INSTANCE = 'global'
+
+const app = new Hono<App>()
+	.use(
+		'*',
+		// middleware
+		(c, next) =>
+			useWorkersLogger(c.env.NAME, {
+				environment: c.env.ENVIRONMENT,
+				release: c.env.SENTRY_RELEASE,
+			})(c, next)
+	)
+
+	.onError(withOnError())
+	.notFound(withNotFound())
+
+	// SignalR negotiation. Clients POST here first; we hand back an id that is
+	// then passed as `?id=` on the WebSocket connect. We don't pre-register it —
+	// the DO adopts whatever id arrives — so negotiate stays stateless.
+	.post('/hub/v1/negotiate', (c) => {
+		const negotiateVersion = Number(c.req.query('negotiateVersion')) || 0
+		const id = crypto.randomUUID()
+		logger.info('signalr negotiate', { negotiateVersion })
+		return c.json({
+			negotiateVersion,
+			connectionId: id,
+			connectionToken: id,
+			availableTransports: [{ transport: 'WebSockets', transferFormats: ['Text'] }],
+		})
+	})
+
+	// The hub WebSocket. Upgrade requests are forwarded to the Durable Object.
+	.get('/hub/v1', (c) => {
+		if ((c.req.header('upgrade') ?? '').toLowerCase() !== 'websocket') {
+			return c.json({ error: 'Expected a WebSocket upgrade request' }, 426)
+		}
+		return c.env.NOTIFICATIONS_HUB.getByName(HUB_INSTANCE).fetch(c.req.raw)
+	})
+
+	// ---- Internal service-to-service send/broadcast --------------------------
+	// Lets other workers push notifications, the way the C# controllers called
+	// the shared NotificationService. TODO: protect these before production.
+	.post('/internal/notify', async (c) => {
+		const body = await c.req
+			.json<{ playerId?: number; notificationType?: number; data?: Record<string, unknown> }>()
+			.catch(() => null)
+		if (!body || typeof body.playerId !== 'number' || typeof body.notificationType !== 'number') {
+			return c.json({ error: 'playerId and notificationType are required' }, 400)
+		}
+		const result = await c.env.NOTIFICATIONS_HUB.getByName(HUB_INSTANCE).notifyPlayer(
+			body.playerId,
+			body.notificationType,
+			body.data
+		)
+		return c.json({ success: true, ...result })
+	})
+
+	.post('/internal/broadcast', async (c) => {
+		const body = await c.req
+			.json<{ notificationType?: number; data?: Record<string, unknown> }>()
+			.catch(() => null)
+		if (!body || typeof body.notificationType !== 'number') {
+			return c.json({ error: 'notificationType is required' }, 400)
+		}
+		const result = await c.env.NOTIFICATIONS_HUB.getByName(HUB_INSTANCE).broadcast(
+			body.notificationType,
+			body.data
+		)
+		return c.json({ success: true, ...result })
+	})
+
+export { NotificationsHub }
+export default app

apps/notify/src/test/integration/api.test.ts

@@ -0,0 +1,150 @@
+import { exports } from 'cloudflare:workers'
+import { describe, expect, test } from 'vitest'
+
+import '../../notify.app'
+
+const ORIGIN = 'https://notify.rec.djdevin.net'
+const RS = '\u001e'
+
+interface HubRecord {
+	type?: number
+	target?: string
+	invocationId?: string
+	result?: unknown
+	arguments?: unknown[]
+}
+
+/** Open a hub WebSocket, accept it, and complete the SignalR handshake. */
+async function connect(id: string): Promise<{ ws: WebSocket; waitFor: (pred: (r: HubRecord) => boolean) => Promise<HubRecord> }> {
+	const res = await exports.default.fetch(`${ORIGIN}/hub/v1?id=${id}`, {
+		headers: { Upgrade: 'websocket' },
+	})
+	expect(res.status).toBe(101)
+	const ws = res.webSocket!
+	ws.accept()
+
+	const records: HubRecord[] = []
+	const waiters: Array<{ pred: (r: HubRecord) => boolean; resolve: (r: HubRecord) => void }> = []
+	ws.addEventListener('message', (e: MessageEvent) => {
+		const text = typeof e.data === 'string' ? e.data : new TextDecoder().decode(e.data as ArrayBuffer)
+		for (const part of text.split(RS)) {
+			if (!part) continue
+			const rec = JSON.parse(part) as HubRecord
+			records.push(rec)
+			for (let i = waiters.length - 1; i >= 0; i--) {
+				if (waiters[i].pred(rec)) {
+					waiters[i].resolve(rec)
+					waiters.splice(i, 1)
+				}
+			}
+		}
+	})
+
+	const waitFor = (pred: (r: HubRecord) => boolean): Promise<HubRecord> => {
+		const existing = records.find(pred)
+		if (existing) return Promise.resolve(existing)
+		return new Promise((resolve, reject) => {
+			const timer = setTimeout(() => reject(new Error('timed out waiting for hub message')), 2000)
+			waiters.push({ pred, resolve: (r) => { clearTimeout(timer); resolve(r) } })
+		})
+	}
+
+	// Handshake, then the C# OnConnect callback.
+	ws.send(`{"protocol":"json","version":1}${RS}`)
+	await waitFor((r) => r.type === 1 && r.target === 'OnConnect')
+
+	return { ws, waitFor }
+}
+
+const send = (ws: WebSocket, msg: HubRecord) => ws.send(JSON.stringify(msg) + RS)
+
+const post = (path: string, body: unknown) =>
+	exports.default.fetch(`${ORIGIN}${path}`, {
+		method: 'POST',
+		headers: { 'Content-Type': 'application/json' },
+		body: JSON.stringify(body),
+	})
+
+describe('negotiate', () => {
+	test('POST /hub/v1/negotiate advertises the WebSocket transport', async () => {
+		const res = await exports.default.fetch(`${ORIGIN}/hub/v1/negotiate?negotiateVersion=1`, { method: 'POST' })
+		expect(res.status).toBe(200)
+		const body = (await res.json()) as {
+			connectionId: string
+			connectionToken: string
+			availableTransports: Array<{ transport: string }>
+		}
+		expect(body.connectionId).toMatch(/^[0-9a-f-]{36}$/)
+		expect(body.connectionToken).toBe(body.connectionId)
+		expect(body.availableTransports[0].transport).toBe('WebSockets')
+	})
+
+	test('GET /hub/v1 without an upgrade header is 426', async () => {
+		const res = await exports.default.fetch(`${ORIGIN}/hub/v1`)
+		expect(res.status).toBe(426)
+	})
+})
+
+describe('hub protocol', () => {
+	test('GetSubscriptions reflects SubscribeToPlayers', async () => {
+		const { ws, waitFor } = await connect('conn-subs')
+
+		send(ws, { type: 1, invocationId: '1', target: 'GetSubscriptions', arguments: [] })
+		const empty = await waitFor((r) => r.type === 3 && r.invocationId === '1')
+		expect(empty.result).toEqual([])
+
+		send(ws, { type: 1, target: 'SubscribeToPlayers', arguments: [{ playerIds: [1, 2, 2] }] })
+		send(ws, { type: 1, invocationId: '2', target: 'GetSubscriptions', arguments: [] })
+		const subscribed = await waitFor((r) => r.type === 3 && r.invocationId === '2')
+		expect((subscribed.result as number[]).sort((a, b) => a - b)).toEqual([1, 2])
+
+		ws.close()
+	})
+
+	test('responds to ping', async () => {
+		const { ws, waitFor } = await connect('conn-ping')
+		send(ws, { type: 6 })
+		const pong = await waitFor((r) => r.type === 6)
+		expect(pong.type).toBe(6)
+		ws.close()
+	})
+})
+
+describe('notification delivery', () => {
+	test('queues for an offline player and flushes on subscribe', async () => {
+		const playerId = 9001
+		const queued = await post('/internal/notify', { playerId, notificationType: 2, data: { messageId: 'm1' } })
+		expect(await queued.json()).toMatchObject({ queued: true, delivered: 0 })
+
+		const { ws, waitFor } = await connect('conn-pending')
+		send(ws, { type: 1, target: 'SubscribeToPlayers', arguments: [{ playerIds: [playerId] }] })
+		const note = await waitFor((r) => r.type === 1 && r.target === 'Notification')
+		expect(JSON.parse((note.arguments as string[])[0])).toEqual({ Id: 2, Msg: { messageId: 'm1' } })
+
+		ws.close()
+	})
+
+	test('delivers live to a subscribed player', async () => {
+		const playerId = 9002
+		const { ws, waitFor } = await connect('conn-live')
+		send(ws, { type: 1, invocationId: 's', target: 'SubscribeToPlayers', arguments: [{ playerIds: [playerId] }] })
+		await waitFor((r) => r.type === 3 && r.invocationId === 's')
+
+		const res = await post('/internal/notify', { playerId, notificationType: 1, data: { accountId: 42 } })
+		expect(await res.json()).toMatchObject({ delivered: 1, queued: false })
+
+		const note = await waitFor((r) => r.type === 1 && r.target === 'Notification')
+		expect(JSON.parse((note.arguments as string[])[0])).toEqual({ Id: 1, Msg: { accountId: 42 } })
+
+		ws.close()
+	})
+
+	test('broadcast reaches connected clients', async () => {
+		const { ws, waitFor } = await connect('conn-broadcast')
+		const res = await post('/internal/broadcast', { notificationType: 25, data: { message: 'maint' } })
+		expect(((await res.json()) as { delivered: number }).delivered).toBeGreaterThanOrEqual(1)
+		const note = await waitFor((r) => r.type === 1 && r.target === 'Notification')
+		expect(JSON.parse((note.arguments as string[])[0])).toEqual({ Id: 25, Msg: { message: 'maint' } })
+		ws.close()
+	})
+})