Merge pull request #129 from mozilla/events

feat(events): add events to delete user data when account is deleted

bin/server.js

@@ -3,10 +3,18 @@
  * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
 
 const config = require('../lib/config').root();
+const db = require('../lib/db');
+const events = require('../lib/events');
 const logger = require('../lib/logging')('bin.server');
 const server = require('../lib/server').create();
 
 logger.debug('config', config);
-server.start(function() {
-  logger.info('listening', server.info.uri);
+db.ping().done(function() {
+  server.start(function() {
+    logger.info('listening', server.info.uri);
+  });
+  events.start();
+}, function(err) {
+  logger.critical('db.ping', err);
+  process.exit(2);
 });

lib/db/memory.js

@@ -127,6 +127,11 @@ MemoryStore.prototype = {
     return P.resolve({ id: id, name: id });
   },
 
+  removeProfile: function removeProfile(uid) {
+    delete this.profile[hex(uid)];
+    return P.resolve();
+  },
+
   getDisplayName: function (uid) {
     var id = hex(uid);
     var name = this.profile[id] ? this.profile[id].displayName : undefined;

lib/db/mysql/index.js

@@ -121,6 +121,7 @@ const Q_PROFILE_DISPLAY_NAME_UPDATE = 'INSERT INTO profile ' +
   'displayName = VALUES(displayName)';
 const Q_PROFILE_DISPLAY_NAME_GET = 'SELECT displayName FROM profile ' +
   'WHERE userId=?';
+const Q_PROFILE_DELETE = 'DELETE FROM profile WHERE userId=?';
 
 function firstRow(rows) {
   return rows[0];
@@ -205,6 +206,10 @@ MysqlStore.prototype = {
     return this._readOne(Q_PROFILE_DISPLAY_NAME_GET, [buf(uid)]);
   },
 
+  removeProfile: function removeProfile(uid) {
+    return this._write(Q_PROFILE_DELETE, [buf(uid)]);
+  },
+
   _write: function _write(sql, params) {
     return this._query(this._pool, sql, params);
   },

lib/env.js

@@ -0,0 +1,10 @@
+/* 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/. */
+
+const config = require('./config');
+
+exports.isProdLike = function isProdLike() {
+  var env = config.get('env');
+  return env === 'prod' || env === 'stage';
+};

lib/events.js

@@ -0,0 +1,65 @@
+/* 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/. */
+
+const Sink = require('fxa-notifier-aws').Sink;
+const P = require('./promise');
+
+const config = require('./config').root();
+const db = require('./db');
+const env = require('./env');
+const logger = require('./logging')('events');
+const workers = require('./img-workers');
+
+const HEX_STRING = require('./validate').hex;
+
+exports.onData = function onData(message) {
+  logger.verbose('data', message);
+  if (message.event === 'delete') {
+    var userId = message.uid.split('@')[0];
+    if (!HEX_STRING.test(userId)) {
+      message.del();
+      return logger.warn('badDelete', { userId: userId });
+    }
+    P.all([
+      db.getAvatars(userId).then(function(avatars) {
+        return P.all(avatars.map(function(avatar) {
+          return workers.delete(avatar.id).then(function() {
+            return db.deleteAvatar(avatar.id);
+          });
+        }));
+      }),
+      db.removeProfile(userId)
+    ]).done(function () {
+        logger.info('delete', { uid: userId });
+        message.del();
+      },
+      function (err) {
+        logger.error('removeUser', err);
+        // The message visibility timeout (in SQS terms) will expire
+        // and be reissued again.
+      });
+  } else {
+    message.del();
+  }
+};
+
+exports.onError = function onError(err) {
+  logger.error('accountEvent', err);
+};
+
+exports.start = function start() {
+  if (!config.events.region || !config.events.queueUrl) {
+    if (env.isProdLike()) {
+      throw new Error('config.events must be included in prod');
+    } else {
+      logger.warn('accountEvent.unconfigured');
+    }
+  } else {
+    var fxaEvents = new Sink(config.events.region, config.events.queueUrl);
+    fxaEvents.on('data', exports.onData);
+    fxaEvents.on('error', exports.onError);
+    fxaEvents.fetch();
+  }
+};
+

lib/img-workers.js

@@ -0,0 +1,64 @@
+/* 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/. */
+
+const buf = require('buf');
+const hex = buf.to.hex;
+const P = require('./promise');
+
+const AppError = require('./error');
+const config = require('./config');
+const logger = require('./logging')('img_workers');
+const request = require('./request');
+
+const WORKER_URL = config.get('worker.url');
+
+exports.upload = function upload(id, payload, headers) {
+  return new P(function(resolve, reject) {
+    var url = WORKER_URL + '/a/' + hex(id);
+    var opts = { headers: headers, json: true };
+    logger.verbose('upload', url);
+    payload.pipe(request.post(url, opts, function(err, res, body) {
+      if (err) {
+        logger.error('upload.network.error', err);
+        reject(AppError.processingError(err));
+        return;
+      }
+
+      if (res.statusCode >= 400 || body.error) {
+        logger.error('upload.worker.error', body);
+        reject(AppError.processingError(body));
+        return;
+      }
+
+      logger.verbose('upload.response', body);
+      resolve(body);
+    }));
+    payload.on('error', function(err) {
+      logger.error('upload.payload.error', err);
+      reject(err);
+    });
+  });
+};
+
+exports.delete = function deleteAvatar(id) {
+  return new P(function(resolve, reject) {
+    var url = WORKER_URL + '/a/' + hex(id);
+    var opts = { method: 'delete', json: true };
+    logger.verbose('delete', url);
+    request(url, opts, function(err, res, body) {
+      if (err) {
+        logger.error('delete.network.error', err);
+        return reject(AppError.processingError(err));
+      }
+      if (res.statusCode >= 400 || body.error) {
+        logger.error('delete.worker.error', body);
+        reject(AppError.processingError(body));
+        return;
+      }
+
+      logger.verbose('delete.worker.response', body);
+      resolve();
+    });
+  });
+};

lib/routes/avatar/delete.js

@@ -7,38 +7,14 @@ const Joi = require('joi');
 const P = require('../../promise');
 
 const AppError = require('../../error');
-const config = require('../../config');
 const db = require('../../db');
 const logger = require('../../logging')('routes.avatar.delete');
-const request = require('../../request');
 const validate = require('../../validate');
+const workers = require('../../img-workers');
 
-const WORKER_URL = config.get('worker.url');
 const EMPTY = Object.create(null);
 const FXA_PROVIDER = 'fxa';
 
-function workerDelete(id) {
-  return new P(function(resolve, reject) {
-    var url = WORKER_URL + '/a/' + id;
-    var opts = { method: 'delete', json: true };
-    logger.verbose('workerDelete', url);
-    request(url, opts, function(err, res, body) {
-      if (err) {
-        logger.error('network.error', err);
-        return reject(AppError.processingError(err));
-      }
-      if (res.statusCode >= 400 || body.error) {
-        logger.error('worker.error', body);
-        reject(AppError.processingError(body));
-        return;
-      }
-
-      logger.verbose('worker', body);
-      resolve();
-    });
-  });
-}
-
 function empty() {
   return EMPTY;
 }
@@ -74,7 +50,7 @@ module.exports = {
       .spread(function(_, provider) {
         logger.debug('provider', provider);
         if (provider.name === FXA_PROVIDER) {
-          return workerDelete(req.params.id);
+          return workers.delete(req.params.id);
         }
       })
       .then(empty)

lib/routes/avatar/upload.js

@@ -5,18 +5,13 @@
 const assert = require('assert');
 
 const Joi = require('joi');
-const P = require('../../promise');
 
-const AppError = require('../../error');
 const config = require('../../config');
 const db = require('../../db');
 const hex = require('buf').to.hex;
 const img = require('../../img');
-const logger = require('../../logging')('routes.avatar.upload');
-const request = require('../../request');
 const validate = require('../../validate');
-
-const WORKER_URL = config.get('worker.url');
+const workers = require('../../img-workers');
 
 const FXA_PROVIDER = 'fxa';
 const FXA_URL_TEMPLATE = config.get('img.url');
@@ -27,35 +22,6 @@ function fxaUrl(id) {
   return FXA_URL_TEMPLATE.replace('{id}', id);
 }
 
-function pipeToWorker(id, payload, headers) {
-  return new P(function(resolve, reject) {
-    var url = WORKER_URL + '/a/' + id;
-    var opts = { headers: headers, json: true };
-    logger.verbose('pipeToWorker', url);
-    payload.pipe(request.post(url, opts, function(err, res, body) {
-      if (err) {
-        logger.error('network.error', err);
-        reject(AppError.processingError(err));
-        return;
-      }
-
-      if (res.statusCode >= 400 || body.error) {
-        logger.error('worker.error', body);
-        reject(AppError.processingError(body));
-        return;
-      }
-
-      logger.verbose('worker', body);
-      resolve(body);
-    }));
-    payload.on('error', function(err) {
-      logger.error('payload', err);
-      reject(err);
-    });
-  });
-}
-
-
 module.exports = {
   auth: {
     strategy: 'oauth',
@@ -83,7 +49,7 @@ module.exports = {
     var id = img.id();
     var url = fxaUrl(id);
     var uid = req.auth.credentials.user;
-    pipeToWorker(id, req.payload, req.headers)
+    workers.upload(id, req.payload, req.headers)
       .then(function save() {
         return db.addAvatar(id, uid, url, FXA_PROVIDER, true);
       })

package.json

@@ -16,10 +16,11 @@
     "checksum": "0.1.1",
     "compute-cluster": "0.0.9",
     "convict": "0.8.0",
+    "fxa-notifier-aws": "^1.0.0",
     "gm": "1.17.0",
     "hapi": "7.5.3",
     "joi": "4.7.0",
-    "mozlog": "1.0.0",
+    "mozlog": "^2.0.0",
     "mysql": "2.5.2",
     "mysql-patcher": "0.7.0",
     "request": "2.47.0",
@@ -43,7 +44,6 @@
     "mocha-text-cov": "^0.1.0",
     "nock": "^0.48.0",
     "pngparse": "2.0.1",
-    "rimraf": "^2.2.8",
     "through": "^2.3.4",
     "time-grunt": "^1.0.0"
   },

test/api.js

@@ -381,10 +381,13 @@ describe('/avatar', function() {
         email: 'user@example.domain',
         scope: ['profile:avatar:write']
       });
-      mock.workerFailure();
-      mock.log('routes.avatar.upload', function(rec) {
+      mock.workerFailure('post');
+      mock.log('img_workers', function(rec) {
         if (rec.levelname === 'ERROR') {
-          assert.equal(rec.message, 'worker.error unexpected server error');
+          assert.equal(
+            rec.message,
+            'upload.worker.error unexpected server error'
+          );
           return true;
         }
       });
@@ -457,8 +460,6 @@ describe('/avatar', function() {
       var s3url;
       var id;
       before(function() {
-        this.slow(2000);
-        this.timeout(3000);
         mock.token({
           user: user,
           email: 'user@example.domain',

test/events.js

@@ -0,0 +1,125 @@
+/* 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/. */
+
+const assert = require('insist');
+const crypto = require('crypto');
+const fs = require('fs');
+const path = require('path');
+
+const db = require('../lib/db');
+const events = require('../lib/events');
+const P = require('../lib/promise');
+
+const UID = crypto.randomBytes(16).toString('hex');
+const mock = require('./lib/mock')({ userid: UID });
+const Server = require('./lib/server');
+const Static = require('./lib/static');
+
+const imagePath = path.join(__dirname, 'lib', 'firefox.png');
+const imageData = fs.readFileSync(imagePath);
+
+const SIZES = require('../lib/img').SIZES;
+const SIZE_SUFFIXES = Object.keys(SIZES).map(function(val) {
+  if (val === 'default') {
+    return '';
+  }
+  return '_' + val;
+});
+
+
+/*global describe,it,beforeEach,afterEach*/
+
+afterEach(function() {
+  mock.done();
+});
+
+
+describe('events', function() {
+  describe('onDeleteMessage', function() {
+    var tok = crypto.randomBytes(32).toString('hex');
+    function Message(type, onDel) {
+      if (typeof type === 'function') {
+        onDel = type;
+        type = 'delete';
+      }
+      return {
+        event: type,
+        uid: UID + '@accounts.firefox.com',
+        del: onDel
+      };
+    }
+
+    describe('avatar', function() {
+      beforeEach(function() {
+        mock.token({
+          user: UID,
+          email: 'user@example.domain',
+          scope: ['profile:avatar:write']
+        });
+        mock.image();
+        return Server.api.post({
+          url: '/avatar/upload',
+          payload: imageData,
+          headers: { authorization: 'Bearer ' + tok,
+            'content-type': 'image/png',
+            'content-length': imageData.length
+          }
+        }).then(function(res) {
+          assert.equal(res.statusCode, 201);
+          assert(res.result.url);
+          assert(res.result.id);
+          return res.result.url;
+        }).then(function(s3url) {
+          return P.all(SIZE_SUFFIXES).map(function(suffix) {
+            return Static.get(s3url + suffix);
+          });
+        }).then(function(responses) {
+          assert.equal(responses.length, SIZE_SUFFIXES.length);
+          responses.forEach(function(res) {
+            assert.equal(res.statusCode, 200);
+          });
+          mock.done();
+        });
+      });
+
+      it('should delete avatars', function(done) {
+        mock.deleteImage();
+        events.onData(new Message(function() {
+          db.getAvatars(UID).then(function(avatars) {
+            assert.equal(avatars.length, 0);
+          }).done(done, done);
+        }));
+      });
+
+      it('should not delete message on error', function(done) {
+        mock.workerFailure('delete');
+        events.onData(new Message(function() {
+          assert(false, 'message.del() should not be called');
+        }));
+        mock.log('events', function(record) {
+          if (record.levelname === 'ERROR' && record.args[0] === 'removeUser') {
+            setTimeout(function() {
+              done();
+            }, 1);
+            return true;
+          }
+          return false;
+        });
+
+      });
+
+    });
+
+    it('should ignore unknown messages', function(done) {
+      db.setDisplayName(UID, 'foo bar').then(function() {
+        events.onData(new Message('baz', function() {
+          db.getDisplayName(UID).then(function(profile) {
+            assert.equal(profile.displayName, 'foo bar');
+          }).done(done, done);
+        }));
+      });
+    });
+
+  });
+});

test/lib/mock.js

@@ -126,19 +126,20 @@ module.exports = function mock(options) {
         .reply(200, JSON.stringify(tok));
     },
 
-    workerFailure: function workerFailure() {
+    workerFailure: function workerFailure(action) {
+      if (action !== 'post' && action !== 'delete') {
+        throw new Error('failure must be post or delete');
+      }
       var parts = url.parse(config.get('worker.url'));
-      var headers = {
+      var headers = action === 'post' ? {
         'content-type': 'image/png',
         'content-length': 12696
-      };
+      } : {};
       return nock(parts.protocol + '//' + parts.host, {
         reqheaders: headers
       })
-        .filteringPath(function filter(_path) {
-          return _path.replace(/\/a\/[0-9a-f]{32}/g, '/a/' + MOCK_ID);
-        })
-        .post('/a/' + MOCK_ID)
+        .filteringPath(/^\/a\/[0-9a-f]{32}$/g, '/a/' + MOCK_ID)
+        [action]('/a/' + MOCK_ID)
         .reply(500, 'unexpected server error');
 
     },