fix(datapath) must be equivalent for pull and push

src/plugins/replication-graphql/helper.ts

@@ -1,5 +1,5 @@
 import type { RxGraphQLReplicationClientState, RxGraphQLReplicationQueryBuilderResponseObject } from '../../types/index.d.ts';
-import { ensureNotFalsy } from '../../plugins/utils/index.ts';
+import { ensureNotFalsy, getProperty } from '../../plugins/utils/index.ts';
 
 export const GRAPHQL_REPLICATION_PLUGIN_IDENTITY_PREFIX = 'graphql';
 
@@ -41,3 +41,12 @@ export function graphQLRequest(
             return body;
         });
 }
+
+export function getDataFromResult(
+    result: { data: object },
+    userDefinedDataPath: string | string[] | undefined
+): any {
+    const dataPath = userDefinedDataPath || ['data', Object.keys(result.data)[0]];
+    const data: any = getProperty(result, dataPath);
+    return data;
+}

src/plugins/replication-graphql/index.ts

@@ -9,6 +9,7 @@ import {
 } from '../../plugins/utils/index.ts';
 
 import {
+    getDataFromResult,
     graphQLRequest
 } from './helper.ts';
 
@@ -131,8 +132,7 @@ export function replicateGraphQL<RxDocType, CheckpointType>(
                 if (result.errors) {
                     throw result.errors;
                 }
-                const dataPath = pull.dataPath || ['data', Object.keys(result.data)[0]];
-                let data: any = getProperty(result, dataPath);
+                let data: any = getDataFromResult(result, pull.dataPath);
                 if (pull.responseModifier) {
                     data = await pull.responseModifier(
                         data,
@@ -166,17 +166,15 @@ export function replicateGraphQL<RxDocType, CheckpointType>(
                 if (result.errors) {
                     throw result.errors;
                 }
-                const dataPath = push.dataPath || Object.keys(result.data)[0];
-                let data: any = getProperty(result.data, dataPath);
-
+                let data: any = getDataFromResult(result, push.dataPath);
                 if (push.responseModifier) {
                     data = await push.responseModifier(
                         data,
                     );
                 }
 
                 return data;
-            },
+            }, 
             batchSize: push.batchSize,
             modifier: push.modifier
         };

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

@@ -38,7 +38,12 @@ export type GraphQLSyncPullOptions<RxDocType, CheckpointType> = Omit<
 > & {
     queryBuilder: RxGraphQLReplicationPullQueryBuilder<CheckpointType>;
     streamQueryBuilder?: RxGraphQLReplicationPullStreamQueryBuilder;
-    dataPath?: string;
+    /**
+     * The path to the data in the GraphQL response.
+     * If set, the data will be taken from the response at this path.
+     * @example ['data', 'foo', 'bar'] or 'data.foo.bar'
+     */
+    dataPath?: string | string[];
     responseModifier?: RxGraphQLPullResponseModifier<RxDocType, CheckpointType>;
     includeWsHeaders?: boolean;
 };
@@ -63,7 +68,12 @@ export type GraphQLSyncPushOptions<RxDocType> = Omit<
     'handler'
 > & {
     queryBuilder: RxGraphQLReplicationPushQueryBuilder;
-    dataPath?: string;
+    /**
+     * The path to the data in the GraphQL response.
+     * If set, the data will be taken from the response at this path.
+     * @example ['data', 'foo', 'bar'] or 'data.foo.bar'
+     */
+    dataPath?: string | string[];
     responseModifier?: RxGraphQLPushResponseModifier<RxDocType>;
 };
 

test/unit/replication-graphql.test.ts

@@ -1187,6 +1187,36 @@ describe('replication-graphql.test.ts', () => {
                     1
                 );
 
+                server.close();
+                c.database.destroy();
+            });
+            it('should push documents from a custom dataPath if one is specified', async () => {
+                const [c, server] = await Promise.all([
+                    humansCollection.createHumanWithTimestamp(batchSize),
+                    SpawnServer.spawn()
+                ]);
+
+                const replicationState = replicateGraphQL({
+                    replicationIdentifier: randomCouchString(10),
+                    collection: c,
+                    url: server.url,
+                    push: {
+                        batchSize,
+                        dataPath: 'data.writeHumans',
+                        queryBuilder: pushQueryBuilder,
+                    },
+                    live: false,
+                    retryTime: 1000,
+                    deletedField: 'deleted'
+                });
+                ensureReplicationHasNoErrors(replicationState);
+                assert.strictEqual(replicationState.isStopped(), false);
+
+                await waitUntil(async () => {
+                    const docs = await c.find().exec();
+                    return docs.length === batchSize;
+                });
+
                 server.close();
                 c.database.destroy();
             });