Merge pull request #207 from isocolsky/feat/email_notifications

Email notifications

config.js

@@ -18,10 +18,10 @@ var config = {
   },
   lockOpts: {
     //  To use locker-server, uncomment this:
-    // lockerServer: {
-    //   host: 'localhost',
-    //   port: 3231,
-    // },
+    lockerServer: {
+      host: 'localhost',
+      port: 3231,
+    },
   },
   messageBrokerOpts: {
     //  To use message broker server, uncomment this:
@@ -39,5 +39,13 @@ var config = {
       url: 'https://test-insight.bitpay.com:443',
     },
   },
+  // To use email notifications uncomment this:
+  emailOpts: {
+    host: 'localhost',
+    port: 25,
+    ignoreTLS: true,
+    subjectPrefix: '[Wallet Service]',
+    from: 'wallet-service@bitcore.io',
+  },
 };
 module.exports = config;

lib/blockchainmonitor.js

@@ -9,6 +9,8 @@ log.debug = log.verbose;
 var BlockchainExplorer = require('./blockchainexplorer');
 var Storage = require('./storage');
 var MessageBroker = require('./messagebroker');
+var Lock = require('./lock');
+var EmailService = require('./emailservice');
 
 var Notification = require('./model/notification');
 
@@ -18,6 +20,8 @@ BlockchainMonitor.prototype.start = function(opts, cb) {
   opts = opts || {};
   $.checkArgument(opts.blockchainExplorerOpts);
   $.checkArgument(opts.storageOpts);
+  $.checkArgument(opts.lockOpts);
+  $.checkArgument(opts.emailOpts);
 
   var self = this;
 
@@ -38,8 +42,20 @@ BlockchainMonitor.prototype.start = function(opts, cb) {
       self.messageBroker = new MessageBroker(opts.messageBrokerOpts);
       done();
     },
-  ], cb);
+    function(done) {
+      self.lock = new Lock(opts.lockOpts);
+    },
+  ], function(err) {
+    if (err) return cb(err);
 
+    self.emailService = new EmailService({
+      lock: self.lock,
+      storage: self.storage,
+      mailer: opts.mailer,
+      emailOpts: opts.emailOpts,
+    });
+    return cb();
+  });
 };
 
 BlockchainMonitor.prototype._initExplorer = function(provider, network, url) {
@@ -105,7 +121,7 @@ BlockchainMonitor.prototype._handleIncommingTx = function(data) {
 BlockchainMonitor.prototype._createNotification = function(walletId, txid, address, amount, cb) {
   var self = this;
 
-  var n = Notification.create({
+  var notification = Notification.create({
     type: 'NewIncomingTx',
     data: {
       txid: txid,
@@ -114,9 +130,13 @@ BlockchainMonitor.prototype._createNotification = function(walletId, txid, addre
     },
     walletId: walletId,
   });
-  self.storage.storeNotification(walletId, n, function() {
-    self.messageBroker.send(n)
-    return cb();
+  self.storage.storeNotification(walletId, notification, function() {
+    self.messageBroker.send(notification)
+    if (self.emailService) {
+      self.emailService.sendEmail(notification, function() {
+        if (cb) return cb();
+      });
+    }
   });
 };
 

lib/emailservice.js

@@ -0,0 +1,200 @@
+'use strict';
+
+var _ = require('lodash');
+var $ = require('preconditions').singleton();
+var async = require('async');
+var log = require('npmlog');
+log.debug = log.verbose;
+var fs = require('fs');
+var nodemailer = require('nodemailer');
+
+var Model = require('./model');
+
+var EMAIL_TYPES = {
+  'NewCopayer': {
+    filename: 'new_copayer',
+    notifyDoer: false,
+  },
+  'NewTxProposal': {
+    filename: 'new_tx_proposal',
+    notifyDoer: false,
+  },
+  'NewOutgoingTx': {
+    filename: 'new_outgoing_tx',
+    notifyDoer: true,
+  },
+  'NewIncomingTx': {
+    filename: 'new_incoming_tx',
+    notifyDoer: true,
+  },
+  'TxProposalFinallyRejected': {
+    filename: 'txp_finally_rejected',
+    notifyDoer: false,
+  },
+};
+
+
+function EmailService(opts) {
+  $.checkArgument(opts);
+
+  opts.emailOpts = opts.emailOpts || {};
+
+  this.storage = opts.storage;
+  this.lock = opts.lock;
+  this.mailer = opts.mailer || nodemailer.createTransport(opts.emailOpts);
+  this.subjectPrefix = opts.emailOpts.subjectPrefix || '[Wallet service]';
+  this.from = opts.emailOpts.from;
+
+  $.checkState(this.mailer);
+  $.checkState(this.from);
+};
+
+// TODO: cache for X minutes
+EmailService.prototype._readTemplate = function(filename, cb) {
+  fs.readFile(__dirname + '/templates/' + filename + '.plain', 'utf8', function(err, template) {
+    if (err) {
+      log.error('Could not read template file ' + filename, err);
+      return cb(err);
+    }
+    var lines = template.split('\n');
+    return cb(null, {
+      subject: _.template(lines[0]),
+      body: _.template(_.rest(lines).join('\n')),
+    });
+  });
+};
+
+EmailService.prototype._applyTemplate = function(template, data, cb) {
+  var result = _.mapValues(template, function(t) {
+    try {
+      return t(data);
+    } catch (e) {
+      log.error('Could not apply data to template', e);
+      return cb(e);
+    }
+  });
+  return cb(null, result);
+};
+
+EmailService.prototype._getRecipientsList = function(notification, emailType, cb) {
+  var self = this;
+
+  self.storage.fetchPreferences(notification.walletId, null, function(err, preferences) {
+    if (err) return cb(err);
+    if (_.isEmpty(preferences)) return cb(null, []);
+
+    var recipients = _.compact(_.map(preferences, function(p) {
+      if (!p.email) return;
+      if (notification.creatorId == p.copayerId && !emailType.notifyDoer) return;
+      return {
+        copayerId: p.copayerId,
+        emailAddress: p.email
+      };
+    }));
+
+    return cb(null, recipients);
+  });
+};
+
+EmailService.prototype._getDataForTemplate = function(notification, cb) {
+  var self = this;
+
+  var data = _.cloneDeep(notification.data);
+  data.subjectPrefix = _.trim(self.subjectPrefix) + ' ';
+  self.storage.fetchWallet(notification.walletId, function(err, wallet) {
+    if (err) return cb(err);
+    data.walletId = wallet.id;
+    data.walletName = wallet.name;
+    data.walletM = wallet.m;
+    data.walletN = wallet.n;
+    var copayer = _.find(wallet.copayers, {
+      id: notification.creatorId
+    });
+    if (copayer) {
+      data.creatorId = copayer.id;
+      data.creatorName = copayer.name;
+    }
+    return cb(null, data);
+  });
+};
+
+EmailService.prototype._send = function(email, cb) {
+  var self = this;
+
+  var mailOptions = {
+    from: email.from,
+    to: email.to,
+    subject: email.subject,
+    text: email.body,
+  };
+  self.mailer.sendMail(mailOptions, function(err, result) {
+    if (err) {
+      log.error('An error occurred when trying to send email to ' + email.to, err);
+      return cb(err);
+    }
+    log.debug('Message sent: ', result || '');
+    return cb(err, result);
+  });
+};
+
+EmailService.prototype.sendEmail = function(notification, cb) {
+  var self = this;
+
+  var emailType = EMAIL_TYPES[notification.type];
+  if (!emailType) return cb();
+
+  self._getRecipientsList(notification, emailType, function(err, recipientsList) {
+    if (_.isEmpty(recipientsList)) return cb();
+
+    async.waterfall([
+
+      function(next) {
+        async.parallel([
+
+          function(next) {
+            self._readTemplate(emailType.filename, next);
+          },
+          function(next) {
+            self._getDataForTemplate(notification, next);
+          },
+        ], function(err, res) {
+          next(err, res[0], res[1]);
+        });
+      },
+      function(template, data, next) {
+        self._applyTemplate(template, data, next);
+      },
+      function(content, next) {
+        async.map(recipientsList, function(recipient, next) {
+          var email = Model.Email.create({
+            walletId: notification.walletId,
+            copayerId: recipient.copayerId,
+            from: self.from,
+            to: recipient.emailAddress,
+            subject: content.subject,
+            body: content.body,
+          });
+          self.storage.storeEmail(email, function(err) {
+            return next(err, email);
+          });
+        }, next);
+      },
+      function(emails, next) {
+        async.each(emails, function(email, next) {
+          self._send(email, function(err) {
+            if (err) {
+              email.setFail();
+            } else {
+              email.setSent();
+            }
+            self.storage.storeEmail(email, next);
+          });
+        }, function(err) {
+          return next();
+        });
+      },
+    ], cb);
+  });
+};
+
+module.exports = EmailService;

lib/expressapp.js

@@ -178,6 +178,13 @@ ExpressApp.prototype.start = function(opts, cb) {
             next();
           });
         },
+        function(next) {
+          server.getPreferences({}, function(err, preferences) {
+            if (err) return next(err);
+            result.preferences = preferences;
+            next();
+          });
+        },
       ], function(err) {
         if (err) return returnError(err, res, req);
         res.json(result);

lib/lock.js

@@ -12,7 +12,7 @@ function Lock(opts) {
   if (opts.lockerServer) {
     this.lock = new RemoteLock(opts.lockerServer.port, opts.lockerServer.host);
 
-    log.info('Using locker server:'  + opts.lockerServer.host + ':' + opts.lockerServer.port);
+    log.info('Using locker server:' + opts.lockerServer.host + ':' + opts.lockerServer.port);
 
     this.lock.on('reset', function() {
       log.debug('Locker server reset');

lib/model/email.js

@@ -0,0 +1,62 @@
+'use strict';
+
+var _ = require('lodash');
+var Uuid = require('uuid');
+
+function Email() {
+  this.version = '1.0.0';
+};
+
+Email.create = function(opts) {
+  opts = opts || {};
+
+  var x = new Email();
+
+  var now = Date.now();
+  x.createdOn = Math.floor(now / 1000);
+  x.id = _.padLeft(now, 14, '0') + Uuid.v4();
+  x.walletId = opts.walletId;
+  x.copayerId = opts.copayerId;
+  x.from = opts.from;
+  x.to = opts.to;
+  x.subject = opts.subject;
+  x.body = opts.body;
+  x.status = 'pending';
+  x.attempts = 0;
+  x.lastAttemptOn = null;
+  return x;
+};
+
+Email.fromObj = function(obj) {
+  var x = new Email();
+
+  x.createdOn = obj.createdOn;
+  x.id = obj.id;
+  x.walletId = obj.walletId;
+  x.copayerId = obj.copayerId;
+  x.from = obj.from;
+  x.to = obj.to;
+  x.subject = obj.subject;
+  x.body = obj.body;
+  x.status = obj.status;
+  x.attempts = obj.attempts;
+  x.lastAttemptOn = obj.lastAttemptOn;
+  return x;
+};
+
+Email.prototype._logAttempt = function(result) {
+  this.attempts++;
+  this.lastAttemptOn = Math.floor(Date.now() / 1000);
+  this.status = result;
+};
+
+Email.prototype.setSent = function() {
+  this._logAttempt('sent');
+};
+
+Email.prototype.setFail = function() {
+  this._logAttempt('fail');
+};
+
+
+module.exports = Email;

lib/model/index.js

@@ -5,5 +5,7 @@ Model.Copayer = require('./copayer');
 Model.TxProposal = require('./txproposal');
 Model.Address = require('./address');
 Model.Notification = require('./notification');
+Model.Preferences = require('./preferences');
+Model.Email = require('./email');
 
 module.exports = Model;

lib/model/preferences.js

@@ -0,0 +1,30 @@
+'use strict';
+
+function Preferences() {
+  this.version = '1.0.0';
+};
+
+Preferences.create = function(opts) {
+  opts = opts || {};
+
+  var x = new Preferences();
+
+  x.createdOn = Math.floor(Date.now() / 1000);
+  x.walletId = opts.walletId;
+  x.copayerId = opts.copayerId;
+  x.email = opts.email;
+  return x;
+};
+
+Preferences.fromObj = function(obj) {
+  var x = new Preferences();
+
+  x.createdOn = obj.createdOn;
+  x.walletId = obj.walletId;
+  x.copayerId = obj.copayerId;
+  x.email = obj.email;
+  return x;
+};
+
+
+module.exports = Preferences;

lib/notificationbroadcaster.js

@@ -0,0 +1,25 @@
+'use strict';
+
+var log = require('npmlog');
+log.debug = log.verbose;
+var inherits = require('inherits');
+var events = require('events');
+var nodeutil = require('util');
+
+function NotificationBroadcaster() {};
+
+nodeutil.inherits(NotificationBroadcaster, events.EventEmitter);
+
+NotificationBroadcaster.prototype.broadcast = function(eventName, notification, walletService) {
+  this.emit(eventName, notification, walletService);
+};
+
+var _instance;
+NotificationBroadcaster.singleton = function() {
+  if (!_instance) {
+    _instance = new NotificationBroadcaster();
+  }
+  return _instance;
+};
+
+module.exports = NotificationBroadcaster.singleton();

lib/server.js

@@ -5,6 +5,7 @@ var async = require('async');
 var log = require('npmlog');
 log.debug = log.verbose;
 log.disableColor();
+var EmailValidator = require('email-validator');
 
 var WalletUtils = require('bitcore-wallet-utils');
 var Bitcore = WalletUtils.Bitcore;
@@ -18,16 +19,19 @@ var Lock = require('./lock');
 var Storage = require('./storage');
 var MessageBroker = require('./messagebroker');
 var BlockchainExplorer = require('./blockchainexplorer');
+var EmailService = require('./emailservice');
 
-var Wallet = require('./model/wallet');
-var Copayer = require('./model/copayer');
-var Address = require('./model/address');
-var TxProposal = require('./model/txproposal');
-var Notification = require('./model/notification');
+var Model = require('./model');
+var Wallet = Model.Wallet;
 
 var initialized = false;
-var lock, storage, blockchainExplorer, blockchainExplorerOpts;
+
+var lock;
+var storage;
+var blockchainExplorer;
+var blockchainExplorerOpts;
 var messageBroker;
+var emailService;
 
 
 /**
@@ -44,6 +48,7 @@ function WalletService() {
   this.blockchainExplorerOpts = blockchainExplorerOpts;
   this.messageBroker = messageBroker;
   this.notifyTicker = 0;
+  this.emailService = emailService;
 };
 
 /**
@@ -61,20 +66,31 @@ WalletService.initialize = function(opts, cb) {
   blockchainExplorer = opts.blockchainExplorer;
   blockchainExplorerOpts = opts.blockchainExplorerOpts;
 
-  if (initialized)
-    return cb();
+  if (initialized) return cb();
 
   function initStorage(cb) {
     if (opts.storage) {
       storage = opts.storage;
       return cb();
+    } else {
+      var newStorage = new Storage();
+      newStorage.connect(opts.storageOpts, function(err) {
+        if (err) return cb(err);
+        storage = newStorage;
+        return cb();
+      });
     }
-    var newStorage = new Storage();
-    newStorage.connect(opts.storageOpts, function(err) {
-      if (err) return cb(err);
-      storage = newStorage;
-      return cb();
+  };
+
+  function initEmailService(cb) {
+    if (!opts.mailer && !opts.emailOpts) return cb();
+    emailService = new EmailService({
+      lock: lock,
+      storage: storage,
+      mailer: opts.mailer,
+      emailOpts: opts.emailOpts,
     });
+    return cb();
   };
 
   function initMessageBroker(cb) {
@@ -94,6 +110,9 @@ WalletService.initialize = function(opts, cb) {
     function(next) {
       initMessageBroker(next);
     },
+    function(next) {
+      initEmailService(next);
+    },
   ], function(err) {
     if (err) {
       log.error('Could not initialize', err);
@@ -322,24 +341,32 @@ WalletService.prototype._notify = function(type, data, opts, cb) {
 
   log.debug('Notification', type, data);
 
+  cb = cb || function() {};
+
   var walletId = self.walletId || data.walletId;
   var copayerId = self.copayerId || data.copayerId;
 
   $.checkState(walletId);
 
-  var n = Notification.create({
+  var notification = Model.Notification.create({
     type: type,
     data: data,
     ticker: this.notifyTicker++,
     creatorId: opts.isGlobal ? null : copayerId,
     walletId: walletId,
   });
-  this.storage.storeNotification(walletId, n, function() {
-    self.messageBroker.send(n);
-    if (cb) return cb();
+
+  this.storage.storeNotification(walletId, notification, function() {
+    self.messageBroker.send(notification);
+    if (self.emailService) {
+      self.emailService.sendEmail(notification, function() {
+        if (cb) return cb();
+      });
+    }
   });
 };
 
+
 /**
  * Joins a wallet in creation.
  * @param {Object} opts
@@ -378,7 +405,7 @@ WalletService.prototype.joinWallet = function(opts, cb) {
       if (wallet.copayers.length == wallet.n)
         return cb(new ClientError('WFULL', 'Wallet full'));
 
-      var copayer = Copayer.create({
+      var copayer = Model.Copayer.create({
         name: opts.name,
         copayerIndex: wallet.copayers.length,
         xPubKey: opts.xPubKey,
@@ -412,6 +439,49 @@ WalletService.prototype.joinWallet = function(opts, cb) {
   });
 };
 
+/**
+ * Save copayer preferences for the current wallet/copayer pair.
+ * @param {Object} opts
+ * @param {string} opts.email - Email address for notifications.
+ */
+WalletService.prototype.savePreferences = function(opts, cb) {
+  var self = this;
+
+  opts = opts || {};
+
+  if (opts.email) {
+    if (!EmailValidator.validate(opts.email)) {
+      return cb(new ClientError('Invalid email address'));
+    }
+  }
+
+  self._runLocked(cb, function(cb) {
+    var preferences = Model.Preferences.create({
+      walletId: self.walletId,
+      copayerId: self.copayerId,
+      email: opts.email,
+    });
+    self.storage.storePreferences(preferences, function(err) {
+      return cb(err);
+    });
+  });
+};
+
+/**
+ * Retrieves a preferences for the current wallet/copayer pair.
+ * @param {Object} opts
+ * @returns {Object} preferences
+ */
+WalletService.prototype.getPreferences = function(opts, cb) {
+  var self = this;
+
+  self.storage.fetchPreferences(self.walletId, self.copayerId, function(err, preferences) {
+    if (err) return cb(err);
+    return cb(null, preferences || {});
+  });
+};
+
+
 /**
  * Creates a new address.
  * @param {Object} opts
@@ -707,7 +777,7 @@ WalletService.prototype.createTx = function(opts, cb) {
 
       var changeAddress = wallet.createAddress(true);
 
-      var txp = TxProposal.create({
+      var txp = Model.TxProposal.create({
         walletId: self.walletId,
         creatorId: self.copayerId,
         toAddress: opts.toAddress,

lib/storage.js

@@ -18,6 +18,8 @@ var collections = {
   ADDRESSES: 'addresses',
   NOTIFICATIONS: 'notifications',
   COPAYERS_LOOKUP: 'copayers_lookup',
+  PREFERENCES: 'preferences',
+  EMAIL_QUEUE: 'email_queue',
 };
 
 var Storage = function(opts) {
@@ -353,6 +355,58 @@ Storage.prototype.fetchAddress = function(address, cb) {
   });
 };
 
+Storage.prototype.fetchPreferences = function(walletId, copayerId, cb) {
+  this.db.collection(collections.PREFERENCES).find({
+    walletId: walletId,
+  }).toArray(function(err, result) {
+    if (err) return cb(err);
+
+    if (copayerId) {
+      result = _.find(result, {
+        copayerId: copayerId
+      });
+    }
+    if (!result) return cb();
+
+    var preferences = _.map([].concat(result), function(r) {
+      return Model.Preferences.fromObj(r);
+    });
+    if (copayerId) {
+      preferences = preferences[0];
+    }
+    return cb(null, preferences);
+  });
+};
+
+Storage.prototype.storePreferences = function(preferences, cb) {
+  this.db.collection(collections.PREFERENCES).update({
+    walletId: preferences.walletId,
+    copayerId: preferences.copayerId,
+  }, preferences, {
+    w: 1,
+    upsert: true,
+  }, cb);
+};
+
+Storage.prototype.storeEmail = function(email, cb) {
+  this.db.collection(collections.EMAIL_QUEUE).update({
+    id: email.id,
+  }, email, {
+    w: 1,
+    upsert: true,
+  }, cb);
+};
+
+Storage.prototype.fetchUnsentEmails = function(cb) {
+  this.db.collection(collections.EMAIL_QUEUE).find({
+    status: 'pending',
+  }).toArray(function(err, result) {
+    if (err) return cb(err);
+    if (!result || _.isEmpty(result)) return cb(null, []);
+    return cb(null, Model.Email.fromObj(result));
+  });
+};
+
 
 Storage.prototype._dump = function(cb, fn) {
   fn = fn || console.log;

lib/templates/new_copayer.plain

@@ -0,0 +1,2 @@
+<%= subjectPrefix %>New copayer
+A new copayer just joined your wallet <%= walletName %>.

lib/templates/new_incoming_tx.plain

@@ -0,0 +1,2 @@
+<%= subjectPrefix %>New payment received
+A Payment has been received into your wallet <%= walletName %>.

lib/templates/new_outgoing_tx.plain

@@ -0,0 +1,2 @@
+<%= subjectPrefix %>Payment sent
+A Payment has been sent from your wallet <%= walletName %>.

lib/templates/new_tx_proposal.plain

@@ -0,0 +1,2 @@
+<%= subjectPrefix %>New spend proposal
+A new spend proposal has been created in your wallet <%= walletName %> by <%= creatorName %>.

lib/templates/txp_finally_rejected.plain

@@ -0,0 +1,2 @@
+<%= subjectPrefix %>Spend proposal rejected
+A spend proposal in your wallet <%= walletName %> has been rejected by <%= creatorName %>.

package.json

@@ -24,6 +24,7 @@
     "bitcore-wallet-utils": "0.0.12",
     "body-parser": "^1.11.0",
     "coveralls": "^2.11.2",
+    "email-validator": "^1.0.1",
     "express": "^4.10.0",
     "inherits": "^2.0.1",
     "locker": "^0.1.0",
@@ -32,6 +33,7 @@
     "mocha-lcov-reporter": "0.0.1",
     "mongodb": "^2.0.27",
     "morgan": "*",
+    "nodemailer": "^1.3.4",
     "npmlog": "^0.1.1",
     "preconditions": "^1.0.7",
     "read": "^1.0.5",

test/integration/server.js

@@ -21,25 +21,21 @@ var Bitcore = WalletUtils.Bitcore;
 var Storage = require('../../lib/storage');
 
 var Model = require('../../lib/model');
-var Wallet = Model.Wallet;
-var TxProposal = Model.TxProposal;
-var Address = Model.Address;
-var Copayer = Model.Copayer;
 
 var WalletService = require('../../lib/server');
 var TestData = require('../testdata');
 
 var helpers = {};
 helpers.getAuthServer = function(copayerId, cb) {
-  var signatureStub = sinon.stub(WalletService.prototype, '_verifySignature');
-  signatureStub.returns(true);
+  var verifyStub = sinon.stub(WalletService.prototype, '_verifySignature');
+  verifyStub.returns(true);
   WalletService.getInstanceWithAuth({
     copayerId: copayerId,
     message: 'dummy',
     signature: 'dummy',
   }, function(err, server) {
+    verifyStub.restore();
     if (err || !server) throw new Error('Could not login as copayerId ' + copayerId);
-    signatureStub.restore();
     return cb(server);
   });
 };
@@ -136,7 +132,6 @@ helpers.toSatoshi = function(btc) {
   }
 };
 
-// Amounts in satoshis 
 helpers.stubUtxos = function(server, wallet, amounts, cb) {
   var amounts = [].concat(amounts);
 
@@ -215,16 +210,34 @@ helpers.createAddresses = function(server, wallet, main, change, cb) {
   });
 };
 
-var db, storage, blockchainExplorer;
+var storage, blockchainExplorer, mailer;
 
-function openDb(cb) {
-  db = new tingodb.Db('./db/test', {});
-  return cb();
+var useMongo = false;
+
+function initStorage(cb) {
+  function getDb(cb) {
+    if (useMongo) {
+      var mongodb = require('mongodb');
+      mongodb.MongoClient.connect('mongodb://localhost:27017/bws_test', function(err, db) {
+        if (err) throw err;
+        return cb(db);
+      });
+    } else {
+      var db = new tingodb.Db('./db/test', {});
+      return cb(db);
+    }
+  }
+  getDb(function(db) {
+    storage = new Storage({
+      db: db
+    });
+    return cb();
+  });
 };
 
-function resetDb(cb) {
-  if (!db) return cb();
-  db.dropDatabase(function(err) {
+function resetStorage(cb) {
+  if (!storage.db) return cb();
+  storage.db.dropDatabase(function(err) {
     return cb();
   });
 };
@@ -232,19 +245,19 @@ function resetDb(cb) {
 
 describe('Wallet service', function() {
   before(function(done) {
-    openDb(function() {
-      storage = new Storage({
-        db: db
-      });
-      done();
-    });
+    initStorage(done);
   });
   beforeEach(function(done) {
-    resetDb(function() {
+    resetStorage(function() {
       blockchainExplorer = sinon.stub();
+      mailer = sinon.stub();
       WalletService.initialize({
         storage: storage,
         blockchainExplorer: blockchainExplorer,
+        mailer: mailer,
+        emailOpts: {
+          from: 'bws@dummy.net',
+        }
       }, done);
     });
   });
@@ -673,7 +686,7 @@ describe('Wallet service', function() {
     it('should create address', function(done) {
       server.createAddress({}, function(err, address) {
         should.not.exist(err);
-        address.should.exist;
+        should.exist(address);
         address.walletId.should.equal(wallet.id);
         address.network.should.equal('livenet');
         address.address.should.equal('3KxttbKQQPWmpsnXZ3rB4mgJTuLnVR7frg');
@@ -718,7 +731,7 @@ describe('Wallet service', function() {
           server.storage.storeAddressAndWallet.restore();
           server.createAddress({}, function(err, address) {
             should.not.exist(err);
-            address.should.exist;
+            should.exist(address);
             address.address.should.equal('3KxttbKQQPWmpsnXZ3rB4mgJTuLnVR7frg');
             address.path.should.equal('m/2147483647/0/0');
             done();
@@ -728,6 +741,61 @@ describe('Wallet service', function() {
     });
   });
 
+  describe('Preferences', function() {
+    var server, wallet;
+    beforeEach(function(done) {
+      helpers.createAndJoinWallet(2, 2, function(s, w) {
+        server = s;
+        wallet = w;
+        done();
+      });
+    });
+
+    it('should save & retrieve preferences', function(done) {
+      server.savePreferences({
+        email: 'dummy@dummy.com'
+      }, function(err) {
+        should.not.exist(err);
+        server.getPreferences({}, function(err, preferences) {
+          should.not.exist(err);
+          should.exist(preferences);
+          preferences.email.should.equal('dummy@dummy.com');
+          done();
+        });
+      });
+    });
+    it('should save preferences only for requesting copayer', function(done) {
+      server.savePreferences({
+        email: 'dummy@dummy.com'
+      }, function(err) {
+        should.not.exist(err);
+        helpers.getAuthServer(wallet.copayers[1].id, function(server2) {
+          server2.getPreferences({}, function(err, preferences) {
+            should.not.exist(err);
+            should.not.exist(preferences.email);
+            done();
+          });
+        });
+      });
+    });
+    it.skip('should save preferences only for requesting wallet', function(done) {});
+    it('should validate email address', function(done) {
+      server.savePreferences({
+        email: ' '
+      }, function(err) {
+        should.exist(err);
+        err.message.should.contain('email');
+        server.savePreferences({
+          email: 'dummy@' + _.repeat('domain', 50),
+        }, function(err) {
+          should.exist(err);
+          err.message.should.contain('email');
+          done();
+        });
+      });
+    });
+  });
+
   describe('#getBalance', function() {
     var server, wallet;
     beforeEach(function(done) {
@@ -1514,7 +1582,7 @@ describe('Wallet service', function() {
       });
     });
 
-    it('should brodcast a tx', function(done) {
+    it('should broadcast a tx', function(done) {
       var clock = sinon.useFakeTimers(1234000);
       helpers.stubBroadcast('999');
       server.broadcastTx({
@@ -3030,4 +3098,50 @@ describe('Wallet service', function() {
       });
     });
   });
+  describe('Email notifications', function() {
+    var server, wallet, sendMailStub;
+    beforeEach(function(done) {
+      helpers.createAndJoinWallet(2, 3, function(s, w) {
+        server = s;
+        wallet = w;
+        sendMailStub = sinon.stub();
+        sendMailStub.yields();
+        server.emailService.mailer.sendMail = sendMailStub;
+
+        var i = 0;
+        async.eachSeries(w.copayers, function(copayer, next) {
+          helpers.getAuthServer(copayer.id, function(server) {
+            server.savePreferences({
+              email: 'copayer' + (i++) + '@domain.com',
+            }, next);
+          });
+        }, done);
+      });
+    });
+
+    it('should notify copayers a new tx proposal has been created', function(done) {
+      helpers.stubUtxos(server, wallet, [1, 1], function() {
+        var txOpts = helpers.createProposalOpts('18PzpUFkFZE8zKWUPvfykkTxmB9oMR8qP7', 0.8, 'some message', TestData.copayers[0].privKey_1H_0);
+        server.createTx(txOpts, function(err, tx) {
+          should.not.exist(err);
+          var calls = sendMailStub.getCalls();
+          calls.length.should.equal(2);
+          var emails = _.map(calls, function(c) {
+            return c.args[0];
+          });
+          _.difference(['copayer1@domain.com', 'copayer2@domain.com'], _.pluck(emails, 'to')).should.be.empty;
+          var one = emails[0];
+          one.from.should.equal('bws@dummy.net');
+          one.subject.should.contain('New spend proposal');
+          one.text.should.contain(wallet.name);
+          one.text.should.contain(wallet.copayers[0].name);
+          server.storage.fetchUnsentEmails(function(err, unsent) {
+            should.not.exist(err);
+            unsent.should.be.empty;
+            done();
+          });
+        });
+      });
+    });
+  });
 });

test/storage.js

@@ -16,9 +16,16 @@ var db, storage;
 
 function openDb(cb) {
   db = new tingodb.Db('./db/test', {});
+  // HACK: There appears to be a bug in TingoDB's close function where the callback is not being executed
+  db.__close = db.close;
+  db.close = function(force, cb) {
+    this.__close(force, cb);
+    return cb();
+  };
   return cb();
 };
 
+
 function resetDb(cb) {
   if (!db) return cb();
   db.dropDatabase(function(err) {