Merge pull request #362 from brave/dup-device-id

Introduce device id v2

client/constants/messages.js

@@ -34,12 +34,17 @@ const messages = {
    * saved. 'config' contains apiVersion, serverUrl, debug;
    * see server/config/default.json
    */
-  GOT_INIT_DATA: _, /* @param {Array} seed, @param {Array} deviceId, @param {Object} config */
+  GOT_INIT_DATA: _, /* @param {Array} seed, @param {Array} deviceId,
+                     * @param {Object} config, @param {string} deviceIdV2
+                     * */
   /**
    * webview -> browser
    * browser must save values in persistent storage if non-empty
+   * It will be called during initial setup or deviceIdV2 migration
    */
-  SAVE_INIT_DATA: _, /* @param {Uint8Array} seed, @param {Uint8Array} deviceId */
+  SAVE_INIT_DATA: _, /* @param {Uint8Array} seed, @param {Uint8Array} deviceId,
+                      * @param {string} deviceIdV2
+                      */
   /**
    * webview -> browser
    * sent when sync has finished initialization

client/init.js

@@ -1,6 +1,7 @@
 'use strict'
 
 const crypto = require('brave-crypto')
+const {generateDeviceIdV2} = require('../lib/crypto')
 const messages = require('./constants/messages')
 const {syncVersion} = require('./config')
 
@@ -12,23 +13,25 @@ module.exports.init = function () {
   return new Promise((resolve, reject) => {
     const ipc = window.chrome.ipcRenderer
     ipc.send(messages.GET_INIT_DATA, syncVersion)
-    ipc.once(messages.GOT_INIT_DATA, (e, seed, deviceId, config) => {
+    ipc.once(messages.GOT_INIT_DATA, (e, seed, deviceId, config, deviceIdV2) => {
       if (seed === null) {
         // Generate a new "persona"
         seed = crypto.getSeed() // 32 bytes
         deviceId = new Uint8Array([0])
-        ipc.send(messages.SAVE_INIT_DATA, seed, deviceId)
+        deviceIdV2 = generateDeviceIdV2()
+        ipc.send(messages.SAVE_INIT_DATA, seed, deviceId, deviceIdV2)
         // TODO: The background process should listen for SAVE_INIT_DATA and emit
         // GOT_INIT_DATA once the save is successful
       }
+
       // XXX: workaround #17
       seed = seed instanceof Array ? new Uint8Array(seed) : seed
       deviceId = deviceId instanceof Array ? new Uint8Array(deviceId) : deviceId
       if (!(seed instanceof Uint8Array) || seed.length !== crypto.DEFAULT_SEED_SIZE) {
         reject(new Error('Invalid crypto seed'))
         return
       }
-      resolve({seed, deviceId, config})
+      resolve({seed, deviceId, config, deviceIdV2})
     })
   })
 }

client/requestUtil.js

@@ -2,6 +2,7 @@
 
 const awsSdk = require('aws-sdk')
 const cryptoUtil = require('./cryptoUtil')
+const deepEqual = require('deep-equal')
 const recordUtil = require('./recordUtil')
 const proto = require('./constants/proto')
 const {limitConcurrency} = require('../lib/promiseHelper')
@@ -30,6 +31,26 @@ const isExpiredCredentialError = (error) => {
   })
 }
 
+const createAndSubscribeSQSforCategory = function (deviceId, category, thisRef) {
+  let newQueueParams = {
+    QueueName: thisRef.sqsName(deviceId, category),
+    Attributes: {
+      'MessageRetentionPeriod': s3Helper.SQS_RETENTION
+    }
+  }
+  return new Promise((resolve, reject) => {
+    thisRef.sqs.createQueue(newQueueParams, (error, data) => {
+      if (error) {
+        console.log('SQS creation failed with error: ' + error)
+        reject(error)
+      } else if (data) {
+        thisRef.SQSUrlByCat[category] = data.QueueUrl
+        resolve([])
+      }
+    })
+  })
+}
+
 /**
  * @param {{
  *   apiVersion: <string>,
@@ -64,6 +85,7 @@ const RequestUtil = function (opts = {}) {
     this.saveAWSCredentials(credentials)
   }
   this.SQSUrlByCat = []
+  this.oldSQSUrlByCat = []
   this.missingObjectsCache = new LRUCache(50)
   // This is used to keep the most recent records for each object id
   this.latestRecordsCache = new LRUCache(100)
@@ -249,8 +271,30 @@ RequestUtil.prototype.list = function (category, startAt, maxRecords, nextContin
       WaitTimeSeconds: CONFIG.SQS_MESSAGES_LONGPOLL_TIMEOUT
     }
 
-    return s3Helper.listNotifications(this.sqs, notificationParams, category,
-      prefix)
+    // We will fetch from both old and new SQS queues until old one gets retired
+    if (this.oldSQSUrlByCat[category]) {
+      let oldNotificationParams = Object.assign({}, notificationParams)
+      oldNotificationParams.QueueUrl = `${this.oldSQSUrlByCat[category]}`
+
+      return s3Helper.listNotifications(
+        this.sqs, notificationParams, category, prefix).then((values) => {
+        if (this.shouldRetireOldSQSQueue(parseInt(values.createdTimeStamp))) {
+          return this.deleteSQSQueue(this.oldSQSUrlByCat[category]).then(() => {
+            delete this.oldSQSUrlByCat[category]
+            return values
+          })
+        }
+        return s3Helper.listNotifications(
+          this.sqs, oldNotificationParams, category, prefix).then((oldValues) => {
+          if (deepEqual(values.contents, oldValues.contents, {strict: true})) {
+            return values
+          }
+          return {contents: values.contents.concat(oldValues.contents),
+            createdTimeStamp: values.createdTimeStamp}
+        })
+      })
+    }
+    return s3Helper.listNotifications(this.sqs, notificationParams, category, prefix)
   })
 }
 
@@ -275,6 +319,21 @@ RequestUtil.prototype.shouldListObject = function (startAt, category) {
       this.listInProgress === true
 }
 
+/**
+ * The retention time of our SQS queue is 24 hours so we will retire old SQS
+ * queue created by device id after 24 hours
+ * @param {number=} createdTimestamp is the when new device id v2 queue was
+ * created
+ * @returns {boolean}
+ */
+RequestUtil.prototype.shouldRetireOldSQSQueue = function (createdTimestamp) {
+  let currentTime = new Date().getTime()
+  let newQueueCreatedTime =
+    this.normalizeTimestampToMs(createdTimestamp, currentTime)
+
+  return (currentTime - newQueueCreatedTime) > parseInt(s3Helper.SQS_RETENTION, 10) * 1000
+}
+
 /**
  * Checks do we need to use s3 list Object or SQS notifications
  * @param {number=} startAt return objects with timestamp >= startAt (e.g. 1482435340). Could be seconds or milliseconds
@@ -339,40 +398,57 @@ RequestUtil.prototype.sqsName = function (deviceId, category) {
   return queueName
 }
 
+/**
+ * Main purpose of this function is to create old SQS queue to test device id v2
+ * migration
+ * @param {string} deviceId
+ * @returns {Promise}
+ */
+RequestUtil.prototype.createAndSubscribeSQSForTest = function (deviceId) {
+  var createSQSPromises = []
+  // Simple for loop instead foreach to capture 'this'
+  for (var i = 0; i < CATEGORIES_FOR_SQS.length; ++i) {
+    createSQSPromises.push(createAndSubscribeSQSforCategory(deviceId, CATEGORIES_FOR_SQS[i], this))
+  }
+
+  return Promise.all(createSQSPromises)
+}
+
 /**
  * Creates SQS for the current device.
  * @param {string} deviceId
+ * @param {string} deviceIdV2
  * @returns {Promise}
  */
-RequestUtil.prototype.createAndSubscribeSQS = function (deviceId) {
+RequestUtil.prototype.createAndSubscribeSQS = function (deviceId, deviceIdV2) {
   // Creating a query for the current userId
-  if (!deviceId) {
-    throw new Error('createSQS failed. deviceId is null!')
+  if (!deviceIdV2) {
+    throw new Error('createSQS failed. deviceIdV2 is null!')
   }
-  this.deviceId = deviceId
-  const createAndSubscribeSQSforCategory = function (deviceId, category, thisRef) {
-    let newQueueParams = {
-      QueueName: thisRef.sqsName(deviceId, category),
-      Attributes: {
-        'MessageRetentionPeriod': s3Helper.SQS_RETENTION
-      }
-    }
+  this.deviceIdV2 = deviceIdV2
+
+  const subscribeOldSQSforCategory = function (deviceId, category, thisRef) {
     return new Promise((resolve, reject) => {
-      thisRef.sqs.createQueue(newQueueParams, (error, data) => {
+      let params = {
+        QueueName: thisRef.sqsName(deviceId, category)
+      }
+      thisRef.sqs.getQueueUrl(params, (error, data) => {
         if (error) {
-          console.log('SQS creation failed with error: ' + error)
-          reject(error)
+          // queue doesn't exist
+          resolve()
         } else if (data) {
-          thisRef.SQSUrlByCat[category] = data.QueueUrl
-          resolve([])
+          thisRef.oldSQSUrlByCat[category] = data.QueueUrl
+          resolve()
         }
       })
     })
   }
   var createSQSPromises = []
   // Simple for loop instead foreach to capture 'this'
   for (var i = 0; i < CATEGORIES_FOR_SQS.length; ++i) {
-    createSQSPromises.push(createAndSubscribeSQSforCategory(deviceId, CATEGORIES_FOR_SQS[i], this))
+    // Doesn't have to create about to deprecated queues
+    createSQSPromises.push(subscribeOldSQSforCategory(deviceId, CATEGORIES_FOR_SQS[i], this))
+    createSQSPromises.push(createAndSubscribeSQSforCategory(deviceIdV2, CATEGORIES_FOR_SQS[i], this))
   }
 
   return Promise.all(createSQSPromises)
@@ -530,7 +606,7 @@ RequestUtil.prototype.deleteUser = function () {
 RequestUtil.prototype.purgeUserCategoryQueue = function (category) {
   return new Promise((resolve, reject) => {
     let params = {
-      QueueName: this.sqsName(this.deviceId, category)
+      QueueName: this.sqsName(this.deviceIdV2, category)
     }
     this.sqs.getQueueUrl(params, (error, data) => {
       if (error) {
@@ -553,6 +629,24 @@ RequestUtil.prototype.purgeUserCategoryQueue = function (category) {
   })
 }
 
+/**
+ * Delete SQS queue by url
+ * @param {string} url - SQS queue url
+ */
+RequestUtil.prototype.deleteSQSQueue = function (url) {
+  return new Promise((resolve, reject) => {
+    let params = {
+      QueueUrl: url
+    }
+    this.sqs.deleteQueue(params, (err, data) => {
+      if (err) {
+        console.log('SQS deleteQueue failed with error: ' + err)
+      }
+      resolve([])
+    })
+  })
+}
+
 RequestUtil.prototype.purgeUserQueue = function () {
   var purgeQueuePromises = []
   for (var i = 0; i < CATEGORIES_FOR_SQS.length; ++i) {

client/sync.js

@@ -7,7 +7,7 @@ const recordUtil = require('./recordUtil')
 const messages = require('./constants/messages')
 const proto = require('./constants/proto')
 const serializer = require('../lib/serializer')
-const {deriveKeys} = require('../lib/crypto')
+const {deriveKeys, generateDeviceIdV2} = require('../lib/crypto')
 
 let ipc = window.chrome.ipcRenderer
 
@@ -18,6 +18,7 @@ const ERROR = 2
 const logElement = document.querySelector('#output')
 
 var clientDeviceId = null
+var clientDeviceIdV2 = null
 var clientUserId = null
 var clientKeys = {}
 var config = {}
@@ -52,6 +53,10 @@ const logSync = (message, logLevel = DEBUG) => {
  */
 const maybeSetDeviceId = (requester) => {
   if (clientDeviceId !== null) {
+    if (clientDeviceIdV2.length === 0) {
+      clientDeviceIdV2 = generateDeviceIdV2()
+      ipc.send(messages.SAVE_INIT_DATA, seed, clientDeviceId, clientDeviceIdV2)
+    }
     return Promise.resolve(requester)
   }
   if (!requester || !requester.s3) {
@@ -71,7 +76,8 @@ const maybeSetDeviceId = (requester) => {
         })
       }
       clientDeviceId = new Uint8Array([maxId + 1])
-      ipc.send(messages.SAVE_INIT_DATA, seed, clientDeviceId)
+      clientDeviceIdV2 = generateDeviceIdV2()
+      ipc.send(messages.SAVE_INIT_DATA, seed, clientDeviceId, clientDeviceIdV2)
       return Promise.resolve(requester)
     })
 }
@@ -252,6 +258,7 @@ const main = () => {
     const clientSerializer = values[0]
     const keys = deriveKeys(values[1].seed)
     const deviceId = values[1].deviceId
+    clientDeviceIdV2 = values[1].deviceIdV2
     seed = values[1].seed
     clientKeys = keys
     config = values[1].config
@@ -288,8 +295,8 @@ const main = () => {
     })
     .then((requester) => {
       if (clientDeviceId !== null && requester && requester.s3) {
-        logSync('set device ID: ' + clientDeviceId)
-        requester.createAndSubscribeSQS(clientDeviceId).then(() => {
+        logSync('set device ID: ' + clientDeviceId + ' device ID V2: ' + clientDeviceIdV2)
+        requester.createAndSubscribeSQS(clientDeviceId, clientDeviceIdV2).then(() => {
           startSync(requester)
         })
           .catch((e) => {

lib/api.proto

@@ -82,6 +82,7 @@ message SyncRecord {
   }
   message Device {
     string name = 1;
+    string deviceIdV2 = 2;
   }
   Action action = 1;
   bytes deviceId = 2;