rename webhook related events

Author: codedrift

src/models/adapter.model.ts

@@ -14,7 +14,7 @@ import {
   LoggedIntegrationEntity,
 } from '.';
 import { IntegrationEntityType } from './integration-entity.model';
-import { ContactChangeEvent } from './contact-change-event.model';
+import { IntegrationsEvent } from './integrations-event.model';
 
 export interface Adapter {
   getToken?: (config: Config) => Promise<{ apiKey: string }>;
@@ -88,6 +88,6 @@ export interface Adapter {
     req: Request,
     res?: Response,
   ) => Promise<{ apiKey: string; apiUrl: string }>;
-  handleWebhook?: (req: Request) => Promise<ContactChangeEvent[]>;
+  handleWebhook?: (req: Request) => Promise<IntegrationsEvent[]>;
   verifyWebhookRequest?: (req: Request) => Promise<boolean>;
 }

src/models/contact-change-event.model.ts

@@ -10,11 +10,13 @@ import {
   CallEventWithIntegrationEntities,
   Contact,
   ContactCache,
-  ContactChangeEvent,
   ContactDelta,
   ContactTemplate,
   ContactUpdate,
   IntegrationEntityType,
+  IntegrationsEvent,
+  PubSubClient,
+  PubSubIntegrationsEventMessage,
   ServerError,
 } from '.';
 import { calendarEventsSchema, contactsSchema } from '../schemas';
@@ -32,12 +34,11 @@ import {
 import { CacheItemStateType } from './cache-item-state.model';
 import { CalendarFilterOptions } from './calendar-filter-options.model';
 import { IntegrationErrorType } from './integration-error.model';
-import { PubSubClient } from './pubsub-client.model';
-import { PubSubContactChangeEventMessage } from './pubsub-contact-change-event-message.model';
+
 import {
   PubSubContactsMessage,
   PubSubContactsState,
-} from './pubsub-contacts-message.model';
+} from './pubsub/pubsub-contacts-message.model';
 
 const CONTACT_FETCH_TIMEOUT = 5000;
 
@@ -57,7 +58,7 @@ export class Controller {
   private ajv: Ajv;
   private pubSubContactStreamingClient: PubSubClient<PubSubContactsMessage> | null =
     null;
-  private pubSubContactChangesClient: PubSubClient<PubSubContactChangeEventMessage> | null =
+  private pubSubIntegrationEventsClient: PubSubClient<PubSubIntegrationsEventMessage> | null =
     null;
   private integrationName: string = 'UNKNOWN';
   private streamingPromises = new Map<string, Promise<void>>();
@@ -107,13 +108,13 @@ export class Controller {
   }
 
   private initContactChanges() {
-    const { PUBSUB_TOPIC_NAME_CONTACT_CHANGES: topicName } = process.env;
+    const { PUBSUB_TOPIC_NAME_INTEGRATION_EVENTS: topicName } = process.env;
 
     if (!topicName) {
-      throw new Error('No PUBSUB_TOPIC_NAME_CONTACT_CHANGES provided.');
+      throw new Error('No PUBSUB_TOPIC_NAME_INTEGRATION_EVENTS provided.');
     }
 
-    this.pubSubContactChangesClient = new PubSubClient(topicName);
+    this.pubSubIntegrationEventsClient = new PubSubClient(topicName);
 
     infoLogger(
       'Controller',
@@ -1411,36 +1412,42 @@ export class Controller {
       const verified = await this.adapter.verifyWebhookRequest(req);
 
       if (!verified) {
-        errorLogger('handleWebhook', 'Webhook verification failed', '');
+        errorLogger('handleWebhook', 'Webhook verification failed');
         throw new ServerError(403, 'Webhook verification failed');
       }
 
-      infoLogger('handleWebhook', 'START', '');
-
-      const changeEvents: ContactChangeEvent[] =
-        await this.adapter.handleWebhook(req);
+      infoLogger('handleWebhook', 'START');
 
-      infoLogger(
-        'handleWebhook',
-        `Got ${changeEvents.length} changed contacts`,
-        '',
-      );
+      const events: IntegrationsEvent[] = await this.adapter.handleWebhook(req);
 
-      const deduplicatedChangeEvents = uniqWith(changeEvents, isEqual);
+      infoLogger('handleWebhook', `Got ${events.length} events`);
 
-      deduplicatedChangeEvents.map((changeEvent: ContactChangeEvent) => {
-        infoLogger(
-          'handleWebhook',
-          `Publishing contact change event with accountId ${changeEvent.accountId} and contactId ${changeEvent.contactId}`,
-          '',
-        );
+      const deduplicatedEvents = uniqWith(events, isEqual);
 
-        const message: PubSubContactChangeEventMessage = {
-          integrationName: this.integrationName,
-          ...changeEvent,
-        };
+      const publishResults = await Promise.allSettled(
+        deduplicatedEvents
+          .map<PubSubIntegrationsEventMessage>((event: IntegrationsEvent) => ({
+            integrationName: this.integrationName,
+            ...event,
+          }))
+          .map((message) => {
+            infoLogger(
+              'handleWebhook',
+              `Publishing event ${message.type} with accountId ${message.accountId}`,
+            );
+            return this.pubSubIntegrationEventsClient?.publishMessage(message);
+          }),
+      );
 
-        this.pubSubContactChangesClient?.publishMessage(message);
+      publishResults.forEach((result) => {
+        if (result.status === 'rejected') {
+          errorLogger(
+            'handleWebhook',
+            `Could not publish event ${result.reason.type} with accountId ${result.reason.accountId}`,
+            '',
+            result.reason,
+          );
+        }
       });
 
       infoLogger('handleWebhook', 'END', '');

src/models/controller.model.ts

@@ -1,17 +1,23 @@
 export * from './adapter.model';
+export * from './api-contact.model';
 export * from './bridge-request.model';
+export * from './cache-item-state.model';
 export * from './calendar-event.model';
 export * from './calendar-filter-options.model';
 export * from './call-direction.enum';
 export * from './call-event.model';
 export * from './config.model';
 export * from './contact-cache.model';
 export * from './contact.model';
-export * from './contact-change-event.model';
 export * from './controller.model';
 export * from './custom-router.model';
-export * from './server-error.model';
-export * from './user.model';
-export * from './integration-error.model';
+export * from './custom-routes.model';
 export * from './integration-entity.model';
+export * from './integration-error.model';
+export * from './integrations-event.model';
+export * from './pubsub';
+export * from './server-error.model';
+export * from './storage-adapter.model';
+export * from './token-cache.model';
 export * from './token.model';
+export * from './user.model';

src/models/index.ts

@@ -0,0 +1,23 @@
+import { IntegrationEntityType } from './integration-entity.model';
+
+export enum IntegrationsEventType {
+  CONTACT_CREATED = 'CONTACT_CREATED',
+  CONTACT_UPDATED = 'CONTACT_UPDATED',
+  CONTACT_DELETED = 'CONTACT_DELETED',
+}
+
+export type IntegrationsBaseEvent<T> = {
+  type: T;
+  accountId: string;
+};
+
+export type IntegrationsContactEvent = IntegrationsBaseEvent<
+  | IntegrationsEventType.CONTACT_CREATED
+  | IntegrationsEventType.CONTACT_UPDATED
+  | IntegrationsEventType.CONTACT_DELETED
+> & {
+  contactId: string;
+  contactType: IntegrationEntityType;
+};
+
+export type IntegrationsEvent = IntegrationsContactEvent;

src/models/integrations-event.model.ts

@@ -0,0 +1,3 @@
+export * from './pubsub-client.model';
+export * from './pubsub-contacts-message.model';
+export * from './pubsub-integrations-event-message.model';

src/models/pubsub-contact-change-event-message.model.ts

@@ -1,5 +1,5 @@
 import { PubSub } from '@google-cloud/pubsub';
-import { timeout } from '../util/timeout';
+import { timeout } from '../../util/timeout';
 
 const PUBLISH_TIMEOUT = 10_000;
 

src/models/pubsub/index.ts

@@ -1,4 +1,4 @@
-import { Contact } from './contact.model';
+import { Contact } from '../contact.model';
 
 export enum PubSubContactsState {
   IN_PROGRESS = 'IN_PROGRESS',

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

@@ -0,0 +1,5 @@
+import { IntegrationsEvent } from '../integrations-event.model';
+
+export type PubSubIntegrationsEventMessage = {
+  integrationName: string;
+} & IntegrationsEvent;