Revert "chore(@rocket.chat/ddp-client): introduce experimental Meteor independent DDP client (RocketChat#40301)"

This reverts commit fa4b2b4.

Author: ggazzo

.changeset/use-rc-sdk-transport-setting.md

@@ -2,23 +2,12 @@ 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 { isSdkTransportEnabled } from '../../../../client/lib/sdk/sdkTransportEnabled';
-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 Meteor's connection. With the SDK transport
-// flag on, *also* register on the SDK socket so presence messages arriving on
-// either WS feed the same streamerCentral. With the flag off, only Meteor's
-// connection is used — duplicating the registration via the meteor-backed sdk
-// proxy would re-feed every frame back through streamerCentral via two paths.
 streamerCentral.getStreamer('user-presence', { ddpConnection: Meteor.connection });
-if (isSdkTransportEnabled()) {
-	streamerCentral.setupDdpConnection('user-presence', createDdpSdkStreamerAdapter(getDdpSdk()));
-}
 
 type args = [username: string, statusChanged?: UserStatus, statusText?: string];
 

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

@@ -28,10 +28,6 @@ Meteor.startup(() => {
 		injectIntoHead('noreferrer', `<meta name="referrer" content="${value}" />`);
 	});
 
-	settings.watch<boolean>('Use_RC_SDK', (value) => {
-		injectIntoHead('Use_RC_SDK', `<meta name="rc-sdk-transport-enabled" content="${value ? 'on' : 'off'}" />`);
-	});
-
 	if (process.env.DISABLE_ANIMATION) {
 		injectIntoHead(
 			'disable-animation',

apps/meteor/app/ui-master/server/index.ts

@@ -6,8 +6,6 @@ import { DDPCommon } from 'meteor/ddp-common';
 import { Meteor } from 'meteor/meteor';
 
 import { APIClient } from './RestApiClient';
-import { ensureConnectedAndAuthenticated, getDdpSdk } from '../../../../client/lib/sdk/ddpSdk';
-import { isSdkTransportEnabled } from '../../../../client/lib/sdk/sdkTransportEnabled';
 
 declare module '@rocket.chat/ddp-client' {
 	// eslint-disable-next-line @typescript-eslint/naming-convention
@@ -21,8 +19,6 @@ declare module '@rocket.chat/ddp-client' {
 	}
 }
 
-const sdkTransportEnabled = isSdkTransportEnabled();
-
 const isChangedCollectionPayload = (
 	msg: any,
 ): msg is { msg: 'changed'; collection: string; fields: { eventName: string; args: unknown[] } } => {
@@ -146,113 +142,13 @@ const createNewMeteorStream = (streamName: StreamNames, key: StreamKeys<StreamNa
 	};
 };
 
-const createNewDdpSdkStream = (
-	streamProxy: Emitter<EventMap>,
-	streamName: StreamNames,
-	key: StreamKeys<StreamNames>,
-	args: unknown[],
-): StreamMapValue => {
-	const ee = new Emitter<{
-		ready: [error: any] | [undefined, any];
-		error: [error: any];
-		stop: undefined;
-	}>();
-	const meta = { ready: false };
-
-	// Defer the actual `subscribe` until DDPSDK is authenticated. Without this,
-	// stream subscriptions fired immediately after re-login (e.g. the
-	// SubscriptionsCachedStore's `notify-user/<uid>/subscriptions-changed`
-	// listener that re-arms via onLoggedIn) hit the SDK socket while it's
-	// still anonymous — server rejects with `not-allowed`/`nosub`, the
-	// stream's `ready` promise emits an error, and the cached store never
-	// receives subsequent server events. The visible failure: an agent that
-	// just took a livechat chat post-relogin sees the chat work but the
-	// "Move to the queue" button never appears, because the new subscription
-	// the server creates for that agent is never replicated to the client's
-	// Subscriptions store, and pseudoRoom (= {...sub, ...room}) ends up with
-	// no `u` for the canMoveQueue check.
-	let subscription: ReturnType<ReturnType<typeof getDdpSdk>['client']['subscribe']> | undefined;
-	let offCollection: (() => void) | undefined;
-	let stopped = false;
-
-	void ensureConnectedAndAuthenticated()
-		.catch(() => undefined)
-		.then(() => {
-			if (stopped) return;
-			const sdk = getDdpSdk();
-			subscription = sdk.client.subscribe(`stream-${streamName}`, key, { useCollection: false, args });
-
-			subscription
-				.ready()
-				.then(() => {
-					if (stopped) return;
-					meta.ready = true;
-					ee.emit('ready', [undefined, { msg: 'ready', subs: [subscription!.id] }]);
-				})
-				.catch((err) => {
-					if (stopped) return;
-					ee.emit('ready', [err]);
-					ee.emit('error', err);
-				});
-
-			offCollection = sdk.client.onCollection(`stream-${streamName}`, (data: any) => {
-				if (data?.msg !== 'changed') return;
-				if (data.collection !== `stream-${streamName}`) return;
-				if (data.fields?.eventName !== key) return;
-				streamProxy.emit(`stream-${streamName}/${key}` as keyof EventMap, data.fields.args);
-			});
-		});
-
-	const onChange: ReturnType<ClientStream['subscribe']>['onChange'] = (cb) => {
-		if (meta.ready) {
-			cb({ msg: 'ready', subs: [] });
-			return;
-		}
-		ee.once('ready', ([error, result]) => {
-			if (error) {
-				cb({ msg: 'nosub', id: '', error });
-				return;
-			}
-			cb(result);
-		});
-	};
-
-	return {
-		stop: () => {
-			// Mirror Meteor's subscription semantics: explicit stop() does not fire the
-			// 'stop' event (onStop is reserved for server-initiated closures).
-			// Emitting it here would recurse through the onStop handler that
-			// createStreamManager registers, which itself iterates the unsubList.
-			stopped = true;
-			offCollection?.();
-			subscription?.stop();
-		},
-		onChange,
-		ready: () => {
-			if (meta.ready) return Promise.resolve();
-			return new Promise<void>((resolve, reject) => {
-				ee.once('ready', ([err]) => {
-					if (err) {
-						reject(err);
-						return;
-					}
-					resolve();
-				});
-			});
-		},
-		onError: (cb: (...args: any[]) => void) => ee.once('error', (error) => cb(error)),
-		onStop: (cb: () => void) => ee.once('stop', cb),
-		get isReady() {
-			return meta.ready;
-		},
-		unsubList: new Set(),
-	};
-};
-
 const createStreamManager = () => {
 	// Emitter that replicates stream messages to registered callbacks
 	const streamProxy = new Emitter<EventMap>();
 
+	// Collection of unsubscribe callbacks for each stream.
+	// const proxyUnsubLists = new Map<string, Set<() => void>>();
+
 	const streams = new Map<string, StreamMapValue>();
 
 	Accounts.onLogout(() => {
@@ -261,20 +157,13 @@ const createStreamManager = () => {
 		});
 	});
 
-	if (!sdkTransportEnabled) {
-		// In legacy Meteor mode, stream frames arrive on Meteor.connection._stream
-		// as `changed` collection messages — bridge them into streamProxy so the
-		// per-stream callbacks fire. With SDK transport on, the frames arrive on
-		// the SDK socket and createNewDdpSdkStream registers its own onCollection
-		// listener instead.
-		Meteor.connection._stream.on('message', (rawMsg: string) => {
-			const msg = DDPCommon.parseDDP(rawMsg);
-			if (!isChangedCollectionPayload(msg)) {
-				return;
-			}
-			streamProxy.emit(`${msg.collection}/${msg.fields.eventName}` as any, msg.fields.args as any);
-		});
-	}
+	Meteor.connection._stream.on('message', (rawMsg: string) => {
+		const msg = DDPCommon.parseDDP(rawMsg);
+		if (!isChangedCollectionPayload(msg)) {
+			return;
+		}
+		streamProxy.emit(`${msg.collection}/${msg.fields.eventName}` as any, msg.fields.args as any);
+	});
 
 	const stream: SDK['stream'] = <N extends StreamNames, K extends StreamKeys<N>>(
 		name: N,
@@ -297,11 +186,7 @@ const createStreamManager = () => {
 
 		streamProxy.on(eventLiteral, proxyCallback);
 
-		const stream =
-			streams.get(eventLiteral) ||
-			(sdkTransportEnabled
-				? createNewDdpSdkStream(streamProxy, name as StreamNames, key as StreamKeys<StreamNames>, args)
-				: createNewMeteorStream(name as StreamNames, key as StreamKeys<StreamNames>, args));
+		const stream = streams.get(eventLiteral) || createNewMeteorStream(name, key, args);
 
 		const stop = (): void => {
 			streamProxy.off(eventLiteral, proxyCallback);
@@ -356,26 +241,30 @@ const createStreamManager = () => {
 export const createSDK = (rest: RestClientInterface) => {
 	const { stream, stopAll } = createStreamManager();
 
-	const publish = sdkTransportEnabled
-		? (name: string, args: unknown[]) => {
-				// DDPSDK queues outbound frames until the WebSocket handshake completes,
-				// so there's no need to gate on an isReady flag here.
-				void getDdpSdk().client.callAsync(`stream-${name}`, ...args);
-			}
-		: (name: string, args: unknown[]) => {
-				Meteor.call(`stream-${name}`, ...args);
-			};
+	const publish = (name: string, args: unknown[]) => {
+		Meteor.call(`stream-${name}`, ...args);
+	};
 
 	const call = <T extends keyof ServerMethods>(method: T, ...args: Parameters<ServerMethods[T]>): Promise<ReturnType<ServerMethods[T]>> => {
 		return Meteor.callAsync(method, ...args);
 	};
 
+	const disconnect = () => {
+		Meteor.disconnect();
+	};
+
+	const reconnect = () => {
+		Meteor.reconnect();
+	};
+
 	return {
 		rest,
 		stop: stopAll,
 		stream,
 		publish,
 		call,
+		disconnect,
+		reconnect,
 	};
 };
 

apps/meteor/app/utils/client/lib/SDKClient.ts

@@ -50,11 +50,7 @@ export abstract class CachedStore<T extends IRocketChatRecord, U = T> implements
 
 	protected eventType: StreamNames;
 
-	// Bumped from 18 → 19 to invalidate caches populated before the DDPSDK
-	// wire encoding was switched from JSON to EJSON. Entries written by the
-	// JSON window stored dates as ISO strings instead of Date instances, so
-	// fields like subscription.ls would fail `.getTime()` when read back.
-	private readonly version = 19;
+	private readonly version = 18;
 
 	private updatedAt = new Date(0);
 

apps/meteor/client/lib/cachedStores/CachedStore.ts

@@ -1,61 +1,20 @@
 import { Accounts } from 'meteor/accounts-base';
 
-import { isSdkTransportEnabled } from './sdk/sdkTransportEnabled';
-import { getUserId, userIdStore } from './user';
-
-const sdkTransportEnabled = isSdkTransportEnabled();
+import { getUserId } from './user';
 
 const isLoggedIn = () => {
 	const uid = getUserId();
 	return !!uid;
 };
 
-/**
- * Fire `cb` whenever the local userId transitions from absent → present.
- *
- * `Accounts.onLogin` would normally cover this, but Meteor only invokes
- * the onLogin hook from inside a Tracker.autorun that waits for
- * `Meteor.userAsync()` to resolve to a real user doc. When a login goes
- * through our REST fallback (e.g. logout → fresh login while DDPSDK is
- * reconnecting), the user document never lands in Meteor.users — it
- * normally arrives as a DDP collection frame, but the REST endpoint
- * only returns the method result. The autorun then sees a null user
- * forever, and onLogin never fires. By piggybacking on userIdStore (which
- * is updated synchronously the moment Accounts.connection.userId() is
- * set), we get a reliable login signal regardless of how the user doc
- * eventually arrives.
- */
-const subscribeToLogin = (handler: () => void): (() => void) => {
-	let lastSeen = userIdStore.getState();
-	return userIdStore.subscribe((next) => {
-		if (next === lastSeen) return;
-		const wasLoggedOut = !lastSeen;
-		lastSeen = next;
-		if (next && wasLoggedOut) {
-			handler();
-		}
-	});
-};
-
 export const whenLoggedIn = () => {
 	if (isLoggedIn()) {
 		return Promise.resolve();
 	}
 
-	if (!sdkTransportEnabled) {
-		// Flag off: develop's exact implementation — wait on Accounts.onLogin only,
-		// no userIdStore bridge.
-		return new Promise<void>((resolve) => {
-			const subscription = Accounts.onLogin(() => {
-				subscription.stop();
-				resolve();
-			});
-		});
-	}
-
 	return new Promise<void>((resolve) => {
-		const stop = subscribeToLogin(() => {
-			stop();
+		const subscription = Accounts.onLogin(() => {
+			subscription.stop();
 			resolve();
 		});
 	});
@@ -71,18 +30,11 @@ export const onLoggedIn = (cb: (() => () => void) | (() => Promise<() => void>)
 		}
 	};
 
-	// With the SDK transport on, login can land via REST (ddpOverREST) without
-	// filling Meteor.users — Accounts.onLogin's autorun would never fire.
-	// Bridge off userIdStore as belt-and-braces. With the flag off, the legacy
-	// DDP path populates Meteor.users and Accounts.onLogin fires reliably; the
-	// extra userIdStore subscription would just double-fire callbacks.
-	const accountsSubscription = Accounts.onLogin(handler);
-	const stopUserIdSubscription = sdkTransportEnabled ? subscribeToLogin(handler) : undefined;
+	const subscription = Accounts.onLogin(handler);
 	if (isLoggedIn()) handler();
 
 	return () => {
-		accountsSubscription.stop();
-		stopUserIdSubscription?.();
+		subscription.stop();
 		cleanup?.();
 	};
 };

apps/meteor/client/lib/loggedIn.ts

@@ -4,30 +4,8 @@ import type { EventHandlerOf } from '@rocket.chat/emitter';
 import { Emitter } from '@rocket.chat/emitter';
 import { Meteor } from 'meteor/meteor';
 
-import { getDdpSdk } from './sdk/ddpSdk';
-import { isSdkTransportEnabled } from './sdk/sdkTransportEnabled';
 import { sdk } from '../../app/utils/client/lib/SDKClient';
 
-const sdkTransportEnabled = isSdkTransportEnabled();
-
-const subscribeUserPresence = (payload: { added?: string[]; removed?: string[] }): void => {
-	if (!sdkTransportEnabled) {
-		// Flag off: route directly through Meteor.subscribe — bit-for-bit develop
-		// behaviour, no DDPSDK socket created, no proxy in the call path.
-		Meteor.subscribe('stream-user-presence', '', payload);
-		return;
-	}
-	const ddp = getDdpSdk();
-	if (ddp.connection.status === 'connected' && ddp.account.uid) {
-		// Fire the command-style subscription over our SDK; it has no lifecycle
-		// (the server registers the added/removed uids and moves on), matching
-		// Meteor.subscribe's behaviour here.
-		ddp.client.subscribe('stream-user-presence', '', payload);
-		return;
-	}
-	Meteor.subscribe('stream-user-presence', '', payload);
-};
-
 type InternalEvents = {
 	remove: IUser['_id'];
 	reset: undefined;
@@ -77,7 +55,7 @@ const getPresence = ((): ((uid: UserPresence['_id']) => void) => {
 			const ids = Array.from(currentUids);
 			const removed = Array.from(deletedUids);
 
-			subscribeUserPresence({
+			Meteor.subscribe('stream-user-presence', '', {
 				...(ids.length > 0 && { added: Array.from(currentUids) }),
 				...(removed.length && { removed: Array.from(deletedUids) }),
 			});