[studio] Complete editor uploads and queued cloud builds (#135)

* [studio] Complete editor uploads and queued cloud builds

* [studio] Protect unpublished source and remove unused notification RPC

apps/notify/src/notifications-hub.ts

@@ -31,6 +31,7 @@ const RS = '\u001e'
 interface SocketState {
 	connectionId: string
 	handshakeDone: boolean
+	channels?: string[]
 	/** The player this socket belongs to, from the token validated at connect. */
 	playerId?: number
 }
@@ -356,6 +357,38 @@ export class NotificationsHub extends DurableObject<Env> {
 
 	private handleInvocation(ws: WebSocket, connectionId: string, msg: HubMessage): void {
 		switch (msg.target) {
+			case 'SubscribeTo':
+			case 'UnsubscribeFrom': {
+				const argument = msg.arguments?.[0] as
+					{ Channels?: unknown; channels?: unknown } | undefined
+				const channels = argument?.Channels ?? argument?.channels
+				if (
+					!Array.isArray(channels) ||
+					channels.length > 128 ||
+					channels.some(
+						(channel) => typeof channel !== 'string' || !channel.length || channel.length > 256
+					)
+				) {
+					if (msg.invocationId)
+						ws.send(this.completionError(msg.invocationId, 'Invalid channel subscriptions.'))
+					break
+				}
+				const state = ws.deserializeAttachment() as SocketState
+				const current = new Set(state.channels ?? [])
+				for (const channel of channels) {
+					if (msg.target === 'SubscribeTo') current.add(channel)
+					else current.delete(channel)
+				}
+				if (current.size > 128) {
+					if (msg.invocationId)
+						ws.send(this.completionError(msg.invocationId, 'Too many channel subscriptions.'))
+					break
+				}
+				state.channels = [...current]
+				ws.serializeAttachment(state)
+				if (msg.invocationId) ws.send(this.completion(msg.invocationId, null))
+				break
+			}
 			case 'SubscribeToPlayers': {
 				const playerIds = parsePlayerIds(msg.arguments)
 				// An argument we can't read is not "subscribe to nobody": subscribing

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

@@ -65,6 +65,7 @@ const DEFAULT_CONNECT_PLAYER = 9000
 
 interface HubRecord {
 	type?: number
+	error?: string
 	target?: string
 	invocationId?: string
 	result?: unknown
@@ -197,6 +198,44 @@ describe('internal endpoint auth', () => {
 })
 
 describe('hub protocol', () => {
+	test('Studio named subscriptions merge and unsubscribe', async () => {
+		const { ws, waitFor } = await connect('conn-studio-channels')
+		const stub = env.RECFLARE_NOTIFICATIONS_HUB.getByName('global')
+		for (const [i, channel] of ['RRS:first', 'RRS:second', 'RRS:first'].entries()) {
+			send(ws, {
+				type: 1,
+				invocationId: `s${i}`,
+				target: 'SubscribeTo',
+				arguments: [{ Channels: [channel] }],
+			})
+			expect(
+				(await waitFor((r) => r.type === 3 && r.invocationId === `s${i}`)).error
+			).toBeUndefined()
+		}
+		const channels = () =>
+			runInDurableObject(
+				stub,
+				(_instance, state) =>
+					state
+						.getWebSockets()
+						.map(
+							(socket) =>
+								socket.deserializeAttachment() as { connectionId: string; channels?: string[] }
+						)
+						.find((attachment) => attachment.connectionId === 'conn-studio-channels')?.channels
+			)
+		expect(await channels()).toEqual(['RRS:first', 'RRS:second'])
+		send(ws, {
+			type: 1,
+			invocationId: 'u',
+			target: 'UnsubscribeFrom',
+			arguments: [{ Channels: ['RRS:first'] }],
+		})
+		expect((await waitFor((r) => r.type === 3 && r.invocationId === 'u')).error).toBeUndefined()
+		expect(await channels()).toEqual(['RRS:second'])
+		ws.close()
+	})
+
 	test('GetSubscriptions reflects SubscribeToPlayers', async () => {
 		const { ws, waitFor } = await connect('conn-subs')