feat(sync): add support for paginated synchronization (time-based)

docs-website/docs/docs/README.md

@@ -204,6 +204,12 @@ The result is fully reactive! Whenever a post or comment is added, changed, or r
 
   <br/>
 
+  <a href="https://ezypack.app/">
+    <img src="https://github.com/Nozbe/WatermelonDB/raw/master/assets/apps/ezypack.png" alt="ezypack" width="300" />
+  </a>
+
+  <br/>
+
 _Does your company or app use 🍉? Open a pull request and add your logo/icon with link here!_
 
 ## Contributing

examples/typescript/yarn.lock

@@ -977,10 +977,10 @@ log-symbols@^2.0.0:
   dependencies:
     chalk "^2.0.1"
 
-"lokijs@npm:@nozbe/lokijs@1.5.12-wmelon6":
-  version "1.5.12-wmelon6"
-  resolved "https://registry.yarnpkg.com/@nozbe/lokijs/-/lokijs-1.5.12-wmelon6.tgz#e457d934d614d5df80105c86314252a6e614df9b"
-  integrity sha512-GXsaqY8qTJ6xdCrGyno2t+ON2aj6PrUDdvhbrkxK/0Fp12C4FGvDg1wS+voLU9BANYHEnr7KRWfItDZnQkjoAg==
+"lokijs@npm:@nozbe/lokijs@1.5.12-wmelon8":
+  version "1.5.12-wmelon8"
+  resolved "https://registry.yarnpkg.com/@nozbe/lokijs/-/lokijs-1.5.12-wmelon8.tgz#38ad7884d9cfd574a645c8201ad0cdbc93076dd6"
+  integrity sha512-WnqtKrWDh48FvuxnFv2LKurxeSAp8Q3TtQ4akwKFC7CkBaZYgn2P7F3YuBXguj4AgXhSEogxJmJWN8xQq7zPRQ==
 
 loud-rejection@^1.0.0:
   version "1.6.0"

src/Collection/index.js

@@ -116,7 +116,7 @@ export default class Collection<Record: Model> {
     })
   }
 
-  /*:: query: ArrayOrSpreadFn<Clause, Query<Record>>  */
+  query: ArrayOrSpreadFn<Clause, Query<Record>>
   /**
    * Returns a `Query` with conditions given.
    *

src/Database/index.js

@@ -84,7 +84,7 @@ export default class Database {
     return this._localStorage
   }
 
-  /*:: batch: ArrayOrSpreadFn<?Model | false, Promise<void>>  */
+  batch: ArrayOrSpreadFn<?Model | false, Promise<void>>
   /**
    * Executes multiple prepared operations
    *

src/Query/index.js

@@ -81,7 +81,7 @@ export default class Query<Record: Model> {
     this.description = Q.queryWithoutDeleted(this._rawDescription)
   }
 
-  /*:: extend: ArrayOrSpreadFn<Clause, Query<Record>>  */
+  extend: ArrayOrSpreadFn<Clause, Query<Record>>
   /**
    * Returns a new Query that contains all clauses (conditions, sorting, etc.) from this Query
    * as well as the ones passed as arguments.

src/observation/subscribeToQueryWithColumns/index.js

@@ -126,42 +126,42 @@ export default function subscribeToQueryWithColumns<Record: Model>(
 
   // Observe the source records list (list of records matching a query)
   // eslint-disable-next-line prefer-arrow-callback
-  const sourceUnsubscribe = subscribeToSource(function observeWithColumnsSourceChanged(
-    recordsOrStatus,
-  ): void {
-    // $FlowFixMe
-    if (recordsOrStatus === false) {
-      sourceIsFetching = true
-      return
-    }
-    sourceIsFetching = false
-
-    // Emit changes if one of observed columns changed OR list of matching records changed
-    const records: Record[] = recordsOrStatus
-    const shouldEmit =
-      firstEmission || hasPendingColumnChanges || !identicalArrays(records, observedRecords)
-
-    hasPendingColumnChanges = false
-    firstEmission = false
-
-    // Find changes, and save current list for comparison on next emission
-    const arrayDifference = require('../../utils/fp/arrayDifference').default
-    const { added, removed } = arrayDifference(observedRecords, records)
-    observedRecords = records
-
-    // Unsubscribe from records removed from list
-    removed.forEach((record) => {
-      recordStates.delete(record.id)
-    })
-
-    // Save current record state for later comparison
-    added.forEach((newRecord) => {
-      recordStates.set(newRecord.id, getRecordState(newRecord, columnNames))
-    })
-
-    // Emit
-    shouldEmit && emitCopy(records)
-  })
+  const sourceUnsubscribe = subscribeToSource(
+    function observeWithColumnsSourceChanged(recordsOrStatus): void {
+      // $FlowFixMe
+      if (recordsOrStatus === false) {
+        sourceIsFetching = true
+        return
+      }
+      sourceIsFetching = false
+
+      // Emit changes if one of observed columns changed OR list of matching records changed
+      const records: Record[] = recordsOrStatus
+      const shouldEmit =
+        firstEmission || hasPendingColumnChanges || !identicalArrays(records, observedRecords)
+
+      hasPendingColumnChanges = false
+      firstEmission = false
+
+      // Find changes, and save current list for comparison on next emission
+      const arrayDifference = require('../../utils/fp/arrayDifference').default
+      const { added, removed } = arrayDifference(observedRecords, records)
+      observedRecords = records
+
+      // Unsubscribe from records removed from list
+      removed.forEach((record) => {
+        recordStates.delete(record.id)
+      })
+
+      // Save current record state for later comparison
+      added.forEach((newRecord) => {
+        recordStates.set(newRecord.id, getRecordState(newRecord, columnNames))
+      })
+
+      // Emit
+      shouldEmit && emitCopy(records)
+    },
+  )
 
   return () => {
     unsubscribed = true

src/sync/impl/__tests__/synchronize.test.js

@@ -180,6 +180,129 @@ describe('synchronize', () => {
     await expectSyncedAndMatches(projects, 'pSynced', { name: 'remote' })
     await expectDoesNotExist(tasks, 'tSynced')
   })
+  it('can pull changes in pages', async () => {
+    const { database, projects } = makeDatabase()
+
+    const pullChanges = jest
+      .fn()
+      .mockReturnValueOnce(
+        Promise.resolve({
+          changes: makeChangeSet({
+            mock_projects: {
+              created: [{ id: 'p1', name: 'remote 1' }],
+            },
+          }),
+          timestamp: 1500,
+          experimentalHasMore: true,
+        }),
+      )
+      .mockReturnValueOnce(
+        Promise.resolve({
+          changes: makeChangeSet({
+            mock_projects: {
+              created: [{ id: 'p2', name: 'remote 2' }],
+            },
+          }),
+          timestamp: 1600,
+          experimentalHasMore: false,
+        }),
+      )
+
+    await synchronize({ database, pullChanges, pushChanges: jest.fn() })
+
+    expect(pullChanges).toHaveBeenCalledTimes(2)
+    expect(pullChanges).toHaveBeenNthCalledWith(1, {
+      lastPulledAt: null,
+      schemaVersion: 1,
+      migration: null,
+    })
+    expect(pullChanges).toHaveBeenNthCalledWith(2, {
+      lastPulledAt: 1500,
+      schemaVersion: 1,
+      migration: null,
+    })
+
+    expect(await getLastPulledAt(database)).toBe(1600)
+    await expectSyncedAndMatches(projects, 'p1', { name: 'remote 1' })
+    await expectSyncedAndMatches(projects, 'p2', { name: 'remote 2' })
+  })
+  it('calls sync hooks for every page', async () => {
+    const { database } = makeDatabase()
+
+    const onWillApplyRemoteChanges = jest.fn()
+    const onDidPullChanges = jest.fn()
+    const pullChanges = jest
+      .fn()
+      .mockReturnValueOnce(
+        Promise.resolve({
+          changes: makeChangeSet({
+            mock_projects: { created: [{ id: 'p1' }] },
+          }),
+          timestamp: 1500,
+          experimentalHasMore: true,
+          extra: 'page 1',
+        }),
+      )
+      .mockReturnValueOnce(
+        Promise.resolve({
+          changes: makeChangeSet({
+            mock_projects: { created: [{ id: 'p2' }] },
+          }),
+          timestamp: 1600,
+          experimentalHasMore: false,
+          extra: 'page 2',
+        }),
+      )
+
+    await synchronize({
+      database,
+      pullChanges,
+      onWillApplyRemoteChanges,
+      onDidPullChanges,
+    })
+
+    expect(onWillApplyRemoteChanges).toHaveBeenCalledTimes(2)
+    expect(onWillApplyRemoteChanges).toHaveBeenNthCalledWith(1, { remoteChangeCount: 1 })
+    expect(onWillApplyRemoteChanges).toHaveBeenNthCalledWith(2, { remoteChangeCount: 1 })
+
+    expect(onDidPullChanges).toHaveBeenCalledTimes(2)
+    expect(onDidPullChanges).toHaveBeenNthCalledWith(1, {
+      extra: 'page 1',
+      timestamp: 1500,
+      experimentalHasMore: true,
+    })
+    expect(onDidPullChanges).toHaveBeenNthCalledWith(2, {
+      extra: 'page 2',
+      timestamp: 1600,
+      experimentalHasMore: false,
+    })
+  })
+  it('does not update lastPulledAt if paginated sync fails midway', async () => {
+    const { database, projects } = makeDatabase()
+
+    const pullChanges = jest
+      .fn()
+      .mockReturnValueOnce(
+        Promise.resolve({
+          changes: makeChangeSet({
+            mock_projects: { created: [{ id: 'p1' }] },
+          }),
+          timestamp: 1500,
+          experimentalHasMore: true,
+        }),
+      )
+      .mockReturnValueOnce(Promise.reject(new Error('page 2 fail')))
+
+    await expect(synchronize({ database, pullChanges, pushChanges: jest.fn() })).rejects.toThrow(
+      'page 2 fail',
+    )
+
+    // Data from first page should be there
+    await expectSyncedAndMatches(projects, 'p1', {})
+
+    // BUT timestamp should still be null
+    expect(await getLastPulledAt(database)).toBe(null)
+  })
   it('can synchronize changes with conflicts', async () => {
     const { database, projects, tasks, comments } = makeDatabase()