apps/notify some notify fixes

apps/notify/src/notifications-hub.ts

@@ -15,10 +15,14 @@ import type { Env } from './context'
  *      3 = completion, 6 = ping, 7 = close.
  *
  * Connection/subscription state lives in SQLite so it survives hibernation:
- *   - `subscriptions(connectionId, playerId)` — both the connection→players and
+ *   - `connection_owner(connectionId, playerId)` — who each socket belongs to,
- *     (queried the other way) the player→connections maps.
+ *     from the token the worker validated at connect. This is how a player's own
+ *     notifications reach them; the client never subscribes to itself.
+ *   - `subscriptions(connectionId, playerId)` — *other* players a connection asked
+ *     for updates about, both the connection→players and (queried the other way)
+ *     the player→connections maps.
  *   - `pending(id, playerId, payload)` — the per-player queue delivered once a
- *     player is subscribed.
+ *     player is reachable again.
  */
 
 /** SignalR record separator (0x1e) that terminates every protocol message. */
@@ -27,6 +31,8 @@ const RS = '\u001e'
 interface SocketState {
 	connectionId: string
 	handshakeDone: boolean
+	/** The player this socket belongs to, from the token validated at connect. */
+	playerId?: number
 }
 
 interface HubMessage {
@@ -36,6 +42,63 @@ interface HubMessage {
 	arguments?: unknown[]
 }
 
+/**
+ * How the worker tells the DO which player a connecting socket belongs to, having
+ * validated the connect request's token. The worker sets it on every connect it lets
+ * through and refuses the rest, so a client can't present its own and be believed.
+ */
+export const OWNER_HEADER = 'x-recflare-connection-owner'
+
+/**
+ * Read the player ids off a `SubscribeToPlayers` invocation. SignalR gives no schema, so
+ * all three plausible spellings are accepted — an options object
+ * (`[{playerIds:[1,2]}]`), a single array argument (`[[1,2]]`), and varargs
+ * (`[1,2]`) — rather than guessing which one the client uses.
+ *
+ * `null` means the argument didn't resolve to a list of ids at all, which the caller
+ * must not treat as an empty subscription; `[]` is a real "subscribe to nobody".
+ */
+function parsePlayerIds(args: unknown[] | undefined): number[] | null {
+	if (args === undefined || args.length === 0) return []
+
+	const first = args[0]
+	const candidate =
+		typeof first === 'object' && first !== null && !Array.isArray(first)
+			? (first as { playerIds?: unknown }).playerIds
+			: Array.isArray(first)
+				? first
+				: args
+
+	if (!Array.isArray(candidate)) return null
+	// Ids arriving as strings ("153") still count — the wire format is the client's
+	// choice, and the subscriptions table is what has to be numeric.
+	const ids = candidate
+		.map((value) =>
+			typeof value === 'number' ? value : typeof value === 'string' ? Number(value) : Number.NaN
+		)
+		.filter((value) => Number.isInteger(value))
+	return ids.length === 0 && candidate.length > 0 ? null : ids
+}
+
+/** What {@link NotificationsHub.inspect} reports — see it for what each field is for. */
+export interface HubState {
+	connections: Array<{
+		connectionId: string
+		/** Whether a socket for this connectionId is still held by the DO. */
+		live: boolean
+		handshakeDone: boolean
+		/**
+		 * The player this connection belongs to. Null only for a connection that predates
+		 * authenticated connects and hasn't closed yet — a new one always has an owner.
+		 */
+		playerId: number | null
+		/** Other players this connection subscribed to updates for. */
+		playerIds: number[]
+	}>
+	/** Queued-while-offline notifications, newest payload per player for identification. */
+	pending: Array<{ playerId: number; count: number; latest: string }>
+}
+
 /** The Coach system account — the `FromPlayerId` on a coach message (see coachMessageAll). */
 const COACH_PLAYER_ID = 1
 
@@ -59,6 +122,11 @@ export class NotificationsHub extends DurableObject<Env> {
 					payload TEXT NOT NULL
 				);
 				CREATE INDEX IF NOT EXISTS idx_pending_player ON pending(playerId);
+				CREATE TABLE IF NOT EXISTS connection_owner (
+					connectionId TEXT PRIMARY KEY,
+					playerId INTEGER NOT NULL
+				);
+				CREATE INDEX IF NOT EXISTS idx_owner_player ON connection_owner(playerId);
 			`)
 		})
 	}
@@ -73,11 +141,30 @@ export class NotificationsHub extends DurableObject<Env> {
 		// negotiate handed the client this id as `connectionToken`/`connectionId`.
 		const connectionId = url.searchParams.get('id') || crypto.randomUUID()
 
+		// Who this socket belongs to, established by the worker from the connect
+		// request's token (see OWNER_HEADER). The worker rejects a connect it can't
+		// identify and always sets the header itself, so an absent one means the DO was
+		// reached by some other path — not something to serve a socket to.
+		const playerId = Number.parseInt(request.headers.get(OWNER_HEADER) ?? '', 10)
+		if (!Number.isInteger(playerId)) {
+			return new Response('Unidentified connection', { status: 401 })
+		}
+
 		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)
+		server.serializeAttachment({
+			connectionId,
+			handshakeDone: false,
+			playerId,
+		} satisfies SocketState)
+
+		this.ctx.storage.sql.exec(
+			'INSERT OR REPLACE INTO connection_owner (connectionId, playerId) VALUES (?, ?)',
+			connectionId,
+			playerId
+		)
 
 		return new Response(null, { status: 101, webSocket: pair[0] })
 	}
@@ -115,6 +202,10 @@ export class NotificationsHub extends DurableObject<Env> {
 				'DELETE FROM subscriptions WHERE connectionId = ?',
 				state.connectionId
 			)
+			this.ctx.storage.sql.exec(
+				'DELETE FROM connection_owner WHERE connectionId = ?',
+				state.connectionId
+			)
 		}
 		try {
 			ws.close()
@@ -145,6 +236,25 @@ export class NotificationsHub extends DurableObject<Env> {
 
 		// Send "OnConnect" to the caller after connecting.
 		ws.send(this.invocation('OnConnect', []))
+
+		// Deliver whatever piled up while this player was away. Done here rather than at
+		// accept because SignalR won't read invocation frames sent before the handshake
+		// reply, and only here because the client never calls SubscribeToPlayers — with
+		// the flush living solely in there, a queued notification was stranded forever.
+		if (state.playerId !== undefined) this.flushPending(ws, state.playerId)
+	}
+
+	/** Send and clear a player's queued notifications on `ws`. */
+	private flushPending(ws: WebSocket, playerId: number): void {
+		const pending = this.ctx.storage.sql
+			.exec<{
+				payload: string
+			}>('SELECT payload FROM pending WHERE playerId = ? ORDER BY id', playerId)
+			.toArray()
+		if (pending.length === 0) return
+
+		for (const row of pending) ws.send(this.invocation('Notification', [row.payload]))
+		this.ctx.storage.sql.exec('DELETE FROM pending WHERE playerId = ?', playerId)
 	}
 
 	private handleMessage(ws: WebSocket, connectionId: string, msg: HubMessage): void {
@@ -166,9 +276,18 @@ export class NotificationsHub extends DurableObject<Env> {
 	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 = parsePlayerIds(msg.arguments)
-				const playerIds = (arg?.playerIds ?? []).filter((n) => typeof n === 'number')
+				// An argument we can't read is not "subscribe to nobody": subscribing
-				this.subscribeToPlayers(ws, connectionId, playerIds)
+				// replaces the connection's whole set, so acting on a misread would wipe
+				// a working connection to zero and silently strand every push.
+				if (playerIds === null) {
+					console.warn('hub: unreadable SubscribeToPlayers argument', {
+						connectionId,
+						arguments: JSON.stringify(msg.arguments),
+					})
+				} else {
+					this.subscribeToPlayers(ws, connectionId, playerIds)
+				}
 				if (msg.invocationId) ws.send(this.completion(msg.invocationId, null))
 				break
 			}
@@ -178,6 +297,14 @@ export class NotificationsHub extends DurableObject<Env> {
 				break
 			}
 			default:
+				// Logged, not just answered with an error completion: a hub method we
+				// don't implement is a client expectation we've missed, and the client
+				// gives no sign of it.
+				console.warn('hub: unknown invocation target', {
+					connectionId,
+					target: msg.target,
+					arguments: JSON.stringify(msg.arguments),
+				})
 				if (msg.invocationId) {
 					ws.send(this.completionError(msg.invocationId, `Unknown method '${msg.target}'`))
 				}
@@ -199,18 +326,26 @@ export class NotificationsHub extends DurableObject<Env> {
 		}
 
 		// Flush any notifications queued while these players were offline.
-		for (const playerId of unique) {
+		for (const playerId of unique) this.flushPending(ws, playerId)
-			const pending = this.ctx.storage.sql
+	}
-				.exec<{
+
-					payload: string
+	/**
-				}>('SELECT payload FROM pending WHERE playerId = ? ORDER BY id', playerId)
+	 * Every connection that receives notifications for a player. Two ways to qualify: the
-				.toArray()
+	 * connection *is* that player (established from the token at connect), or it
-			if (pending.length === 0) continue
+	 * subscribed to them. The client only ever uses the first — it never calls
-			for (const row of pending) {
+	 * SubscribeToPlayers — so a player's own notifications reach them through
-				ws.send(this.invocation('Notification', [row.payload]))
+	 * connection_owner, and subscriptions carry other players' updates.
-			}
+	 */
-			this.ctx.storage.sql.exec('DELETE FROM pending WHERE playerId = ?', playerId)
+	private connectionIdsFor(playerId: number): string[] {
-		}
+		return this.ctx.storage.sql
+			.exec<{ connectionId: string }>(
+				`SELECT connectionId FROM connection_owner WHERE playerId = ?1
+				UNION
+				SELECT connectionId FROM subscriptions WHERE playerId = ?1`,
+				playerId
+			)
+			.toArray()
+			.map((r) => r.connectionId)
 	}
 
 	private getSubscribedPlayers(connectionId: string): number[] {
@@ -237,6 +372,17 @@ export class NotificationsHub extends DurableObject<Env> {
 		const payload = this.buildNotificationPayload(notificationType, data)
 		const delivered = this.deliverToPlayer(playerId, payload)
 
+		if (delivered === 0) {
+			// Distinguishes the two ways this fails: no connection is registered for the
+			// player at all, versus one is registered but has no live socket behind it.
+			console.warn('hub: notification queued, nobody to deliver to', {
+				playerId,
+				notificationType,
+				connectionIds: this.connectionIdsFor(playerId),
+				liveSockets: this.ctx.getWebSockets().length,
+			})
+		}
+
 		if (delivered === 0) {
 			this.ctx.storage.sql.exec(
 				'INSERT INTO pending (playerId, payload) VALUES (?, ?)',
@@ -270,7 +416,97 @@ export class NotificationsHub extends DurableObject<Env> {
 		notificationType: string | number,
 		data?: Record<string, unknown>
 	): Promise<{ delivered: number }> {
-		return { delivered: this.broadcastToConnected(this.buildNotificationPayload(notificationType, data)) }
+		return {
+			delivered: this.broadcastToConnected(this.buildNotificationPayload(notificationType, data)),
+		}
+	}
+
+	/**
+	 * Dump the hub's routing state for debugging delivery. A notification only reaches a
+	 * player through a `subscriptions` row (see {@link deliverToPlayer}), so when a push
+	 * doesn't arrive this answers the two questions that matter: is the player subscribed
+	 * on a live connection, and is the frame sitting in `pending` instead?
+	 *
+	 * Connections are listed even when one side is missing — a socket that has yet to
+	 * subscribe (`playerIds: []`) and a subscription set whose socket is gone
+	 * (`live: false`, a close we never saw) are both delivery failures worth seeing.
+	 */
+	async inspect(): Promise<HubState> {
+		const live = new Map<string, boolean>()
+		for (const ws of this.ctx.getWebSockets()) {
+			const state = ws.deserializeAttachment() as SocketState | null
+			if (state) live.set(state.connectionId, state.handshakeDone)
+		}
+
+		const subscribed = new Map<string, number[]>()
+		const rows = this.ctx.storage.sql
+			.exec<{
+				connectionId: string
+				playerId: number
+			}>('SELECT connectionId, playerId FROM subscriptions ORDER BY connectionId, playerId')
+			.toArray()
+		for (const row of rows) {
+			const players = subscribed.get(row.connectionId) ?? []
+			players.push(row.playerId)
+			subscribed.set(row.connectionId, players)
+		}
+
+		const owners = new Map(
+			this.ctx.storage.sql
+				.exec<{
+					connectionId: string
+					playerId: number
+				}>('SELECT connectionId, playerId FROM connection_owner')
+				.toArray()
+				.map((row) => [row.connectionId, row.playerId] as const)
+		)
+
+		const connections = [...new Set([...live.keys(), ...subscribed.keys(), ...owners.keys()])].map(
+			(connectionId) => ({
+				connectionId,
+				live: live.has(connectionId),
+				handshakeDone: live.get(connectionId) ?? false,
+				playerId: owners.get(connectionId) ?? null,
+				playerIds: subscribed.get(connectionId) ?? [],
+			})
+		)
+
+		const pending = this.ctx.storage.sql
+			.exec<{ playerId: number; count: number; latest: string }>(
+				`SELECT playerId, COUNT(*) AS count,
+					(SELECT payload FROM pending AS newest WHERE newest.playerId = pending.playerId
+						ORDER BY id DESC LIMIT 1) AS latest
+				FROM pending GROUP BY playerId ORDER BY playerId`
+			)
+			.toArray()
+
+		return { connections, pending }
+	}
+
+	/**
+	 * Drop queued notifications — for one player, or (`playerId` omitted) the whole
+	 * queue. Anything pending is delivered the moment that player next subscribes, so a
+	 * frame queued by a bug that has since been fixed would otherwise arrive, out of
+	 * context, at the next reconnect. Returns how many were discarded.
+	 */
+	async clearPending(playerId?: number): Promise<{ cleared: number }> {
+		// Counted first rather than read off the cursor: the delete's own row count isn't
+		// reported, and this is an admin-facing number we want to be exact.
+		const [counted] =
+			playerId === undefined
+				? this.ctx.storage.sql
+						.exec<{ count: number }>('SELECT COUNT(*) AS count FROM pending')
+						.toArray()
+				: this.ctx.storage.sql
+						.exec<{
+							count: number
+						}>('SELECT COUNT(*) AS count FROM pending WHERE playerId = ?', playerId)
+						.toArray()
+
+		if (playerId === undefined) this.ctx.storage.sql.exec('DELETE FROM pending')
+		else this.ctx.storage.sql.exec('DELETE FROM pending WHERE playerId = ?', playerId)
+
+		return { cleared: counted?.count ?? 0 }
 	}
 
 	// ---- Helpers -------------------------------------------------------------
@@ -281,13 +517,7 @@ export class NotificationsHub extends DurableObject<Env> {
 	 * shared send path for {@link notifyPlayer} and {@link coachMessageAll}.
 	 */
 	private deliverToPlayer(playerId: number, payload: string): number {
-		const connectionIds = this.ctx.storage.sql
+		const connectionIds = this.connectionIdsFor(playerId)
-			.exec<{ connectionId: string }>(
-				'SELECT DISTINCT connectionId FROM subscriptions WHERE playerId = ?',
-				playerId
-			)
-			.toArray()
-			.map((r) => r.connectionId)
 
 		let delivered = 0
 		for (const connectionId of connectionIds) {

apps/notify/src/notify.app.ts

@@ -2,12 +2,12 @@ import { Hono } from 'hono'
 import { useWorkersLogger } from 'workers-tagged-logger'
 
 import { logger, withNotFound, withOnError } from '@repo/hono-helpers'
-import { validateAndGetRoles } from '@repo/jwt'
+import { validateAndGetAccountId, validateAndGetRoles } from '@repo/jwt'
 
-import { NotificationsHub } from './notifications-hub'
+import { NotificationsHub, OWNER_HEADER } from './notifications-hub'
 
+import type { Context, MiddlewareHandler } from 'hono'
 import type { App } from './context'
-import type { MiddlewareHandler } from 'hono'
 
 /**
  * Maps a SignalR hub at `/hub/v1`. The hub itself — WebSocket transport, the
@@ -37,6 +37,26 @@ function isNotificationType(value: unknown): value is string | number {
  */
 const ADMIN_ROLES = new Set(['developer', 'moderator'])
 
+/**
+ * The player opening a hub WebSocket, or null when the connect carries no valid token.
+ *
+ * A WebSocket connect can't always carry an `Authorization` header — SignalR clients
+ * that can't set headers on the upgrade put the token in an `access_token` query
+ * param instead — so both are accepted, header first.
+ */
+async function connectionOwner(c: Context<App>): Promise<number | null> {
+	const secret = await c.env.JWT_SECRET.get()
+	const id = await validateAndGetAccountId(c.req.raw, secret)
+	if (id !== null) return id
+
+	const token = c.req.query('access_token')
+	if (!token) return null
+	return validateAndGetAccountId(
+		new Request(c.req.url, { headers: { Authorization: `Bearer ${token}` } }),
+		secret
+	)
+}
+
 /**
  * Gates the `/internal/*` endpoints on a valid Bearer token that carries one of the
  * {@link ADMIN_ROLES} in its `role` claim. 401 for a missing/invalid token, 403 for a
@@ -78,12 +98,26 @@ const app = new Hono<App>()
 		})
 	})
 
-	// The hub WebSocket. Upgrade requests are forwarded to the Durable Object.
+	// The hub WebSocket. Upgrade requests are forwarded to the Durable Object, tagged
+	// with the connecting player so the hub can route their own notifications to them.
 	.get('/hub/v1', async (c) => {
 		if ((c.req.header('upgrade') ?? '').toLowerCase() !== 'websocket') {
 			return c.json({ error: 'Expected a WebSocket upgrade request' }, 426)
 		}
-		return c.env.RECFLARE_NOTIFICATIONS_HUB.getByName(HUB_INSTANCE).fetch(c.req.raw)
+
+		const playerId = await connectionOwner(c)
+		// Every notification the hub sends is either for a specific player or a
+		// broadcast to logged-in clients, so a connection we can't identify has nothing
+		// to receive. Refusing it here keeps unidentified sockets out of the hub
+		// entirely rather than letting them sit there collecting broadcasts.
+		if (playerId === null) return c.json({ error: 'Unauthorized' }, 401)
+
+		const request = new Request(c.req.raw)
+		// Always set from the validated token, never passed through: the header is the
+		// DO's proof of identity, so a client sending its own must not be believed.
+		request.headers.set(OWNER_HEADER, String(playerId))
+
+		return c.env.RECFLARE_NOTIFICATIONS_HUB.getByName(HUB_INSTANCE).fetch(request)
 	})
 
 	// ---- Internal service-to-service send/broadcast --------------------------
@@ -134,5 +168,31 @@ const app = new Hono<App>()
 		return c.json({ success: true, ...result })
 	})
 
+	// Read-only view of the hub's routing state, for working out why a notification
+	// didn't arrive: which connections are live, which players each one receives for,
+	// and what's queued for a player who wasn't reachable.
+	.get('/internal/hub-state', async (c) => {
+		return c.json(await c.env.RECFLARE_NOTIFICATIONS_HUB.getByName(HUB_INSTANCE).inspect())
+	})
+
+	// Discard queued notifications, for `?playerId=` or — with the explicit `?all=true`,
+	// so a bare call can't do it by accident — the whole queue. Anything left pending is
+	// delivered on the player's next subscribe, so stale frames need a way out.
+	.delete('/internal/hub-state/pending', async (c) => {
+		const raw = c.req.query('playerId')
+		const playerId = raw === undefined ? undefined : Number.parseInt(raw, 10)
+		if (playerId !== undefined && !Number.isInteger(playerId)) {
+			return c.json({ error: 'playerId must be an integer' }, 400)
+		}
+		if (playerId === undefined && c.req.query('all') !== 'true') {
+			return c.json({ error: 'pass playerId, or all=true to clear every queue' }, 400)
+		}
+
+		const result =
+			await c.env.RECFLARE_NOTIFICATIONS_HUB.getByName(HUB_INSTANCE).clearPending(playerId)
+		logger.info('cleared pending notifications', { playerId: playerId ?? null, ...result })
+		return c.json({ success: true, ...result })
+	})
+
 export { NotificationsHub }
 export default app

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

@@ -5,6 +5,7 @@ import { beforeAll, describe, expect, test } from 'vitest'
 import '../../notify.app'
 
 import type { Env } from '../../context'
+import type { HubState } from '../../notifications-hub'
 
 declare module 'cloudflare:test' {
 	interface ProvidedEnv extends Env {}
@@ -22,9 +23,13 @@ function b64url(input: ArrayBuffer | string): string {
 	for (const byte of bytes) binary += String.fromCharCode(byte)
 	return btoa(binary).replace(/\+/g, '-').replace(/\//g, '_').replace(/=+$/, '')
 }
-async function bearer(sub: string, roles?: string[]): Promise<Record<string, string>> {
+async function bearer(
+	sub: string,
+	roles?: string[],
+	expiresIn = 3600
+): Promise<Record<string, string>> {
 	const now = Math.floor(Date.now() / 1000)
-	const claims: Record<string, unknown> = { sub, exp: now + 3600 }
+	const claims: Record<string, unknown> = { sub, exp: now + expiresIn }
 	if (roles) claims.role = roles
 	const signingInput = `${b64url(JSON.stringify({ alg: 'HS256', typ: 'JWT' }))}.${b64url(
 		JSON.stringify(claims)
@@ -45,6 +50,9 @@ beforeAll(async () => {
 	await adminSecretsStore(env.JWT_SECRET).create(TEST_SECRET)
 })
 
+/** Who `connect` is by default — kept clear of the player ids the tests notify. */
+const DEFAULT_CONNECT_PLAYER = 9000
+
 interface HubRecord {
 	type?: number
 	target?: string
@@ -55,10 +63,14 @@ interface HubRecord {
 
 /** Open a hub WebSocket, accept it, and complete the SignalR handshake. */
 async function connect(
-	id: string
+	id: string,
+	opts: { headers?: Record<string, string>; query?: string } = {}
 ): Promise<{ ws: WebSocket; waitFor: (pred: (r: HubRecord) => boolean) => Promise<HubRecord> }> {
-	const res = await exports.default.fetch(`${ORIGIN}/hub/v1?id=${id}`, {
+	// The hub only accepts identified connections, so default to a token; pass explicit
-		headers: { Upgrade: 'websocket' },
+	// `headers` (`{}` for none) where the test is about who is connecting.
+	const auth = opts.headers ?? (await bearer(String(DEFAULT_CONNECT_PLAYER), ['gameClient']))
+	const res = await exports.default.fetch(`${ORIGIN}/hub/v1?id=${id}${opts.query ?? ''}`, {
+		headers: { Upgrade: 'websocket', ...auth },
 	})
 	expect(res.status).toBe(101)
 	const ws = res.webSocket!
@@ -185,6 +197,48 @@ describe('hub protocol', () => {
 		ws.close()
 	})
 
+	// The client's argument spelling isn't specified anywhere, and getting it wrong is
+	// silent: subscribing replaces the connection's set, so a misread leaves it at zero
+	// and every notification queues instead of being delivered.
+	test.each([
+		['an options object', [{ playerIds: [11, 12] }]],
+		['a single array argument', [[11, 12]]],
+		['varargs', [11, 12]],
+		['string ids', [{ playerIds: ['11', '12'] }]],
+	])('SubscribeToPlayers accepts %s', async (label, args) => {
+		const { ws, waitFor } = await connect(`conn-args-${label.replace(/\s+/g, '-')}`)
+		send(ws, { type: 1, target: 'SubscribeToPlayers', arguments: args })
+		send(ws, { type: 1, invocationId: 'g', target: 'GetSubscriptions', arguments: [] })
+		const result = await waitFor((r) => r.type === 3 && r.invocationId === 'g')
+		expect((result.result as number[]).sort((a, b) => a - b)).toEqual([11, 12])
+		ws.close()
+	})
+
+	test('an unreadable argument leaves existing subscriptions alone', async () => {
+		const { ws, waitFor } = await connect('conn-args-bad')
+		send(ws, { type: 1, target: 'SubscribeToPlayers', arguments: [{ playerIds: [21] }] })
+
+		// Answered, but the connection keeps the players it already had rather than
+		// being wiped to zero.
+		send(ws, { type: 1, invocationId: 'b', target: 'SubscribeToPlayers', arguments: ['nonsense'] })
+		await waitFor((r) => r.type === 3 && r.invocationId === 'b')
+
+		send(ws, { type: 1, invocationId: 'g', target: 'GetSubscriptions', arguments: [] })
+		const result = await waitFor((r) => r.type === 3 && r.invocationId === 'g')
+		expect(result.result).toEqual([21])
+		ws.close()
+	})
+
+	test('an explicitly empty list still clears the connection', async () => {
+		const { ws, waitFor } = await connect('conn-args-empty')
+		send(ws, { type: 1, target: 'SubscribeToPlayers', arguments: [{ playerIds: [31] }] })
+		send(ws, { type: 1, target: 'SubscribeToPlayers', arguments: [{ playerIds: [] }] })
+		send(ws, { type: 1, invocationId: 'g', target: 'GetSubscriptions', arguments: [] })
+		const result = await waitFor((r) => r.type === 3 && r.invocationId === 'g')
+		expect(result.result).toEqual([])
+		ws.close()
+	})
+
 	test('responds to ping', async () => {
 		const { ws, waitFor } = await connect('conn-ping')
 		send(ws, { type: 6 })
@@ -274,7 +328,9 @@ describe('notification delivery', () => {
 		expect(payloadA.Msg).toMatchObject({ FromPlayerId: 1, Type: 100, Data: 'hello all' })
 
 		const noteB = await b.waitFor((r) => r.type === 1 && r.target === 'Notification')
-		const payloadB = JSON.parse((noteB.arguments as string[])[0]) as { Msg: Record<string, unknown> }
+		const payloadB = JSON.parse((noteB.arguments as string[])[0]) as {
+			Msg: Record<string, unknown>
+		}
 		expect(payloadB.Msg).toMatchObject({ Data: 'hello all' })
 
 		a.ws.close()
@@ -309,3 +365,219 @@ describe('notification delivery', () => {
 		ws.close()
 	})
 })
+
+// The client never calls SubscribeToPlayers, so a player's own notifications reach them
+// only because the connect established who they are.
+describe('connection ownership', () => {
+	const hubState = async () =>
+		(await (
+			await exports.default.fetch(`${ORIGIN}/internal/hub-state`, {
+				headers: await bearer('1', ['gameClient', 'moderator']),
+			})
+		).json()) as HubState
+
+	// The client never calls SubscribeToPlayers, so connecting is the only moment a
+	// queue can drain — with the flush living only in there, anything that landed while
+	// the player was reconnecting stayed queued forever.
+	test('drains what queued while the player was away, on connect', async () => {
+		const playerId = 9305
+		await post('/internal/notify', { playerId, notificationType: 90, data: { n: 1 } })
+		await post('/internal/notify', { playerId, notificationType: 90, data: { n: 2 } })
+
+		const { ws, waitFor } = await connect('conn-flush', {
+			headers: await bearer(String(playerId), ['gameClient']),
+		})
+
+		// Delivered in the order they queued, and cleared once sent.
+		const first = await waitFor(
+			(r) => r.target === 'Notification' && (r.arguments as string[])[0].includes('"n":1')
+		)
+		expect(JSON.parse((first.arguments as string[])[0])).toEqual({ Id: '90', Msg: { n: 1 } })
+		await waitFor(
+			(r) => r.target === 'Notification' && (r.arguments as string[])[0].includes('"n":2')
+		)
+		expect((await hubState()).pending.find((p) => p.playerId === playerId)).toBeUndefined()
+
+		ws.close()
+	})
+
+	test('delivers to the connecting player without any subscription', async () => {
+		const playerId = 9301
+		const { ws, waitFor } = await connect('conn-owned', {
+			headers: await bearer(String(playerId), ['gameClient']),
+		})
+
+		const res = await post('/internal/notify', {
+			playerId,
+			notificationType: 90,
+			data: { chatThreadId: 22 },
+		})
+		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: '90',
+			Msg: { chatThreadId: 22 },
+		})
+
+		const connection = (await hubState()).connections.find((c) => c.connectionId === 'conn-owned')
+		expect(connection).toMatchObject({ playerId, playerIds: [] })
+
+		ws.close()
+	})
+
+	// SignalR clients that can't set headers on the upgrade put the token here instead.
+	test('accepts the token as an access_token query param', async () => {
+		const playerId = 9302
+		const auth = await bearer(String(playerId), ['gameClient'])
+		const token = auth.Authorization.slice('Bearer '.length)
+		const { ws, waitFor } = await connect('conn-owned-query', {
+			headers: {},
+			query: `&access_token=${token}`,
+		})
+
+		await post('/internal/notify', { playerId, notificationType: 90, data: { a: 1 } })
+		const note = await waitFor((r) => r.type === 1 && r.target === 'Notification')
+		expect(JSON.parse((note.arguments as string[])[0])).toMatchObject({ Id: '90' })
+
+		ws.close()
+	})
+
+	// The header is the DO's proof of identity, so presenting one without a token must
+	// not get you in — otherwise anyone could name themselves and receive that player's
+	// notifications.
+	test('a client-supplied owner header does not authenticate a connect', async () => {
+		const res = await exports.default.fetch(`${ORIGIN}/hub/v1?id=conn-spoofed`, {
+			headers: { Upgrade: 'websocket', 'x-recflare-connection-owner': '9303' },
+		})
+		expect(res.status).toBe(401)
+
+		const notified = await post('/internal/notify', { playerId: 9303, notificationType: 90 })
+		expect(await notified.json()).toMatchObject({ delivered: 0, queued: true })
+		expect(
+			(await hubState()).connections.find((c) => c.connectionId === 'conn-spoofed')
+		).toBeUndefined()
+	})
+
+	test('an unidentified connect is refused', async () => {
+		const res = await exports.default.fetch(`${ORIGIN}/hub/v1?id=conn-anon`, {
+			headers: { Upgrade: 'websocket' },
+		})
+		expect(res.status).toBe(401)
+	})
+
+	test('an expired token is refused', async () => {
+		const expired = await bearer(String(9304), ['gameClient'], -60)
+		const res = await exports.default.fetch(`${ORIGIN}/hub/v1?id=conn-expired`, {
+			headers: { Upgrade: 'websocket', ...expired },
+		})
+		expect(res.status).toBe(401)
+	})
+})
+
+describe('hub state', () => {
+	const hubState = async (auth?: Record<string, string>) =>
+		exports.default.fetch(`${ORIGIN}/internal/hub-state`, {
+			headers: auth ?? (await bearer('1', ['gameClient', 'moderator'])),
+		})
+
+	test('reports a connection, who it receives for, and what is queued', async () => {
+		const subscribed = 9101
+		const offline = 9102
+		const { ws, waitFor } = await connect('conn-state')
+		send(ws, {
+			type: 1,
+			invocationId: 's',
+			target: 'SubscribeToPlayers',
+			arguments: [{ playerIds: [subscribed] }],
+		})
+		await waitFor((r) => r.type === 3 && r.invocationId === 's')
+
+		// Nobody is subscribed to `offline`, so this one queues instead of delivering —
+		// the case a missing push is usually hiding in.
+		await post('/internal/notify', { playerId: offline, notificationType: 90, data: { a: 1 } })
+
+		// One DO is shared across the file, so other tests' connections are in here too.
+		const state = (await (await hubState()).json()) as HubState
+		expect(state.connections.find((c) => c.connectionId === 'conn-state')).toEqual({
+			connectionId: 'conn-state',
+			live: true,
+			handshakeDone: true,
+			playerId: DEFAULT_CONNECT_PLAYER,
+			playerIds: [subscribed],
+		})
+		const queued = state.pending.find((p) => p.playerId === offline)
+		expect(queued?.count).toBe(1)
+		expect(JSON.parse(queued!.latest)).toEqual({ Id: '90', Msg: { a: 1 } })
+		expect(state.pending.find((p) => p.playerId === subscribed)).toBeUndefined()
+
+		ws.close()
+	})
+
+	test('is admin-gated like the other internal endpoints', async () => {
+		expect((await exports.default.fetch(`${ORIGIN}/internal/hub-state`)).status).toBe(401)
+		expect((await hubState(await bearer('3', ['gameClient']))).status).toBe(403)
+	})
+})
+
+describe('clearing pending notifications', () => {
+	const clear = async (query: string, auth?: Record<string, string>) =>
+		exports.default.fetch(`${ORIGIN}/internal/hub-state/pending?${query}`, {
+			method: 'DELETE',
+			headers: auth ?? (await bearer('1', ['gameClient', 'moderator'])),
+		})
+
+	const pendingFor = async (playerId: number) => {
+		const state = (await (
+			await exports.default.fetch(`${ORIGIN}/internal/hub-state`, {
+				headers: await bearer('1', ['gameClient', 'moderator']),
+			})
+		).json()) as HubState
+		return state.pending.find((p) => p.playerId === playerId)
+	}
+
+	const queue = async (playerId: number, count: number) => {
+		for (let i = 0; i < count; i++) {
+			await post('/internal/notify', { playerId, notificationType: 90, data: { i } })
+		}
+	}
+
+	test('clears one player, leaving everyone else queued', async () => {
+		await queue(9201, 2)
+		await queue(9202, 1)
+
+		const res = await clear('playerId=9201')
+		expect(res.status).toBe(200)
+		expect(await res.json()).toEqual({ success: true, cleared: 2 })
+		expect(await pendingFor(9201)).toBeUndefined()
+		expect((await pendingFor(9202))?.count).toBe(1)
+
+		// Clearing a queue that's already empty is a no-op, not an error.
+		expect(await (await clear('playerId=9201')).json()).toEqual({ success: true, cleared: 0 })
+	})
+
+	test('clears every queue only when all=true is explicit', async () => {
+		await queue(9203, 1)
+
+		const guarded = await clear('')
+		expect(guarded.status).toBe(400)
+		expect((await pendingFor(9203))?.count).toBe(1)
+
+		const res = await clear('all=true')
+		expect(res.status).toBe(200)
+		expect(((await res.json()) as { cleared: number }).cleared).toBeGreaterThanOrEqual(1)
+		expect(await pendingFor(9203)).toBeUndefined()
+	})
+
+	test('400s on a non-numeric playerId', async () => {
+		expect((await clear('playerId=nope')).status).toBe(400)
+	})
+
+	test('is admin-gated like the other internal endpoints', async () => {
+		const res = await exports.default.fetch(`${ORIGIN}/internal/hub-state/pending?all=true`, {
+			method: 'DELETE',
+		})
+		expect(res.status).toBe(401)
+		expect((await clear('all=true', await bearer('3', ['gameClient']))).status).toBe(403)
+	})
+})