fixup notifications

apps/notify/package.json

@@ -16,6 +16,7 @@
 	},
 	"dependencies": {
 		"@repo/hono-helpers": "workspace:*",
+		"@repo/jwt": "workspace:*",
 		"hono": "4.12.27",
 		"workers-tagged-logger": "1.0.1"
 	},

apps/notify/src/context.ts

@@ -5,6 +5,12 @@ import type { NotificationsHub } from './notifications-hub'
 export type Env = SharedHonoEnv & {
 	/** Durable Object hosting the SignalR notifications hub. */
 	RECFLARE_NOTIFICATIONS_HUB: DurableObjectNamespace<NotificationsHub>
+	/**
+	 * HS256 JWT signing key, read with `await env.JWT_SECRET.get()`; all workers
+	 * bind the same store so tokens signed by `auth` verify here. Used to gate the
+	 * internal send/broadcast endpoints.
+	 */
+	JWT_SECRET: SecretsStoreSecret
 }
 
 /** Variables can be extended */

apps/notify/src/notifications-hub.ts

@@ -276,7 +276,10 @@ export class NotificationsHub extends DurableObject<Env> {
 	/**
 	 * Build the `Notification` argument: a JSON string `{ Id, Msg }`
 	 * (null values are dropped from `Msg`). `Id` is a client-defined tag — a
-	 * string name (e.g. "AccountUpdate") or a numeric code.
+	 * string name (e.g. "AccountUpdate") or a numeric code. It is always emitted
+	 * as a string: the client dispatches on a string `Id`, so a numeric frame
+	 * (e.g. the `NotificationType` enum values sent by `econ`) would otherwise be
+	 * silently dropped.
 	 */
 	private buildNotificationPayload(
 		notificationType: string | number,
@@ -289,7 +292,7 @@ export class NotificationsHub extends DurableObject<Env> {
 				msg[key] = value
 			}
 		}
-		return JSON.stringify({ Id: notificationType ?? '', Msg: msg })
+		return JSON.stringify({ Id: String(notificationType), Msg: msg })
 	}
 
 	private invocation(target: string, args: unknown[]): string {

apps/notify/src/notify.app.ts

@@ -2,10 +2,12 @@ import { Hono } from 'hono'
 import { useWorkersLogger } from 'workers-tagged-logger'
 
 import { logger, withNotFound, withOnError } from '@repo/hono-helpers'
+import { validateAndGetAccountId } from '@repo/jwt'
 
 import { NotificationsHub } from './notifications-hub'
 
 import type { App } from './context'
+import type { MiddlewareHandler } from 'hono'
 
 /**
  * Maps a SignalR hub at `/hub/v1`. The hub itself — WebSocket transport, the
@@ -26,6 +28,25 @@ function isNotificationType(value: unknown): value is string | number {
 	return (typeof value === 'string' && value !== '') || typeof value === 'number'
 }
 
+/**
+ * Account ids allowed to call the internal send/broadcast endpoints. Temporary
+ * lockdown until these are properly gated — for now only these admins can push
+ * notifications through the shared hub.
+ */
+const ADMIN_ACCOUNT_IDS = new Set([1, 2])
+
+/**
+ * Gates the `/internal/*` endpoints on a valid Bearer token whose `sub` is an
+ * allowed admin account. 401 for a missing/invalid token, 403 for a valid token
+ * that isn't an admin.
+ */
+const requireAdmin: MiddlewareHandler<App> = async (c, next) => {
+	const accountId = await validateAndGetAccountId(c.req.raw, await c.env.JWT_SECRET.get())
+	if (accountId === null) return c.json({ error: 'Unauthorized' }, 401)
+	if (!ADMIN_ACCOUNT_IDS.has(accountId)) return c.json({ error: 'Forbidden' }, 403)
+	await next()
+}
+
 const app = new Hono<App>()
 	.use(
 		'*',
@@ -64,8 +85,10 @@ const app = new Hono<App>()
 	})
 
 	// ---- Internal service-to-service send/broadcast --------------------------
-	// Lets other workers push notifications through the shared hub.
+	// Lets other workers push notifications through the shared hub. Gated to admin
-	// TODO: protect these before production.
+	// accounts (see requireAdmin) as a temporary lockdown.
+	.use('/internal/*', requireAdmin)
+
 	.post('/internal/notify', async (c) => {
 		const body = await c.req
 			.json<{

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

@@ -1,11 +1,48 @@
+import { adminSecretsStore, env } from 'cloudflare:test'
 import { exports } from 'cloudflare:workers'
-import { describe, expect, test } from 'vitest'
+import { beforeAll, describe, expect, test } from 'vitest'
 
 import '../../notify.app'
 
+import type { Env } from '../../context'
+
+declare module 'cloudflare:test' {
+	interface ProvidedEnv extends Env {}
+}
+
 const ORIGIN = 'https://example.com'
 const RS = '\u001e'
 
+// Mint a token the way the `auth` worker does, signing with the shared test key
+// seeded into the JWT_SECRET store. Accounts 1 and 2 are the internal-endpoint admins.
+const TEST_SECRET = 'test-signing-key'
+function b64url(input: ArrayBuffer | string): string {
+	const bytes = typeof input === 'string' ? new TextEncoder().encode(input) : new Uint8Array(input)
+	let binary = ''
+	for (const byte of bytes) binary += String.fromCharCode(byte)
+	return btoa(binary).replace(/\+/g, '-').replace(/\//g, '_').replace(/=+$/, '')
+}
+async function bearer(sub: string): Promise<Record<string, string>> {
+	const now = Math.floor(Date.now() / 1000)
+	const signingInput = `${b64url(JSON.stringify({ alg: 'HS256', typ: 'JWT' }))}.${b64url(
+		JSON.stringify({ sub, exp: now + 3600 })
+	)}`
+	const key = await crypto.subtle.importKey(
+		'raw',
+		new TextEncoder().encode(TEST_SECRET),
+		{ name: 'HMAC', hash: 'SHA-256' },
+		false,
+		['sign']
+	)
+	const sig = await crypto.subtle.sign('HMAC', key, new TextEncoder().encode(signingInput))
+	return { Authorization: `Bearer ${signingInput}.${b64url(sig)}` }
+}
+
+// Seed the shared JWT signing key into the local Secrets Store so .get() resolves.
+beforeAll(async () => {
+	await adminSecretsStore(env.JWT_SECRET).create(TEST_SECRET)
+})
+
 interface HubRecord {
 	type?: number
 	target?: string
@@ -67,10 +104,15 @@ async function connect(
 
 const send = (ws: WebSocket, msg: HubRecord) => ws.send(JSON.stringify(msg) + RS)
 
-const post = (path: string, body: unknown) =>
+// The /internal/* endpoints are admin-gated, so default to an admin (account 1)
+// Bearer token; pass `auth` to override (e.g. to test the 401/403 paths).
+const post = async (path: string, body: unknown, auth?: Record<string, string>) =>
 	exports.default.fetch(`${ORIGIN}${path}`, {
 		method: 'POST',
-		headers: { 'Content-Type': 'application/json' },
+		headers: {
+			'Content-Type': 'application/json',
+			...(auth ?? (await bearer('1'))),
+		},
 		body: JSON.stringify(body),
 	})
 
@@ -96,6 +138,31 @@ describe('negotiate', () => {
 	})
 })
 
+describe('internal endpoint auth', () => {
+	const body = { playerId: 1, notificationType: 1, data: {} }
+
+	test('401 without a Bearer token', async () => {
+		const res = await exports.default.fetch(`${ORIGIN}/internal/notify`, {
+			method: 'POST',
+			headers: { 'Content-Type': 'application/json' },
+			body: JSON.stringify(body),
+		})
+		expect(res.status).toBe(401)
+	})
+
+	test('403 for a valid token that is not an admin', async () => {
+		const res = await post('/internal/notify', body, await bearer('3'))
+		expect(res.status).toBe(403)
+	})
+
+	test('admin accounts 1 and 2 are allowed', async () => {
+		for (const sub of ['1', '2']) {
+			const res = await post('/internal/broadcast', { notificationType: 1 }, await bearer(sub))
+			expect(res.status).toBe(200)
+		}
+	})
+})
+
 describe('hub protocol', () => {
 	test('GetSubscriptions reflects SubscribeToPlayers', async () => {
 		const { ws, waitFor } = await connect('conn-subs')
@@ -134,7 +201,10 @@ describe('notification delivery', () => {
 		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' } })
+		expect(JSON.parse((note.arguments as string[])[0])).toEqual({
+			Id: '2',
+			Msg: { messageId: 'm1' },
+		})
 
 		ws.close()
 	})
@@ -158,7 +228,10 @@ describe('notification delivery', () => {
 		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 } })
+		expect(JSON.parse((note.arguments as string[])[0])).toEqual({
+			Id: '1',
+			Msg: { accountId: 42 },
+		})
 
 		ws.close()
 	})
@@ -172,9 +245,32 @@ describe('notification delivery', () => {
 		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,
+			Id: '25',
 			Msg: { message: 'maint' },
 		})
 		ws.close()
 	})
+
+	test('emits a numeric notificationType as a string Id', async () => {
+		// The client dispatches on a string Id, so numeric codes (e.g. econ's
+		// NotificationType enum) must be serialized as strings or they're dropped.
+		const playerId = 9003
+		const { ws, waitFor } = await connect('conn-numeric-id')
+		send(ws, {
+			type: 1,
+			invocationId: 's',
+			target: 'SubscribeToPlayers',
+			arguments: [{ playerIds: [playerId] }],
+		})
+		await waitFor((r) => r.type === 3 && r.invocationId === 's')
+
+		await post('/internal/notify', { playerId, notificationType: 71, data: { itemId: 5 } })
+
+		const note = await waitFor((r) => r.type === 1 && r.target === 'Notification')
+		const payload = JSON.parse((note.arguments as string[])[0]) as { Id: unknown }
+		expect(payload.Id).toBe('71')
+		expect(typeof payload.Id).toBe('string')
+
+		ws.close()
+	})
 })

apps/notify/wrangler.jsonc

@@ -8,6 +8,16 @@
 		"bindings": [{ "name": "RECFLARE_NOTIFICATIONS_HUB", "class_name": "NotificationsHub" }]
 	},
 	"migrations": [{ "tag": "v1", "new_sqlite_classes": ["NotificationsHub"] }],
+	// Shared Secrets Store holding the HS256 JWT signing key. Every worker binds the
+	// same store as JWT_SECRET so tokens signed by `auth` verify here. The "local"
+	// store_id placeholder is replaced with RECFLARE_SECRETS_STORE at deploy time.
+	"secrets_store_secrets": [
+		{
+			"binding": "JWT_SECRET",
+			"store_id": "local",
+			"secret_name": "JWT_SECRET"
+		}
+	],
 	"logpush": false,
 	"upload_source_maps": true,
 	"observability": {