fix(compass-app-stores): fixes stale listeners getting triggered on globalAppRegistry events (#6019)

Author: himanshusinghs

packages/compass-aggregations/src/modules/aggregation.ts

@@ -400,7 +400,7 @@ const fetchAggregationData = (
       logger: { log, mongoLogId },
       track,
       connectionInfoAccess,
-      globalAppRegistry,
+      connectionScopedAppRegistry,
     }
   ) => {
     const {
@@ -452,7 +452,7 @@ const fetchAggregationData = (
       });
 
       if (isMergeOrOut) {
-        globalAppRegistry.emit(
+        connectionScopedAppRegistry.emit(
           'agg-pipeline-out-executed',
           getDestinationNamespaceFromStage(
             namespace,

packages/compass-aggregations/src/modules/index.ts

@@ -109,7 +109,9 @@ export type PipelineBuilderExtraArgs = {
   instance: MongoDBInstance;
   dataService: DataService;
   connectionInfoAccess: ConnectionInfoAccess;
-  connectionScopedAppRegistry: ConnectionScopedAppRegistry<'open-export'>;
+  connectionScopedAppRegistry: ConnectionScopedAppRegistry<
+    'open-export' | 'view-edited' | 'agg-pipeline-out-executed'
+  >;
 };
 
 export type PipelineBuilderThunkDispatch<A extends Action = AnyAction> =

packages/compass-aggregations/src/modules/pipeline-builder/stage-editor.ts

@@ -389,7 +389,7 @@ export const runStage = (
   return async (
     dispatch,
     getState,
-    { pipelineBuilder, preferences, globalAppRegistry }
+    { pipelineBuilder, preferences, connectionScopedAppRegistry }
   ) => {
     const {
       dataService: { dataService },
@@ -442,7 +442,7 @@ export const runStage = (
         id: idx,
         previewDocs: result.map((doc) => new HadronDocument(doc)),
       });
-      globalAppRegistry.emit(
+      connectionScopedAppRegistry.emit(
         'agg-pipeline-out-executed',
         getDestinationNamespaceFromStage(
           namespace,

packages/compass-aggregations/src/modules/pipeline-builder/text-editor-output-stage.ts

@@ -131,7 +131,7 @@ export const runPipelineWithOutputStage = (): PipelineBuilderThunkAction<
   return async (
     dispatch,
     getState,
-    { pipelineBuilder, preferences, globalAppRegistry }
+    { pipelineBuilder, preferences, connectionScopedAppRegistry }
   ) => {
     const {
       autoPreview,
@@ -173,7 +173,7 @@ export const runPipelineWithOutputStage = (): PipelineBuilderThunkAction<
       dispatch({
         type: OutputStageActionTypes.FetchSucceded,
       });
-      globalAppRegistry.emit(
+      connectionScopedAppRegistry.emit(
         'agg-pipeline-out-executed',
         getDestinationNamespaceFromStage(
           namespace,

packages/compass-aggregations/src/modules/update-view.spec.ts

@@ -4,11 +4,26 @@ import { ERROR_UPDATING_VIEW, updateView } from './update-view';
 import { createNoopLogger } from '@mongodb-js/compass-logging/provider';
 import { createNoopTrack } from '@mongodb-js/compass-telemetry/provider';
 import AppRegistry from 'hadron-app-registry';
-import { TEST_CONNECTION_INFO } from '@mongodb-js/compass-connections/provider';
+import {
+  type ConnectionInfoAccess,
+  ConnectionScopedAppRegistryImpl,
+  TEST_CONNECTION_INFO,
+} from '@mongodb-js/compass-connections/provider';
 
 describe('update-view module', function () {
+  const globalAppRegistry = new AppRegistry();
+  const connectionInfoAccess: ConnectionInfoAccess = {
+    getCurrentConnectionInfo() {
+      return TEST_CONNECTION_INFO;
+    },
+  };
+  const connectionScopedAppRegistry = new ConnectionScopedAppRegistryImpl(
+    globalAppRegistry.emit.bind(globalAppRegistry),
+    connectionInfoAccess
+  );
   const thunkArg = {
-    globalAppRegistry: new AppRegistry(),
+    globalAppRegistry,
+    connectionScopedAppRegistry,
     localAppRegistry: new AppRegistry(),
     pipelineBuilder: {
       getPipelineFromStages() {

packages/compass-aggregations/src/modules/update-view.ts

@@ -87,7 +87,7 @@ export const updateView = (): PipelineBuilderThunkAction<Promise<void>> => {
       workspaces,
       logger: { debug },
       track,
-      globalAppRegistry,
+      connectionScopedAppRegistry,
       connectionInfoAccess,
     }
   ) => {
@@ -123,7 +123,7 @@ export const updateView = (): PipelineBuilderThunkAction<Promise<void>> => {
         connectionInfo
       );
       debug('selecting namespace', viewNamespace);
-      globalAppRegistry.emit('view-edited', viewNamespace);
+      connectionScopedAppRegistry.emit('view-edited', viewNamespace);
       workspaces.openCollectionWorkspace(connectionInfo.id, viewNamespace);
     } catch (e: any) {
       debug('Unexpected error updating view', e);

packages/compass-app-stores/src/stores/instance-store.spec.ts

@@ -121,6 +121,9 @@ describe('InstanceStore [Store]', function () {
     let connectedInstance: MongoDBInstance;
     let initialInstanceRefreshedPromise: Promise<unknown>;
     beforeEach(function () {
+      sinon
+        .stub(connectionsManager, 'getDataServiceForConnection')
+        .returns(dataService);
       connectionsManager.emit(
         ConnectionsManagerEvents.ConnectionAttemptSuccessful,
         connectedConnectionInfoId,
@@ -330,35 +333,35 @@ describe('InstanceStore [Store]', function () {
         });
       });
 
-      const createdEvents = ['agg-pipeline-out-executed'];
-
-      for (const evt of createdEvents) {
-        context(`on '${evt}' event`, function () {
-          it('should add collection to the databases collections', function () {
-            globalAppRegistry.emit(evt, 'foo.qux');
-            expect(
-              connectedInstance.databases.get('foo')?.collections
-            ).to.have.lengthOf(3);
-            expect(
-              connectedInstance.databases
-                .get('foo')
-                ?.collections.get('foo.qux', '_id')
-            ).to.exist;
+      context(`on 'agg-pipeline-out-executed' event`, function () {
+        it('should add collection to the databases collections', function () {
+          globalAppRegistry.emit('agg-pipeline-out-executed', 'foo.qux', {
+            connectionId: connectedConnectionInfoId,
           });
+          expect(
+            connectedInstance.databases.get('foo')?.collections
+          ).to.have.lengthOf(3);
+          expect(
+            connectedInstance.databases
+              .get('foo')
+              ?.collections.get('foo.qux', '_id')
+          ).to.exist;
+        });
 
-          it("should add new database and add collection to its collections if database doesn't exist yet", function () {
-            globalAppRegistry.emit(evt, 'bar.qux');
-            expect(connectedInstance.databases).to.have.lengthOf(2);
-            expect(connectedInstance.databases.get('bar')).to.exist;
-            expect(
-              connectedInstance.databases.get('bar')?.collections
-            ).to.have.lengthOf(1);
-            expect(
-              connectedInstance.databases.get('bar')?.collections.get('bar.qux')
-            ).to.exist;
+        it("should add new database and add collection to its collections if database doesn't exist yet", function () {
+          globalAppRegistry.emit('agg-pipeline-out-executed', 'bar.qux', {
+            connectionId: connectedConnectionInfoId,
           });
+          expect(connectedInstance.databases).to.have.lengthOf(2);
+          expect(connectedInstance.databases.get('bar')).to.exist;
+          expect(
+            connectedInstance.databases.get('bar')?.collections
+          ).to.have.lengthOf(1);
+          expect(
+            connectedInstance.databases.get('bar')?.collections.get('bar.qux')
+          ).to.exist;
         });
-      }
+      });
 
       context(`on 'collection-created' event`, function () {
         it('should not respond when the connectionId does not matches the connectionId for the instance', function () {
@@ -469,4 +472,68 @@ describe('InstanceStore [Store]', function () {
       });
     });
   });
+
+  context('when disconnected', function () {
+    it('should remove the instance from InstancesManager and should not perform any actions on the stale instance', async function () {
+      // first connect
+      connectionsManager.emit(
+        ConnectionsManagerEvents.ConnectionAttemptSuccessful,
+        connectedConnectionInfoId,
+        dataService
+      );
+
+      // setup a spy on old instance
+      const oldInstance = instancesManager.getMongoDBInstanceForConnection(
+        connectedConnectionInfoId
+      );
+      const oldFetchDatabasesSpy = sinon.spy(oldInstance, 'fetchDatabases');
+
+      // now disconnect
+      connectionsManager.emit(
+        ConnectionsManagerEvents.ConnectionDisconnected,
+        connectedConnectionInfoId
+      );
+
+      // there is no instance in store InstancesManager now
+      expect(() =>
+        instancesManager.getMongoDBInstanceForConnection(
+          connectedConnectionInfoId
+        )
+      ).to.throw;
+
+      // lets connect again and ensure that old instance does not receive events anymore
+      const newDataService = createDataService();
+      sinon
+        .stub(connectionsManager, 'getDataServiceForConnection')
+        .returns(dataService);
+      connectionsManager.emit(
+        ConnectionsManagerEvents.ConnectionAttemptSuccessful,
+        connectedConnectionInfoId,
+        newDataService
+      );
+
+      // setup a spy on new instance
+      const newInstance = instancesManager.getMongoDBInstanceForConnection(
+        connectedConnectionInfoId
+      );
+
+      let resolveFetchDatabasePromise: (() => void) | undefined;
+      const fetchDatabasePromise = new Promise<void>((resolve) => {
+        resolveFetchDatabasePromise = resolve;
+      });
+      const newFetchDatabasesSpy = sinon
+        .stub(newInstance, 'fetchDatabases')
+        .callsFake(() => {
+          resolveFetchDatabasePromise?.();
+          return Promise.resolve();
+        });
+
+      globalAppRegistry.emit('refresh-databases', {
+        connectionId: connectedConnectionInfoId,
+      });
+      await fetchDatabasePromise;
+      expect(oldFetchDatabasesSpy).to.not.be.called;
+      expect(newFetchDatabasesSpy).to.be.called;
+    });
+  });
 });