diff --git a/.changeset/realtime-wire-key.md b/.changeset/realtime-wire-key.md new file mode 100644 index 00000000..de3f28a9 --- /dev/null +++ b/.changeset/realtime-wire-key.md @@ -0,0 +1,5 @@ +--- +'@aws-blocks/bb-realtime': patch +--- + +Align local Realtime WebSocket message envelopes with the AWS runtime by using `data` for published messages. diff --git a/packages/bb-realtime/DESIGN.md b/packages/bb-realtime/DESIGN.md index f1600f3a..e04eeae6 100644 --- a/packages/bb-realtime/DESIGN.md +++ b/packages/bb-realtime/DESIGN.md @@ -62,7 +62,7 @@ Both are HMAC-SHA256 signed with the same shared secret. The connect token's cha - Connection pool keyed by `wsUrl` (e.g., `ws://localhost:3001/realtime`) - Subscribe: `{ action: 'subscribe', channel, token }` - Response: `{ type: 'subscribe_success', channel }` or `{ type: 'error', channel, message }` -- Data: `{ type: 'message', channel, payload }` +- Data: `{ type: 'message', channel, data }` ### AWS Middleware @@ -177,7 +177,6 @@ Channel path: my-app-collab/chat/room-123 | Behavior Difference | Impact | Mitigation | |------------|--------|------------| -| WS message uses `payload` field (mock) vs `data` (AWS) | Native mobile clients see different wire format | Abstracted by SDK middlewares; document for native implementors | | Mock skips connect-time token auth (`?token=` query param) | A client can still *open* a WS locally, but can no longer *subscribe* without a valid channel token (enforced at subscribe time) | Recommend sandbox testing for connect-time auth-sensitive flows | | No connection duration limit locally | Mock WS stays open indefinitely | Document the 2-hour AWS limit in README | | Single-process only | No cross-process pub/sub | Local dev is single-process | diff --git a/packages/bb-realtime/src/index.ts b/packages/bb-realtime/src/index.ts index 0eab3b80..f1367954 100644 --- a/packages/bb-realtime/src/index.ts +++ b/packages/bb-realtime/src/index.ts @@ -130,7 +130,7 @@ export const Realtime: { validatePublishSize(fullChannel, data); globalEmitter.emit(fullChannel, data); if (getBroadcastBus()) { - getBroadcastBus()!.emit('broadcast', { channel: fullChannel, payload: data }); + getBroadcastBus()!.emit('broadcast', { channel: fullChannel, data }); } } diff --git a/packages/bb-realtime/src/mock-middleware.ts b/packages/bb-realtime/src/mock-middleware.ts index 908ee285..85f24090 100644 --- a/packages/bb-realtime/src/mock-middleware.ts +++ b/packages/bb-realtime/src/mock-middleware.ts @@ -93,7 +93,7 @@ function doConnect(wsUrl: string) { if (data.type === 'subscribe_success' && data.channel) { const pending = conn.pendingEstablished.get(data.channel); if (pending) { - pending.forEach(p => p.resolve()); + pending.forEach(p => { p.resolve(); }); conn.pendingEstablished.delete(data.channel); } } else if (data.type === 'error' && data.channel) { @@ -101,7 +101,7 @@ function doConnect(wsUrl: string) { if (pending) { const err = new Error(data.message || 'Subscription rejected'); err.name = 'ConnectionFailedException'; - pending.forEach(p => p.reject(err)); + pending.forEach(p => { p.reject(err); }); conn.pendingEstablished.delete(data.channel); } // Remove the subscription entry since it was rejected @@ -110,7 +110,7 @@ function doConnect(wsUrl: string) { } else if (data.type === 'message' && data.channel) { const handlers = conn.subscriptions.get(data.channel); if (handlers) { - handlers.forEach(h => { try { h(data.payload); } catch (e) { console.error('[Realtime] Handler error:', e); } }); + handlers.forEach(h => { try { h(data.data); } catch (e) { console.error('[Realtime] Handler error:', e); } }); } } } catch (e) { console.error('[Realtime] Parse error:', e); } diff --git a/packages/bb-realtime/src/ws-server.test.ts b/packages/bb-realtime/src/ws-server.test.ts index 1b9b81dc..ce04c1e8 100644 --- a/packages/bb-realtime/src/ws-server.test.ts +++ b/packages/bb-realtime/src/ws-server.test.ts @@ -13,7 +13,7 @@ import { describe, it, before, after } from 'node:test'; import assert from 'node:assert'; import { createServer, type Server } from 'node:http'; -import { AddressInfo } from 'node:net'; +import type { AddressInfo } from 'node:net'; import { WebSocket } from 'ws'; import { attach, closeWebSocketServer, localRealtimeBus } from './ws-server.js'; import { LOCAL_TOKEN_SECRET } from './local-dev.js'; @@ -98,21 +98,48 @@ describe('WebSocket server: subscribe authorization', () => { // Give the intruder's rejected subscribe time to settle, then broadcast. await new Promise((r) => setTimeout(r, 50)); - const authedGotMessage = new Promise((resolve) => { + const authedGotMessage = new Promise<{ type: string; channel: string; data: { sender: string; text: string } }>((resolve) => { authed.on('message', (data) => { const msg = JSON.parse(data.toString()); - if (msg.type === 'message') resolve(msg.payload); + if (msg.type === 'message') resolve(msg); }); }); - localRealtimeBus.emit('broadcast', { channel: CHANNEL, payload: { sender: 'alice', text: 'Top secret' } }); + localRealtimeBus.emit('broadcast', { channel: CHANNEL, data: { sender: 'alice', text: 'Top secret' } }); - const payload = await authedGotMessage; - assert.strictEqual(payload.text, 'Top secret', 'authorized client should receive the broadcast'); + const message = await authedGotMessage; + assert.deepStrictEqual(message.data, { sender: 'alice', text: 'Top secret' }); + assert.strictEqual('payload' in message, false, 'local messages must use the AWS runtime data key'); assert.strictEqual(intruderGotMessage, false, 'unauthorized client must not receive the broadcast'); authed.close(); intruder.close(); }); + + it('uses the data key when forwarding a client publish', async () => { + const token = mintChannelToken(CHANNEL, LOCAL_TOKEN_SECRET); + const { ws: subscriber } = await subscribe({ channel: CHANNEL, token }); + + const received = new Promise<{ type: string; channel: string; data: { text: string } }>((resolve) => { + subscriber.on('message', (data) => { + const message = JSON.parse(data.toString()); + if (message.type === 'message') resolve(message); + }); + }); + + const publisher = new WebSocket(`ws://localhost:${port}/realtime`); + await new Promise((resolve, reject) => { + publisher.on('open', resolve); + publisher.on('error', reject); + }); + publisher.send(JSON.stringify({ action: 'publish', channel: CHANNEL, payload: { text: 'from client' } })); + + const message = await received; + assert.deepStrictEqual(message.data, { text: 'from client' }); + assert.strictEqual('payload' in message, false, 'local messages must use the AWS runtime data key'); + + publisher.close(); + subscriber.close(); + }); }); describe('Mock middleware: token replay on reconnect (regression)', () => { @@ -140,7 +167,7 @@ describe('Mock middleware: token replay on reconnect (regression)', () => { // Wait for the middleware to reconnect and resubscribe (backoff starts ~1s). await new Promise((r) => setTimeout(r, 1500)); - localRealtimeBus.emit('broadcast', { channel: CHANNEL, payload: { text: 'after reconnect' } }); + localRealtimeBus.emit('broadcast', { channel: CHANNEL, data: { text: 'after reconnect' } }); // Allow the broadcast to round-trip to the reconnected client. await new Promise((r) => setTimeout(r, 200)); diff --git a/packages/bb-realtime/src/ws-server.ts b/packages/bb-realtime/src/ws-server.ts index 9aa62345..119661be 100644 --- a/packages/bb-realtime/src/ws-server.ts +++ b/packages/bb-realtime/src/ws-server.ts @@ -59,7 +59,7 @@ export function attach(httpServer: Server) { } else if (msg.action === 'unsubscribe' && msg.channel) { subscription.channels.delete(msg.channel); } else if (msg.action === 'publish' && msg.channel && msg.payload !== undefined) { - const outMsg = JSON.stringify({ type: 'message', channel: msg.channel, payload: msg.payload }); + const outMsg = JSON.stringify({ type: 'message', channel: msg.channel, data: msg.payload }); for (const [otherWs, otherSub] of clients) { if (otherWs !== ws && otherSub.channels.has(msg.channel) && otherWs.readyState === WebSocket.OPEN) { otherWs.send(outMsg); @@ -75,8 +75,8 @@ export function attach(httpServer: Server) { ws.on('error', () => { clients.delete(ws); }); }); - localRealtimeBus.on('broadcast', ({ channel, payload }) => { - const message = JSON.stringify({ type: 'message', channel, payload }); + localRealtimeBus.on('broadcast', ({ channel, data }) => { + const message = JSON.stringify({ type: 'message', channel, data }); for (const [ws, sub] of clients) { if (sub.channels.has(channel) && ws.readyState === WebSocket.OPEN) { ws.send(message);