✨ add message ordering to contact streaming messages

Author: vogelskamp

src/models/controller.model.ts

@@ -50,6 +50,8 @@ export class Controller {
   private pubSubIntegrationEventsClient: PubSubClient<PubSubIntegrationsEventMessage> | null =
     null;
   private integrationName: string = 'UNKNOWN';
+
+  // used for garbage collection reasons, to prevent long running promises from getting canceled
   private streamingPromises = new Map<string, Promise<void>>();
 
   constructor(adapter: Adapter, contactCache: ContactCache | null) {
@@ -288,10 +290,11 @@ export class Controller {
       }
 
       const timestamp = Date.now();
+      const orderingKey = `${userId}:${timestamp}`;
 
       infoLogger(
         'streamContacts',
-        `Starting contact streaming ${timestamp}`,
+        `Starting contact streaming ${timestamp} - orderingKey ${orderingKey}`,
         providerConfig.apiKey,
       );
 
@@ -303,7 +306,6 @@ export class Controller {
         const iterator = this.adapter.streamContacts(providerConfig);
 
         let result = await iterator.next();
-
         while (!result.done) {
           const { value: contacts } = result;
 
@@ -312,18 +314,19 @@ export class Controller {
               throw new Error('Invalid contacts received');
             }
 
-            const message: PubSubContactsMessage = {
-              userId,
-              timestamp,
-              contacts: contacts.map((contact) =>
-                sanitizeContact(contact, providerConfig.locale),
-              ),
-              state: PubSubContactsState.IN_PROGRESS,
-              integrationName: this.integrationName,
-              // traceparent: tracer.getTraceParent(),
-            };
-
-            await this.pubSubContactStreamingClient?.publishMessage(message);
+            await this.pubSubContactStreamingClient?.publishMessage(
+              {
+                userId,
+                timestamp,
+                contacts: contacts.map((contact) =>
+                  sanitizeContact(contact, providerConfig.locale),
+                ),
+                state: PubSubContactsState.IN_PROGRESS,
+                integrationName: this.integrationName,
+                // traceparent: tracer.getTraceParent(),
+              },
+              orderingKey,
+            );
           } catch (error) {
             errorLogger(
               'streamContacts',
@@ -339,14 +342,17 @@ export class Controller {
 
       const streamingPromise = streamContacts()
         .then(() => {
-          return this.pubSubContactStreamingClient?.publishMessage({
-            userId: providerConfig.userId,
-            timestamp,
-            contacts: [],
-            state: PubSubContactsState.COMPLETE,
-            integrationName: this.integrationName,
-            // traceparent: tracer.getTraceParent(),
-          });
+          return this.pubSubContactStreamingClient?.publishMessage(
+            {
+              userId: providerConfig.userId,
+              timestamp,
+              contacts: [],
+              state: PubSubContactsState.COMPLETE,
+              integrationName: this.integrationName,
+              // traceparent: tracer.getTraceParent(),
+            },
+            orderingKey,
+          );
         })
         .catch(async (error) => {
           errorLogger(
@@ -355,14 +361,17 @@ export class Controller {
             providerConfig.apiKey,
             error,
           );
-          return this.pubSubContactStreamingClient?.publishMessage({
-            userId: providerConfig.userId,
-            timestamp,
-            contacts: [],
-            state: PubSubContactsState.FAILED,
-            integrationName: this.integrationName,
-            // traceparent: tracer.getTraceParent(),
-          });
+          return this.pubSubContactStreamingClient?.publishMessage(
+            {
+              userId: providerConfig.userId,
+              timestamp,
+              contacts: [],
+              state: PubSubContactsState.FAILED,
+              integrationName: this.integrationName,
+              // traceparent: tracer.getTraceParent(),
+            },
+            orderingKey,
+          );
         })
         .catch((error) => {
           errorLogger(

src/models/pubsub/pubsub-client.model.ts

@@ -1,6 +1,6 @@
 import { PubSub } from '@google-cloud/pubsub';
-import { timeout } from '../../util/timeout';
 import { context, propagation } from '@opentelemetry/api';
+import { timeout } from '../../util/timeout';
 
 const PUBLISH_TIMEOUT = 10_000;
 
@@ -13,7 +13,7 @@ export class PubSubClient<T> {
     this.topicName = topicName;
   }
 
-  async publishMessage(message: T) {
+  async publishMessage(message: T, orderingKey?: string) {
     if (!this.topicName) {
       throw new Error('No pubsub topic name provided.');
     }
@@ -22,10 +22,12 @@ export class PubSubClient<T> {
 
     const json = JSON.stringify(message);
     const dataBuffer = Buffer.from(json);
-    const topic = this.client.topic(this.topicName);
+    const topic = this.client.topic(this.topicName, {
+      messageOrdering: orderingKey !== undefined,
+    });
 
     await Promise.race([
-      topic.publishMessage({ data: dataBuffer }),
+      topic.publishMessage({ data: dataBuffer, orderingKey }),
       timeout(
         PUBLISH_TIMEOUT,
         'Could not publish message in time. Did you forget to authenticate with GCP? (gcloud auth application-default login)',