FIX RxStorage should never emit an eventBulk with an empty events array

Author: pubkey

CHANGELOG.md

@@ -3,6 +3,8 @@
 
 <!-- CHANGELOG NEWEST -->
 
+- FIX: RxStorage should never emit an eventBulk with an empty events array.
+
 <!-- ADD new changes here! -->
 
 <!-- /CHANGELOG NEWEST -->

src/plugins/dexie/rx-storage-instance-dexie.ts

@@ -255,9 +255,11 @@ export class RxStorageInstanceDexie<RxDocType> implements RxStorageInstance<
                 ]);
             });
 
-        const endTime = now();
-        eventBulk.events.forEach(event => event.endTime = endTime);
-        this.changes$.next(eventBulk);
+        if (eventBulk.events.length > 0) {
+            const endTime = now();
+            eventBulk.events.forEach(event => event.endTime = endTime);
+            this.changes$.next(eventBulk);
+        }
 
         return ret;
     }

src/plugins/lokijs/rx-storage-instance-loki.ts

@@ -146,7 +146,10 @@ export class RxStorageInstanceLoki<RxDocType> implements RxStorageInstance<
             ret.error[err.documentId] = err;
         });
         localState.databaseState.saveQueue.addWrite();
-        this.changes$.next(categorized.eventBulk);
+
+        if (categorized.eventBulk.events.length > 0) {
+            this.changes$.next(categorized.eventBulk);
+        }
 
         return ret;
     }

src/plugins/memory/rx-storage-instance-memory.ts

@@ -143,7 +143,10 @@ export class RxStorageInstanceMemory<RxDocType> implements RxStorageInstance<
             );
         });
 
-        this.changes$.next(categorized.eventBulk);
+        if (categorized.eventBulk.events.length > 0) {
+            this.changes$.next(categorized.eventBulk);
+        }
+
         return Promise.resolve(ret);
     }
 

src/plugins/pouchdb/rx-storage-instance-pouch.ts

@@ -85,17 +85,17 @@ export class RxStorageInstancePouch<RxDocType> implements RxStorageInstance<
          */
         const emittedEventBulkIds: ObliviousSet<string> = new ObliviousSet(60 * 1000);
 
-        const eventSub = emitter.subject.subscribe(async (ev) => {
+        const eventSub = emitter.subject.subscribe(async (eventBulk) => {
             if (
-                ev.events.length === 0 ||
-                emittedEventBulkIds.has(ev.id)
+                eventBulk.events.length === 0 ||
+                emittedEventBulkIds.has(eventBulk.id)
             ) {
                 return;
             }
-            emittedEventBulkIds.add(ev.id);
+            emittedEventBulkIds.add(eventBulk.id);
 
             // rewrite primaryPath of all events
-            ev.events.forEach(event => {
+            eventBulk.events.forEach(event => {
                 if (event.change.doc) {
                     event.change.doc = pouchSwapIdToPrimary(
                         this.primaryPath,
@@ -110,7 +110,7 @@ export class RxStorageInstancePouch<RxDocType> implements RxStorageInstance<
                 }
             });
 
-            this.changes$.next(ev);
+            this.changes$.next(eventBulk);
         });
         this.subs.push(eventSub);
     }

test/unit/rx-storage-implementations.test.ts

@@ -1687,6 +1687,49 @@ config.parallel('rx-storage-implementations.test.js (implementation: ' + config.
                 assert.strictEqual(lastEvent.change.operation, 'DELETE');
                 assert.ok(lastEvent.change.previous);
 
+                sub.unsubscribe();
+                storageInstance.close();
+            });
+            it('it should not emit an empty eventBulk when the write had only errors', async () => {
+                const storageInstance = await config.storage.getStorage().createStorageInstance<TestDocType>({
+                    databaseName: randomCouchString(12),
+                    collectionName: randomCouchString(12),
+                    schema: getPseudoSchemaForVersion<TestDocType>(0, 'key'),
+                    options: {},
+                    multiInstance: false
+                });
+
+                const emitted: EventBulk<RxStorageChangeEvent<RxDocumentData<TestDocType>>>[] = [];
+                const sub = storageInstance.changeStream().subscribe(x => {
+                    emitted.push(x);
+                });
+
+                const writeData = {
+                    key: 'foobar',
+                    value: 'one',
+                    _rev: EXAMPLE_REVISION_1,
+                    _deleted: false,
+                    _attachments: {},
+                    _meta: {
+                        lwt: now()
+                    }
+                };
+
+                // insert
+                const firstWriteResult = await storageInstance.bulkWrite([{
+                    document: writeData
+                }]);
+                assert.deepStrictEqual(firstWriteResult.error, {});
+
+                // insert again to cause conflict error
+                const secondWriteResult = await storageInstance.bulkWrite([{
+                    document: writeData
+                }]);
+                assert.deepStrictEqual(secondWriteResult.success, {});
+
+                assert.strictEqual(emitted.length, 1);
+                assert.strictEqual(emitted[0].events.length, 1);
+
                 sub.unsubscribe();
                 storageInstance.close();
             });