OUT-1985 | OUT-2058 | Support companyIds in webhook event Notifications

src/app/api/notification/notification.service.ts

@@ -6,7 +6,7 @@ import {
   NotificationRequestBody,
   Uuid,
 } from '@/types/common'
-import { bottleneck } from '@/utils/bottleneck'
+import { copilotBottleneck } from '@/utils/bottleneck'
 import { CopilotAPI } from '@/utils/CopilotAPI'
 import APIError from '@api/core/exceptions/api'
 import { BaseService } from '@api/core/services/base.service'
@@ -334,7 +334,7 @@ export class NotificationService extends BaseService {
     const copilot = new CopilotAPI(this.user.token)
     const markAsReadPromises = []
     for (const id of notificationIds) {
-      markAsReadPromises.push(bottleneck.schedule(() => copilot.markNotificationAsRead(id)))
+      markAsReadPromises.push(copilotBottleneck.schedule(() => copilot.markNotificationAsRead(id)))
     }
     await Promise.all(markAsReadPromises)
   }

src/app/api/notification/validate-count/validateCount.service.ts

@@ -1,7 +1,7 @@
 import { MAX_NOTIFICATIONS_COUNT } from '@/constants/notifications'
 import { DuplicateNotificationsQuerySchema } from '@/types/client-notifications'
 import { getArrayDifference } from '@/utils/array'
-import { bottleneck } from '@/utils/bottleneck'
+import { copilotBottleneck } from '@/utils/bottleneck'
 import { CopilotAPI } from '@/utils/CopilotAPI'
 import User from '@api/core/models/User.model'
 import { NotificationService } from '@api/notification/notification.service'
@@ -154,7 +154,7 @@ export class ValidateCountService extends NotificationService {
         continue
       }
       createNotificationPromises.push(
-        bottleneck.schedule(() => {
+        copilotBottleneck.schedule(() => {
           console.info(`ValidateCount :: Creating missing notification for task ${task.id} - ${task.title}`)
           // @ts-expect-error SDK types for new notification payload is not up to datelike always, SDK types are not up to date
           return this.copilot.createNotification({

src/app/api/webhook/webhook.service.ts

@@ -1,18 +1,19 @@
+import { MAX_FETCH_ASSIGNEE_COUNT } from '@/constants/users'
+import { NotificationRequestBody } from '@/types/common'
 import { ClientUpdatedEventData, HANDLEABLE_EVENT, WebhookEntitySchema, WebhookEvent, WebhookSchema } from '@/types/webhook'
-import { BaseService } from '@api/core/services/base.service'
-import { NextRequest } from 'next/server'
-import APIError from '@api/core/exceptions/api'
-import httpStatus from 'http-status'
-import { AssigneeType, StateType } from '@prisma/client'
+import { getArrayDifference } from '@/utils/array'
+import { copilotBottleneck, dbBottleneck } from '@/utils/bottleneck'
 import { CopilotAPI } from '@/utils/CopilotAPI'
-import { TasksService } from '@api/tasks/tasks.service'
-import Bottleneck from 'bottleneck'
-import { getInProductNotificationDetails } from '@api/notification/notification.helpers'
+import APIError from '@api/core/exceptions/api'
+import User from '@api/core/models/User.model'
+import { BaseService } from '@api/core/services/base.service'
 import { NotificationTaskActions } from '@api/core/types/tasks'
+import { getInProductNotificationDetails } from '@api/notification/notification.helpers'
 import { NotificationService } from '@api/notification/notification.service'
-import User from '@api/core/models/User.model'
-import { MAX_FETCH_ASSIGNEE_COUNT } from '@/constants/users'
-import { NotificationRequestBody } from '@/types/common'
+import { TasksService } from '@api/tasks/tasks.service'
+import { AssigneeType, StateType } from '@prisma/client'
+import httpStatus from 'http-status'
+import { NextRequest } from 'next/server'
 
 // TODO: This will be broken for a while until OUT-1985
 class WebhookService extends BaseService {
@@ -56,18 +57,25 @@ class WebhookService extends BaseService {
     return { assigneeId, assigneeType }
   }
 
-  async handleClientCreated(assigneeId: string) {
+  async handleClientCreated(clientId: string, forCompanyId?: string) {
     // First fetch all the current non-complete tasks for client's company, if exists
-    const client = await this.copilot.getClient(assigneeId)
+    const client = await this.copilot.getClient(clientId)
     let company
     try {
       // Accomodate indididual client's company id behavior for copilot
-      company = await this.copilot.getCompany(client.companyId)
+      // Note: When a client is created, only one company can be assigned to it.
+      // This is why we don't need to check for multiple companies here.
+      //
+      // NOTE: forCompanyId is used when a client is created for a company that is not the first company in the client's companyIds array
+      const companyId = forCompanyId || client.companyIds?.[0] || client.companyId
+      company = await this.copilot.getCompany(companyId)
     } catch (e: unknown) {
       company = null
     }
     if (!company?.name) {
-      console.info("Ignoring client.activated event for client that doesn't have a company")
+      console.info(
+        `WebhookService#handleClientCreated :: Ignoring event for client that doesn't have a company (has placeholder company)`,
+      )
       return
     }
 
@@ -76,18 +84,17 @@ class WebhookService extends BaseService {
     if (!tasks.length) return
 
     // Then trigger appropriate notifications
-    const bottleneck = new Bottleneck({ minTime: 250, maxConcurrent: 2 })
     const createNotificationPromises = []
 
     const internalUsers = (await this.copilot.getInternalUsers({ limit: MAX_FETCH_ASSIGNEE_COUNT })).data
     const existingNotifications = await this.db.clientNotification.findMany({
-      where: { clientId: assigneeId },
+      where: { clientId, companyId: company.id },
     })
 
     for (let task of tasks) {
       const actionUser = internalUsers.find((iu) => iu.id === task.createdById)
       if (!actionUser) {
-        console.error(`WebhookService#handleClientCreated :: could not get action user for company task ${task.id}`)
+        console.error(`WebhookService#handleClientCreated :: Could not get action user for company task ${task.id}`)
         continue
       }
 
@@ -101,23 +108,22 @@ class WebhookService extends BaseService {
       const notificationDetails: NotificationRequestBody = {
         senderId: task.createdById,
         senderType: 'internalUser',
-        recipientClientId: assigneeId,
+        recipientClientId: clientId,
         recipientCompanyId: company.id,
         // If any of the given action is not present in details obj, that type of notification is not sent
         deliveryTargets: { inProduct },
       }
-      createNotificationPromises.push(bottleneck.schedule(() => this.copilot.createNotification(notificationDetails)))
+      createNotificationPromises.push(copilotBottleneck.schedule(() => this.copilot.createNotification(notificationDetails)))
     }
     const notifications = await Promise.all(createNotificationPromises)
 
     // Now add appropriate records to our ClientNotifications table
-    const insertBottleneck = new Bottleneck({ minTime: 100, maxConcurrent: 4 })
     const insertPromises = []
     const notificationService = new NotificationService(this.user)
     for (let i = 0; i < notifications.length; i++) {
       insertPromises.push(
         // This is assuming a 1:1 map for tasks and notifications
-        insertBottleneck.schedule(() => notificationService.addToClientNotifications(tasks[i], notifications[i])),
+        dbBottleneck.schedule(() => notificationService.addToClientNotifications(tasks[i], notifications[i])),
       )
     }
     await Promise.all(insertPromises)
@@ -126,7 +132,7 @@ class WebhookService extends BaseService {
   async handleUserDeleted(assigneeId: string, assigneeType: AssigneeType) {
     const tasksService = new TasksService(this.user)
     // Delete corresponding tasks
-    console.info(`Deleting all tasks for ${assigneeType} ${assigneeId}`)
+    console.info(`WebhookService#handleUserDeleted :: Deleting all tasks for ${assigneeType} ${assigneeId}`)
     const tasks = await tasksService.deleteAllAssigneeTasks(assigneeId, assigneeType)
 
     // Now delete any and all associated notifications triggered on behalf of company tasks for clients.
@@ -154,42 +160,35 @@ class WebhookService extends BaseService {
   async handleClientUpdated(data: ClientUpdatedEventData) {
     const {
       id: clientId,
-      companyId: newCompanyId,
-      previousAttributes: { companyId: prevCompanyId },
+      companyIds: newCompanyIds,
+      previousAttributes: { companyIds: prevCompanyIds },
     } = data
-    // If company hasn't been changed - don't bother with any of this
-    if (prevCompanyId === newCompanyId || !prevCompanyId) return
-
-    // Only delete prev notifications if client had a valid company before
-    let prevCompany, newCompany
-    try {
-      prevCompany = await this.copilot.getCompany(prevCompanyId)
-    } catch (e: unknown) {
-      prevCompany = null
-    }
-    try {
-      newCompany = await this.copilot.getCompany(newCompanyId)
-    } catch (e: unknown) {
-      newCompany = null
+    // If companies haven't been changed then we don't have to migrate any tasks / notifications
+    // When companies have been changed, the previousAttributes object will contain `companyIds` array with the previous company ids
+    if (!prevCompanyIds || !newCompanyIds) return
+
+    const removedCompanies = getArrayDifference(prevCompanyIds, newCompanyIds || [])
+    const addedCompanies = getArrayDifference(newCompanyIds || [], prevCompanyIds)
+
+    // Technically having multiple companies updated in one webhook event is not possible and accessing
+    // removedCompanies[0] or addedCompanies[0] should be enough,
+    // However I'm keeping this for future-proofing
+    if (removedCompanies.length) {
+      for (let companyId of removedCompanies) {
+        await this.handleCompanyUnassignment(clientId, companyId)
+      }
     }
-
-    // Cases for company change - assignment / unassignment / reassignment
-    if (!prevCompany?.name && newCompany?.name) {
-      await this.handleCompanyAssignment(clientId)
-    } else if (!newCompany?.name && prevCompany?.name) {
-      await this.handleCompanyUnassignment(clientId, prevCompanyId)
-    } else if (prevCompany?.name && newCompany?.name) {
-      await this.handleCompanyUnassignment(clientId, prevCompanyId)
-      await this.handleCompanyAssignment(clientId)
-    } else {
-      // This should never happen, still we want to be reported if this ever does
-      throw new APIError(httpStatus.INTERNAL_SERVER_ERROR, 'Could not identify company change')
+    if (addedCompanies.length) {
+      for (let companyId of addedCompanies) {
+        await this.handleCompanyAssignment(clientId, companyId)
+      }
     }
+    return
   }
 
-  private async handleCompanyAssignment(clientId: string) {
+  private async handleCompanyAssignment(clientId: string, companyId: string) {
     // Notifications logic is the same as when a new client is created
-    await this.handleClientCreated(clientId)
+    await this.handleClientCreated(clientId, companyId)
   }
 
   private async handleCompanyUnassignment(clientId: string, prevCompanyId: string) {
@@ -198,19 +197,23 @@ class WebhookService extends BaseService {
     // NOTE: If prev company was not a valid company, instead a randomly generated placeholder company,
     // prevCompany.name will be an empty string
     if (!company.name) {
-      throw new APIError(httpStatus.INTERNAL_SERVER_ERROR, 'Cannot unassign from a company that does not exist')
+      // Fail safely since there can be multiple unassignments in one webhook event
+      console.error(`WebhookService#handleCompanyUnassignment :: Cannot unassign from a company that does not exist`)
+      return
     }
 
     // First find all tasks related to previous company
     const prevCompanyTasks = await this.db.task.findMany({
       where: {
-        assigneeId: company.id,
-        assigneeType: AssigneeType.company,
+        companyId: prevCompanyId,
+        clientId: null,
         workflowState: {
           type: { not: StateType.completed },
         },
       },
     })
+    if (!prevCompanyTasks.length) return
+
     const prevCompanyTaskIds = prevCompanyTasks.map((task) => task.id)
 
     // Find all triggered notifications for this client, on behalf of prev company
@@ -224,14 +227,8 @@ class WebhookService extends BaseService {
 
     // Delete all task notifications triggered for client for previous company
     const deletePromises = []
-    const bottleneck = new Bottleneck({ minTime: 250, maxConcurrent: 2 })
-
-    for (let notification of prevCompanyNotifications) {
-      deletePromises.push(
-        bottleneck.schedule(() => {
-          return this.copilot.markNotificationAsRead(notification.notificationId)
-        }),
-      )
+    for (let { notificationId } of prevCompanyNotifications) {
+      deletePromises.push(copilotBottleneck.schedule(() => this.copilot.markNotificationAsRead(notificationId)))
     }
     await Promise.all(deletePromises)
     await this.db.clientNotification.deleteMany({ where: { id: { in: notificationIds } } })

src/jobs/notifications/send-reply-create-notifications.ts

@@ -1,6 +1,6 @@
 import { NotificationSender, NotificationSenderSchema } from '@/types/common'
 import { getAssigneeName } from '@/utils/assignee'
-import { bottleneck } from '@/utils/bottleneck'
+import { copilotBottleneck } from '@/utils/bottleneck'
 import { CopilotAPI } from '@/utils/CopilotAPI'
 import { CommentRepository } from '@api/comment/comment.repository'
 import { CommentService } from '@api/comment/comment.service'
@@ -44,7 +44,7 @@ export const sendReplyCreateNotifications = task({
 
     const notificationPromises: Promise<unknown>[] = []
     const queueNotificationPromise = <T>(promise: Promise<T>): void => {
-      notificationPromises.push(bottleneck.schedule(() => promise))
+      notificationPromises.push(copilotBottleneck.schedule(() => promise))
     }
 
     // Get all initiators involved in thread except the current user

src/types/webhook.ts

@@ -26,9 +26,9 @@ export type WebhookEvent = z.infer<typeof WebhookSchema>
 
 export const ClientUpdatedEventDataSchema = z.object({
   id: z.string(),
-  companyId: z.string(),
+  companyIds: z.array(z.string()).nullish(),
   previousAttributes: z.object({
-    companyId: z.string().optional(),
+    companyIds: z.array(z.string()).optional(),
   }),
 })
 export type ClientUpdatedEventData = z.infer<typeof ClientUpdatedEventDataSchema>

src/utils/bottleneck.ts

@@ -1,3 +1,20 @@
+/**
+ * Bottleneck is used to limit the number of concurrent requests to an external service.
+ * This external service can be Copilot API, the database, or any other service that has a rate limit in place
+ * This library can be used to prevent overwhelming the external service with requests
+ *
+ * The maxConcurrent value is the maximum number of concurrent requests to the external service at a single time.
+ * The minTime value is the minimum time between requests to the external service.
+ */
+
 import Bottleneck from 'bottleneck'
 
-export const bottleneck = new Bottleneck({ maxConcurrent: 2, minTime: 250 })
+// For production, max peak rate is 6 * (1000 / 200) = 30 requests per second.
+// For preview / development, max peak rate is 3 * (1000 / 250) = 12 requests per second.
+// We have a ratelimit of 100req/s in prod, with a burst rate of 200req/s.
+// For preview / development, we have a ratelimit of 20req/s, with a burst rate of 20req/s.
+const maxConcurrent = process.env.VERCEL_ENV === 'production' ? 6 : 3
+const minTime = process.env.VERCEL_ENV === 'production' ? 200 : 250
+export const copilotBottleneck = new Bottleneck({ maxConcurrent, minTime })
+
+export const dbBottleneck = new Bottleneck({ maxConcurrent: 20, minTime: 100 })