Merge branch 'contacts-streaming'

Author: fgladisch

package-lock.json

@@ -19,6 +19,7 @@
   "license": "UNLICENSED",
   "scripts": {
     "dev": "npm run build && npm link && tsc --watch",
+    "watch": "tsc --watch",
     "prepare": "husky install",
     "test": "jest",
     "compile": "tsc",
@@ -63,6 +64,7 @@
     "typescript": "4.9.5"
   },
   "dependencies": {
+    "@google-cloud/pubsub": "^3.7.1",
     "ajv": "^8.11.0",
     "awesome-phonenumber": "^3.2.0",
     "axios": "^0.27.2",

package.json

@@ -12,8 +12,8 @@ import {
 } from "./models";
 import { CustomRouter } from "./models/custom-router.model";
 import { CustomRoute } from "./models/custom-routes.model";
-import { getContactCache } from "./util/get-contact-cache";
 import { errorLogger, infoLogger } from "./util";
+import { getContactCache } from "./util/get-contact-cache";
 
 const PORT: number = Number(process.env.PORT) || 8080;
 
@@ -43,51 +43,71 @@ export function start(
   app.get("/contacts", (req, res, next) =>
     controller.getContacts(req, res, next)
   );
+
   app.post("/contacts", (req, res, next) =>
     controller.createContact(req, res, next)
   );
+
   app.put("/contacts/:id", (req, res, next) =>
     controller.updateContact(req, res, next)
   );
+
   app.delete("/contacts/:id", (req, res, next) =>
     controller.deleteContact(req, res, next)
   );
+
+  app.post("/contacts/stream", (req, res, next) =>
+    controller.streamContacts(req, res, next)
+  );
+
   app.get(
     "/entity/:type/:id",
     (req: IntegrationEntityBridgeRequest, res, next) =>
       controller.getEntity(req, res, next)
   );
+
   app.get("/calendar", (req, res, next) =>
     controller.getCalendarEvents(req, res, next)
   );
+
   app.post("/calendar", (req, res, next) =>
     controller.createCalendarEvent(req, res, next)
   );
+
   app.put("/calendar/:id", (req, res, next) =>
     controller.updateCalendarEvent(req, res, next)
   );
+
   app.delete("/calendar/:id", (req, res, next) =>
     controller.deleteCalendarEvent(req, res, next)
   );
+
   app.post("/events/calls", (req, res, next) =>
     controller.handleCallEvent(req, res, next)
   );
+
   app.put("/events/calls/:id", (req, res, next) =>
     controller.updateCallEvent(req, res, next)
   );
+
   app.post("/events/connected", (req, res, next) =>
     controller.handleConnectedEvent(req, res, next)
   );
+
   app.put("/call-log", (req, res, next) =>
     controller.createOrUpdateCallLogsForEntities(req, res, next)
   );
+
   app.put("/call-log/phoneNumber", (req, res, next) =>
     controller.createCallLogForPhoneNumber(req, res, next)
   );
+
   app.get("/health", (req, res, next) => controller.getHealth(req, res, next));
+
   app.get("/oauth2/redirect", (req, res, next) =>
     controller.oAuth2Redirect(req, res, next)
   );
+
   app.get("/oauth2/callback", (req, res) =>
     controller.oAuth2Callback(req, res)
   );

src/index.ts

@@ -8,11 +8,13 @@ export function extractHeaderMiddleware(
   res: Response,
   next: NextFunction
 ): void {
+  const userId = req.get("x-user-id") || "";
   const apiKey = req.get("x-provider-key") || "";
   const apiUrl = req.get("x-provider-url") || "";
   const locale = req.get("x-provider-locale") || DEFAULT_LOCALE;
 
   req.providerConfig = {
+    userId,
     apiKey,
     apiUrl,
     locale,

src/middlewares/extract-header.middleware.ts

@@ -17,6 +17,7 @@ import { IntegrationEntityType } from "./integration-entity.model";
 export interface Adapter {
   getToken?: (config: Config) => Promise<{ apiKey: string }>;
   getContacts?: (config: Config) => Promise<Contact[]>;
+  streamContacts?: (config: Config) => AsyncGenerator<Contact[], void, unknown>;
   createContact?: (
     config: Config,
     contact: ContactTemplate

src/models/adapter.model.ts

@@ -1,4 +1,5 @@
 export interface Config {
+  userId: string;
   apiKey: string;
   apiUrl: string;
   locale: string;

src/models/config.model.ts

@@ -1,9 +1,9 @@
 import { Request, Response } from "express";
 import {
-  createRequest,
-  createResponse,
   MockRequest,
   MockResponse,
+  createRequest,
+  createResponse,
 } from "node-mocks-http";
 import {
   CalendarEvent,

src/models/controller.model.test.ts

@@ -14,6 +14,7 @@ import {
   ServerError,
 } from ".";
 import { calendarEventsSchema, contactsSchema } from "../schemas";
+import { isProduction } from "../util";
 import { shouldSkipCallEvent } from "../util/call-event.util";
 import { errorLogger, infoLogger } from "../util/logger.util";
 import { parsePhoneNumber } from "../util/phone-number-utils";
@@ -27,8 +28,10 @@ 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 { PubSubContactsMessage } from "./pubsub-contacts-message.model";
 
-const CONTACT_FETCH_TIMEOUT: number = 3000;
+const CONTACT_FETCH_TIMEOUT = 3000;
 
 function sanitizeContact(contact: Contact, locale: string): Contact {
   const result: APIContact = {
@@ -44,11 +47,25 @@ export class Controller {
   private adapter: Adapter;
   private contactCache: ContactCache | null;
   private ajv: Ajv;
+  private pubSubClient: PubSubClient | null = null;
 
   constructor(adapter: Adapter, contactCache: ContactCache | null) {
     this.adapter = adapter;
     this.contactCache = contactCache;
     this.ajv = new Ajv();
+
+    const { PUBSUB_TOPIC_NAME: topicName } = process.env;
+
+    if (isProduction() && typeof this.adapter.streamContacts === "function") {
+      if (!topicName) {
+        throw new Error("No pubsub topic name provided.");
+      }
+      this.pubSubClient = new PubSubClient(topicName);
+      infoLogger(
+        "Controller",
+        `Initialized PubSub client with topic ${topicName}`
+      );
+    }
   }
 
   public async getContacts(
@@ -149,6 +166,102 @@ export class Controller {
     }
   }
 
+  public async streamContacts(
+    req: BridgeRequest<unknown>,
+    res: Response,
+    next: NextFunction
+  ): Promise<void> {
+    const { providerConfig } = req;
+
+    if (!providerConfig) {
+      throw new ServerError(400, "Missing parameters");
+    }
+
+    const timestamp = Date.now();
+
+    try {
+      infoLogger(
+        "streamContacts",
+        `Starting contact streaming ${timestamp}`,
+        providerConfig.apiKey
+      );
+
+      const streamContacts = async () => {
+        if (!this.adapter.streamContacts) {
+          throw new ServerError(501, "Streaming contacts is not implemented");
+        }
+
+        const iterator = this.adapter.streamContacts(providerConfig);
+
+        let result = await iterator.next();
+
+        while (!result.done) {
+          const { value: contacts } = result;
+
+          try {
+            if (!validate(this.ajv, contactsSchema, contacts)) {
+              throw new Error("Invalid contacts received");
+            }
+
+            const message: PubSubContactsMessage = {
+              userId: providerConfig.userId,
+              timestamp,
+              contacts: contacts.map((contact) =>
+                sanitizeContact(contact, providerConfig.locale)
+              ),
+            };
+
+            await this.pubSubClient?.publishMessage(message);
+          } catch (error) {
+            errorLogger(
+              "streamContacts",
+              "Could not publish contacts",
+              providerConfig.apiKey,
+              error
+            );
+          } finally {
+            result = await iterator.next();
+          }
+        }
+      };
+
+      streamContacts().catch((error) =>
+        errorLogger(
+          "streamContacts",
+          "Could not stream contacts",
+          providerConfig.apiKey,
+          error
+        )
+      );
+
+      if (this.adapter.getToken && req.providerConfig) {
+        const { apiKey } = await this.adapter.getToken(req.providerConfig);
+        res.header("X-Provider-Key", apiKey);
+      }
+
+      infoLogger("streamContacts", "END", providerConfig.apiKey);
+
+      res.status(200).send({ timestamp });
+    } catch (error: any) {
+      // prevent logging of refresh errors
+      if (
+        error instanceof ServerError &&
+        error.message === IntegrationErrorType.INTEGRATION_REFRESH_ERROR
+      ) {
+        next(error);
+        return;
+      }
+
+      errorLogger(
+        "streamContacts",
+        "Could not stream contacts",
+        providerConfig.apiKey,
+        error
+      );
+      next(error);
+    }
+  }
+
   public async createContact(
     req: BridgeRequest<ContactTemplate>,
     res: Response,

src/models/controller.model.ts

@@ -0,0 +1,35 @@
+import { PubSub } from "@google-cloud/pubsub";
+import { errorLogger, infoLogger } from "../util";
+import { PubSubContactsMessage } from "./pubsub-contacts-message.model";
+
+export class PubSubClient {
+  private client: PubSub;
+  private topicName: string;
+
+  constructor(topicName: string) {
+    this.client = new PubSub();
+    this.topicName = topicName;
+  }
+
+  async publishMessage(message: PubSubContactsMessage) {
+    try {
+      if (!this.topicName) {
+        throw new Error("No pubsub topic name provided.");
+      }
+
+      const json = JSON.stringify(message);
+      const dataBuffer = Buffer.from(json);
+      const topic = this.client.topic(this.topicName);
+      await topic.publishMessage({ data: dataBuffer });
+
+      infoLogger(
+        PubSubClient.name,
+        `Published ${message.contacts.length} contacts for user ${message.userId} to topic ${this.topicName}`
+      );
+    } catch (error) {
+      console.error(error);
+      const message = (error as Error).message;
+      errorLogger(PubSubClient.name, `Could not publish to pubsub: ${message}`);
+    }
+  }
+}

src/models/pubsub-client.model.ts

@@ -0,0 +1,7 @@
+import { Contact } from "./contact.model";
+
+export type PubSubContactsMessage = {
+  userId: string;
+  timestamp: number;
+  contacts: Contact[];
+};