feat(sdk): bridge user-presence streamer onto DDPSDK listener

The client/lib/presence.ts subscribe already routes through DDPSDK when
the SDK is authenticated, but app/notifications/client/lib/Presence.ts
was still registering streamerCentral against Meteor.connection only.
That meant user-presence stream-changed frames arrived on the DDPSDK
socket while the listener was wired to Meteor — the UI would miss
status updates whenever the subscribe path picked DDPSDK.

Add a thin DDPSDK → StreamerDDPConnection adapter (createDdpSdkStreamerAdapter)
that exposes a compatible _stream.on('message', ...) surface by
re-serialising DDPSDK's parsed payloads with EJSON so StreamerCentral's
JSON-parse path round-trips dates identically to the raw Meteor frames.
Call streamerCentral.setupDdpConnection a second time with the adapter:
StreamerCentral tracks registration per connection object via
hasMeteorStreamerEventListeners, so both transports get a listener and
the streamerCentral emitter fires once per frame regardless of which
socket delivered it.

Author: ggazzo

apps/meteor/app/notifications/client/lib/Presence.ts

@@ -2,12 +2,21 @@ import { UserStatus } from '@rocket.chat/core-typings';
 import { Meteor } from 'meteor/meteor';
 
 import { Presence } from '../../../../client/lib/presence';
+import { getDdpSdk } from '../../../../client/lib/sdk/ddpSdk';
+import { createDdpSdkStreamerAdapter } from '../../../../client/lib/sdk/streamerAdapter';
 import { streamerCentral } from '../../../../client/lib/streamer';
 
 // TODO implement API on Streamer to be able to listen to all streamed data
 // this is a hacky way to listen to all streamed data from user-presence Streamer
 
+// Register the presence streamer on BOTH transports. The subscribe call in
+// client/lib/presence.ts routes through DDPSDK when it's ready and falls back
+// to Meteor otherwise, so the corresponding messages can arrive on either WS.
+// StreamerCentral uses a per-connection `hasMeteorStreamerEventListeners` flag,
+// so calling setupDdpConnection twice with distinct connection objects
+// installs both listeners without duplicating within the same transport.
 streamerCentral.getStreamer('user-presence', { ddpConnection: Meteor.connection });
+streamerCentral.setupDdpConnection('user-presence', createDdpSdkStreamerAdapter(getDdpSdk()));
 
 type args = [username: string, statusChanged?: UserStatus, statusText?: string];
 

apps/meteor/client/lib/sdk/streamerAdapter.ts

@@ -0,0 +1,49 @@
+import type { DDPSDK } from '@rocket.chat/ddp-client';
+import EJSON from 'ejson';
+
+type StreamerDDPConnection = {
+	_stream: {
+		on: {
+			(key: 'message', callback: (data: string) => void): void;
+			(key: 'reset', callback: () => void): void;
+		};
+	};
+	subscribe(name: string, ...args: unknown[]): { stop: () => void };
+	call(methodName: string, ...args: unknown[]): void;
+	hasMeteorStreamerEventListeners?: boolean;
+};
+
+/**
+ * Presents a DDPSDK instance with the shape `StreamerCentral` expects
+ * (the subset of Meteor.connection exposed as `StreamerDDPConnection`).
+ * Lets us run the existing streamer infrastructure against our SDK's
+ * WebSocket without rewriting StreamerCentral.
+ */
+export const createDdpSdkStreamerAdapter = (sdk: DDPSDK): StreamerDDPConnection => {
+	const ddp = (
+		sdk.client as unknown as { ddp: { onMessage: (cb: (payload: unknown) => void) => () => void } }
+	).ddp;
+
+	return {
+		_stream: {
+			on: ((key: 'message' | 'reset', callback: ((data: string) => void) | (() => void)): void => {
+				if (key === 'message') {
+					ddp.onMessage((payload) => {
+						// StreamerCentral re-parses the string with JSON.parse; hand it
+						// an EJSON-serialised payload so Dates/undefined round-trip the
+						// same way Meteor.connection's raw WS frames did.
+						(callback as (data: string) => void)(EJSON.stringify(payload));
+					});
+					return;
+				}
+				if (key === 'reset') {
+					sdk.connection.on('disconnected', callback as () => void);
+				}
+			}) as StreamerDDPConnection['_stream']['on'],
+		},
+		subscribe: (name: string, ...args: unknown[]) => sdk.client.subscribe(name, ...args),
+		call: (methodName: string, ...args: unknown[]): void => {
+			void sdk.client.callAsync(methodName, ...args);
+		},
+	};
+};