Cache messages on receipt, remove from cache when processed

FREEBIE
signalapp · Aug 4, 2017 · bd0050b · bd0050b
1 parent e6859a3
commit bd0050b
Show file tree

Hide file tree

Showing 13 changed files with 683 additions and 118 deletions.
diff --git a/Gruntfile.js b/Gruntfile.js
@@ -47,6 +47,7 @@ module.exports = function(grunt) {
           'libtextsecure/storage.js',
           'libtextsecure/storage/user.js',
           'libtextsecure/storage/groups.js',
+          'libtextsecure/storage/unprocessed.js',
           'libtextsecure/protobufs.js',
           'libtextsecure/websocket-resources.js',
           'libtextsecure/helpers.js',

diff --git a/js/background.js b/js/background.js
@@ -162,7 +162,7 @@
           return;
         }
 
-        ConversationController.create(c).save();
+        ConversationController.create(c).save().then(ev.confirm);
     }
 
     function onGroupReceived(ev) {
@@ -179,14 +179,24 @@
         } else {
             attributes.left = true;
         }
+
         var conversation = ConversationController.create(attributes);
-        conversation.save();
+        conversation.save().then(ev.confirm);
     }
 
     function onMessageReceived(ev) {
         var data = ev.data;
-        var message = initIncomingMessage(data.source, data.timestamp);
-        message.handleDataMessage(data.message);
+        var message = initIncomingMessage(data);
+
+        isMessageDuplicate(message).then(function(isDuplicate) {
+            if (isDuplicate) {
+                console.log('Received duplicate message', message.idForLogging());
+                ev.confirm();
+                return;
+            }
+
+            message.handleDataMessage(data.message, ev.confirm);
+        });
     }
 
     function onSentMessage(ev) {
@@ -195,6 +205,7 @@
 
         var message = new Whisper.Message({
             source         : textsecure.storage.user.getNumber(),
+            sourceDevice   : data.device,
             sent_at        : data.timestamp,
             received_at    : now,
             conversationId : data.destination,
@@ -203,17 +214,53 @@
             expirationStartTimestamp: data.expirationStartTimestamp,
         });
 
-        message.handleDataMessage(data.message);
+        isMessageDuplicate(message).then(function(isDuplicate) {
+            if (isDuplicate) {
+                console.log('Received duplicate message', message.idForLogging());
+                ev.confirm();
+                return;
+            }
+
+            message.handleDataMessage(data.message, ev.confirm);
+        });
     }
 
-    function initIncomingMessage(source, timestamp) {
+    function isMessageDuplicate(message) {
+        return new Promise(function(resolve) {
+            var fetcher = new Whisper.Message();
+            var options = {
+                index: {
+                    name: 'unique',
+                    value: [
+                        message.get('source'),
+                        message.get('sourceDevice'),
+                        message.get('sent_at')
+                    ]
+                }
+            };
+
+            fetcher.fetch(options).always(function() {
+                if (fetcher.get('id')) {
+                    return resolve(true);
+                }
+
+                return resolve(false);
+            });
+        }).catch(function(error) {
+            console.log('isMessageDuplicate error:', error && error.stack ? error.stack : error);
+            return false;
+        });
+    }
+
+    function initIncomingMessage(data) {
         var now = new Date().getTime();
 
         var message = new Whisper.Message({
-            source         : source,
-            sent_at        : timestamp,
+            source         : data.source,
+            sourceDevice   : data.sourceDevice,
+            sent_at        : data.timestamp,
             received_at    : now,
-            conversationId : source,
+            conversationId : data.source,
             type           : 'incoming',
             unread         : 1
         });
@@ -284,11 +331,12 @@
         var timestamp = ev.read.timestamp;
         var sender    = ev.read.sender;
         console.log('read receipt ', sender, timestamp);
-        Whisper.ReadReceipts.add({
+        var receipt = Whisper.ReadReceipts.add({
             sender    : sender,
             timestamp : timestamp,
             read_at   : read_at
         });
+        receipt.on('remove', ev.confirm);
     }
 
     function onVerified(ev) {
@@ -323,11 +371,11 @@
         };
 
         if (state === 'VERIFIED') {
-            contact.setVerified(options);
+            contact.setVerified(options).then(ev.confirm);
         } else if (state === 'DEFAULT') {
-            contact.setVerifiedDefault(options);
+            contact.setVerifiedDefault(options).then(ev.confirm);
         } else {
-            contact.setUnverified(options);
+            contact.setUnverified(options).then(ev.confirm);
         }
     }
 
@@ -340,9 +388,11 @@
             timestamp
         );
 
-        Whisper.DeliveryReceipts.add({
-            timestamp: timestamp, source: pushMessage.source
+        var receipt = Whisper.DeliveryReceipts.add({
+            timestamp: timestamp,
+            source: pushMessage.source
         });
+        receipt.on('remove', ev.confirm);
     }
 
     window.owsDesktopApp = {

diff --git a/js/database.js b/js/database.js
@@ -254,6 +254,26 @@
                   console.log(event);
                 };
             }
+        },
+        {
+            version: "14.0",
+            migrate: function(transaction, next) {
+                console.log('migration 14.0');
+                console.log('Adding unprocessed message store');
+                var unprocessed = transaction.db.createObjectStore('unprocessed');
+                unprocessed.createIndex('received', 'timestamp', { unique: false });
+                next();
+            }
+        },
+        {
+            version: "15.0",
+            migrate: function(transaction, next) {
+                console.log('migration 15.0');
+                console.log('Adding messages index for de-duplication');
+                var messages = transaction.objectStore('messages');
+                messages.createIndex('unique', ['source', 'sourceDevice', 'sent_at'], { unique: true });
+                next();
+            }
         }
     ];
 }());
diff --git a/js/debugLog.js b/js/debugLog.js
@@ -40,7 +40,7 @@
         }
     });
 
-    var MAX_MESSAGES = 1000;
+    var MAX_MESSAGES = 3000;
     var PHONE_REGEX = /\+\d{7,12}(\d{3})/g;
     var log = new DebugLog();
     if (window.console) {

diff --git a/js/delivery_receipts.js b/js/delivery_receipts.js
@@ -43,7 +43,6 @@
                 });
             }).then(function(message) {
                 if (message) {
-                    this.remove(receipt);
                     var deliveries = message.get('delivered') || 0;
                     message.save({
                         delivered: deliveries + 1
@@ -55,7 +54,9 @@
                         if (conversation) {
                             conversation.trigger('delivered', message);
                         }
-                    });
+
+                        this.remove(receipt);
+                    }.bind(this));
                     // TODO: consider keeping a list of numbers we've
                     // successfully delivered to?
                 }