diff --git a/src/app/api/notification/__snapshots__/notification.helpers.test.ts.snap b/src/app/api/notification/__snapshots__/notification.helpers.test.ts.snap index 0904c2538..81f1621a7 100644 --- a/src/app/api/notification/__snapshots__/notification.helpers.test.ts.snap +++ b/src/app/api/notification/__snapshots__/notification.helpers.test.ts.snap @@ -10,7 +10,7 @@ Please make sure to complete this task by the due date.", "taskId": "task_1", }, "header": "A task was assigned to your company", - "subject": "Acme portal: [Due Soon] Task due in 3 days", + "subject": "[Due Soon] Task due in 3 days", "title": "View task", }, "DUE_DATE_OVERDUE_3D": { @@ -19,7 +19,7 @@ Please make sure to complete this task by the due date.", "taskId": "task_1", }, "header": "A task was assigned to your company", - "subject": "Acme portal: [Overdue] Task was due 3 days ago", + "subject": "[Overdue] Task was due 3 days ago", "title": "View task", }, "DUE_DATE_OVERDUE_7D": { @@ -30,7 +30,7 @@ Please complete this task as soon as possible.", "taskId": "task_1", }, "header": "A task was assigned to your company", - "subject": "Acme portal: [Overdue] Task overdue by one week", + "subject": "[Overdue] Task overdue by one week", "title": "View task", }, "DUE_DATE_TODAY": { @@ -41,7 +41,7 @@ Please complete this task as soon as possible.", "taskId": "task_1", }, "header": "A task was assigned to your company", - "subject": "Acme portal: [Due Soon] Task due today", + "subject": "[Due Soon] Task due today", "title": "View task", }, "NO_DUE_DATE_3D": { @@ -52,7 +52,7 @@ If you've already completed this task, please mark it as done in the portal.", "taskId": "task_1", }, "header": "A task was assigned to your company", - "subject": "Acme portal: [Reminder] You have a task to complete", + "subject": "[Reminder] You have a task to complete", "title": "View task", }, "NO_DUE_DATE_7D": { @@ -63,7 +63,7 @@ If you've already completed this task, please mark it as done in the portal.", "taskId": "task_1", }, "header": "A task was assigned to your company", - "subject": "Acme portal: [Reminder] Task still pending", + "subject": "[Reminder] Task still pending", "title": "View task", }, } @@ -79,7 +79,7 @@ Please make sure to complete this task by the due date.", "taskId": "task_1", }, "header": "A task was assigned to you", - "subject": "Acme portal: [Due Soon] Task due in 3 days", + "subject": "[Due Soon] Task due in 3 days", "title": "View task", }, "DUE_DATE_OVERDUE_3D": { @@ -88,7 +88,7 @@ Please make sure to complete this task by the due date.", "taskId": "task_1", }, "header": "A task was assigned to you", - "subject": "Acme portal: [Overdue] Task was due 3 days ago", + "subject": "[Overdue] Task was due 3 days ago", "title": "View task", }, "DUE_DATE_OVERDUE_7D": { @@ -99,7 +99,7 @@ Please complete this task as soon as possible.", "taskId": "task_1", }, "header": "A task was assigned to you", - "subject": "Acme portal: [Overdue] Task overdue by one week", + "subject": "[Overdue] Task overdue by one week", "title": "View task", }, "DUE_DATE_TODAY": { @@ -110,7 +110,7 @@ Please complete this task as soon as possible.", "taskId": "task_1", }, "header": "A task was assigned to you", - "subject": "Acme portal: [Due Soon] Task due today", + "subject": "[Due Soon] Task due today", "title": "View task", }, "NO_DUE_DATE_3D": { @@ -121,7 +121,7 @@ If you've already completed this task, please mark it as done in the portal.", "taskId": "task_1", }, "header": "A task was assigned to you", - "subject": "Acme portal: [Reminder] You have a task to complete", + "subject": "[Reminder] You have a task to complete", "title": "View task", }, "NO_DUE_DATE_7D": { @@ -132,7 +132,7 @@ If you've already completed this task, please mark it as done in the portal.", "taskId": "task_1", }, "header": "A task was assigned to you", - "subject": "Acme portal: [Reminder] Task still pending", + "subject": "[Reminder] Task still pending", "title": "View task", }, } diff --git a/src/app/api/notification/notification.helpers.test.ts b/src/app/api/notification/notification.helpers.test.ts index 414fadc2b..5874c181e 100644 --- a/src/app/api/notification/notification.helpers.test.ts +++ b/src/app/api/notification/notification.helpers.test.ts @@ -39,10 +39,11 @@ describe('getReminderEmailDetails', () => { expect(result[TaskReminderType.NO_DUE_DATE_3D].header).toBe('A task was assigned to your team') }) - it('falls back gracefully when brandName is missing', () => { - const noBrand: WorkspaceResponse = { ...workspace, brandName: undefined } - const result = getReminderEmailDetails(noBrand, task, false) - expect(result[TaskReminderType.NO_DUE_DATE_3D].subject).toBe('portal: [Reminder] You have a task to complete') + it('omits any ` portal:` prefix from subjects (Copilot prepends it server-side)', () => { + const result = getReminderEmailDetails(workspace, task, false) + for (const variant of Object.values(TaskReminderType)) { + expect(result[variant].subject).not.toMatch(/portal:/i) + } }) it('emits ctaParams with the task id for every variant', () => { diff --git a/src/app/api/notification/notification.helpers.ts b/src/app/api/notification/notification.helpers.ts index 708af2eae..7b889f761 100644 --- a/src/app/api/notification/notification.helpers.ts +++ b/src/app/api/notification/notification.helpers.ts @@ -202,16 +202,8 @@ export const getEmailDetails = ( } } -/** - * Helper function that returns reminder email content for each TaskReminderType variant. - * Lifecycle is independent from `getEmailDetails` (which is keyed by NotificationTaskActions). - * @param {WorkspaceResponse} workspace - Workspace whose brandName fronts the subject and - * whose labels resolve the company term. - * @param {Pick} task - Task being reminded about. Used for ctaParams and body interpolation. - * @param {boolean} isCompanyRecipient - True if recipient is a company (header reads "your {groupTerm}"), - * false for an individual recipient (header reads "you"). - * @returns Reminder email content keyed by TaskReminderType. - */ +// Subjects intentionally omit any ` portal:` prefix — Copilot's email +// service prepends that itself, and adding it here results in a duplicated prefix. export const getReminderEmailDetails = ( workspace: WorkspaceResponse, task: Pick, @@ -226,7 +218,6 @@ export const getReminderEmailDetails = ( ctaParams: { taskId: string } } > => { - const portalPrefix = `${workspace.brandName ?? ''} portal:`.trimStart() const labels = getWorkspaceLabels(workspace) const header = isCompanyRecipient ? `A task was assigned to your ${labels.groupTerm}` : 'A task was assigned to you' const ctaParams = { taskId: task.id } @@ -234,42 +225,42 @@ export const getReminderEmailDetails = ( return { [TaskReminderType.NO_DUE_DATE_3D]: { - subject: `${portalPrefix} [Reminder] You have a task to complete`, + subject: '[Reminder] You have a task to complete', header, title, body: `This is a friendly reminder that you have a task ‘${task.title}’ assigned to you that's still pending completion.\n\nIf you've already completed this task, please mark it as done in the portal.`, ctaParams, }, [TaskReminderType.NO_DUE_DATE_7D]: { - subject: `${portalPrefix} [Reminder] Task still pending`, + subject: '[Reminder] Task still pending', header, title, body: `This is a friendly reminder that you have a task ‘${task.title}’ that was assigned to you a week ago and is still pending.\n\nIf you've already completed this task, please mark it as done in the portal.`, ctaParams, }, [TaskReminderType.DUE_DATE_BEFORE_3D]: { - subject: `${portalPrefix} [Due Soon] Task due in 3 days`, + subject: '[Due Soon] Task due in 3 days', header, title, body: `This is a friendly reminder that you have a task ‘${task.title}’ due in 3 days.\n\nPlease make sure to complete this task by the due date.`, ctaParams, }, [TaskReminderType.DUE_DATE_TODAY]: { - subject: `${portalPrefix} [Due Soon] Task due today`, + subject: '[Due Soon] Task due today', header, title, body: `This is a friendly reminder that you have a task ‘${task.title}’ due today.\n\nPlease complete this task as soon as possible.`, ctaParams, }, [TaskReminderType.DUE_DATE_OVERDUE_3D]: { - subject: `${portalPrefix} [Overdue] Task was due 3 days ago`, + subject: '[Overdue] Task was due 3 days ago', header, title, body: `This is a friendly reminder that the task ‘${task.title}’ is now overdue. It was due 3 days ago and is still pending completion.`, ctaParams, }, [TaskReminderType.DUE_DATE_OVERDUE_7D]: { - subject: `${portalPrefix} [Overdue] Task overdue by one week`, + subject: '[Overdue] Task overdue by one week', header, title, body: `This is a friendly reminder that the task ‘${task.title}’ is now one week overdue.\n\nPlease complete this task as soon as possible.`, diff --git a/src/jobs/notifications/dispatch-reminder-email.test.ts b/src/jobs/notifications/dispatch-reminder-email.test.ts new file mode 100644 index 000000000..a9188e6dc --- /dev/null +++ b/src/jobs/notifications/dispatch-reminder-email.test.ts @@ -0,0 +1,110 @@ +import { TaskReminderType } from '@prisma/client' + +const mockSendReminderEmail = jest.fn() +const mockTaskReminderSentDelete = jest.fn() +const mockCopilotApiCtor = jest.fn() + +jest.mock('@trigger.dev/sdk/v3', () => ({ + task: ({ run }: { run: (payload: unknown) => unknown }) => ({ run }), + tasks: { onFailure: () => undefined }, + logger: { log: jest.fn(), error: jest.fn(), warn: jest.fn() }, +})) + +jest.mock('@/config', () => ({ copilotAPIKey: 'test-api-key' })) + +jest.mock('@/lib/db', () => ({ + __esModule: true, + default: { + getInstance: () => ({ + taskReminderSent: { delete: mockTaskReminderSentDelete }, + }), + }, +})) + +jest.mock('@/utils/CopilotAPI', () => ({ + CopilotAPI: jest.fn().mockImplementation((...args: unknown[]) => { + mockCopilotApiCtor(...args) + return {} + }), +})) + +jest.mock('./send-reminder-email', () => ({ + sendReminderEmail: (...args: unknown[]) => mockSendReminderEmail(...args), +})) + +import { + DispatchReminderEmailPayload, + dispatchReminderEmailOnFailure, + dispatchReminderEmailRun, +} from './dispatch-reminder-email' + +const buildPayload = (overrides: Partial = {}): DispatchReminderEmailPayload => ({ + ledgerId: 'ledger_1', + workspaceId: 'ws_1', + task: { id: 'task_1', title: 'Submit timesheet', createdById: 'iu_1' }, + recipientClientId: 'client_1', + recipientCompanyId: 'company_1', + reminderType: TaskReminderType.NO_DUE_DATE_3D, + isCompanyRecipient: false, + workspace: { id: 'ws_1', brandName: 'Acme' }, + ...overrides, +}) + +describe('dispatchReminderEmail', () => { + beforeEach(() => { + jest.clearAllMocks() + mockSendReminderEmail.mockReset() + mockTaskReminderSentDelete.mockReset() + mockCopilotApiCtor.mockReset() + }) + + describe('run', () => { + it('mints a workspace-scoped CopilotAPI and forwards the payload to sendReminderEmail', async () => { + mockSendReminderEmail.mockResolvedValueOnce('notif_1') + + const result = await dispatchReminderEmailRun(buildPayload()) + + expect(mockCopilotApiCtor).toHaveBeenCalledWith('', 'ws_1/test-api-key') + expect(mockSendReminderEmail).toHaveBeenCalledTimes(1) + expect(mockSendReminderEmail.mock.calls[0][0]).toMatchObject({ + task: { id: 'task_1', title: 'Submit timesheet', createdById: 'iu_1' }, + recipientClientId: 'client_1', + recipientCompanyId: 'company_1', + reminderType: TaskReminderType.NO_DUE_DATE_3D, + isCompanyRecipient: false, + }) + expect(result).toEqual({ ledgerId: 'ledger_1', notificationId: 'notif_1', sent: true }) + }) + + it('rethrows so Trigger.dev can apply its retry policy', async () => { + mockSendReminderEmail.mockRejectedValueOnce(new Error('copilot 5xx')) + + await expect(dispatchReminderEmailRun(buildPayload())).rejects.toThrow('copilot 5xx') + expect(mockTaskReminderSentDelete).not.toHaveBeenCalled() // compensation is onFailure's job, not run's + }) + }) + + describe('onFailure', () => { + it('deletes the ledger row so the next cron run can retry', async () => { + mockTaskReminderSentDelete.mockResolvedValueOnce({ id: 'ledger_1' }) + + await dispatchReminderEmailOnFailure({ + payload: buildPayload(), + error: new Error('all retries exhausted'), + }) + + expect(mockTaskReminderSentDelete).toHaveBeenCalledWith({ where: { id: 'ledger_1' } }) + }) + + it('does not throw if the ledger DELETE itself fails (logs and moves on)', async () => { + mockTaskReminderSentDelete.mockRejectedValueOnce(new Error('db blew up')) + + await expect( + dispatchReminderEmailOnFailure({ + payload: buildPayload(), + error: new Error('all retries exhausted'), + }), + ).resolves.toBeUndefined() + }) + }) +}) diff --git a/src/jobs/notifications/dispatch-reminder-email.ts b/src/jobs/notifications/dispatch-reminder-email.ts new file mode 100644 index 000000000..8a80ff0db --- /dev/null +++ b/src/jobs/notifications/dispatch-reminder-email.ts @@ -0,0 +1,72 @@ +import 'server-only' + +import { copilotAPIKey } from '@/config' +import DBClient from '@/lib/db' +import { WorkspaceResponse } from '@/types/common' +import { CopilotAPI } from '@/utils/CopilotAPI' +import { serializeError } from '@/utils/serializeError' +import { Task, TaskReminderType } from '@prisma/client' +import { logger, task, tasks } from '@trigger.dev/sdk/v3' + +import { sendReminderEmail } from './send-reminder-email' + +export type DispatchReminderEmailPayload = { + ledgerId: string + workspaceId: string + task: Pick + recipientClientId: string + recipientCompanyId: string | null + reminderType: TaskReminderType + isCompanyRecipient: boolean + workspace: WorkspaceResponse +} + +const TASK_ID = 'dispatch-reminder-email' + +export const dispatchReminderEmailRun = async (payload: DispatchReminderEmailPayload) => { + const copilot = new CopilotAPI('', `${payload.workspaceId}/${copilotAPIKey}`) + const notificationId = await sendReminderEmail({ + task: payload.task, + recipientClientId: payload.recipientClientId, + recipientCompanyId: payload.recipientCompanyId, + reminderType: payload.reminderType, + isCompanyRecipient: payload.isCompanyRecipient, + workspace: payload.workspace, + copilot, + }) + return { ledgerId: payload.ledgerId, notificationId, sent: true as const } +} + +// Fires after Trigger.dev exhausts all retries. Compensating here (instead of inside run's +// catch) avoids dropping the ledger row on transient failures a retry would have recovered. +// The SDK types the hook's payload as `unknown`; we cast once via destructure. +export const dispatchReminderEmailOnFailure = async ({ payload, error }: { payload: unknown; error: unknown }) => { + const { ledgerId, workspaceId, task, recipientClientId, reminderType } = payload as DispatchReminderEmailPayload + logger.error('dispatch-reminder-email: retries exhausted, compensating ledger', { + ledgerId, + workspaceId, + taskId: task.id, + recipientClientId, + reminderType, + error: serializeError(error), + }) + const db = DBClient.getInstance() + try { + await db.taskReminderSent.delete({ where: { id: ledgerId } }) + } catch (deleteErr) { + logger.error('dispatch-reminder-email: ledger compensation DELETE failed, reminder will not retry', { + ledgerId, + error: serializeError(deleteErr), + }) + } +} + +export const dispatchReminderEmail = task({ + id: TASK_ID, + queue: { concurrencyLimit: 5 }, + retry: { maxAttempts: 3, factor: 2, minTimeoutInMs: 1_000, maxTimeoutInMs: 15_000, randomize: true }, + maxDuration: 30, + run: dispatchReminderEmailRun, +}) + +tasks.onFailure(TASK_ID, dispatchReminderEmailOnFailure) diff --git a/src/jobs/notifications/eligibility.ts b/src/jobs/notifications/eligibility.ts index 87e7915b6..028dde4fe 100644 --- a/src/jobs/notifications/eligibility.ts +++ b/src/jobs/notifications/eligibility.ts @@ -4,40 +4,24 @@ import { AssigneeType, TaskReminderType } from '@prisma/client' export type EligibilityRow = { taskId: string workspaceId: string + title: string + createdById: string assigneeId: string assigneeType: AssigneeType - /** - * Company context for the recipient. - * - assigneeType='client' → task.companyId (a client may belong to multiple companies on Copilot; - * this disambiguates which "hat" they're wearing for this task) - * - assigneeType='company' → assigneeId (the company IS the assignee; caller fans out to members) - * - assigneeType='internalUser'→ null (IUs have no company concept and never receive email notifications) - * - * Required for ClientNotifications inserts (unique key includes companyId) and for - * Copilot's recipientCompanyId on email-bearing notifications. See - * src/app/api/notification/notification.service.ts:558. - */ companyId: string | null reminderType: TaskReminderType } -/** - * Returns one row per (task, assignee, reminderType) eligible for a reminder today. - * - * Company-assigned tasks emit a single row with assigneeType='company' and assigneeId - * set to the company id. Caller fans those out to individual members via Copilot — - * the SQL deliberately stops at the company boundary so DB has no Copilot dependency. - * - * Already-sent reminders are NOT filtered here. The TaskReminderSents unique constraint - * is the dedupe primitive at insert time, so a retried cron run is idempotent without - * an extra NOT EXISTS check (the windows are exact-day so day-N reminders don't repeat - * in normal operation). - */ +// Company-assigned tasks emit one row at the company level; caller fans out to members. +// Already-sent reminders are not filtered here — TaskReminderSents' unique constraint is +// the dedupe primitive at insert time. export const getEligibleReminders = async (db: ReturnType): Promise => { return db.$queryRaw` SELECT t.id::text AS "taskId", t."workspaceId", + t."title", + t."createdById"::text AS "createdById", t."assigneeId"::text AS "assigneeId", t."assigneeType" AS "assigneeType", (CASE @@ -54,11 +38,8 @@ export const getEligibleReminders = async (db: ReturnType { recipientCompanyId: 'company_1', }) expect(payload.deliveryTargets.email).toEqual({ - subject: 'Acme portal: [Reminder] You have a task to complete', + subject: '[Reminder] You have a task to complete', header: 'A task was assigned to you', title: 'View task', body: expect.stringContaining('‘Submit timesheet’'), @@ -80,7 +80,7 @@ describe('sendReminderEmail', () => { const payload = createNotification.mock.calls[0][0] expect(payload.deliveryTargets.email.header).toBe('A task was assigned to your company') - expect(payload.deliveryTargets.email.subject).toBe('Acme portal: [Due Soon] Task due today') + expect(payload.deliveryTargets.email.subject).toBe('[Due Soon] Task due today') }) it('omits recipientCompanyId when null', async () => { diff --git a/src/jobs/notifications/send-reminder-email.ts b/src/jobs/notifications/send-reminder-email.ts index f30441e7d..7b247d4d5 100644 --- a/src/jobs/notifications/send-reminder-email.ts +++ b/src/jobs/notifications/send-reminder-email.ts @@ -15,18 +15,8 @@ export type SendReminderEmailArgs = { copilot: CopilotAPI } -/** - * Dispatches a single task reminder email via Copilot's notification API. - * - * Email-only delivery: omits `deliveryTargets.inProduct` so no in-product notification - * is created. We also deliberately skip writing to `ClientNotification` — - * `ClientNotification` tracks read-state for in-product notifications, which reminders - * don't create. Reminder dedupe state lives in `TaskReminderSent`, which the caller - * inserts on success (the unique constraint is the idempotency primitive). - * - * Throws on Copilot failure. Callers compensate by NOT inserting into - * `TaskReminderSent`, so a future cron run will retry the same `(task, recipient, type)`. - */ +// Email-only: omits deliveryTargets.inProduct and does not write to ClientNotification. +// Reminder dedupe lives in TaskReminderSent (caller's responsibility). export const sendReminderEmail = async ({ task, recipientClientId, diff --git a/src/jobs/notifications/send-task-reminders.test.ts b/src/jobs/notifications/send-task-reminders.test.ts new file mode 100644 index 000000000..257dc5d5c --- /dev/null +++ b/src/jobs/notifications/send-task-reminders.test.ts @@ -0,0 +1,292 @@ +import { AssigneeType, TaskReminderType } from '@prisma/client' + +const mockTaskReminderSentCreateManyAndReturn = jest.fn() +const mockTaskReminderSentDeleteMany = jest.fn() +const mockGetEligibleReminders = jest.fn() +const mockBatchTrigger = jest.fn() +const mockGetWorkspace = jest.fn() +const mockGetCompanyClients = jest.fn() +const mockCopilotApiCtor = jest.fn() + +jest.mock('@trigger.dev/sdk/v3', () => ({ + schedules: { + task: ({ run }: { run: (payload: unknown, ctx?: unknown) => unknown }) => ({ run }), + }, + logger: { log: jest.fn(), error: jest.fn(), warn: jest.fn() }, +})) + +jest.mock('@/config', () => ({ copilotAPIKey: 'test-api-key' })) + +jest.mock('@/lib/db', () => ({ + __esModule: true, + default: { + getInstance: () => ({ + taskReminderSent: { + createManyAndReturn: mockTaskReminderSentCreateManyAndReturn, + deleteMany: mockTaskReminderSentDeleteMany, + }, + }), + }, +})) + +jest.mock('@/utils/CopilotAPI', () => ({ + CopilotAPI: jest.fn().mockImplementation((...args: unknown[]) => { + mockCopilotApiCtor(...args) + return { getWorkspace: mockGetWorkspace, getCompanyClients: mockGetCompanyClients } + }), +})) + +jest.mock('./eligibility', () => ({ + getEligibleReminders: (...args: unknown[]) => mockGetEligibleReminders(...args), +})) + +jest.mock('./dispatch-reminder-email', () => ({ + dispatchReminderEmail: { batchTrigger: (...args: unknown[]) => mockBatchTrigger(...args) }, +})) + +jest.mock('bottleneck', () => ({ + __esModule: true, + default: jest.fn().mockImplementation(() => { + let chain: Promise = Promise.resolve() + return { + schedule: (fn: () => Promise) => { + const next = chain.then(() => fn()) + chain = next.catch(() => undefined) + return next + }, + } + }), +})) + +import { sendTaskReminders } from './send-task-reminders' + +type RunResult = { enqueued: number; skipped: number; workspaceCount: number } +const runJob = async (): Promise => { + const { run } = sendTaskReminders as unknown as { run: (payload: { timestamp: Date }) => Promise } + return run({ timestamp: new Date() }) +} + +const workspace = { id: 'ws_1', brandName: 'Acme' } + +const buildRow = (overrides: Partial[1]> = {}) => ({ + taskId: 'task_1', + workspaceId: 'ws_1', + title: 'Submit timesheet', + createdById: 'iu_1', + assigneeId: 'client_1', + assigneeType: AssigneeType.client, + companyId: 'company_1', + reminderType: TaskReminderType.NO_DUE_DATE_3D, + ...overrides, +}) + +describe('sendTaskReminders', () => { + beforeEach(() => { + jest.clearAllMocks() + mockTaskReminderSentCreateManyAndReturn.mockReset() + mockTaskReminderSentDeleteMany.mockReset() + mockGetEligibleReminders.mockReset() + mockBatchTrigger.mockReset() + mockGetWorkspace.mockReset() + mockGetCompanyClients.mockReset() + mockCopilotApiCtor.mockReset() + mockGetWorkspace.mockResolvedValue(workspace) + mockBatchTrigger.mockResolvedValue({ batchId: 'b1' }) + }) + + it('exits cleanly when no rows are eligible', async () => { + mockGetEligibleReminders.mockResolvedValueOnce([]) + + const result = await runJob() + + expect(result).toEqual({ enqueued: 0, skipped: 0, workspaceCount: 0 }) + expect(mockTaskReminderSentCreateManyAndReturn).not.toHaveBeenCalled() + expect(mockBatchTrigger).not.toHaveBeenCalled() + expect(mockCopilotApiCtor).not.toHaveBeenCalled() + }) + + it('filters out internalUser rows before any Copilot work', async () => { + mockGetEligibleReminders.mockResolvedValueOnce([ + buildRow({ assigneeType: AssigneeType.internalUser, assigneeId: 'iu_1', companyId: null }), + ]) + + const result = await runJob() + + expect(result.workspaceCount).toBe(0) + expect(mockBatchTrigger).not.toHaveBeenCalled() + }) + + it('enqueues one dispatch per net-new ledger row (client-assigned)', async () => { + mockGetEligibleReminders.mockResolvedValueOnce([buildRow()]) + mockTaskReminderSentCreateManyAndReturn.mockResolvedValueOnce([ + { id: 'ledger_1', taskId: 'task_1', recipientId: 'client_1', reminderType: TaskReminderType.NO_DUE_DATE_3D }, + ]) + + const result = await runJob() + + expect(result).toEqual({ enqueued: 1, skipped: 0, workspaceCount: 1 }) + expect(mockBatchTrigger).toHaveBeenCalledTimes(1) + const batch = mockBatchTrigger.mock.calls[0][0] + expect(batch).toHaveLength(1) + expect(batch[0].payload).toMatchObject({ + ledgerId: 'ledger_1', + workspaceId: 'ws_1', + task: { id: 'task_1', title: 'Submit timesheet', createdById: 'iu_1' }, + recipientClientId: 'client_1', + recipientCompanyId: 'company_1', + reminderType: TaskReminderType.NO_DUE_DATE_3D, + isCompanyRecipient: false, + workspace, + }) + }) + + it('initializes CopilotAPI with a workspace-scoped apiKey', async () => { + mockGetEligibleReminders.mockResolvedValueOnce([buildRow()]) + mockTaskReminderSentCreateManyAndReturn.mockResolvedValueOnce([ + { id: 'ledger_1', taskId: 'task_1', recipientId: 'client_1', reminderType: TaskReminderType.NO_DUE_DATE_3D }, + ]) + + await runJob() + + expect(mockCopilotApiCtor).toHaveBeenCalledWith('', 'ws_1/test-api-key') + }) + + it('treats ON CONFLICT returning zero rows as fully-skipped (re-run idempotency)', async () => { + mockGetEligibleReminders.mockResolvedValueOnce([buildRow()]) + mockTaskReminderSentCreateManyAndReturn.mockResolvedValueOnce([]) + + const result = await runJob() + + expect(result).toEqual({ enqueued: 0, skipped: 1, workspaceCount: 1 }) + expect(mockBatchTrigger).not.toHaveBeenCalled() + }) + + it('fans out a company-assigned task to one dispatch per current member', async () => { + mockGetEligibleReminders.mockResolvedValueOnce([ + buildRow({ assigneeType: AssigneeType.company, assigneeId: 'company_1', companyId: 'company_1' }), + ]) + mockGetCompanyClients.mockResolvedValueOnce([{ id: 'm_1' }, { id: 'm_2' }, { id: 'm_3' }]) + mockTaskReminderSentCreateManyAndReturn.mockResolvedValueOnce([ + { id: 'l_1', taskId: 'task_1', recipientId: 'm_1', reminderType: TaskReminderType.NO_DUE_DATE_3D }, + { id: 'l_2', taskId: 'task_1', recipientId: 'm_2', reminderType: TaskReminderType.NO_DUE_DATE_3D }, + { id: 'l_3', taskId: 'task_1', recipientId: 'm_3', reminderType: TaskReminderType.NO_DUE_DATE_3D }, + ]) + + const result = await runJob() + + expect(result).toEqual({ enqueued: 3, skipped: 0, workspaceCount: 1 }) + expect(mockBatchTrigger).toHaveBeenCalledTimes(1) + const batch = mockBatchTrigger.mock.calls[0][0] + expect(batch).toHaveLength(3) + expect(batch.map((b: { payload: { recipientClientId: string } }) => b.payload.recipientClientId).sort()).toEqual([ + 'm_1', + 'm_2', + 'm_3', + ]) + expect(batch[0].payload.isCompanyRecipient).toBe(true) + }) + + it('chunks batchTrigger calls so a workspace with >500 fanned-out sends still enqueues', async () => { + // One company task fanning out to 1200 members → 1200 dispatch payloads → 3 chunks of 500. + const members = Array.from({ length: 1200 }, (_, i) => ({ id: `m_${i}` })) + const ledgerRows = members.map((m, i) => ({ + id: `l_${i}`, + taskId: 'task_1', + recipientId: m.id, + reminderType: TaskReminderType.NO_DUE_DATE_3D, + })) + mockGetEligibleReminders.mockResolvedValueOnce([ + buildRow({ assigneeType: AssigneeType.company, assigneeId: 'company_1', companyId: 'company_1' }), + ]) + mockGetCompanyClients.mockResolvedValueOnce(members) + mockTaskReminderSentCreateManyAndReturn.mockResolvedValueOnce(ledgerRows) + + const result = await runJob() + + expect(result).toEqual({ enqueued: 1200, skipped: 0, workspaceCount: 1 }) + expect(mockBatchTrigger).toHaveBeenCalledTimes(3) + expect(mockBatchTrigger.mock.calls[0][0]).toHaveLength(500) + expect(mockBatchTrigger.mock.calls[1][0]).toHaveLength(500) + expect(mockBatchTrigger.mock.calls[2][0]).toHaveLength(200) + expect(mockTaskReminderSentDeleteMany).not.toHaveBeenCalled() + }) + + it('compensates the ledger when a batchTrigger chunk fails', async () => { + const members = Array.from({ length: 800 }, (_, i) => ({ id: `m_${i}` })) + const ledgerRows = members.map((m, i) => ({ + id: `l_${i}`, + taskId: 'task_1', + recipientId: m.id, + reminderType: TaskReminderType.NO_DUE_DATE_3D, + })) + mockGetEligibleReminders.mockResolvedValueOnce([ + buildRow({ assigneeType: AssigneeType.company, assigneeId: 'company_1', companyId: 'company_1' }), + ]) + mockGetCompanyClients.mockResolvedValueOnce(members) + mockTaskReminderSentCreateManyAndReturn.mockResolvedValueOnce(ledgerRows) + // First chunk (500) succeeds, second (300) fails. + mockBatchTrigger.mockResolvedValueOnce({ batchId: 'b1' }).mockRejectedValueOnce(new Error('trigger.dev 5xx')) + + const result = await runJob() + + expect(result.enqueued).toBe(500) + expect(mockTaskReminderSentDeleteMany).toHaveBeenCalledTimes(1) + const deleteArgs = mockTaskReminderSentDeleteMany.mock.calls[0][0] + expect(deleteArgs.where.id.in).toHaveLength(300) // failed chunk's ledger rows + expect(deleteArgs.where.id.in[0]).toBe('l_500') + expect(deleteArgs.where.id.in[299]).toBe('l_799') + }) + + it('skips a single task whose getCompanyClients fails without dropping siblings', async () => { + // Two company tasks in the same workspace. The first one's fan-out throws (Copilot + // exhausted its own retries); the second one should still get its reminder enqueued. + mockGetEligibleReminders.mockResolvedValueOnce([ + buildRow({ + taskId: 'task_bad', + assigneeType: AssigneeType.company, + assigneeId: 'company_bad', + companyId: 'company_bad', + }), + buildRow({ + taskId: 'task_good', + assigneeType: AssigneeType.company, + assigneeId: 'company_good', + companyId: 'company_good', + }), + ]) + mockGetCompanyClients + .mockRejectedValueOnce(new Error('copilot 5xx')) + .mockResolvedValueOnce([{ id: 'm_1' }, { id: 'm_2' }]) + mockTaskReminderSentCreateManyAndReturn.mockResolvedValueOnce([ + { id: 'l_1', taskId: 'task_good', recipientId: 'm_1', reminderType: TaskReminderType.NO_DUE_DATE_3D }, + { id: 'l_2', taskId: 'task_good', recipientId: 'm_2', reminderType: TaskReminderType.NO_DUE_DATE_3D }, + ]) + + const result = await runJob() + + expect(result).toEqual({ enqueued: 2, skipped: 0, workspaceCount: 1 }) + expect(mockBatchTrigger).toHaveBeenCalledTimes(1) + const batch = mockBatchTrigger.mock.calls[0][0] + expect(batch.map((b: { payload: { recipientClientId: string } }) => b.payload.recipientClientId).sort()).toEqual([ + 'm_1', + 'm_2', + ]) + }) + + it('does not abort the sweep when one workspace throws (per-workspace isolation)', async () => { + mockGetEligibleReminders.mockResolvedValueOnce([ + buildRow({ workspaceId: 'ws_bad', taskId: 'task_bad' }), + buildRow({ workspaceId: 'ws_good', taskId: 'task_good', assigneeId: 'client_good' }), + ]) + mockTaskReminderSentCreateManyAndReturn + .mockRejectedValueOnce(new Error('db blew up')) + .mockResolvedValueOnce([ + { id: 'ledger_g', taskId: 'task_good', recipientId: 'client_good', reminderType: TaskReminderType.NO_DUE_DATE_3D }, + ]) + + const result = await runJob() + + expect(result.workspaceCount).toBe(2) + expect(result.enqueued).toBe(1) + }) +}) diff --git a/src/jobs/notifications/send-task-reminders.ts b/src/jobs/notifications/send-task-reminders.ts new file mode 100644 index 000000000..cca3c79e7 --- /dev/null +++ b/src/jobs/notifications/send-task-reminders.ts @@ -0,0 +1,204 @@ +import 'server-only' + +import { copilotAPIKey } from '@/config' +import DBClient from '@/lib/db' +import { CopilotAPI } from '@/utils/CopilotAPI' +import { serializeError } from '@/utils/serializeError' +import { AssigneeType, TaskReminderType } from '@prisma/client' +import { logger, schedules } from '@trigger.dev/sdk/v3' +import Bottleneck from 'bottleneck' + +import { dispatchReminderEmail, DispatchReminderEmailPayload } from './dispatch-reminder-email' +import { EligibilityRow, getEligibleReminders } from './eligibility' + +const WORKSPACE_CONCURRENCY = 5 +// Trigger.dev caps batchTrigger at 500 items per call; chunk so a single workspace with +// thousands of fanned-out sends still gets enqueued. +const BATCH_TRIGGER_CHUNK_SIZE = 500 + +type WorkspaceTotals = { enqueued: number; skipped: number } + +type Recipient = { clientId: string; companyId: string | null } + +type LedgerPlanEntry = { + task: EligibilityRow + recipient: Recipient +} + +export const sendTaskReminders = schedules.task({ + id: 'send-task-reminders', + cron: '0 0 * * *', + maxDuration: 3000, + run: async (payload) => { + const db = DBClient.getInstance() + + const eligibleTasks = await getEligibleReminders(db) + const tasks = eligibleTasks.filter((t) => t.assigneeType !== AssigneeType.internalUser) + + const tasksByWorkspace = new Map() + for (const task of tasks) { + const bucket = tasksByWorkspace.get(task.workspaceId) + if (bucket) bucket.push(task) + else tasksByWorkspace.set(task.workspaceId, [task]) + } + + logger.log('send-task-reminders: sweep starting', { + totalEligible: eligibleTasks.length, + afterIuFilter: tasks.length, + eligibleWorkspaces: tasksByWorkspace.size, + workspaceConcurrency: WORKSPACE_CONCURRENCY, + runAt: payload.timestamp, + }) + + const totals = { enqueued: 0, skipped: 0 } + let processed = 0 + const workspaceCount = tasksByWorkspace.size + + const workspaceBottleneck = new Bottleneck({ maxConcurrent: WORKSPACE_CONCURRENCY }) + + await Promise.allSettled( + Array.from(tasksByWorkspace.entries()).map(([workspaceId, workspaceTasks]) => + workspaceBottleneck.schedule(async () => { + let wsTotals: WorkspaceTotals = { enqueued: 0, skipped: 0 } + try { + wsTotals = await processWorkspace(db, workspaceId, workspaceTasks) + } catch (err) { + logger.error('send-task-reminders: workspace failed', { + workspaceId, + error: serializeError(err), + }) + } finally { + totals.enqueued += wsTotals.enqueued + totals.skipped += wsTotals.skipped + processed += 1 + logger.log( + `[${processed}/${workspaceCount}] workspace ${workspaceId}: enqueued ${wsTotals.enqueued}, skipped ${wsTotals.skipped}`, + { workspaceId, ...wsTotals, processed, eligibleWorkspaces: workspaceCount }, + ) + } + }), + ), + ) + + logger.log('send-task-reminders: sweep complete', { + ...totals, + workspaceCount, + totalEligible: eligibleTasks.length, + }) + + return { ...totals, workspaceCount } + }, +}) + +const processWorkspace = async ( + db: ReturnType, + workspaceId: string, + tasks: EligibilityRow[], +): Promise => { + // Workspace-scoped apiKey: empty token + `${workspaceId}/${apiKey}` is accepted by the + // SDK when COPILOT_ENV is set on the Trigger.dev runtime. + const copilot = new CopilotAPI('', `${workspaceId}/${copilotAPIKey}`) + const workspace = await copilot.getWorkspace() + + const plan: LedgerPlanEntry[] = [] + for (const task of tasks) { + let recipients: Recipient[] + try { + recipients = await resolveRecipients(copilot, task) + } catch (err) { + // Contain blast radius to this task. Copilot is already wrapped in withRetry, so + // a thrown error means retries are exhausted — propagating would drop unrelated + // sibling tasks in the same workspace for the day. + logger.error('send-task-reminders: failed to resolve recipients, skipping task', { + workspaceId, + taskId: task.taskId, + assigneeType: task.assigneeType, + assigneeId: task.assigneeId, + error: serializeError(err), + }) + continue + } + for (const recipient of recipients) { + plan.push({ task, recipient }) + } + } + + if (plan.length === 0) return { enqueued: 0, skipped: 0 } + + // Ledger insert before send: the unique constraint is the dedupe primitive. + const inserted = await db.taskReminderSent.createManyAndReturn({ + data: plan.map((entry) => ({ + taskId: entry.task.taskId, + workspaceId, + recipientId: entry.recipient.clientId, + reminderType: entry.task.reminderType, + })), + skipDuplicates: true, + }) + + const insertedKey = (taskId: string, recipientId: string, type: TaskReminderType) => `${taskId}|${recipientId}|${type}` + const planByKey = new Map( + plan.map((e) => [insertedKey(e.task.taskId, e.recipient.clientId, e.task.reminderType), e]), + ) + + const triggers: { payload: DispatchReminderEmailPayload }[] = [] + for (const row of inserted) { + const entry = planByKey.get(insertedKey(row.taskId, row.recipientId, row.reminderType)) + if (!entry) continue + triggers.push({ + payload: { + ledgerId: row.id, + workspaceId, + task: { id: entry.task.taskId, title: entry.task.title, createdById: entry.task.createdById }, + recipientClientId: entry.recipient.clientId, + recipientCompanyId: entry.recipient.companyId, + reminderType: entry.task.reminderType, + isCompanyRecipient: entry.task.assigneeType === AssigneeType.company, + workspace, + }, + }) + } + + // Returns the number actually enqueued. On failure, drops the chunk's ledger rows + // so the next cron run can retry — without this, the unique constraint blocks any + // future insert but no dispatcher exists to consume them. + const dispatchChunk = async (chunk: { payload: DispatchReminderEmailPayload }[]): Promise => { + try { + await dispatchReminderEmail.batchTrigger(chunk) + return chunk.length + } catch (err) { + const ledgerIds = chunk.map((t) => t.payload.ledgerId) + logger.error('send-task-reminders: batchTrigger failed, compensating ledger', { + workspaceId, + chunkSize: chunk.length, + error: serializeError(err), + }) + try { + await db.taskReminderSent.deleteMany({ where: { id: { in: ledgerIds } } }) + } catch (deleteErr) { + logger.error('send-task-reminders: ledger compensation deleteMany failed, ledger rows orphaned', { + workspaceId, + ledgerIds, + error: serializeError(deleteErr), + }) + } + return 0 + } + } + + let enqueued = 0 + for (let i = 0; i < triggers.length; i += BATCH_TRIGGER_CHUNK_SIZE) { + enqueued += await dispatchChunk(triggers.slice(i, i + BATCH_TRIGGER_CHUNK_SIZE)) + } + + return { enqueued, skipped: plan.length - inserted.length } +} + +// IU rows are filtered upstream so only client/company assignees reach here. +const resolveRecipients = async (copilot: CopilotAPI, task: EligibilityRow): Promise => { + if (task.assigneeType !== AssigneeType.company) { + return [{ clientId: task.assigneeId, companyId: task.companyId }] + } + const members = await copilot.getCompanyClients(task.assigneeId) + return members.map((m) => ({ clientId: m.id, companyId: task.assigneeId })) +} diff --git a/src/utils/serializeError.ts b/src/utils/serializeError.ts new file mode 100644 index 000000000..1bcfff086 --- /dev/null +++ b/src/utils/serializeError.ts @@ -0,0 +1,2 @@ +// JS can throw anything; this turns the unknown into something safe to log. +export const serializeError = (err: unknown) => (err instanceof Error ? { message: err.message, stack: err.stack } : err)