#2606: Allow 'dataPath' to be specified on the GraphQL replication pull object to specify where to fetch result documents from

CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ### X.X.X (coming soon)
 
+Features:
+  - Added `dataPath` property to GraphQL replication pull options to allow the document JSON lookup path to configured instead of assuming the document data is always the first child of the response [#2606](https://github.com/pubkey/rxdb/issues/2606)
+
 Types:
   - `getLocal()` can return `undefined`. Thanks [@chrisdrackett](https://github.com/chrisdrackett)
 
@@ -72,7 +75,7 @@ Other:
 
 ### 9.13.0 (10 February 2021)
 
-Features: 
+Features:
 
   - Added `RxCollection().bulkRemove()` [#2845](https://github.com/pubkey/rxdb/pull/2845) Thanks [@qinyang912](https://github.com/qinyang912)
 
@@ -143,7 +146,7 @@ Features:
   - Added [RxDocument.atomicPatch()](https://rxdb.info/rx-document.html#atomicpatch)
 
 Bugfixes:
-  - (types) Returned values of `syncGraphQL()` did not type-match with `RxGraphQLReplicationState` 
+  - (types) Returned values of `syncGraphQL()` did not type-match with `RxGraphQLReplicationState`
   - `RxDocument.atomicUpdate()` now does a retry on 409 write conflicts
 
 Other:
@@ -170,7 +173,7 @@ Features:
 Other:
   - Refactored GraphQL replication to run faster [#2524](https://github.com/pubkey/rxdb/pull/2524/) Thanks [@corinv](https://github.com/corinv)
 
-### 9.6.0 (7 September 2020) 
+### 9.6.0 (7 September 2020)
 
 Features:
   - Add `RxReplicationState.setHeaders()` [#2399](https://github.com/pubkey/rxdb/pull/2399/) Thanks [@DDoerner](https://github.com/DDoerner)
@@ -696,7 +699,7 @@ Other:
 
 ## 6.0.0 (September 19, 2017) BREAKING
 
-Breaking:    
+Breaking:
   - Filenames are now kebab-case
   - `pouchdb-replication`-plugin is now imported by default, do not import it by your own.
   - `RxDB.create()` throws if you create the same database twice. (You can use [ignoreDuplicate](https://pubkey.github.io/rxdb/rx-database.html#ignoreduplicate))

docs-src/replication-graphql.md

@@ -161,6 +161,7 @@ const replicationState = myCollection.syncGraphQL({
     pull: {
         queryBuilder: pullQueryBuilder, // the queryBuilder from above
         modifier: doc => doc // (optional) modifies all pulled documents before they are handeled by RxDB. Returning null will skip the document.
+        dataPath: undefined // (optional) specifies the object path to access the document(s). Otherwise, the first result of the response data is used.
     },
     deletedFlag: 'deleted', // the flag which indicates if a pulled document is deleted
     live: true // if this is true, rxdb will watch for ongoing changes and sync them, when false, a one-time-replication will be done

src/plugins/replication-graphql/index.ts

@@ -14,6 +14,7 @@ import {
     filter
 } from 'rxjs/operators';
 import GraphQLClient from 'graphql-client';
+import objectPath from 'object-path';
 import {
     promiseWait,
     flatClone,
@@ -235,9 +236,8 @@ export class RxGraphQLReplicationState {
             return false;
         }
 
-        // this assumes that there will be always only one property in the response
-        // is this correct?
-        const data: any[] = result.data[Object.keys(result.data)[0]];
+        const dataPath = (this.pull as any).dataPath || ['data', Object.keys(result.data)[0]];
+        const data: any[] = objectPath.get(result, dataPath);
         const modified: any[] = (await Promise.all(data
             .map(async (doc: any) => await (this.pull as any).modifier(doc))
         )).filter(doc => !!doc);

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

@@ -13,6 +13,7 @@ export type RxGraphQLReplicationQueryBuilder = (doc: any) =>
 export interface GraphQLSyncPullOptions {
     queryBuilder: RxGraphQLReplicationQueryBuilder;
     modifier?: (doc: any) => Promise<any> | any;
+    dataPath?: string;
 }
 export interface GraphQLSyncPushOptions {
     queryBuilder: RxGraphQLReplicationQueryBuilder;

test/helper/graphql-server.ts

@@ -86,6 +86,7 @@ export async function spawn(
         type Query {
             info: Int
             feedForRxDBReplication(lastId: String!, minUpdatedAt: Int!, limit: Int!): [Human!]!
+            collectionFeedForRxDBReplication(lastId: String!, minUpdatedAt: Int!, offset: Int, limit: Int!): HumanCollection!
             getAll: [Human!]!
         }
         type Mutation {
@@ -118,6 +119,10 @@ export async function spawn(
             _rev: String,
             _revisions: Revision,
         }
+        type HumanCollection {
+            collection: [Human!]
+            totalCount: Int!
+        }
         type Subscription {
             humanChanged: Human
         }
@@ -138,25 +143,36 @@ export async function spawn(
     // The root provides a resolver function for each API endpoint
     const root = {
         info: () => 1,
+        collectionFeedForRxDBReplication: (args: any) => {
+            const { limit, offset = 0, ...feedForRxDBReplicationArgs } = args;
+            const collection = root.feedForRxDBReplication(feedForRxDBReplicationArgs);
+
+            // console.log('collection');
+            // console.dir(collection);
+
+            return {
+                totalCount: collection.length,
+                collection: collection.slice(offset, offset + limit)
+            };
+        },
         feedForRxDBReplication: (args: any) => {
             // console.log('## feedForRxDBReplication');
             // console.dir(args);
             // sorted by updatedAt and primary
             const sortedDocuments = documents.sort(sortByUpdatedAtAndPrimary);
 
             // only return where updatedAt >= minUpdatedAt
-            const filterForMinUpdatedAtAndId = sortedDocuments.filter((doc) => {
+            const filteredByMinUpdatedAtAndId = sortedDocuments.filter((doc) => {
                 if (doc.updatedAt < args.minUpdatedAt) return false;
                 if (doc.updatedAt > args.minUpdatedAt) return true;
                 if (doc.updatedAt === args.minUpdatedAt) {
                     if (doc.id > args.lastId) return true;
                     else return false;
                 }
-
             });
 
-            // limit
-            const limited = filterForMinUpdatedAtAndId.slice(0, args.limit);
+            // limit if requested
+            const limited = args.limit ? filteredByMinUpdatedAtAndId.slice(0, args.limit) : filteredByMinUpdatedAtAndId;
 
             /*
             console.log('sortedDocuments:');

test/unit/replication-graphql.test.ts

@@ -818,6 +818,63 @@ describe('replication-graphql.test.js', () => {
                 server.close();
                 c.database.destroy();
             });
+            it('should pull documents from a custom dataPath if one is specified', async () => {
+                const [c, server] = await Promise.all([
+                    humansCollection.createHumanWithTimestamp(0),
+                    SpawnServer.spawn(getTestData(batchSize))
+                ]);
+
+                const collectionQueryBuilder = (doc: any) => {
+                    if (!doc) {
+                        doc = {
+                            id: '',
+                            updatedAt: 0
+                        };
+                    }
+
+                    const query = `query($lastId: String!, $updatedAt: Int!, $batchSize: Int!)
+                    {
+                        collectionFeedForRxDBReplication(lastId: $lastId, minUpdatedAt: $updatedAt, limit: $batchSize) {
+                            collection {
+                                id
+                                name
+                                age
+                                updatedAt
+                                deleted
+                            }
+                        }
+                    }`;
+
+                    const variables = {
+                        lastId: doc.id,
+                        updatedAt: doc.updatedAt,
+                        batchSize
+                    };
+
+                    return {
+                        query,
+                        variables
+                    };
+                };
+
+                const replicationState = c.syncGraphQL({
+                    url: server.url,
+                    pull: {
+                        queryBuilder: collectionQueryBuilder,
+                        dataPath: 'data.collectionFeedForRxDBReplication.collection'
+                    },
+                    deletedFlag: 'deleted'
+                });
+                assert.strictEqual(replicationState.isStopped(), false);
+
+            await AsyncTestUtil.waitUntil(async () => {
+                    const docs = await c.find().exec();
+                    return docs.length === batchSize;
+                });
+
+                server.close();
+                c.database.destroy();
+            });
             it('pulled docs should be marked with a special revision if syncRevisions is false', async () => {
                 const [c, server] = await Promise.all([
                     humansCollection.createHumanWithTimestamp(0),