From 4a464944e92c1f94e60e1d77a35fc05c838fd036 Mon Sep 17 00:00:00 2001 From: Rojan Rajbhandari Date: Wed, 23 Jul 2025 14:06:45 +0545 Subject: [PATCH 1/3] feat(OUT-1985): implement new bottleneck --- src/utils/bottleneck.ts | 19 ++++++++++++++++++- 1 file changed, 18 insertions(+), 1 deletion(-) diff --git a/src/utils/bottleneck.ts b/src/utils/bottleneck.ts index e8f8a19d4..8c07326de 100644 --- a/src/utils/bottleneck.ts +++ b/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 }) From 5e217f4fb3683106e8c270ce4d3c4a05adbc200a Mon Sep 17 00:00:00 2001 From: Rojan Rajbhandari Date: Wed, 23 Jul 2025 15:42:57 +0545 Subject: [PATCH 2/3] fix(OUT-2058): use new copilot botteneck for higher req limit --- src/app/api/notification/notification.service.ts | 4 ++-- .../api/notification/validate-count/validateCount.service.ts | 4 ++-- src/jobs/notifications/send-reply-create-notifications.ts | 4 ++-- 3 files changed, 6 insertions(+), 6 deletions(-) diff --git a/src/app/api/notification/notification.service.ts b/src/app/api/notification/notification.service.ts index 1ee556a7d..a7ade508b 100644 --- a/src/app/api/notification/notification.service.ts +++ b/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) } diff --git a/src/app/api/notification/validate-count/validateCount.service.ts b/src/app/api/notification/validate-count/validateCount.service.ts index 3283ce2c4..12a7871de 100644 --- a/src/app/api/notification/validate-count/validateCount.service.ts +++ b/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({ diff --git a/src/jobs/notifications/send-reply-create-notifications.ts b/src/jobs/notifications/send-reply-create-notifications.ts index 2bed2a506..e9aa01479 100644 --- a/src/jobs/notifications/send-reply-create-notifications.ts +++ b/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[] = [] const queueNotificationPromise = (promise: Promise): void => { - notificationPromises.push(bottleneck.schedule(() => promise)) + notificationPromises.push(copilotBottleneck.schedule(() => promise)) } // Get all initiators involved in thread except the current user From 006d6b26a2f406e1f89b0551205edb90b7575023 Mon Sep 17 00:00:00 2001 From: Rojan Rajbhandari Date: Wed, 23 Jul 2025 15:43:35 +0545 Subject: [PATCH 3/3] feat(OUT-1985): implement webhook for companyIds updates --- src/app/api/webhook/webhook.service.ts | 125 ++++++++++++------------- src/types/webhook.ts | 4 +- 2 files changed, 63 insertions(+), 66 deletions(-) diff --git a/src/app/api/webhook/webhook.service.ts b/src/app/api/webhook/webhook.service.ts index f19b707d5..a7fe6b34b 100644 --- a/src/app/api/webhook/webhook.service.ts +++ b/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 } } }) diff --git a/src/types/webhook.ts b/src/types/webhook.ts index e71b22e83..c396087dd 100644 --- a/src/types/webhook.ts +++ b/src/types/webhook.ts @@ -26,9 +26,9 @@ export type WebhookEvent = z.infer 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