feat: add queue prefix config and fix login validation

Because:

* The queue prefix is not configurable and FxA reuses client ID's in
  the dev/stage environment.
* Login messages don't always include a service property.

This commit:

* Allows the queue prefix to be configured.
* Makes the service property optional for login messages.

Closes #2819

packages/fxa-event-broker/bin/worker.ts

@@ -67,6 +67,7 @@ async function main() {
     capabilityService,
     webhookService,
     pubsub,
+    Config.get('topicPrefix'),
     serviceNotificationQueueUrl,
     new SQS({ region })
   );

packages/fxa-event-broker/bin/workerDev.ts

@@ -139,6 +139,7 @@ async function main() {
     capabilityService,
     webhookService,
     pubsub,
+    Config.get('topicPrefix'),
     Config.get('serviceNotificationQueueUrl'),
     new SQS()
   );

packages/fxa-event-broker/config/index.ts

@@ -181,6 +181,12 @@ const conf = convict({
       'The queue URL to use (should include https://sqs.<region>.amazonaws.com/<account-id>/<queue-name>)',
     env: 'SERVICE_NOTIFICATION_QUEUE_URL',
     format: String
+  },
+  topicPrefix: {
+    default: 'rpQueue-',
+    doc: 'GCP PubSub Queue prefix',
+    env: 'PUBSUB_QUEUE_PREFIX',
+    format: String
   }
 });
 

packages/fxa-event-broker/lib/db/firestore.ts

@@ -1,7 +1,7 @@
 /* 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/. */
-import { DocumentData, Firestore, Settings } from '@google-cloud/firestore';
+import { Firestore, Settings } from '@google-cloud/firestore';
 import { TypedCollectionReference, TypedDocumentReference } from 'typesafe-node-firestore';
 
 import { ClientWebhooks } from '../selfUpdatingService/clientWebhookService';

packages/fxa-event-broker/lib/notificationProcessor.ts

@@ -31,7 +31,6 @@ function unhandledEventType(e: ServiceNotification) {
 
 class ServiceNotificationProcessor {
   public readonly app: Consumer;
-  private readonly topicPrefix = 'rpQueue-';
 
   constructor(
     private readonly logger: Logger,
@@ -40,6 +39,7 @@ class ServiceNotificationProcessor {
     private readonly capabilityService: ClientCapabilityService,
     private readonly webhookService: ClientWebhookService,
     private readonly pubsub: PubSub,
+    private readonly topicPrefix: string,
     queueUrl: string,
     sqs: SQS
   ) {

packages/fxa-event-broker/lib/serviceNotifications.ts

@@ -2,6 +2,7 @@
  * 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/. */
 
+import * as sentry from '@sentry/node';
 import { Logger } from 'mozlog';
 import * as joi from 'typesafe-joi';
 
@@ -32,7 +33,7 @@ const LOGIN_SCHEMA = joi
       .string()
       .valid(LOGIN_EVENT)
       .required(),
-    service: joi.string().required(),
+    service: joi.string().optional(),
     timestamp: joi.number().optional(),
     ts: joi.number().required(),
     uid: joi.string().required(),
@@ -121,7 +122,7 @@ function multiSchemaAttempt(
 }
 
 export const ServiceNotification = {
-  from(logger: Logger, rawMessage: object): ServiceNotification {
+  from(logger: Logger, rawMessage: object): ServiceNotification | undefined {
     try {
       const validMessage = multiSchemaAttempt(eventSchemas, rawMessage);
       if (validMessage) {
@@ -129,6 +130,7 @@ export const ServiceNotification = {
         return validMessage;
       }
     } catch (err) {
+      sentry.captureException(err);
       logger.error('from.sqsMessage', { message: 'Invalid message', err });
     }
   }

packages/fxa-event-broker/test/lib/notificationProcessor.spec.ts

@@ -95,6 +95,7 @@ describe('ServiceNotificationProcessor', () => {
       capabilityService,
       webhookService,
       pubsub,
+      'rpQueue-',
       'https://sqs.eu-west-1.amazonaws.com/account-id/queue-name',
       sqs
     );