Pend and confirm sync records to allow pause/resume

Depends on brave/sync#122

Fix brave/sync#112

app/browser/reducers/syncReducer.js

@@ -0,0 +1,25 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this file,
+ * You can obtain one at http://mozilla.org/MPL/2.0/. */
+
+'use strict'
+
+const appConstants = require('../../../js/constants/appConstants')
+const syncPendState = require('../../common/state/syncPendState')
+
+const syncReducer = (state, action) => {
+  switch (action.actionType) {
+    case appConstants.APP_SET_STATE:
+      state = syncPendState.initPendingRecords(state)
+      break
+    case appConstants.APP_PEND_SYNC_RECORDS:
+      state = syncPendState.pendRecords(state, action.records)
+      break
+    case appConstants.APP_CONFIRM_SYNC_RECORDS:
+      state = syncPendState.confirmRecords(state, action.records)
+      break
+  }
+  return state
+}
+
+module.exports = syncReducer

app/common/state/syncPendState.js

@@ -0,0 +1,70 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public * License, v. 2.0. If a copy of the MPL was not distributed with this file,
+ * You can obtain one at http://mozilla.org/MPL/2.0/. */
+
+'use strict'
+
+const Immutable = require('immutable')
+const syncUtil = require('../../../js/state/syncUtil')
+
+/**
+ * Turn app state pendingRecords into an OrderedMap.
+ * @param {Immutable.Map} state app state
+ * @returns {Immutable.OrderedMap} new app state
+ */
+module.exports.initPendingRecords = (state) => {
+  const pendingRecords = state.getIn(['sync', 'pendingRecords'])
+  const orderedPendingRecords = new Immutable.OrderedMap(pendingRecords)
+  return state.setIn(['sync', 'pendingRecords'], orderedPendingRecords)
+}
+
+/**
+ * Given records sent via SEND_SYNC_RECORDS, insert them in the app state.
+ * @param {Immutable.Map} state app state
+ * @param {Array.<Object>} records
+ * @returns {Immutable.Map} new app state
+ */
+module.exports.pendRecords = (state, records) => {
+  const enqueueTimestamp = new Date().getTime()
+  records.forEach(record => {
+    const pendingRecord = Immutable.fromJS({enqueueTimestamp, record})
+    const key = record.objectId.toString()
+    console.log(`APP_PEND_SYNC_RECORDS: ${key} => ${JSON.stringify(pendingRecord.toJS())}`)
+    state = state.setIn(['sync', 'pendingRecords', key], pendingRecord)
+  })
+  return state
+}
+
+/**
+ * Given app state, extract records to be sent via SEND_SYNC_RECORDS.
+ * @param {Immutable.Map} state app state
+ * @returns {Array.<Object>} records
+ */
+module.exports.getPendingRecords = (state) => {
+  const pendingRecords = state.getIn(['sync', 'pendingRecords'])
+  return pendingRecords.map(pendingRecord => pendingRecord.get('record').toJS()).toArray()
+}
+
+/**
+ * Confirm downloaded records and remove them from the app state sync
+ * pending records.
+ * In case of multiple updates to the same object, the newest update takes
+ * precedence. Thus, if a downloaded record has a newer timestamp it
+ * dequeues any pending record.
+ * @param {Immutable.Map} state app state
+ * @param {Array.<Object>} downloadedRecord
+ * @returns {Immutable.Map} new app state
+ */
+module.exports.confirmRecords = (state, downloadedRecords) => {
+  downloadedRecords.forEach(record => {
+    // browser-laptop stores byte arrays like objectId as Arrays.
+    // downloaded records use Uint8Arrays which we should convert back.
+    const fixedRecord = syncUtil.deepArrayify(record)
+    const key = fixedRecord.objectId.toString()
+    const enqueueTimestamp = state.getIn(['sync', 'pendingRecords', key, 'enqueueTimestamp'])
+    if (!enqueueTimestamp || (enqueueTimestamp > fixedRecord.syncTimestamp)) {
+      return
+    }
+    state = state.deleteIn(['sync', 'pendingRecords', key])
+  })
+  return state
+}

app/sessionStore.js

@@ -622,7 +622,8 @@ module.exports.defaultAppState = () => {
     sync: {
       devices: {},
       lastFetchTimestamp: 0,
-      objectsById: {}
+      objectsById: {},
+      pendingRecords: {}
     },
     locationSiteKeysCache: undefined,
     sites: getTopSiteMap(),

app/sync.js

@@ -21,6 +21,7 @@ const appDispatcher = require('../js/dispatcher/appDispatcher')
 const AppStore = require('../js/stores/appStore')
 const siteUtil = require('../js/state/siteUtil')
 const syncUtil = require('../js/state/syncUtil')
+const syncPendState = require('./common/state/syncPendState')
 const getSetting = require('../js/settings').getSetting
 const settings = require('../js/constants/settings')
 const extensions = require('./extensions')
@@ -59,6 +60,7 @@ const appStoreChangeCallback = function (diffs) {
     return
   }
 
+  let pendingRecords = []
   diffs.forEach((diff) => {
     if (!diff || !diff.path) {
       return
@@ -116,42 +118,52 @@ const appStoreChangeCallback = function (diffs) {
         const entryJS = entry.toJS()
         entryJS.objectId = entryJS.objectId || syncUtil.newObjectId(statePath)
 
-        sendSyncRecords(backgroundSender, action,
-          [type === 'sites' ? syncUtil.createSiteData(entryJS) : syncUtil.createSiteSettingsData(statePath[1], entryJS)])
+        pendingRecords = pendingRecords.concat(
+          sendSyncRecords(backgroundSender, action,
+            [type === 'sites' ? syncUtil.createSiteData(entryJS) : syncUtil.createSiteSettingsData(statePath[1], entryJS)])
+        )
       }
     }
   })
+  if (pendingRecords.length > 0) {
+    appActions.pendSyncRecords(pendingRecords)
+  }
 }
 
 /**
  * Sends sync records of the same category to the sync server.
  * @param {event.sender} sender
  * @param {number} action
  * @param {Array.<{name: string, value: Object}>} data
+ * @returns {Array.<object>} records which were sent
  */
 const sendSyncRecords = (sender, action, data) => {
   if (!deviceId) {
     throw new Error('Cannot build a sync record because deviceId is not set')
   }
   if (!data || !data.length || !data[0]) {
-    return
+    return []
   }
   const category = CATEGORY_MAP[data[0].name]
   if (!category ||
     (category.settingName && !getSetting(settings[category.settingName]))) {
-    return
+    return []
   }
-  sender.send(syncMessages.SEND_SYNC_RECORDS, category.categoryName, data.map((item) => {
-    if (!item || !item.name || !item.value) {
+  let records = []
+  data.forEach(item => {
+    if (!item || !item.name || !item.value || !item.objectId) {
       return
     }
-    return {
+    records.push({
       action,
       deviceId,
       objectId: item.objectId,
       [item.name]: item.value
-    }
-  }))
+    })
+  })
+  console.log(`Sending ${records.length} sync records`)
+  sender.send(syncMessages.SEND_SYNC_RECORDS, category.categoryName, records)
+  return records
 }
 
 /**
@@ -222,16 +234,20 @@ module.exports.onSyncReady = (isFirstRun, e) => {
     return
   }
   AppStore.addChangeListener(appStoreChangeCallback)
+  let pendingRecords = []
 
   if (!deviceIdSent && isFirstRun) {
     // Sync the device id for this device
-    sendSyncRecords(e.sender, writeActions.CREATE, [{
+    const deviceRecord = {
       name: 'device',
       objectId: syncUtil.newObjectId(['sync']),
       value: {
         name: getSetting(settings.SYNC_DEVICE_NAME)
       }
-    }])
+    }
+    pendingRecords = pendingRecords.concat(
+      sendSyncRecords(e.sender, writeActions.CREATE, [deviceRecord])
+    )
     deviceIdSent = true
   }
   const appState = AppStore.getState()
@@ -247,6 +263,7 @@ module.exports.onSyncReady = (isFirstRun, e) => {
    * @param {Immutable.Map} site
    */
   const folderToObjectId = {}
+  const bookmarksToSync = []
   const shouldSyncBookmark = (site) => {
     if (!site) { return false }
     // originalSeed is set on reset to prevent synced bookmarks on a device
@@ -278,12 +295,19 @@ module.exports.onSyncReady = (isFirstRun, e) => {
       folderToObjectId[folderId] = record.objectId
     }
 
-    sendSyncRecords(e.sender, writeActions.CREATE, [record])
+    bookmarksToSync.push(record)
   }
 
   // Sync bookmarks that have not been synced yet.
-  siteUtil.getBookmarks(sites).filter(site => shouldSyncBookmark(site))
-    .forEach(syncBookmark)
+  sites.forEach((site) => {
+    if (siteUtil.isSiteBookmarked(sites, site) !== true || shouldSyncBookmark(site) !== true) {
+      return
+    }
+    syncBookmark(site)
+  })
+  pendingRecords = pendingRecords.concat(
+    sendSyncRecords(e.sender, writeActions.CREATE, bookmarksToSync)
+  )
 
   // Sync site settings that have not been synced yet
   // FIXME: If Sync was disabled and settings were changed, those changes
@@ -293,10 +317,24 @@ module.exports.onSyncReady = (isFirstRun, e) => {
       return !value.get('objectId') && syncUtil.isSyncable('siteSetting', value)
     }).toJS()
   if (siteSettings) {
-    sendSyncRecords(e.sender, writeActions.UPDATE,
-      Object.keys(siteSettings).map((item) => {
-        return syncUtil.createSiteSettingsData(item, siteSettings[item])
-      }))
+    const siteSettingsData = Object.keys(siteSettings).map((item) => {
+      return syncUtil.createSiteSettingsData(item, siteSettings[item])
+    })
+    pendingRecords = pendingRecords.concat(
+      sendSyncRecords(e.sender, writeActions.UPDATE, siteSettingsData)
+    )
+  }
+
+  // Sync records which haven't been confirmed yet.
+  const oldPendingRecords = syncPendState.getPendingRecords(appState)
+  if (oldPendingRecords.length > 0) {
+    log(`Sending ${oldPendingRecords.length} pending records`)
+    e.sender.send(syncMessages.SEND_SYNC_RECORDS, undefined, oldPendingRecords)
+  }
+
+  // Pend records sent during init
+  if (pendingRecords.length > 0) {
+    appActions.pendSyncRecords(pendingRecords)
   }
 
   appActions.createSyncCache()
@@ -421,6 +459,9 @@ module.exports.init = function (appState) {
     appActions.setSyncSetupError(error || locale.translation('unknownError'))
   })
   ipcMain.on(syncMessages.GET_EXISTING_OBJECTS, (event, categoryName, records) => {
+    if (records.length > 0) {
+      appActions.confirmSyncRecords(records)
+    }
     if (!syncEnabled()) {
       return
     }

docs/appActions.md

@@ -777,6 +777,30 @@ Dispatches a message to set objectId for a syncable object.
 
 
 
+### pendSyncRecords(records:) 
+
+Add records sent with sync lib's SEND_SYNC_RECORDS to the appState
+records pending upload. After we download records via the sync lib
+we run confirmSyncRecords.
+
+**Parameters**
+
+**records:**: `Object`, Array.<object>
+
+
+
+### confirmSyncRecords(records:) 
+
+Remove records from the appState's records pending upload.
+This function is called after we download the records from the sync
+library.
+
+**Parameters**
+
+**records:**: `Object`, Array.<object>
+
+
+
 ### saveSyncDevices(devices) 
 
 Dispatch to update sync devices cache.

docs/state.md

@@ -302,6 +302,12 @@ AppStore
     objectsById: {
       [string of objectId joined by pipes |]: Array.<string> // array key path within appState, so we can do appState.getIn({key path})
     },
+    pendingRecords: { // OrderedMap of unconfirmed (not yet downloaded) sync records.
+      [objectId]: {
+        enqueueTimestamp: number // new Date().getTime() when record was submitted
+        record: object, // Sync record sent with SEND_SYNC_RECORDS
+      }
+    },
     seed: Array.<number>,
     seedQr: string, // data URL of QR code representing the seed
     setupError: string? // indicates that an error occurred during sync setup

js/actions/appActions.js

@@ -958,6 +958,32 @@ const appActions = {
     })
   },
 
+  /**
+   * Add records sent with sync lib's SEND_SYNC_RECORDS to the appState
+   * records pending upload. After we download records via the sync lib
+   * we run confirmSyncRecords.
+   * @param {Object} records: Array.<object>
+   */
+  pendSyncRecords: function (records) {
+    dispatch({
+      actionType: appConstants.APP_PEND_SYNC_RECORDS,
+      records
+    })
+  },
+
+  /**
+   * Remove records from the appState's records pending upload.
+   * This function is called after we download the records from the sync
+   * library.
+   * @param {Object} records: Array.<object>
+   */
+  confirmSyncRecords: function (records) {
+    dispatch({
+      actionType: appConstants.APP_CONFIRM_SYNC_RECORDS,
+      records
+    })
+  },
+
   /**
    * Dispatch to update sync devices cache.
    * NOTE: deviceId is a string! Normally it's Array.<number> but that can't

js/constants/appConstants.js

@@ -97,6 +97,8 @@ const appConstants = {
   APP_TAB_CLONED: _,
   APP_SET_OBJECT_ID: _,
   APP_CREATE_SYNC_CACHE: _,
+  APP_PEND_SYNC_RECORDS: _,
+  APP_CONFIRM_SYNC_RECORDS: _,
   APP_SAVE_SYNC_DEVICES: _,
   APP_SAVE_SYNC_INIT_DATA: _,
   APP_RESET_SYNC_DATA: _,

js/constants/sync/messages.js

@@ -82,7 +82,7 @@ const messages = {
    * browser sends this to the webview with the data that needs to be synced
    * to the sync server.
    */
-  SEND_SYNC_RECORDS: _, /* @param {string} categoryName, @param {Array.<Object>} records */
+  SEND_SYNC_RECORDS: _, /* @param {string=} categoryName, @param {Array.<Object>} records */
   /**
    * browser -> webview
    * browser sends this to delete the current user.

js/constants/sync/proto.js

@@ -11,3 +11,10 @@ module.exports.actions = {
   UPDATE: 1,
   DELETE: 2
 }
+
+module.exports.categoryMap = {
+  bookmark: 'BOOKMARKS',
+  historySite: 'HISTORY_SITES',
+  siteSetting: 'PREFERENCES',
+  device: 'PREFERENCES'
+}