feat(event-broker): add remaining RP events

Because: * Only the subscription state and delete events were propagated. * RPs need to know about password reset/change and profile updates. This commit: * Sends password change and profile change events to RPs. Closes #3481
2019-12-05 14:15:50 -08:00 · 2019-12-05 14:15:50 -08:00 · d5c42ccd8b
--- a/packages/fxa-event-broker/README.md
+++ b/packages/fxa-event-broker/README.md
@ -15,6 +15,61 @@ A relying party will get webhook calls for events. These events are encoded in
 [SET][set]s with the following formats. See the [SET RFC][set] for definitions and other
 examples.

+### Password Change
+
+Sent when a user has reset or changed their password. Services receiving this event
+should terminate user login sessions that were established prior to the event.
+
+- Event Identifier
+  - `https://schemas.accounts.firefox.com/event/password-change`
+- Event Payload
+  - [Password Event Identifier]
+    - changeTime
+      - Time when the password reset took place. All logins established before this
+        time should be terminated.
+
+### Example Password Change Event
+
+    {
+     "iss": "https://accounts.firefox.com/",
+     "sub": "FXA_USER_ID",
+     "aud": "REMOTE_SYSTEM",
+     "iat": 1565720808,
+     "jti": "e19ed6c5-4816-4171-aa43-56ffe80dbda1",
+     "events": {
+       "https://schemas.accounts.firefox.com/event/password-change": {
+           "changeTime": 1565721242227
+       }
+     }
+
+### Profile Change
+
+Sent when a user has changed their profile data in some manner. Updates to their
+profile may include a new primary email address, display name, or 2FA status. This
+event does not include what changed and the profile data a service has access to
+may not show any changes if the data changed was outside the OAuth scope the service
+was granted.
+
+Services should update any cached profile data they hold about the user.
+
+- Event Identifier
+  - `https://schemas.accounts.firefox.com/event/profile-change`
+- Event Payload
+  - [Profile Event Identifier]
+    - `{}`
+
+### Example Profile Change Event
+
+    {
+     "iss": "https://accounts.firefox.com/",
+     "sub": "FXA_USER_ID",
+     "aud": "REMOTE_SYSTEM",
+     "iat": 1565720808,
+     "jti": "e19ed6c5-4816-4171-aa43-56ffe80dbda1",
+     "events": {
+       "https://schemas.accounts.firefox.com/event/profile-change": {}
+     }
+
 ### Subscription State Change

 Sent when a user's subscription state has changed to RPs that provide the changed
@ -112,10 +167,12 @@ Where `CAPABILITIES` is a comma-seperated string of capabilities to include.

 #### Usage

-    $ npm run build
-    $ export LOG_FORMAT=pretty
-    $ node dist/bin/simulate-webhook-call.js a9238ba https://example.com/webhook capability_1
-    fxa-event-broker.simulateWebhookCall.INFO: webhookCall {"statusCode":200,"body":"ok\n"}
-    $
+```bash
+$ npm run build
+$ export LOG_FORMAT=pretty
+$ node dist/bin/simulate-webhook-call.js a9238ba https://example.com/webhook capability_1
+fxa-event-broker.simulateWebhookCall.INFO: webhookCall {"statusCode":200,"body":"ok\n"}
+$
+```

 [set]: https://tools.ietf.org/html/rfc8417
--- a/packages/fxa-event-broker/architecture.md
+++ b/packages/fxa-event-broker/architecture.md
@ -40,7 +40,7 @@ sequenceDiagram

 <br /><br /><br /><br /><br />

-## Subscription State Change Events
+## Relying Party Events

 Note: SQS participant not shown here to save space.

@ -53,13 +53,13 @@ sequenceDiagram
    participant RP as Relying Party
    participant FS as Google Firestore

-    Auth->>Ev: SubscriptionEvent via SQS
+    Auth->>Ev: RelyingPartyEvent via SQS
    Ev-->>+FS: Get RPs the User has logged into (FetchClientIds)
    FS-->>-Ev: List of Client Ids

-    loop on clientIds
-    Ev->>PS: StateChangeEvent
+    Ev->>PS: RPEvent

+    loop on clientIds
    PS-->>+Ev: POST /proxy/{clientId}
    Ev-->>+RP: POST /client/webhook
    RP-->>-Ev: {Status: 200}
--- a/packages/fxa-event-broker/lib/api/proxy-controller.ts
+++ b/packages/fxa-event-broker/lib/api/proxy-controller.ts
@ -9,7 +9,14 @@ import requests from 'request-promise-native';

 import { JWT } from '../jwts';
 import { ClientWebhookService } from '../selfUpdatingService/clientWebhookService';
-import { DELETE_EVENT, SUBSCRIPTION_UPDATE_EVENT } from '../serviceNotifications';
+import {
+  DELETE_EVENT,
+  PASSWORD_CHANGE_EVENT,
+  PASSWORD_RESET_EVENT,
+  PRIMARY_EMAIL_EVENT,
+  PROFILE_CHANGE_EVENT,
+  SUBSCRIPTION_UPDATE_EVENT
+} from '../serviceNotifications';
 import { version } from '../version';
 import { proxyPayload } from './proxy-validator';

@ -121,6 +128,21 @@ export default class ProxyController {
          uid: message.uid
        });
      }
+      case PASSWORD_RESET_EVENT:
+      case PASSWORD_CHANGE_EVENT: {
+        return await this.jwt.generatePasswordSET({
+          changeTime: message.changeTime,
+          clientId,
+          uid: message.uid
+        });
+      }
+      case PRIMARY_EMAIL_EVENT:
+      case PROFILE_CHANGE_EVENT: {
+        return await this.jwt.generateProfileSET({
+          clientId,
+          uid: message.uid
+        });
+      }
      default:
        throw Error(`Invalid event: ${message.event}`);
    }
--- a/packages/fxa-event-broker/lib/jwts.ts
+++ b/packages/fxa-event-broker/lib/jwts.ts
@ -6,16 +6,25 @@ import jwtool from 'fxa-jwtool';
 import uuid from 'uuid';

 // SET Event identifiers
+export const DELETE_EVENT_ID = 'https://schemas.accounts.firefox.com/event/delete-user';
+export const PASSWORD_EVENT_ID = 'https://schemas.accounts.firefox.com/event/password-change';
+export const PROFILE_EVENT_ID = 'https://schemas.accounts.firefox.com/event/profile-change';
 export const SUBSCRIPTION_STATE_EVENT_ID =
  'https://schemas.accounts.firefox.com/event/subscription-state-change';

-export const DELETE_USER_EVENT_ID = 'https://schemas.accounts.firefox.com/event/delete-user';
-
 type deleteEvent = {
  clientId: string;
  uid: string;
 };

+type passwordEvent = {
+  uid: string;
+  clientId: string;
+  changeTime: number;
+};
+
+type profileEvent = deleteEvent;
+
 type securityEvent = {
  uid: string;
  clientId: string;
@ -66,6 +75,38 @@ export class JWT {
    return this.tokenKey.sign(claims);
  }

+  /**
+   * Generate a Password Security Event Token.
+   *
+   * @param  passEvent
+   */
+  public generatePasswordSET(passEvent: passwordEvent): Promise<string> {
+    return this.generateSET({
+      clientId: passEvent.clientId,
+      events: {
+        [PASSWORD_EVENT_ID]: {
+          changeTime: passEvent.changeTime
+        }
+      },
+      uid: passEvent.uid
+    });
+  }
+
+  /**
+   * Generate a Profile Security Event Token.
+   *
+   * @param  proEvent
+   */
+  public generateProfileSET(proEvent: profileEvent): Promise<string> {
+    return this.generateSET({
+      clientId: proEvent.clientId,
+      events: {
+        [PROFILE_EVENT_ID]: {}
+      },
+      uid: proEvent.uid
+    });
+  }
+
  /**
   * Generate a Subscription Security Event Token.
   *
@ -94,7 +135,7 @@ export class JWT {
    return this.generateSET({
      clientId: delEvent.clientId,
      events: {
-        [DELETE_USER_EVENT_ID]: {}
+        [DELETE_EVENT_ID]: {}
      },
      uid: delEvent.uid
    });
--- a/packages/fxa-event-broker/lib/notificationProcessor.ts
+++ b/packages/fxa-event-broker/lib/notificationProcessor.ts
@ -17,14 +17,17 @@ import {
  deleteSchema,
  LOGIN_EVENT,
  loginSchema,
+  PASSWORD_CHANGE_EVENT,
+  PASSWORD_RESET_EVENT,
+  passwordSchema,
+  PRIMARY_EMAIL_EVENT,
+  PROFILE_CHANGE_EVENT,
+  profileSchema,
  ServiceNotification,
  SUBSCRIPTION_UPDATE_EVENT,
  subscriptionUpdateSchema
 } from './serviceNotifications';

-// Use never type to force exhaustive switch handling for defined
-// ServiceNotifications.
-function unhandledEventType(e: never): never;
 function unhandledEventType(e: ServiceNotification) {
  throw new Error('Unhandled message event type: ' + e);
 }
@ -97,6 +100,33 @@ class ServiceNotificationProcessor {
    }
  }

+  private async handleProfileEvent(message: profileSchema) {
+    this.metrics.increment('message.type.profile');
+    const clientIds = await this.db.fetchClientIds(message.uid);
+    for (const clientId of clientIds) {
+      const topicName = this.topicPrefix + clientId;
+      const messageId = await this.pubsub
+        .topic(topicName)
+        .publishJSON({ event: message.event, uid: message.uid, timestamp: Date.now() });
+      this.logger.debug('publishedMessage', { topicName, messageId });
+    }
+  }
+
+  private async handlePasswordEvent(message: passwordSchema) {
+    this.metrics.increment('message.type.password');
+    const clientIds = await this.db.fetchClientIds(message.uid);
+    for (const clientId of clientIds) {
+      const topicName = this.topicPrefix + clientId;
+      const messageId = await this.pubsub.topic(topicName).publishJSON({
+        changeTime: message.timestamp ? message.timestamp : message.ts * 1000,
+        event: message.event,
+        timestamp: Date.now(),
+        uid: message.uid
+      });
+      this.logger.debug('publishedMessage', { topicName, messageId });
+    }
+  }
+
  private async handleLoginEvent(message: loginSchema) {
    // Sync and some logins don't emit a clientId, so we have nothing to track
    if (!message.clientId) {
@ -180,6 +210,16 @@ class ServiceNotificationProcessor {
        await this.handleDeleteEvent(message);
        break;
      }
+      case PRIMARY_EMAIL_EVENT:
+      case PROFILE_CHANGE_EVENT: {
+        await this.handleProfileEvent(message);
+        break;
+      }
+      case PASSWORD_CHANGE_EVENT:
+      case PASSWORD_RESET_EVENT: {
+        await this.handlePasswordEvent(message);
+        break;
+      }
      default:
        unhandledEventType(message);
    }
--- a/packages/fxa-event-broker/lib/serviceNotifications.ts
+++ b/packages/fxa-event-broker/lib/serviceNotifications.ts
@ -7,9 +7,13 @@ import { Logger } from 'mozlog';
 import joi from 'typesafe-joi';

 // Event strings
-export const LOGIN_EVENT = 'login';
-export const SUBSCRIPTION_UPDATE_EVENT = 'subscription:update';
 export const DELETE_EVENT = 'delete';
+export const LOGIN_EVENT = 'login';
+export const PASSWORD_CHANGE_EVENT = 'passwordChange';
+export const PASSWORD_RESET_EVENT = 'reset';
+export const PRIMARY_EMAIL_EVENT = 'primaryEmailChanged';
+export const PROFILE_CHANGE_EVENT = 'profileDataChange';
+export const SUBSCRIPTION_UPDATE_EVENT = 'subscription:update';

 // Message schemas
 const BASE_MESSAGE_SCHEMA = joi
@ -84,11 +88,45 @@ const DELETE_SCHEMA = joi
  .unknown(true)
  .required();

-export type loginSchema = joi.Literal<typeof LOGIN_SCHEMA>;
-export type subscriptionUpdateSchema = joi.Literal<typeof SUBSCRIPTION_UPDATE_SCHEMA>;
-export type deleteSchema = joi.Literal<typeof DELETE_SCHEMA>;
+// Whether its a password change or reset, the service must handle it the
+// same so we only pass that the prior password is no longer valid.
+const PASSWORD_CHANGE_SCHEMA = joi
+  .object()
+  .keys({
+    event: joi.string().valid(PASSWORD_CHANGE_EVENT, PASSWORD_RESET_EVENT),
+    timestamp: joi.number().optional(),
+    ts: joi.number().required(),
+    uid: joi.string().required()
+  })
+  .unknown(true)
+  .required();

-export type ServiceNotification = loginSchema | subscriptionUpdateSchema | deleteSchema | undefined;
+// Whether its the primary email, or some other data in the profile changing, the
+// profile has changed and a new request should be made to handle it.
+const PROFILE_CHANGE_SCHEMA = joi
+  .object()
+  .keys({
+    event: joi.string().valid(PRIMARY_EMAIL_EVENT, PROFILE_CHANGE_EVENT),
+    timestamp: joi.number().optional(),
+    ts: joi.number().required(),
+    uid: joi.string().required()
+  })
+  .unknown(true)
+  .required();
+
+export type deleteSchema = joi.Literal<typeof DELETE_SCHEMA>;
+export type loginSchema = joi.Literal<typeof LOGIN_SCHEMA>;
+export type passwordSchema = joi.Literal<typeof PASSWORD_CHANGE_SCHEMA>;
+export type profileSchema = joi.Literal<typeof PROFILE_CHANGE_SCHEMA>;
+export type subscriptionUpdateSchema = joi.Literal<typeof SUBSCRIPTION_UPDATE_SCHEMA>;
+
+export type ServiceNotification =
+  | deleteSchema
+  | loginSchema
+  | passwordSchema
+  | profileSchema
+  | subscriptionUpdateSchema
+  | undefined;

 interface SchemaTable {
  [key: string]: joi.ObjectSchema<any>;
@ -98,7 +136,11 @@ interface SchemaTable {
 const eventSchemas = {
  [LOGIN_EVENT]: LOGIN_SCHEMA,
  [SUBSCRIPTION_UPDATE_EVENT]: SUBSCRIPTION_UPDATE_SCHEMA,
-  [DELETE_EVENT]: DELETE_SCHEMA
+  [DELETE_EVENT]: DELETE_SCHEMA,
+  [PROFILE_CHANGE_EVENT]: PROFILE_CHANGE_SCHEMA,
+  [PRIMARY_EMAIL_EVENT]: PROFILE_CHANGE_SCHEMA,
+  [PASSWORD_CHANGE_EVENT]: PASSWORD_CHANGE_SCHEMA,
+  [PASSWORD_RESET_EVENT]: PASSWORD_CHANGE_SCHEMA
 };

 /**
--- a/packages/fxa-event-broker/package-lock.json
+++ b/packages/fxa-event-broker/package-lock.json
@ -68,28 +68,28 @@
      }
    },
    "@google-cloud/paginator": {
-      "version": "2.0.1",
-      "resolved": "https://registry.npmjs.org/@google-cloud/paginator/-/paginator-2.0.1.tgz",
-      "integrity": "sha512-HZ6UTGY/gHGNriD7OCikYWL/Eu0sTEur2qqse2w6OVsz+57se3nTkqH14JIPxtf0vlEJ8IJN5w3BdZ22pjCB8g==",
+      "version": "2.0.2",
+      "resolved": "https://registry.npmjs.org/@google-cloud/paginator/-/paginator-2.0.2.tgz",
+      "integrity": "sha512-PCddVtZWvw0iZ3BLIsCXMBQvxUcS9O5CgfHBu8Zd8T3DCiML+oQED1odsbl3CQ9d3RrvBaj+eIh7Dv12D15PbA==",
      "requires": {
        "arrify": "^2.0.0",
        "extend": "^3.0.2"
      }
    },
    "@google-cloud/precise-date": {
-      "version": "1.0.1",
-      "resolved": "https://registry.npmjs.org/@google-cloud/precise-date/-/precise-date-1.0.1.tgz",
-      "integrity": "sha512-nmH/UG87qUHc4OH7cwxVUNBd+5jGaNr6muUAbF0meauqborh/5qUGiz4AVmin6SBnFUazndldDbozU2zpWTfSw=="
+      "version": "1.0.2",
+      "resolved": "https://registry.npmjs.org/@google-cloud/precise-date/-/precise-date-1.0.2.tgz",
+      "integrity": "sha512-EXk9MYAoKz3hD0ITklFIe4aPK+tHk/3WL2DvTD28wAPpYiIMClXTUqX9fanhdurhO1KUU06HBoU3+Rks32yTTQ=="
    },
    "@google-cloud/projectify": {
-      "version": "1.0.1",
-      "resolved": "https://registry.npmjs.org/@google-cloud/projectify/-/projectify-1.0.1.tgz",
-      "integrity": "sha512-xknDOmsMgOYHksKc1GPbwDLsdej8aRNIA17SlSZgQdyrcC0lx0OGo4VZgYfwoEU1YS8oUxF9Y+6EzDOb0eB7Xg=="
+      "version": "1.0.2",
+      "resolved": "https://registry.npmjs.org/@google-cloud/projectify/-/projectify-1.0.2.tgz",
+      "integrity": "sha512-WnkGxvk4U1kAJpoS/Ehk+3MZXVW+XHHhwc/QyD6G8Za4xml3Fv+NRn/bYffl1TxSg+gE0N0mj9Shgc7e8+fl8A=="
    },
    "@google-cloud/promisify": {
-      "version": "1.0.2",
-      "resolved": "https://registry.npmjs.org/@google-cloud/promisify/-/promisify-1.0.2.tgz",
-      "integrity": "sha512-7WfV4R/3YV5T30WRZW0lqmvZy9hE2/p9MvpI34WuKa2Wz62mLu5XplGTFEMK6uTbJCLWUxTcZ4J4IyClKucE5g=="
+      "version": "1.0.3",
+      "resolved": "https://registry.npmjs.org/@google-cloud/promisify/-/promisify-1.0.3.tgz",
+      "integrity": "sha512-Rufgfl3TnkIil3CjsH33Q6093zeoVqyqCdvtvgHuCqRJxCZYfaVPIyr8JViMeLTD4Ja630pRKKZVSjKggoVbNg=="
    },
    "@google-cloud/pubsub": {
      "version": "1.1.5",
@ -1679,9 +1679,9 @@
      }
    },
    "deep-equal": {
-      "version": "1.1.0",
-      "resolved": "https://registry.npmjs.org/deep-equal/-/deep-equal-1.1.0.tgz",
-      "integrity": "sha512-ZbfWJq/wN1Z273o7mUSjILYqehAktR2NVoSrOukDkU9kg2v/Uv89yU4Cvz8seJeAmtN5oqiefKq8FPuXOboqLw==",
+      "version": "1.1.1",
+      "resolved": "https://registry.npmjs.org/deep-equal/-/deep-equal-1.1.1.tgz",
+      "integrity": "sha512-yd9c5AdiqVcR+JjcwUQb9DkhJc8ngNr0MahEBGvDiJw8puWab2yZlh+nkasOnZP+EGTAP6rRp2JzJhJZzvNF8g==",
      "requires": {
        "is-arguments": "^1.0.4",
        "is-date-object": "^1.0.1",
--- a/packages/fxa-event-broker/test/lib/api/proxy-controller.spec.ts
+++ b/packages/fxa-event-broker/test/lib/api/proxy-controller.spec.ts
@ -14,10 +14,20 @@ import sinon from 'sinon';
 import { stubInterface } from 'ts-sinon';

 import Config from '../../../config';
-import { DELETE_USER_EVENT_ID, SUBSCRIPTION_STATE_EVENT_ID } from '../../../lib/jwts';
+import {
+  DELETE_EVENT_ID,
+  PASSWORD_EVENT_ID,
+  PROFILE_EVENT_ID,
+  SUBSCRIPTION_STATE_EVENT_ID
+} from '../../../lib/jwts';
 import * as proxyserver from '../../../lib/proxy-server';
 import { ClientWebhookService } from '../../../lib/selfUpdatingService/clientWebhookService';
-import { DELETE_EVENT, SUBSCRIPTION_UPDATE_EVENT } from '../../../lib/serviceNotifications';
+import {
+  DELETE_EVENT,
+  PASSWORD_CHANGE_EVENT,
+  PROFILE_CHANGE_EVENT,
+  SUBSCRIPTION_UPDATE_EVENT
+} from '../../../lib/serviceNotifications';

 const TEST_KEY = {
  d:
@ -47,6 +57,7 @@ const TEST_PUBLIC_KEY = {
 };
 const TEST_CLIENT_ID = 'abc1234';
 const PUBLIC_JWT = jwtool.JWK.fromObject(TEST_PUBLIC_KEY);
+const CHANGE_TIME = Date.now();

 describe('Proxy Controller', () => {
  let logger: Logger;
@ -86,6 +97,25 @@ describe('Proxy Controller', () => {
    ).toString('base64');
  };

+  const createValidProfileMessage = (): string => {
+    return Buffer.from(
+      JSON.stringify({
+        event: PROFILE_CHANGE_EVENT,
+        uid: 'uid1234'
+      })
+    ).toString('base64');
+  };
+
+  const createValidPasswordMessage = (): string => {
+    return Buffer.from(
+      JSON.stringify({
+        changeTime: CHANGE_TIME,
+        event: PASSWORD_CHANGE_EVENT,
+        uid: 'uid1234'
+      })
+    ).toString('base64');
+  };
+
  beforeEach(async () => {
    logger = stubInterface<Logger>();
    metrics = new StatsD({ mock: true });
@ -160,26 +190,40 @@ describe('Proxy Controller', () => {
    cassert.equal(payload.events[SUBSCRIPTION_STATE_EVENT_ID].isActive, true);
  });

-  it('notifies successfully on delete', async () => {
-    mockWebhook();
-    const message = createValidDeleteMessage();
+  describe('handles common RP events: ', () => {
+    const eventTypes: { [key: string]: [string, () => string] } = {
+      delete: [DELETE_EVENT_ID, createValidDeleteMessage],
+      password: [PASSWORD_EVENT_ID, createValidPasswordMessage],
+      profile: [PROFILE_EVENT_ID, createValidProfileMessage]
+    };
+    for (const [key, value] of Object.entries(eventTypes)) {
+      const [eventId, creatFunc] = value;
+      it(`notifies successfully on ${key}`, async () => {
+        mockWebhook();
+        const message = creatFunc();

-    const result = await server.inject({
-      method: 'POST',
-      payload: JSON.stringify({
-        message: { data: message, messageId: 'test-message' },
-        subscription: 'test-sub'
-      }),
-      url: '/v1/proxy/abc1234'
-    });
-    const bearer = result.headers['x-auth'] as string;
-    const token = (bearer.match(/Bearer (.*)/) as string[])[1];
+        const result = await server.inject({
+          method: 'POST',
+          payload: JSON.stringify({
+            message: { data: message, messageId: 'test-message' },
+            subscription: 'test-sub'
+          }),
+          url: '/v1/proxy/abc1234'
+        });
+        const bearer = result.headers['x-auth'] as string;
+        const token = (bearer.match(/Bearer (.*)/) as string[])[1];

-    const payload = await PUBLIC_JWT.verify(token);
-    cassert.equal(payload.aud, 'abc1234');
-    cassert.equal(payload.sub, 'uid1234');
-    cassert.equal(payload.iss, 'testing');
-    cassert.deepEqual(payload.events[DELETE_USER_EVENT_ID], {});
+        const payload = await PUBLIC_JWT.verify(token);
+        cassert.equal(payload.aud, 'abc1234');
+        cassert.equal(payload.sub, 'uid1234');
+        cassert.equal(payload.iss, 'testing');
+        if (key === 'password') {
+          cassert.deepEqual(payload.events[PASSWORD_EVENT_ID], { changeTime: CHANGE_TIME });
+        } else {
+          cassert.deepEqual(payload.events[eventId], {});
+        }
+      });
+    }
  });

  it('logs an error on invalid message payloads', async () => {
--- a/packages/fxa-event-broker/test/lib/notificationProcessor.spec.ts
+++ b/packages/fxa-event-broker/test/lib/notificationProcessor.spec.ts
@ -62,6 +62,26 @@ const baseDeleteMessage = {
  event: 'delete'
 };

+const basePasswordResetMessage = {
+  ...baseMessage,
+  event: 'reset'
+};
+
+const basePasswordChangeMessage = {
+  ...baseMessage,
+  event: 'passwordChange'
+};
+
+const basePrimaryEmailMessage = {
+  ...baseMessage,
+  event: 'primaryEmailChanged'
+};
+
+const baseProfileMessage = {
+  ...baseMessage,
+  event: 'profileDataChange'
+};
+
 describe('ServiceNotificationProcessor', () => {
  let sqs: any;
  let db: Datastore;
@ -137,51 +157,47 @@ describe('ServiceNotificationProcessor', () => {
    assert.calledWith(db.storeLogin as SinonSpy, baseLoginMessage.uid, baseLoginMessage.clientId);
  });

-  it('fetches on valid legacy subscription message', async () => {
-    updateStubMessage(baseSubscriptionUpdateLegacyMessage);
-    consumer.start();
-    await pEvent(consumer.app, 'message_processed');
-    consumer.stop();
-    assert.calledWith(db.fetchClientIds as SinonSpy);
-  });
+  const fetchOnValidMessage = {
+    'delete message': baseDeleteMessage,
+    'legacy subscription message': baseSubscriptionUpdateLegacyMessage,
+    'password change message': basePasswordChangeMessage,
+    'password reset message': basePasswordResetMessage,
+    'primary email message': basePrimaryEmailMessage,
+    'profile change message': baseProfileMessage,
+    'subscription message': baseSubscriptionUpdateMessage
+  };

-  it('fetches on valid subscription message', async () => {
-    updateStubMessage(baseSubscriptionUpdateMessage);
-    consumer.start();
-    await pEvent(consumer.app, 'message_processed');
-    consumer.stop();
-    assert.calledWith(db.fetchClientIds as SinonSpy);
-  });
+  // Ensure that all our message types can be handled without error.
+  for (const [key, value] of Object.entries(fetchOnValidMessage)) {
+    it(`fetches on valid ${key}`, async () => {
+      updateStubMessage(value);
+      consumer.start();
+      await pEvent(consumer.app, 'message_processed');
+      consumer.stop();
+      assert.calledWith(db.fetchClientIds as SinonSpy);
+    });
+  }

-  it('fetches on valid delete message', async () => {
-    updateStubMessage(baseDeleteMessage);
-    consumer.start();
-    await pEvent(consumer.app, 'message_processed');
-    consumer.stop();
-    assert.calledWith(db.fetchClientIds as SinonSpy);
-  });
-
-  it('logs an error on invalid login message', async () => {
-    updateStubMessage(Object.assign({}, { ...baseLoginMessage, email: false }));
-    consumer.start();
-    await pEvent(consumer.app, 'message_processed');
-    consumer.stop();
-    assert.calledOnce(logger.error as SinonSpy);
-    const callArgs = (logger.error as SinonSpy).getCalls()[0].args;
-    cassert.equal(callArgs[0], 'from.sqsMessage');
-    cassert.equal(callArgs[1].message, 'Invalid message');
-  });
-
-  it('logs an error on invalid subscription message', async () => {
-    updateStubMessage(Object.assign({}, { ...baseSubscriptionUpdateMessage, productId: false }));
-    consumer.start();
-    await pEvent(consumer.app, 'message_processed');
-    consumer.stop();
-    assert.calledOnce(logger.error as SinonSpy);
-    assert.calledWithMatch(logger.error as SinonSpy, 'from.sqsMessage');
-    const callArgs = (logger.error as SinonSpy).getCalls()[0].args;
-    cassert.equal(callArgs[1].message, 'Invalid message');
-  });
+  const invalidMessages = {
+    login: { ...baseLoginMessage, ts: false },
+    'password change': { ...basePasswordChangeMessage, ts: false },
+    'password reset': { ...basePasswordResetMessage, ts: false },
+    'primary email change': { ...basePrimaryEmailMessage, ts: false },
+    'profile change': { ...baseProfileMessage, ts: false },
+    subscription: { ...baseSubscriptionUpdateMessage, productId: false }
+  };
+  for (const [key, value] of Object.entries(invalidMessages)) {
+    it(`logs an error on invalid ${key} message`, async () => {
+      updateStubMessage(value);
+      consumer.start();
+      await pEvent(consumer.app, 'message_processed');
+      consumer.stop();
+      assert.calledOnce(logger.error as SinonSpy);
+      const callArgs = (logger.error as SinonSpy).getCalls()[0].args;
+      cassert.equal(callArgs[0], 'from.sqsMessage');
+      cassert.equal(callArgs[1].message, 'Invalid message');
+    });
+  }

  it('logs on message its not interested in', async () => {
    updateStubMessage(Object.assign({}, { ...baseLoginMessage, event: 'logout' }));