fix: multiple stream subscribe after reconnection

.changeset/light-beans-sparkle.md

@@ -0,0 +1,5 @@
+---
+"@rocket.chat/meteor": patch
+---
+
+Fixes an issue, where multiple reconnections would subscribe multiple times to the same stream, only a frontend issue, since the stream cache prevents sending multiple times to the backend, but does not prevent running the callback multiple times

apps/meteor/client/hooks/useLoadRoomForAllowedAnonymousRead.ts

@@ -11,6 +11,10 @@ export const useLoadRoomForAllowedAnonymousRead = () => {
 		if (!userId && accountsAllowAnonymousRead === true) {
 			CachedChatRoom.init();
 			CachedChatSubscription.ready.set(true);
+			return () => {
+				CachedChatRoom.ready.set(false);
+				CachedChatSubscription.ready.set(false);
+			};
 		}
 	}, [accountsAllowAnonymousRead, userId]);
 };

apps/meteor/client/lib/cachedCollections/CachedCollection.ts

@@ -199,8 +199,8 @@ export abstract class CachedCollection<T extends IRocketChatRecord, U = T> {
 		this.collection.state.replaceAll([]);
 	}
 
-	protected async setupListener() {
-		sdk.stream(this.eventType, [this.eventName], (async (action: 'removed' | 'changed', record: U) => {
+	protected setupListener() {
+		return sdk.stream(this.eventType, [this.eventName], (async (action: 'removed' | 'changed', record: U) => {
 			this.log('record received', action, record);
 			await this.handleRecordEvent(action, record);
 		}) as (...args: unknown[]) => void);
@@ -295,7 +295,9 @@ export abstract class CachedCollection<T extends IRocketChatRecord, U = T> {
 		return true;
 	}
 
-	async init() {
+	private listenerUnsubscriber: (() => void) | undefined;
+
+	private async performInitialization() {
 		if (await this.loadFromCache()) {
 			this.trySync();
 		} else {
@@ -318,7 +320,34 @@ export abstract class CachedCollection<T extends IRocketChatRecord, U = T> {
 			}
 		});
 
-		return this.setupListener();
+		const subscription = this.setupListener();
+		this.listenerUnsubscriber = () => {
+			subscription.stop();
+			this.listenerUnsubscriber = undefined;
+		};
+	}
+
+	private initializationPromise: Promise<void> | undefined;
+
+	init() {
+		if (this.initializationPromise) {
+			return this.initializationPromise;
+		}
+
+		this.initializationPromise = this.performInitialization().finally(() => {
+			this.initializationPromise = undefined;
+		});
+
+		return this.initializationPromise;
+	}
+
+	async release() {
+		if (this.initializationPromise) {
+			await this.initializationPromise;
+		}
+
+		this.listenerUnsubscriber?.();
+		this.ready.set(false);
 	}
 
 	private reconnectionComputation: Tracker.Computation | undefined;
@@ -353,7 +382,7 @@ export class PrivateCachedCollection<T extends IRocketChatRecord, U = T> extends
 		});
 
 		Accounts.onLogout(() => {
-			this.ready.set(false);
+			this.release();
 		});
 	}
 }

apps/meteor/client/lib/cachedCollections/DocumentMapStore.ts

@@ -249,8 +249,9 @@ export const createDocumentMapStore = <T extends { _id: string }>({
 				const records = new Map<T['_id'], T>();
 				for (const record of state.records.values()) {
 					if (predicate(record)) {
-						if (onInvalidate) affected.push(record);
-						records.set(record._id, modifier(record));
+						const newRecord = modifier(record);
+						records.set(record._id, newRecord);
+						if (onInvalidate) affected.push(newRecord);
 					} else {
 						records.set(record._id, record);
 					}
@@ -267,8 +268,9 @@ export const createDocumentMapStore = <T extends { _id: string }>({
 
 			for await (const record of get().records.values()) {
 				if (predicate(record)) {
-					if (onInvalidate) affected.push(record);
-					records.set(record._id, await modifier(record));
+					const newRecord = await modifier(record);
+					records.set(record._id, newRecord);
+					if (onInvalidate) affected.push(newRecord);
 				} else {
 					records.set(record._id, record);
 				}

apps/meteor/client/lib/settings/PrivateSettingsCachedCollection.ts

@@ -11,12 +11,16 @@ class PrivateSettingsCachedCollection extends PrivateCachedCollection<ISetting>
 		});
 	}
 
-	async setupListener(): Promise<void> {
-		sdk.stream('notify-logged', [this.eventName as 'private-settings-changed'], async (t: string, { _id, ...record }: { _id: string }) => {
-			this.log('record received', t, { _id, ...record });
-			this.collection.update({ _id }, { $set: record }, { upsert: true });
-			this.sync();
-		});
+	override setupListener() {
+		return sdk.stream(
+			'notify-logged',
+			[this.eventName as 'private-settings-changed'],
+			async (t: string, { _id, ...record }: { _id: string }) => {
+				this.log('record received', t, { _id, ...record });
+				this.collection.update({ _id }, { $set: record }, { upsert: true });
+				this.sync();
+			},
+		);
 	}
 }