fix(bb-realtime): align local message envelopes with AWS runtime

.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.

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 |

packages/bb-realtime/src/index.ts

@@ -14,7 +14,7 @@
 import { Scope, registerSdkIdentifiers } from '@aws-blocks/core';
 import type { ScopeParent } from '@aws-blocks/core';
 import type { StandardSchemaV1 } from '@standard-schema/spec';
 import { EventEmitter } from 'events';
 import type {
 	NamespaceConfig,
 	NamespaceDefs,
@@ -24,7 +24,7 @@
 	RealtimeOptions,
 	SubscribeOptions,
 } from './types.js';
 import { RealtimeErrors } from './errors.js';
 import { blocksError, validateSchema, mintChannelToken, mintConnectToken, validateChannelPath, validatePublishSize } from './utils.js';
 import { getBroadcastBus, LOCAL_TOKEN_SECRET } from './local-dev.js';
 import { Logger } from '@aws-blocks/bb-logger';
@@ -130,7 +130,7 @@
 		validatePublishSize(fullChannel, data);
 		globalEmitter.emit(fullChannel, data);
 		if (getBroadcastBus()) {
-			getBroadcastBus()!.emit('broadcast', { channel: fullChannel, payload: data });
+			getBroadcastBus()!.emit('broadcast', { channel: fullChannel, data });
 		}
 	}
 

packages/bb-realtime/src/mock-middleware.ts

@@ -76,15 +76,15 @@
 			conn.isConnected = true;
 			conn.reconnectAttempts = 0;
 			for (const ch of conn.subscriptions.keys()) {
 				conn.ws!.send(JSON.stringify({ action: 'subscribe', channel: ch, token: conn.channelTokens.get(ch) }));
 			}
 			for (const { channel, token } of conn.pendingSubs) {
 				if (!conn.subscriptions.has(channel)) {
 					conn.ws!.send(JSON.stringify({ action: 'subscribe', channel, token }));
 				}
 			}
 			conn.pendingSubs.length = 0;
 			for (const msg of conn.pendingMessages) { conn.ws!.send(msg); }
 			conn.pendingMessages.length = 0;
 		};
 		conn.ws.onmessage = (event) => {
@@ -93,15 +93,15 @@
 				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) {
 					const pending = conn.pendingEstablished.get(data.channel);
 					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 @@
 				} 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); }
@@ -188,9 +188,9 @@
 	if (!conn.pendingEstablished.has(channel)) {
 		conn.pendingEstablished.set(channel, []);
 	}
 	conn.pendingEstablished.get(channel)!.push({ resolve: establishedResolve!, reject: establishedReject! });
 
 	conn.subscriptions.get(channel)!.add(handler);
 
 	return {
 		unsubscribe() {
@@ -205,7 +205,7 @@
 					conn.subscriptions.delete(channel);
 					conn.channelTokens.delete(channel);
 					if (conn.isConnected && conn.ws?.readyState === WebSocket.OPEN) {
 						conn.ws!.send(JSON.stringify({ action: 'unsubscribe', channel }));
 					}
 				}
 			}

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<any>((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<void>((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));

packages/bb-realtime/src/ws-server.ts

@@ -9,7 +9,7 @@
 
 import { WebSocketServer, WebSocket } from 'ws';
 import type { Server } from 'node:http';
 import { EventEmitter } from 'events';
 import { setBroadcastBus, LOCAL_TOKEN_SECRET } from './local-dev.js';
 import { validateChannelToken } from './utils.js';
 
@@ -59,7 +59,7 @@
 				} 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 @@
 		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);