Merge pull request #651 from isocolsky/fix/notif-replay

Avoid replaying notification for incoming txs

lib/blockchainmonitor.js

@@ -160,18 +160,30 @@ BlockchainMonitor.prototype._handleTxOuts = function(data) {
       var walletId = address.walletId;
       log.info('Incoming tx for wallet ' + walletId + ' [' + out.amount + 'sat -> ' + out.address + ']');
 
-      var notification = Notification.create({
-        type: 'NewIncomingTx',
-        data: {
-          txid: data.txid,
-          address: out.address,
-          amount: out.amount,
-        },
-        walletId: walletId,
-      });
-      self.storage.softResetTxHistoryCache(walletId, function() {
-        self._updateActiveAddresses(address, function() {
-          self._storeAndBroadcastNotification(notification, next);
+      var fromTs = Date.now() - 24 * 3600 * 1000;
+      self.storage.fetchNotifications(walletId, null, fromTs, function(err, notifications) {
+        if (err) return next(err);
+        var alreadyNotified = _.any(notifications, function(n) {
+          return n.type == 'NewIncomingTx' && n.data && n.data.txid == data.txid;
+        });
+        if (alreadyNotified) {
+          log.info('The incoming tx ' + data.txid + ' was already notified');
+          return next();
+        }
+
+        var notification = Notification.create({
+          type: 'NewIncomingTx',
+          data: {
+            txid: data.txid,
+            address: out.address,
+            amount: out.amount,
+          },
+          walletId: walletId,
+        });
+        self.storage.softResetTxHistoryCache(walletId, function() {
+          self._updateActiveAddresses(address, function() {
+            self._storeAndBroadcastNotification(notification, next);
+          });
         });
       });
     });

test/integration/bcmonitor.js

@@ -87,4 +87,31 @@ describe('Blockchain monitor', function() {
       }, 100);
     });
   });
+
+  it('should not notify copayers of incoming txs more than once', function(done) {
+    server.createAddress({}, function(err, address) {
+      should.not.exist(err);
+
+      var incoming = {
+        txid: '123',
+        vout: [{}],
+      };
+      incoming.vout[0][address.address] = 1500;
+      socket.handlers['tx'](incoming);
+      setTimeout(function() {
+        socket.handlers['tx'](incoming);
+
+        setTimeout(function() {
+          server.getNotifications({}, function(err, notifications) {
+            should.not.exist(err);
+            var notification = _.filter(notifications, {
+              type: 'NewIncomingTx'
+            });
+            notification.length.should.equal(1);
+            done();
+          });
+        }, 100);
+      }, 50);
+    });
+  });
 });