FIX #5721 RxDB does not respond to collection.remove() changes (#6013)

* FIX #5721 RxDB does not respond to collection.remove() changes

* FIX lint

CHANGELOG.md

@@ -2,6 +2,8 @@
 # RxDB Changelog
 
 <!-- CHANGELOG NEWEST -->
+- FIX `collection.remove()` must end up with the correct RxCollection state across tabs. [5721](https://github.com/pubkey/rxdb/issues/5721)
+- ADD `RxCollection.onRemove` hooks to detect the removing of a RxCollection across tabs.
 - IMPROVE performance of insert to [IndexedDB](https://rxdb.info/rx-storage-indexeddb.html)
 
 <!-- ADD new changes here! -->

docs-src/docs/rx-collection.md

@@ -278,6 +278,18 @@ Destroys the collection's object instance. This is to free up memory and stop al
 await myCollection.destroy();
 ```
 
+### onDestroy / onRemove()
+With these you can add a function that is run when the collection was destroyed or removed.
+This works even across multiple browser tabs so you can detect when another tab removes the collection
+and you application can behave accordingly.
+
+```js
+await myCollection.onDestroy(() => console.log('I am destroyed'));
+await myCollection.onRemove(() => console.log('I am removed'));
+```
+
+
+
 
 ### isRxCollection
 Returns true if the given object is an instance of RxCollection. Returns false if not.

src/plugins/dev-mode/error-messages.ts

@@ -96,6 +96,7 @@ export const ERROR_MESSAGES = {
     COL18: 'collection-method not allowed because fieldname is in the schema',
     // removed in 14.0.0, use CONFLICT instead - COL19: 'Document update conflict. When changing a document you must work on the previous revision',
     COL20: 'Storage write error',
+    COL21: 'The RxCollection is destroyed or removed already, either from this JavaScript realm or from another, like a browser tab',
     CONFLICT: 'Document update conflict. When changing a document you must work on the previous revision',
 
     // rx-document.js

src/rx-collection-helper.ts

@@ -1,6 +1,7 @@
 import type {
     HashFunction,
     InternalStoreDocType,
+    RxCollection,
     RxDatabase,
     RxDocumentData,
     RxJsonSchema,
@@ -24,6 +25,8 @@ import { runAsyncPluginHooks } from './hooks.ts';
 import { getAllCollectionDocuments } from './rx-database-internal-store.ts';
 import { flatCloneDocWithMeta } from './rx-storage-helper.ts';
 import { overwritable } from './overwritable.ts';
+import type { RxCollectionBase } from './rx-collection.ts';
+import { newRxError } from './rx-error.ts';
 
 /**
  * fills in the default data.
@@ -167,3 +170,18 @@ export async function removeCollectionStorages(
         );
     }
 }
+
+
+export function ensureRxCollectionIsNotDestroyed(
+    collection: RxCollection | RxCollectionBase<any, any, any, any, any>
+) {
+    if (collection.destroyed) {
+        throw newRxError(
+            'COL21',
+            {
+                collection: collection.name,
+                version: collection.schema.version
+            }
+        );
+    }
+}

src/rx-collection.ts

@@ -17,7 +17,8 @@ import {
 import {
     fillObjectDataBeforeInsert,
     createRxCollectionStorageInstance,
-    removeCollectionStorages
+    removeCollectionStorages,
+    ensureRxCollectionIsNotDestroyed
 } from './rx-collection-helper.ts';
 import {
     createRxQuery,
@@ -184,10 +185,12 @@ export class RxCollectionBase<
      * these functions will be called an awaited.
      * Used to automatically clean up stuff that
      * belongs to this collection.
-     */
+    */
     public onDestroy: (() => MaybePromise<any>)[] = [];
     public destroyed = false;
 
+    public onRemove: (() => MaybePromise<any>)[] = [];
+
     public async prepare(): Promise<void> {
         this.storageInstance = getWrappedStorageInstance(
             this.database,
@@ -224,8 +227,28 @@ export class RxCollectionBase<
             }
         );
 
+
+        const listenToRemoveSub = this.database.internalStore.changeStream().pipe(
+            filter(bulk => {
+                const key = this.name + '-' + this.schema.version;
+                const found = bulk.events.find(event => {
+                    return (
+                        event.documentData.context === 'collection' &&
+                        event.documentData.key === key &&
+                        event.operation === 'DELETE'
+                    );
+                });
+                return !!found;
+            })
+        ).subscribe(async () => {
+            await this.destroy();
+            await Promise.all(this.onRemove.map(fn => fn()));
+        });
+        this._subs.push(listenToRemoveSub);
+
+
         /**
-         * Instead of resolving the EventBulk array here and spit it into
+         * TODO Instead of resolving the EventBulk array here and spit it into
          * single events, we should fully work with event bulks internally
          * to save performance.
          */
@@ -290,6 +313,7 @@ export class RxCollectionBase<
      * @link https://rxdb.info/cleanup.html
      */
     cleanup(_minimumDeletedTime?: number): Promise<boolean> {
+        ensureRxCollectionIsNotDestroyed(this);
         throw pluginMissing('cleanup');
     }
 
@@ -301,6 +325,7 @@ export class RxCollectionBase<
         throw pluginMissing('migration-schema');
     }
     startMigration(batchSize: number = 10): Promise<void> {
+        ensureRxCollectionIsNotDestroyed(this);
         return this.getMigrationState().startMigration(batchSize);
     }
     migratePromise(batchSize: number = 10): Promise<any> {
@@ -310,6 +335,7 @@ export class RxCollectionBase<
     async insert(
         json: RxDocumentType | RxDocument
     ): Promise<RxDocument<RxDocumentType, OrmMethods>> {
+        ensureRxCollectionIsNotDestroyed(this);
         const writeResult = await this.bulkInsert([json as any]);
 
         const isError = writeResult.error[0];
@@ -324,6 +350,7 @@ export class RxCollectionBase<
         success: RxDocument<RxDocumentType, OrmMethods>[];
         error: RxStorageWriteError<RxDocumentType>[];
     }> {
+        ensureRxCollectionIsNotDestroyed(this);
         /**
          * Optimization shortcut,
          * do nothing when called with an empty array
@@ -400,6 +427,7 @@ export class RxCollectionBase<
         success: RxDocument<RxDocumentType, OrmMethods>[];
         error: RxStorageWriteError<RxDocumentType>[];
     }> {
+        ensureRxCollectionIsNotDestroyed(this);
         const primaryPath = this.schema.primaryPath;
         /**
          * Optimization shortcut,
@@ -469,6 +497,7 @@ export class RxCollectionBase<
         success: RxDocument<RxDocumentType, OrmMethods>[];
         error: RxStorageWriteError<RxDocumentType>[];
     }> {
+        ensureRxCollectionIsNotDestroyed(this);
         const insertData: RxDocumentType[] = [];
         const useJsonByDocId: Map<string, RxDocumentType> = new Map();
         docsData.forEach(docData => {
@@ -514,6 +543,7 @@ export class RxCollectionBase<
      * same as insert but overwrites existing document with same primary
      */
     async upsert(json: Partial<RxDocumentType>): Promise<RxDocument<RxDocumentType, OrmMethods>> {
+        ensureRxCollectionIsNotDestroyed(this);
         const bulkResult = await this.bulkUpsert([json]);
         throwIfIsStorageWriteError<RxDocumentType>(
             this.asRxCollection,
@@ -528,6 +558,7 @@ export class RxCollectionBase<
      * upserts to a RxDocument, uses incrementalModify if document already exists
      */
     incrementalUpsert(json: Partial<RxDocumentType>): Promise<RxDocument<RxDocumentType, OrmMethods>> {
+        ensureRxCollectionIsNotDestroyed(this);
         const useJson = fillObjectDataBeforeInsert(this.schema, json);
         const primary: string = useJson[this.schema.primaryPath] as any;
         if (!primary) {
@@ -560,6 +591,7 @@ export class RxCollectionBase<
         OrmMethods,
         Reactivity
     > {
+        ensureRxCollectionIsNotDestroyed(this);
         if (typeof queryObj === 'string') {
             throw newRxError('COL5', {
                 queryObj
@@ -582,6 +614,7 @@ export class RxCollectionBase<
         OrmMethods,
         Reactivity
     > {
+        ensureRxCollectionIsNotDestroyed(this);
 
         // TODO move this check to dev-mode plugin
         if (
@@ -628,6 +661,7 @@ export class RxCollectionBase<
         OrmMethods,
         Reactivity
     > {
+        ensureRxCollectionIsNotDestroyed(this);
         if (!queryObj) {
             queryObj = _getDefaultQuery();
         }
@@ -647,6 +681,7 @@ export class RxCollectionBase<
         OrmMethods,
         Reactivity
     > {
+        ensureRxCollectionIsNotDestroyed(this);
         const mangoQuery: MangoQuery<RxDocumentType> = {
             selector: {
                 [this.schema.primaryPath]: {
@@ -849,6 +884,7 @@ export class RxCollectionBase<
      */
     async remove(): Promise<any> {
         await this.destroy();
+        await Promise.all(this.onRemove.map(fn => fn()));
         await removeCollectionStorages(
             this.database.storage,
             this.database.internalStore,

test/unit/rx-collection.test.ts

@@ -1120,6 +1120,54 @@ describe('rx-collection.test.ts', () => {
 
                     db2.destroy();
                 });
+                it('#5721 should remove the RxCollection instance across tabs and emit the .$removed event', async () => {
+                    if (!config.storage.hasMultiInstance) {
+                        return;
+                    }
+
+                    const dbName = randomCouchString();
+
+                    async function createDb() {
+                        const db = await createRxDatabase<{ humans: RxCollection<HumanDocumentType>; }>({
+                            name: dbName,
+                            storage: config.storage.getStorage(),
+                            ignoreDuplicate: true
+                        });
+                        await db.addCollections({
+                            humans: { schema: schemas.human }
+                        });
+                        await db.collections.humans.insert(schemaObjects.humanData());
+                        return db;
+                    }
+
+                    const db1 = await createDb();
+                    const db2 = await createDb();
+                    const col1 = db1.humans;
+                    const col2 = db2.humans;
+
+                    // remember the emitted events
+                    let emitted1 = false;
+                    let emitted2 = false;
+                    col1.onRemove.push(() => emitted1 = true);
+                    col2.onRemove.push(() => emitted2 = true);
+
+                    // remove collection2
+                    await col2.remove();
+
+                    await waitUntil(() => emitted1);
+                    await waitUntil(() => emitted2);
+
+                    // calling operations on other collection should also fail
+                    await assertThrows(
+                        () => col1.insert(schemaObjects.humanData()),
+                        'RxError',
+                        'COL21'
+                    );
+
+                    assert.deepStrictEqual(emitted1, emitted2);
+                    db1.destroy();
+                    db2.destroy();
+                });
             });
             describeParallel('.bulkRemove()', () => {
                 describe('positive', () => {