Push sync conflict resolving

src/RawRecord/index.d.ts

@@ -11,6 +11,8 @@ export type DirtyRaw = { [key: string]: any }
 type _RawRecord = {
   id: RecordId
   _status: SyncStatus
+  // _changed is used by default pull conflict resolution and determines columns for which local
+  // changes will override remote changes
   _changed: string
 }
 

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

@@ -131,7 +131,7 @@ describe('markLocalChangesAsSynced', () => {
     })
 
     // test that second push will mark all as synced
-    await markLocalChangesAsSynced(database, localChanges2)
+    await markLocalChangesAsSynced(database, localChanges2, false)
     expect(destroyDeletedRecordsSpy).toHaveBeenCalledTimes(2)
     expect(await fetchLocalChanges(database)).toEqual(emptyLocalChanges)
 
@@ -146,7 +146,7 @@ describe('markLocalChangesAsSynced', () => {
     const localChanges = await fetchLocalChanges(database)
 
     // mark as synced
-    await markLocalChangesAsSynced(database, localChanges, {
+    await markLocalChangesAsSynced(database, localChanges, false, {
       mock_projects: ['pCreated1', 'pUpdated'],
       mock_comments: ['cDeleted'],
     })
@@ -161,6 +161,30 @@ describe('markLocalChangesAsSynced', () => {
     )
     expect(await allDeletedRecords([comments])).toEqual(['cDeleted'])
   })
+  it(`marks only acceptedIds as synced`, async () => {
+    const { database, comments } = makeDatabase()
+
+    const { pCreated1, pUpdated } = await makeLocalChanges(database)
+    const localChanges = await fetchLocalChanges(database)
+
+    // mark as synced
+    await markLocalChangesAsSynced(database, localChanges, true, {}, {
+      // probably better solution exists (we essentially list all but expected in verify)
+      mock_projects: ['pCreated2', 'pDeleted'],
+      mock_comments: ['cCreated', 'cUpdated'],
+      mock_tasks: ['tCreated', 'tUpdated', 'tDeleted'],
+    })
+
+    // verify
+    const localChanges2 = await fetchLocalChanges(database)
+    expect(localChanges2.changes).toEqual(
+      makeChangeSet({
+        mock_projects: { created: [pCreated1._raw], updated: [pUpdated._raw] },
+        mock_comments: { deleted: ['cDeleted'] },
+      }),
+    )
+    expect(await allDeletedRecords([comments])).toEqual(['cDeleted'])
+  })
   it(`can mark records as synced when ids are per-table not globally unique`, async () => {
     const { database, projects, tasks, comments } = makeDatabase()
 
@@ -202,4 +226,45 @@ describe('markLocalChangesAsSynced', () => {
   it.skip('only returns changed fields', async () => {
     // TODO: Possible future improvement?
   })
+  describe('pushConflictResolver', () => {
+    it('marks local changes as synced', async () => {
+      const { database, tasks } = makeDatabase()
+
+      await makeLocalChanges(database)
+
+      await markLocalChangesAsSynced(database, await fetchLocalChanges(database), false, null, null,
+        (_table, local, remote, resolved) => {
+          if (local.id !== 'tCreated' || (remote && remote.changeMe !== true)) {
+            return resolved
+          }
+          resolved.name = remote.name
+          resolved._status = 'updated'
+          return resolved
+        },
+        {
+          mock_tasks: [
+            {
+              id: 'tCreated',
+              name: 'I shall prevail',
+              changeMe: true,
+            },
+          ],
+        })
+
+        await expectSyncedAndMatches(tasks, 'tCreated', {
+          _status: 'updated',
+          name: 'I shall prevail', // concat of remote and local change
+        })
+
+        // should be untouched
+        await expectSyncedAndMatches(tasks, 'tUpdated', {
+          _status: 'synced',
+          name: 'local',
+          position: 100,
+          description: 'orig',
+          project_id: 'orig',
+        })
+
+    })
+  })
 })

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

@@ -62,4 +62,71 @@ describe('synchronize - partial push rejections', () => {
       }),
     )
   })
+  it(`can partially accept a push`, async () => {
+    const { database } = makeDatabase()
+
+    const { tCreated, tUpdated } = await makeLocalChanges(database)
+
+    const acceptedIds = Object.freeze({
+      // probably better solution exists (we essentially list all but expected in expect below)
+      mock_projects: ['pCreated1', 'pCreated2', 'pDeleted', 'pUpdated'],
+      mock_comments: ['cCreated', 'cUpdated'],
+      mock_tasks: ['tDeleted'],
+    })
+    const rejectedIds = Object.freeze({
+      mock_tasks: ['tCreated', 'tUpdated'],
+      mock_comments: ['cDeleted'],
+    })
+    const log = {}
+    await synchronize({
+      database,
+      pullChanges: jest.fn(emptyPull()),
+      pushChanges: jest.fn(() => ({ experimentalAcceptedIds: acceptedIds })),
+      pushShouldConfirmOnlyAccepted: true,
+      log,
+    })
+    expect((await fetchLocalChanges(database)).changes).toEqual(
+      makeChangeSet({
+        mock_tasks: { created: [tCreated._raw], updated: [tUpdated._raw] },
+        mock_comments: { deleted: ['cDeleted'] },
+      }),
+    )
+    expect(log.rejectedIds).toStrictEqual(rejectedIds)
+  })
+  it(`can partially accept a push and make changes during push`, async () => {
+    const { database, comments } = makeDatabase()
+
+    const { pCreated1, tUpdated } = await makeLocalChanges(database)
+    const pCreated1Raw = { ...pCreated1._raw }
+    let newComment
+    await synchronize({
+      database,
+      pullChanges: jest.fn(emptyPull()),
+      pushChanges: jest.fn(async () => {
+        await database.write(async () => {
+          await pCreated1.update((p) => {
+            p.name = 'updated!'
+          })
+          newComment = await comments.create((c) => {
+            c.body = 'bazinga'
+          })
+        })
+        return {
+          experimentalAcceptedIds: {
+            mock_projects: ['pCreated1', 'pCreated2', 'pDeleted', 'pUpdated'],
+            mock_comments: ['cCreated', 'cUpdated'],
+            mock_tasks: ['tCreated', 'tDeleted'],
+          },
+        }
+      }),
+      pushShouldConfirmOnlyAccepted: true,
+    })
+    expect((await fetchLocalChanges(database)).changes).toEqual(
+      makeChangeSet({
+        mock_projects: { created: [{ ...pCreated1Raw, _changed: 'name', name: 'updated!' }] },
+        mock_tasks: { updated: [tUpdated._raw] },
+        mock_comments: { created: [newComment._raw], deleted: ['cDeleted'] },
+      }),
+    )
+  })
 })

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

@@ -571,4 +571,38 @@ describe('synchronize', () => {
   it.skip(`only emits one collection batch change`, async () => {
     // TODO: unskip when batch change emissions are implemented
   })
+  it(`allows push conflict resolution to be customized`, async () => {
+    const { database, tasks } = makeDatabase()
+    const task = tasks.prepareCreateFromDirtyRaw({
+      id: 't1',
+      name: 'Task name',
+      position: 1,
+      is_completed: false,
+      project_id: 'p1',
+    })
+    await database.write(() => database.batch(task))
+
+    const pushConflictResolver = jest.fn((_table, local, remote, resolved) => {
+      return resolved
+    })
+
+    await synchronize({
+      database,
+      pullChanges: () => ({
+        timestamp: 1500,
+      }),
+      pushChanges: async () => {
+        return {
+          pushResultSet: {
+            mock_tasks: [
+              {id: 't1'},
+            ],
+          },
+        }
+      },
+      pushConflictResolver,
+    })
+
+    expect(pushConflictResolver).toHaveBeenCalledTimes(1)
+  })
 })

src/sync/impl/applyRemote.js

@@ -11,7 +11,6 @@ import type {
   Collection,
   Model,
   TableName,
-  DirtyRaw,
   Query,
   RawRecord,
 } from '../..'
@@ -25,7 +24,7 @@ import type {
   SyncConflictResolver,
   SyncPullStrategy,
 } from '../index'
-import { prepareCreateFromRaw, prepareUpdateFromRaw, recordFromRaw } from './helpers'
+import { prepareCreateFromRaw, prepareUpdateFromRaw, recordFromRaw, validateRemoteRaw } from './helpers'
 
 type ApplyRemoteChangesContext = $Exact<{
   db: Database,
@@ -249,15 +248,6 @@ const getAllRecordsToApply = (
   )
 }
 
-function validateRemoteRaw(raw: DirtyRaw): void {
-  // TODO: I think other code is actually resilient enough to handle illegal _status and _changed
-  // would be best to change that part to a warning - but tests are needed
-  invariant(
-    raw && typeof raw === 'object' && 'id' in raw && !('_status' in raw || '_changed' in raw),
-    `[Sync] Invalid raw record supplied to Sync. Records must be objects, must have an 'id' field, and must NOT have a '_status' or '_changed' fields`,
-  )
-}
-
 function prepareApplyRemoteChangesToCollection<T: Model>(
   recordsToApply: RecordsToApplyRemoteChangesTo<T>,
   collection: Collection<T>,

src/sync/impl/helpers.d.ts

@@ -1,12 +1,20 @@
 import type { Model, Collection, Database } from '../..'
 import type { RawRecord, DirtyRaw } from '../../RawRecord'
-import type { SyncLog, SyncDatabaseChangeSet, SyncConflictResolver } from '../index'
+import type {
+  SyncLog,
+  SyncDatabaseChangeSet,
+  SyncShouldUpdateRecord,
+  SyncConflictResolver,
+  SyncPushResultSet,
+} from '../index'
 
 // Returns raw record with naive solution to a conflict based on local `_changed` field
 // This is a per-column resolution algorithm. All columns that were changed locally win
 // and will be applied on top of the remote version.
 export function resolveConflict(local: RawRecord, remote: DirtyRaw): DirtyRaw
 
+export function validateRemoteRaw(raw: DirtyRaw): void
+
 export function prepareCreateFromRaw<T extends Model = Model>(
   collection: Collection<T>,
   dirtyRaw: DirtyRaw,
@@ -19,7 +27,11 @@ export function prepareUpdateFromRaw<T extends Model = Model>(
   conflictResolver?: SyncConflictResolver,
 ): T
 
-export function prepareMarkAsSynced<T extends Model = Model>(record: T): T
+export function prepareMarkAsSynced<T extends Model = Model>(
+  record: T,
+  pushConflictResolver?: SyncConflictResolver,
+  remoteDirtyRaw?: DirtyRaw,
+): T
 
 export function ensureSameDatabase(database: Database, initialResetCount: number): void
 

src/sync/impl/helpers.js

@@ -4,9 +4,14 @@ import { values } from '../../utils/fp'
 import areRecordsEqual from '../../utils/fp/areRecordsEqual'
 import { invariant } from '../../utils/common'
 
-import type { Model, Collection, Database } from '../..'
+import type { Model, Collection, Database, TableName, RecordId } from '../..'
 import { type RawRecord, type DirtyRaw, sanitizedRaw } from '../../RawRecord'
-import type { SyncLog, SyncDatabaseChangeSet, SyncConflictResolver } from '../index'
+import type {
+  SyncIds,
+  SyncLog,
+  SyncDatabaseChangeSet,
+  SyncConflictResolver,
+} from '../index'
 
 // Returns raw record with naive solution to a conflict based on local `_changed` field
 // This is a per-column resolution algorithm. All columns that were changed locally win
@@ -37,6 +42,15 @@ export function resolveConflict(local: RawRecord, remote: DirtyRaw): DirtyRaw {
   return resolved
 }
 
+export function validateRemoteRaw(raw: DirtyRaw): void {
+  // TODO: I think other code is actually resilient enough to handle illegal _status and _changed
+  // would be best to change that part to a warning - but tests are needed
+  invariant(
+    raw && typeof raw === 'object' && 'id' in raw && !('_status' in raw || '_changed' in raw),
+    `[Sync] Invalid raw record supplied to Sync. Records must be objects, must have an 'id' field, and must NOT have a '_status' or '_changed' fields`,
+  )
+}
+
 function replaceRaw(record: Model, dirtyRaw: DirtyRaw): void {
   record._raw = sanitizedRaw(dirtyRaw, record.collection.schema)
 }
@@ -120,9 +134,16 @@ export function prepareUpdateFromRaw<T: Model>(
   })
 }
 
-export function prepareMarkAsSynced<T: Model>(record: T): T {
+export function prepareMarkAsSynced<T: Model>(
+  record: T,
+  pushConflictResolver?: ?SyncConflictResolver,
+  remoteDirtyRaw?: ?DirtyRaw,
+): T {
   // $FlowFixMe
-  const newRaw = Object.assign({}, record._raw, { _status: 'synced', _changed: '' }) // faster than object spread
+  let newRaw = Object.assign({}, record._raw, { _status: 'synced', _changed: '' }) // faster than object spread
+  if (pushConflictResolver) {
+    newRaw = pushConflictResolver(record.collection.table, record._raw, remoteDirtyRaw, newRaw)
+  }
   // $FlowFixMe
   return record.prepareUpdate(() => {
     replaceRaw(record, newRaw)
@@ -148,3 +169,46 @@ export const changeSetCount: (SyncDatabaseChangeSet) => number = (changeset) =>
       ({ created, updated, deleted }) => created.length + updated.length + deleted.length,
     ),
   )
+
+const extractChangeSetIds: (SyncDatabaseChangeSet) => { [TableName<any>]: RecordId[] } = (changeset) =>
+  Object.keys(changeset).reduce((acc: { [TableName<any>]: RecordId[] }, key: string) => {
+    // $FlowFixMe
+    const { created, updated, deleted } = changeset[key]
+    // $FlowFixMe
+    acc[key] = [
+      ...created.map(it => it.id),
+      ...updated.map(it => it.id),
+      ...deleted,
+    ]
+    return acc
+  }, {})
+
+// Returns all rejected ids and is used when accepted ids are used 
+export const findRejectedIds:
+  (?SyncIds, ?SyncIds, SyncDatabaseChangeSet) => SyncIds =
+    (experimentalRejectedIds, experimentalAcceptedIds, changeset) => {
+      const localIds = extractChangeSetIds(changeset)
+
+      const acceptedIdsSets = Object.keys(changeset).reduce(
+        (acc: { [TableName<any>]: Set<RecordId> }, key: string) => {
+          // $FlowFixMe
+          acc[key] = new Set(experimentalAcceptedIds[key])
+          return acc
+      }, {})
+
+      return Object.keys(changeset).reduce((acc: { [TableName<any>]: RecordId[] }, key: string) => {
+        const rejectedIds = [
+          // $FlowFixMe
+          ...(experimentalRejectedIds ? experimentalRejectedIds[key] || [] : []),
+          // $FlowFixMe
+          ...(localIds[key] || []),
+          // $FlowFixMe
+        ].filter(it => !acceptedIdsSets[key].has(it))
+
+        if (rejectedIds.length > 0) {
+          // $FlowFixMe
+          acc[key] = rejectedIds
+        }
+        return acc
+      }, {})
+    }

src/sync/impl/markAsSynced.d.ts

@@ -1,9 +1,13 @@
 import type { Database, Model, TableName } from '../..'
 
-import type { SyncLocalChanges, SyncRejectedIds } from '../index'
+import type { SyncLocalChanges, SyncIds, SyncConflictResolver, SyncPushResultSet } from '../index'
 
 export default function markLocalChangesAsSynced(
   db: Database,
   syncedLocalChanges: SyncLocalChanges,
-  rejectedIds?: SyncRejectedIds,
+  allowOnlyAcceptedIds: boolean,
+  rejectedIds?: SyncIds,
+  allAcceptedIds?: SyncIds,
+  pushConflictResolver?: SyncConflictResolver,
+  remoteDirtyRaws?: SyncPushResultSet,
 ): Promise<void>