Add new hooks in the replication write (pubkey#4754)

* add new hooks in the replication write

* Fix build not passing

* Fix code style

Author: jonathan-neugber

src/hooks.ts

@@ -79,7 +79,21 @@ export const HOOKS: { [k: string]: any[]; } = {
      * runs after a database has been removed
      * @async
      */
-    postRemoveRxDatabase: []
+    postRemoveRxDatabase: [],
+
+    /**
+     * runs before the replication writes the rows to master
+     * but before the rows have been modified
+     * @async
+     */
+    preReplicationMasterWrite: [],
+
+    /**
+     * runs after the replication has been sent to the server
+     * but before the new documents have been handled
+     * @async
+     */
+    preReplicationMasterWriteDocumentsHandle: [],
 };
 
 export function runPluginHooks(hookKey: string, obj: any) {

src/plugins/replication/index.ts

@@ -59,6 +59,9 @@ import {
 import { addRxPlugin } from '../../plugin';
 import { hasEncryption } from '../../rx-storage-helper';
 import { overwritable } from '../../overwritable';
+import {
+    runAsyncPluginHooks
+} from '../../hooks';
 
 
 export const REPLICATION_STATE_BY_COLLECTION: WeakMap<RxCollection, RxReplicationState<any, any>[]> = new WeakMap();
@@ -240,6 +243,12 @@ export class RxReplicationState<RxDocType, CheckpointType> {
                         return [];
                     }
                     let done = false;
+
+                    await runAsyncPluginHooks('preReplicationMasterWrite', {
+                        rows,
+                        collection: this.collection
+                    });
+
                     const useRows = await Promise.all(
                         rows.map(async (row) => {
                             row.newDocumentState = await pushModifier(row.newDocumentState);
@@ -257,6 +266,13 @@ export class RxReplicationState<RxDocType, CheckpointType> {
                     );
 
                     let result: WithDeleted<RxDocType>[] = null as any;
+
+                    // In case all the rows have been filtered and nothing has to be sent
+                    if (useRows.length === 0) {
+                        done = true;
+                        result = [];
+                    }
+
                     while (!done && !this.isStopped()) {
                         try {
                             result = await this.push.handler(useRows);
@@ -290,6 +306,12 @@ export class RxReplicationState<RxDocType, CheckpointType> {
                     if (this.isStopped()) {
                         return [];
                     }
+
+                    await runAsyncPluginHooks('preReplicationMasterWriteDocumentsHandle', {
+                        result,
+                        collection: this.collection
+                    });
+
                     const conflicts = handlePulledDocuments(this.collection, this.deletedField, ensureNotFalsy(result));
                     return conflicts;
                 }

src/types/rx-plugin.d.ts

@@ -16,7 +16,9 @@ import type {
     RxDatabase,
     RxDatabaseCreator,
     RxDocument,
-    RxStorage
+    RxStorage,
+    RxReplicationWriteToMasterRow,
+    WithDeleted
 } from '../types';
 import type { RxSchema } from '../rx-schema';
 
@@ -128,5 +130,13 @@ export interface RxPlugin {
         preCreateRxStorageInstance?: RxPluginHooks<RxStorageInstanceCreationParams<any, any>>;
         preMigrateDocument?: RxPluginHooks<any>;
         postMigrateDocument?: RxPluginHooks<any>;
+        preReplicationMasterWrite?: RxPluginHooks<{
+            rows: RxReplicationWriteToMasterRow<any>[];
+            collection: RxCollection;
+        }>;
+        preReplicationMasterWriteDocumentsHandle?: RxPluginHooks<{
+            result: WithDeleted<any>[];
+            collection: RxCollection;
+        }>;
     };
 }