feat(email): Add flow events for email delivery notifications (#1626), r=@philbooth

Adds support for handling and processing `flowEvents` for email delivery.

bin/email_notifications.js

@@ -7,7 +7,8 @@ var log = require('../lib/log')(config.log.level, 'fxa-email-bouncer')
 var error = require('../lib/error')
 var Token = require('../lib/tokens')(log, config)
 var SQSReceiver = require('../lib/sqs')(log)
-var bounces = require('../lib/bounces')(log, error)
+var bounces = require('../lib/email/bounces')(log, error)
+var delivery = require('../lib/email/delivery')(log)
 
 var DB = require('../lib/db')(
   config,
@@ -20,14 +21,19 @@ var DB = require('../lib/db')(
   Token.PasswordChangeToken
 )
 
-var bounceQueue = new SQSReceiver(config.bounces.region, [
-  config.bounces.bounceQueueUrl,
-  config.bounces.complaintQueueUrl
+var bounceQueue = new SQSReceiver(config.emailNotifications.region, [
+  config.emailNotifications.bounceQueueUrl,
+  config.emailNotifications.complaintQueueUrl
+])
+
+var deliveryQueue = new SQSReceiver(config.emailNotifications.region, [
+  config.emailNotifications.deliveryQueueUrl
 ])
 
 DB.connect(config[config.db.backend])
   .done(
     function (db) {
       bounces(bounceQueue, db)
+      delivery(deliveryQueue)
     }
   )

config/index.js

@@ -306,7 +306,7 @@ var conf = convict({
     env: 'SNS_TOPIC_ARN',
     default: ''
   },
-  bounces: {
+  emailNotifications: {
     region: {
       doc: 'The region where the queues live, most likely the same region we are sending email e.g. us-east-1, us-west-2',
       format: String,
@@ -324,6 +324,12 @@ var conf = convict({
       format: String,
       env: 'COMPLAINT_QUEUE_URL',
       default: ''
+    },
+    deliveryQueueUrl: {
+      doc: 'The email delivery queue URL to use (should include https://sqs.<region>.amazonaws.com/<account-id>/<queue-name>)',
+      format: String,
+      env: 'DELIVERY_QUEUE_URL',
+      default: ''
     }
   },
   verificationReminders: {

lib/email/bounces.js

@@ -3,7 +3,8 @@
  * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
 
 var eaddrs = require('email-addresses')
-var P = require('./promise')
+var P = require('./../promise')
+var utils = require('./utils/helpers')
 
 module.exports = function (log, error) {
 
@@ -49,26 +50,7 @@ module.exports = function (log, error) {
       }
     }
 
-    function getHeaderValue(headerName, message){
-      var value = ''
-      if (message.mail && message.mail.headers) {
-        message.mail.headers.some(function (header) {
-          if (header.name === headerName) {
-            value = header.value
-            return true
-          }
-
-          return false
-        })
-      }
-
-      return value
-    }
-
-
     function handleBounce(message) {
-      const currentTime = Date.now()
-
       var recipients = []
       if (message.bounce && message.bounce.bounceType === 'Permanent') {
         recipients = message.bounce.bouncedRecipients
@@ -81,7 +63,7 @@ module.exports = function (log, error) {
       // Headers are stored as an array of name/value pairs.
       // Log the `X-Template-Name` header to help track the email template that bounced.
       // Ref: http://docs.aws.amazon.com/ses/latest/DeveloperGuide/notification-contents.html
-      var templateName = getHeaderValue('X-Template-Name', message)
+      var templateName = utils.getHeaderValue('X-Template-Name', message)
 
       return P.each(recipients, function (recipient) {
 
@@ -123,27 +105,8 @@ module.exports = function (log, error) {
           }
         }
 
-        // Log flow metrics if `flowId` and `flowBeginTime` specified in headers
-        const flowId = getHeaderValue('X-Flow-Id', message)
-        const flowBeginTime = getHeaderValue('X-Flow-Begin-Time', message)
-        const elapsedTime = currentTime - flowBeginTime
-
-        if (flowId && flowBeginTime && (elapsedTime > 0)) {
-          const eventName = `email.${templateName}.bounced`
-
-          // Flow events have a specific event and structure that must be emitted.
-          // Ref `gather` in https://github.com/mozilla/fxa-auth-server/blob/master/lib/metrics/context.js
-          const flowEventInfo = {
-            event: eventName,
-            time: currentTime,
-            flow_id: flowId,
-            flow_time: elapsedTime
-          }
-
-          log.flowEvent(flowEventInfo)
-        } else {
-          log.error({ op: 'handleBounce.flowEvent', templateName, flowId, flowBeginTime, currentTime })
-        }
+        // Log the bounced flowEvent metrics if available
+        utils.logFlowEventFromMessage(log, message, 'bounced')
 
         log.info(logData)
         log.increment('account.email_bounced')

lib/email/delivery.js

@@ -0,0 +1,60 @@
+/* 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/. */
+
+var P = require('./../promise')
+var utils = require('./utils/helpers')
+
+module.exports = function (log) {
+
+  return function start(deliveryQueue) {
+
+    function handleDelivery(message) {
+
+      var recipients = []
+      if (message.delivery && message.notificationType === 'Delivery') {
+        recipients = message.delivery.recipients
+      }
+
+      // SES can now send custom headers if enabled on topic.
+      // Headers are stored as an array of name/value pairs.
+      // Log the `X-Template-Name` header to help track the email template that delivered.
+      // Ref: http://docs.aws.amazon.com/ses/latest/DeveloperGuide/notification-contents.html
+      const templateName = utils.getHeaderValue('X-Template-Name', message)
+
+      return P.each(recipients, function (recipient) {
+
+        var email = recipient
+        var logData = {
+          op: 'handleDelivery',
+          email: email,
+          processingTimeMillis: message.delivery.processingTimeMillis
+        }
+
+        // Template name corresponds directly with the email template that was used
+        if (templateName) {
+          logData.template = templateName
+        }
+
+        // Log the delivery flowEvent metrics if available
+        utils.logFlowEventFromMessage(log, message, 'delivered')
+
+        log.info(logData)
+        log.increment('account.email_delivered')
+      }).then(
+        function () {
+          // We always delete the message, even if handling some addrs failed.
+          message.del()
+        }
+      )
+    }
+
+    deliveryQueue.on('data', handleDelivery)
+    deliveryQueue.start()
+
+    return {
+      deliveryQueue: deliveryQueue,
+      handleDelivery: handleDelivery
+    }
+  }
+}

lib/email/utils/helpers.js

@@ -0,0 +1,51 @@
+/* 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/. */
+
+function getHeaderValue(headerName, message){
+  var value = ''
+  if (message.mail && message.mail.headers) {
+    message.mail.headers.some(function (header) {
+      if (header.name === headerName) {
+        value = header.value
+        return true
+      }
+
+      return false
+    })
+  }
+
+  return value
+}
+
+function logFlowEventFromMessage(log, message, type) {
+  const currentTime = Date.now()
+  var templateName = getHeaderValue('X-Template-Name', message)
+
+  // Log flow metrics if `flowId` and `flowBeginTime` specified in headers
+  const flowId = getHeaderValue('X-Flow-Id', message)
+  const flowBeginTime = getHeaderValue('X-Flow-Begin-Time', message)
+  const elapsedTime = currentTime - flowBeginTime
+
+  if (flowId && flowBeginTime && (elapsedTime > 0) && type && templateName) {
+    const eventName = `email.${templateName}.${type}`
+
+    // Flow events have a specific event and structure that must be emitted.
+    // Ref `gather` in https://github.com/mozilla/fxa-auth-server/blob/master/lib/metrics/context.js
+    const flowEventInfo = {
+      event: eventName,
+      time: currentTime,
+      flow_id: flowId,
+      flow_time: elapsedTime
+    }
+
+    log.flowEvent(flowEventInfo)
+  } else {
+    log.error({ op: 'handleBounce.flowEvent', templateName, type, flowId, flowBeginTime, currentTime})
+  }
+}
+
+module.exports = {
+  logFlowEventFromMessage: logFlowEventFromMessage,
+  getHeaderValue: getHeaderValue
+}

test/local/bounce_tests.js

@@ -9,7 +9,7 @@ var sinon = require('sinon')
 var spyLog = require('../mocks').spyLog
 var error = require('../../lib/error')
 var P = require('../../lib/promise')
-var bounces = require('../../lib/bounces')
+var bounces = require('../../lib/email/bounces')
 
 var mockBounceQueue = new EventEmitter()
 mockBounceQueue.start = function start() {}

test/local/delivery_tests.js

@@ -0,0 +1,115 @@
+/* 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 EventEmitter = require('events').EventEmitter
+const sinon = require('sinon')
+const spyLog = require('../mocks').spyLog
+const delivery = require('../../lib/email/delivery')
+
+const mockDeliveryQueue = new EventEmitter()
+mockDeliveryQueue.start = function start() {
+}
+
+function mockMessage(msg) {
+  msg.del = sinon.spy()
+  return msg
+}
+
+function mockedDelivery(log) {
+  return delivery(log)(mockDeliveryQueue)
+}
+
+describe('delivery messages', () => {
+  it(
+    'should ignore unknown message types',
+    () => {
+      const mockLog = spyLog()
+      return mockedDelivery(mockLog).handleDelivery(mockMessage({
+        junk: 'message'
+      })).then(function () {
+        assert.equal(mockLog.messages.length, 0)
+      })
+    }
+  )
+
+  it(
+    'should log delivery',
+    () => {
+      const mockLog = spyLog()
+      const mockMsg = mockMessage({
+        notificationType: 'Delivery',
+        delivery: {
+          timestamp: '2016-01-27T14:59:38.237Z',
+          recipients: ['jane@example.com'],
+          processingTimeMillis: 546,
+          reportingMTA: 'a8-70.smtp-out.amazonses.com',
+          smtpResponse: '250 ok:  Message 64111812 accepted',
+          remoteMtaIp: '127.0.2.0'
+        },
+        mail: {
+          headers: [
+            {
+              name: 'X-Template-Name',
+              value: 'verifyLoginEmail'
+            }
+          ]
+        }
+      })
+
+      return mockedDelivery(mockLog).handleDelivery(mockMsg).then(function () {
+        assert.equal(mockLog.messages.length, 3)
+        assert.equal(mockLog.messages[1].args[0]['email'], 'jane@example.com')
+        assert.equal(mockLog.messages[1].args[0]['op'], 'handleDelivery')
+        assert.equal(mockLog.messages[1].args[0]['template'], 'verifyLoginEmail')
+        assert.equal(mockLog.messages[1].args[0]['processingTimeMillis'], 546)
+        assert.equal(mockLog.messages[2].args[0], 'account.email_delivered')
+        assert.equal(mockLog.messages[2].level, 'increment')
+      })
+    }
+  )
+
+  it(
+    'should emit flow metrics',
+    () => {
+      const mockLog = spyLog()
+      const mockMsg = mockMessage({
+        notificationType: 'Delivery',
+        delivery: {
+          timestamp: '2016-01-27T14:59:38.237Z',
+          recipients: ['jane@example.com'],
+          processingTimeMillis: 546,
+          reportingMTA: 'a8-70.smtp-out.amazonses.com',
+          smtpResponse: '250 ok:  Message 64111812 accepted',
+          remoteMtaIp: '127.0.2.0'
+        },
+        mail: {
+          headers: [
+            {
+              name: 'X-Template-Name',
+              value: 'verifyLoginEmail'
+            },
+            {
+              name: 'X-Flow-Id',
+              value: 'someFlowId'
+            },
+            {
+              name: 'X-Flow-Begin-Time',
+              value: '1234'
+            }
+          ]
+        }
+      })
+
+      return mockedDelivery(mockLog).handleDelivery(mockMsg).then(function () {
+        assert.equal(mockLog.messages.length, 3)
+        assert.equal(mockLog.messages[0].args[0]['event'], 'email.verifyLoginEmail.delivered')
+        assert.equal(mockLog.messages[0].args[0]['flow_id'], 'someFlowId')
+        assert.equal(mockLog.messages[0].args[0]['flow_time'] > 0, true)
+        assert.equal(mockLog.messages[0].args[0]['time'] > 0, true)
+      })
+    }
+  )
+})