Feature: Ability to use a responseModifier for GraphQL push replication

docs-src/replication-graphql.md

@@ -361,7 +361,7 @@ const replicationState: RxGraphQLReplicationState<RxDocType> = replicateGraphQL(
 With the `pull.responseModifier` you can modify the whole response from the GraphQL endpoint **before** it is processed by RxDB.
 For example if your endpoint is not capable of returning a valid checkpoint, but instead only returns the plain document array, you can use the `responseModifier` to aggregate the checkpoint from the returned documents.
 
-```js
+```ts
 import {
 
 } from 'rxdb';
@@ -396,7 +396,42 @@ const replicationState: RxGraphQLReplicationState<RxDocType> = replicateGraphQL(
 );
 ```
 
+### push.responseModifier
+
+It's also possible to modify the response of a push mutation. For example if your server returns more than the just conflicting docs:
+
+```graphql
+type PushResponse {
+    conflicts: [Human]
+    conflictMessages: [ReplicationConflictMessage]
+}
+
+type Mutation {
+    # Returns a PushResponse type that contains the conflicts along with other information
+    pushHuman(rows: [HumanInputPushRow!]): PushResponse!
+}
+```
 
+```ts
+import {} from "rxdb";
+const replicationState: RxGraphQLReplicationState<RxDocType> = replicateGraphQL(
+    {
+        collection: myRxCollection,
+        url: {/* ... */},
+        headers: {/* ... */},
+        push: {
+            responseModifier: async function (plainResponse) {
+                /**
+                 * In this example we aggregate the conflicting documents from a response object
+                 */
+                return plainResponse.conflicts;
+            },
+        },
+        pull: {/* ... */},
+        /* ... */
+    }
+);
+```
 
 #### Helper Functions
 
@@ -442,4 +477,3 @@ See [the fetch spec](https://fetch.spec.whatwg.org/#concept-request-credentials-
 
 
 **NOTICE:** To play around, check out the full example of the RxDB [GraphQL replication with server and client](https://github.com/pubkey/rxdb/tree/master/examples/graphql)
-

src/plugins/replication-graphql/index.ts

@@ -161,8 +161,15 @@ export function replicateGraphQL<RxDocType, CheckpointType>(
                 if (result.errors) {
                     throw result.errors;
                 }
-                const dataPath = Object.keys(result.data)[0];
-                const data: any = getProperty(result.data, dataPath);
+                const dataPath = push.dataPath || Object.keys(result.data)[0];
+                let data: any = getProperty(result.data, dataPath);
+
+                if (push.responseModifier) {
+                    data = await push.responseModifier(
+                        data,
+                    );
+                }
+
                 return data;
             },
             batchSize: push.batchSize,

src/types/plugins/replication-graphql.d.ts

@@ -1,6 +1,6 @@
 import { RxReplicationWriteToMasterRow } from '../replication-protocol';
 import { ById, MaybePromise } from '../util';
-import { ReplicationOptions, ReplicationPullHandlerResult, ReplicationPullOptions, ReplicationPushOptions } from './replication';
+import { ReplicationOptions, ReplicationPullHandlerResult, ReplicationPullOptions, ReplicationPushHandlerResult, ReplicationPushOptions } from './replication';
 
 export interface RxGraphQLReplicationQueryBuilderResponseObject {
     query: string;
@@ -43,13 +43,20 @@ export type RxGraphQLPullResponseModifier<RxDocType, CheckpointType> = (
     requestCheckpoint?: CheckpointType
 ) => MaybePromise<ReplicationPullHandlerResult<RxDocType, CheckpointType>>;
 
+export type RxGraphQLPushResponseModifier<RxDocType> = (
+    // the exact response that was returned from the server
+    plainResponse: ReplicationPushHandlerResult<RxDocType> | any,
+) => MaybePromise<ReplicationPushHandlerResult<RxDocType>>;
+
 export type RxGraphQLReplicationPullStreamQueryBuilder = (headers: { [k: string]: string; }) => RxGraphQLReplicationQueryBuilderResponse;
 
 export type GraphQLSyncPushOptions<RxDocType> = Omit<
 ReplicationPushOptions<RxDocType>,
 'handler'
 > & {
     queryBuilder: RxGraphQLReplicationPushQueryBuilder;
+    dataPath?: string;
+    responseModifier?: RxGraphQLPushResponseModifier<RxDocType>;
 };
 
 export type GraphQLServerUrl = {

src/types/plugins/replication.d.ts

@@ -22,6 +22,8 @@ export type ReplicationPullHandlerResult<RxDocType, CheckpointType> = {
     documents: WithDeleted<RxDocType>[];
 };
 
+export type ReplicationPushHandlerResult<RxDocType> = RxDocType[];
+
 export type ReplicationPullHandler<RxDocType, CheckpointType> = (
     lastPulledCheckpoint: CheckpointType,
     batchSize: number

test/unit/replication-graphql.test.ts

@@ -61,7 +61,7 @@ import {
     buildSchema,
     parse as parseQuery
 } from 'graphql';
-import { RxDocumentData } from '../../src/types';
+import { ReplicationPushHandlerResult, RxDocumentData } from '../../src/types';
 import { enableKeyCompression } from '../helper/schemas';
 
 declare type WithDeleted<T> = T & { deleted: boolean; };
@@ -1047,6 +1047,45 @@ describe('replication-graphql.test.ts', () => {
                 const docsOnServer = server.getDocuments();
                 assert.strictEqual(docsOnServer.length, batchSize);
 
+                server.close();
+                c.database.destroy();
+            });
+            it('should respect the push.responseModifier', async () => {
+                const [c, server] = await Promise.all([
+                    humansCollection.createHumanWithTimestamp(batchSize),
+                    SpawnServer.spawn()
+                ]);
+
+                let responseHaveBeenCalledTimes = 0;
+                const replicationState = replicateGraphQL({
+                    collection: c,
+                    url: server.url,
+                    push: {
+                        batchSize,
+                        queryBuilder: pushQueryBuilder,
+                        responseModifier(
+                            originalResponse: ReplicationPushHandlerResult<HumanWithTimestampDocumentType>,
+                        ) {
+                            responseHaveBeenCalledTimes += 1;
+                            return originalResponse;
+                        }
+                    },
+                    live: false,
+                    retryTime: 1000,
+                    deletedField: 'deleted'
+                });
+                ensureReplicationHasNoErrors(replicationState);
+
+
+                await replicationState.awaitInitialReplication();
+
+                const docsOnServer = server.getDocuments();
+                assert.strictEqual(docsOnServer.length, batchSize);
+                assert.strictEqual(
+                    responseHaveBeenCalledTimes,
+                    1
+                );
+
                 server.close();
                 c.database.destroy();
             });