Merge pull request #20245 from mozilla/FXA-13287

fix(event-broker): Use generation for event changeTime

Author: nshirley

packages/fxa-event-broker/src/queueworker/queueworker.service.spec.ts

@@ -59,11 +59,13 @@ const baseDeleteMessage = {
 const basePasswordResetMessage = {
   ...baseMessage,
   event: 'reset',
+  generation: now - 5000,
 };
 
 const basePasswordChangeMessage = {
   ...baseMessage,
   event: 'passwordChange',
+  generation: now - 5000,
 };
 
 const basePrimaryEmailMessage = {
@@ -313,6 +315,32 @@ describe('QueueworkerService', () => {
       });
     }
 
+    it('uses generation as changeTime for password change events', async () => {
+      const msg = updateStubMessage(basePasswordChangeMessage);
+      await (service as any).handleMessage(msg);
+      expect(topic.publishMessage).toHaveBeenCalledTimes(1);
+      const published = topic.publishMessage.mock.calls[0][0].json;
+      expect(published.changeTime).toBe(basePasswordChangeMessage.generation);
+    });
+
+    it('uses generation as changeTime for password reset events', async () => {
+      const msg = updateStubMessage(basePasswordResetMessage);
+      await (service as any).handleMessage(msg);
+      expect(topic.publishMessage).toHaveBeenCalledTimes(1);
+      const published = topic.publishMessage.mock.calls[0][0].json;
+      expect(published.changeTime).toBe(basePasswordResetMessage.generation);
+    });
+
+    it('falls back to timestamp when generation is absent for password events', async () => {
+      const { generation, ...messageWithoutGeneration } =
+        basePasswordChangeMessage;
+      const msg = updateStubMessage(messageWithoutGeneration);
+      await (service as any).handleMessage(msg);
+      expect(topic.publishMessage).toHaveBeenCalledTimes(1);
+      const published = topic.publishMessage.mock.calls[0][0].json;
+      expect(published.changeTime).toBe(messageWithoutGeneration.timestamp);
+    });
+
     const invalidMessages = {
       login: { ...baseLoginMessage, clientId: 'test1234' },
       'password change': { ...basePasswordChangeMessage, ts: false },

packages/fxa-event-broker/src/queueworker/queueworker.service.ts

@@ -192,9 +192,16 @@ export class QueueworkerService
     const clientIds = await this.clientIdsForUserAndResourceServers(
       message.uid
     );
+
+    const changeTime =
+      (eventType === 'password' &&
+        (message as dto.passwordSchema).generation) ||
+      message.timestamp ||
+      message.ts * 1000;
+
     for (const clientId of clientIds) {
       const messageId = await this.publishMessage(clientId, {
-        changeTime: message.timestamp ? message.timestamp : message.ts * 1000,
+        changeTime,
         event: message.event,
         timestamp: Date.now(),
         uid: message.uid,

packages/fxa-event-broker/src/queueworker/sqs.dto.ts

@@ -71,6 +71,7 @@ export const PASSWORD_CHANGE_SCHEMA = joi
   .object()
   .keys({
     event: joi.string().valid(PASSWORD_CHANGE_EVENT, PASSWORD_RESET_EVENT),
+    generation: joi.number().optional(),
     timestamp: joi.number().optional(),
     ts: joi.number().required(),
     uid: joi.string().required(),
@@ -118,6 +119,7 @@ export type loginSchema = {
 
 export type passwordSchema = {
   event: typeof PASSWORD_CHANGE_EVENT | typeof PASSWORD_RESET_EVENT;
+  generation?: number;
   timestamp?: number;
   ts: number;
   uid: string;