From 8f63a1befc46ab2ef1a9353b16d7d5689a33cddf Mon Sep 17 00:00:00 2001 From: Prios Shrestha <30313649+priosshrsth@users.noreply.github.com> Date: Fri, 7 Aug 2026 10:33:34 +0545 Subject: [PATCH 1/5] OUT-4027 | Fix 500 when deleting a task whose label row is missing (#1397) deleteLabel passed `id: currentLabel?.id` straight into label.delete, so when findFirst matched nothing Prisma got `{ id: undefined }` and threw PrismaClientValidationError, failing the whole delete transaction. Return early instead. --- .../label-mapping.service.test.ts | 44 +++++++++++++++++++ .../label-mapping/label-mapping.service.ts | 3 +- 2 files changed, 46 insertions(+), 1 deletion(-) create mode 100644 src/app/api/label-mapping/label-mapping.service.test.ts diff --git a/src/app/api/label-mapping/label-mapping.service.test.ts b/src/app/api/label-mapping/label-mapping.service.test.ts new file mode 100644 index 000000000..24a195ede --- /dev/null +++ b/src/app/api/label-mapping/label-mapping.service.test.ts @@ -0,0 +1,44 @@ +const mockLabelFindFirst = jest.fn() +const mockLabelDelete = jest.fn() + +jest.mock('@/lib/db', () => ({ + __esModule: true, + default: { + getInstance: () => ({ + label: { findFirst: mockLabelFindFirst, delete: mockLabelDelete }, + }), + }, +})) + +jest.mock('@/utils/CopilotAPI', () => ({ CopilotAPI: jest.fn() })) + +import { LabelMappingService } from '@api/label-mapping/label-mapping.service' +import User from '@api/core/models/User.model' +import { UserRole } from '@api/core/types/user' + +const user = { + workspaceId: 'ws-1', + role: UserRole.IU, + internalUserId: 'iu-1', + token: 'token', +} as unknown as User + +describe('LabelMappingService#deleteLabel', () => { + beforeEach(() => jest.clearAllMocks()) + + it('deletes the matching label row', async () => { + mockLabelFindFirst.mockResolvedValue({ id: 'label-1' }) + + await new LabelMappingService(user).deleteLabel('ASS10-009') + + expect(mockLabelDelete).toHaveBeenCalledWith({ where: { id: 'label-1' } }) + }) + + it('no-ops when the label row is already gone', async () => { + mockLabelFindFirst.mockResolvedValue(null) + + await new LabelMappingService(user).deleteLabel('ASS10-009') + + expect(mockLabelDelete).not.toHaveBeenCalled() + }) +}) diff --git a/src/app/api/label-mapping/label-mapping.service.ts b/src/app/api/label-mapping/label-mapping.service.ts index 42007a03c..b0d9c2abd 100644 --- a/src/app/api/label-mapping/label-mapping.service.ts +++ b/src/app/api/label-mapping/label-mapping.service.ts @@ -175,9 +175,10 @@ export class LabelMappingService extends BaseService { label, }, }) + if (!currentLabel) return await this.db.label.delete({ where: { - id: currentLabel?.id, + id: currentLabel.id, }, }) } From 1327cf7f3da758f8cca45c4378021e5a8e7edcd0 Mon Sep 17 00:00:00 2001 From: Prios Shrestha <30313649+priosshrsth@users.noreply.github.com> Date: Mon, 24 Aug 2026 16:46:01 +0545 Subject: [PATCH 2/5] OUT-4093 | Stop reading localStorage in the assignee cache (#1434) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * OUT-4093 | Remove the localStorage→localForage assignee migration --- src/app/_cache/AssigneeCacheGetter.tsx | 3 +- src/app/_cache/forageStorage.test.ts | 45 ++++++++++++++++++++++++++ src/app/_cache/forageStorage.ts | 35 ++------------------ 3 files changed, 49 insertions(+), 34 deletions(-) create mode 100644 src/app/_cache/forageStorage.test.ts diff --git a/src/app/_cache/AssigneeCacheGetter.tsx b/src/app/_cache/AssigneeCacheGetter.tsx index ccb7cda7b..1d665706f 100644 --- a/src/app/_cache/AssigneeCacheGetter.tsx +++ b/src/app/_cache/AssigneeCacheGetter.tsx @@ -3,7 +3,7 @@ import { setAssigneeList } from '@/redux/features/taskBoardSlice' import store from '@/redux/store' import { useEffect } from 'react' -import { getAssignees, migrateAssignees } from '@/app/_cache/forageStorage' +import { getAssignees } from '@/app/_cache/forageStorage' interface ClientAssigneeCacheGetterProps { lookupKey: string @@ -12,7 +12,6 @@ interface ClientAssigneeCacheGetterProps { export const AssigneeCacheGetter = ({ lookupKey }: ClientAssigneeCacheGetterProps) => { useEffect(() => { const run = async () => { - await migrateAssignees(lookupKey) //migrate from localStorage to localForage if required. Remember to remove this after a while. const assignee = await getAssignees(lookupKey) if (assignee.length) { store.dispatch(setAssigneeList(assignee)) diff --git a/src/app/_cache/forageStorage.test.ts b/src/app/_cache/forageStorage.test.ts new file mode 100644 index 000000000..dbfe00100 --- /dev/null +++ b/src/app/_cache/forageStorage.test.ts @@ -0,0 +1,45 @@ +const mockGetItem = jest.fn() +const mockSetItem = jest.fn() + +jest.mock('localforage', () => ({ + __esModule: true, + default: { + config: jest.fn(), + getItem: (...args: unknown[]) => mockGetItem(...args), + setItem: (...args: unknown[]) => mockSetItem(...args), + }, +})) + +import { getAssignees, setAssignees } from '@/app/_cache/forageStorage' + +const denied = () => new DOMException('Access is denied for this document.', 'SecurityError') + +describe('assignee cache', () => { + beforeAll(() => { + globalThis.window = {} as Window & typeof globalThis + }) + afterAll(() => { + delete (globalThis as { window?: unknown }).window + }) + beforeEach(() => jest.clearAllMocks()) + + it('returns an empty list when storage access is denied', async () => { + mockGetItem.mockRejectedValue(denied()) + + await expect(getAssignees('lookup-key')).resolves.toEqual([]) + expect(mockGetItem).toHaveBeenCalledWith('assignees.lookup-key') + }) + + it('swallows write failures when storage access is denied', async () => { + mockSetItem.mockRejectedValue(denied()) + + await expect(setAssignees('lookup-key', [])).resolves.toBeUndefined() + expect(mockSetItem).toHaveBeenCalledWith('assignees.lookup-key', []) + }) + + it('returns an empty list when nothing is cached', async () => { + mockGetItem.mockResolvedValue(null) + + await expect(getAssignees('lookup-key')).resolves.toEqual([]) + }) +}) diff --git a/src/app/_cache/forageStorage.ts b/src/app/_cache/forageStorage.ts index e8baf6391..e4c23abe0 100644 --- a/src/app/_cache/forageStorage.ts +++ b/src/app/_cache/forageStorage.ts @@ -8,52 +8,23 @@ localforage.config({ storeName: 'assignees', }) -export async function migrateAssignees(lookupKey: string) { - const lKey = `assignees.${lookupKey}` - const existing = localStorage.getItem(lKey) - - if (existing) { - try { - const parsed = JSON.parse(existing) - await localforage.setItem(lKey, parsed) - localStorage.removeItem(lKey) - } catch (err) { - console.error('Migration failed', err) - } - } -} //a utility function to migrate existing assignee data from localStorage to localForage - export async function getAssignees(lookupKey: string): Promise { if (typeof window === 'undefined') return [] try { - if (!(await document.hasStorageAccess())) { - console.info('Browswer has no storage access') - await document.requestStorageAccess() - } - return (await localforage.getItem(`assignees.${lookupKey}`)) ?? [] } catch (error: unknown) { - console.error( - "Storage access not granted. Under Chrome's Settings > Privacy and Security, make sure 'Third-party cookies' is allowed.", - ) + console.info('Assignee cache unavailable, falling back to network', error) return [] } } -export async function setAssignees(lookupKey: string, value: any) { +export async function setAssignees(lookupKey: string, value: IAssigneeCombined[]) { if (typeof window === 'undefined') return try { - if (!(await document.hasStorageAccess())) { - console.info('Browswer has no storage access') - await document.requestStorageAccess() - } - return await localforage.setItem(`assignees.${lookupKey}`, value) } catch (error: unknown) { - console.error( - "Storage access not granted. Under Chrome's Settings > Privacy and Security, make sure 'Third-party cookies' is allowed.", - ) + console.info('Assignee cache write skipped, storage unavailable', error) } } From 1c096b7121018cccfb1d57605287955f2bf737db Mon Sep 17 00:00:00 2001 From: priosshrsth Date: Wed, 26 Aug 2026 03:39:55 +0000 Subject: [PATCH 3/5] OUT-4105 | Sweep stale grouped-email windows instead of deleting them on failure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When the Supabase pooler went unreachable on 8/24 every flush window burned its ~3s retry budget on the first $queryRaw, and onFailure then DELETEd the whole window — unsent rows included — so those grouped emails were destroyed, or orphaned when the delete failed too. Nothing ever re-flushed them. onFailure now leaves the rows alone, and an hourly sweeper re-enqueues any window still unsent after 30 minutes. Rows are pruned after 15 days rather than at the first sign of trouble. The flush run is already idempotent, so a re-enqueue only sends what is still outstanding. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Y4tF84tzJW7Bo1DehsDX22 --- .../notifications/flush-grouped-email.test.ts | 4 +- src/jobs/notifications/flush-grouped-email.ts | 14 +---- src/jobs/notifications/index.ts | 1 + .../sweep-grouped-email-windows.test.ts | 52 +++++++++++++++++++ .../sweep-grouped-email-windows.ts | 36 +++++++++++++ 5 files changed, 93 insertions(+), 14 deletions(-) create mode 100644 src/jobs/notifications/sweep-grouped-email-windows.test.ts create mode 100644 src/jobs/notifications/sweep-grouped-email-windows.ts diff --git a/src/jobs/notifications/flush-grouped-email.test.ts b/src/jobs/notifications/flush-grouped-email.test.ts index e05b33a31..a520f79cb 100644 --- a/src/jobs/notifications/flush-grouped-email.test.ts +++ b/src/jobs/notifications/flush-grouped-email.test.ts @@ -348,9 +348,9 @@ describe('flushGroupedEmailOnFailure', () => { }) }) - it('deletes all rows for the window so orphaned events do not accumulate', async () => { + it('leaves unsent rows in place for the sweeper to re-enqueue', async () => { await flushGroupedEmailOnFailure({ payload, error: new Error('terminal') }) - expect(mockExecuteRaw).toHaveBeenCalledTimes(1) + expect(mockExecuteRaw).not.toHaveBeenCalled() }) }) diff --git a/src/jobs/notifications/flush-grouped-email.ts b/src/jobs/notifications/flush-grouped-email.ts index b40a69784..39001f6c1 100644 --- a/src/jobs/notifications/flush-grouped-email.ts +++ b/src/jobs/notifications/flush-grouped-email.ts @@ -321,27 +321,17 @@ export const flushGroupedEmailRun = async (payload: FlushGroupedEmailPayload) => export const flushGroupedEmailOnFailure = async ({ payload, error }: { payload: unknown; error: unknown }) => { const { workspaceId, windowKey } = payload as FlushGroupedEmailPayload Sentry.captureException(error, { tags: { job: TASK_ID, workspaceId, windowKey } }) - logger.error('flush-grouped-email: retries exhausted, cleaning up window', { + logger.error('flush-grouped-email: retries exhausted, leaving the window for the sweeper', { workspaceId, windowKey, error: serializeError(error), }) - const db = DBClient.getInstance() - try { - await db.$executeRaw`DELETE FROM "GroupedEmailEvents" WHERE "windowKey" = ${windowKey}` - } catch (deleteErr) { - logger.error('flush-grouped-email: window cleanup on failure failed, rows are orphaned', { - workspaceId, - windowKey, - error: serializeError(deleteErr), - }) - } } export const flushGroupedEmail = task({ id: TASK_ID, queue: { concurrencyLimit: 5 }, - retry: { maxAttempts: 3, factor: 2, minTimeoutInMs: 1_000, maxTimeoutInMs: 15_000, randomize: true }, + retry: { maxAttempts: 5, factor: 2, minTimeoutInMs: 5_000, maxTimeoutInMs: 60_000, randomize: true }, maxDuration: 60, run: flushGroupedEmailRun, }) diff --git a/src/jobs/notifications/index.ts b/src/jobs/notifications/index.ts index 80161fbd6..ab2880125 100644 --- a/src/jobs/notifications/index.ts +++ b/src/jobs/notifications/index.ts @@ -6,3 +6,4 @@ export { sendTaskReminders } from './send-task-reminders' export { dispatchReminderEmail } from './dispatch-reminder-email' export { dispatchGroupedReminderEmail } from './dispatch-grouped-reminder-email' export { flushGroupedEmail } from './flush-grouped-email' +export { sweepGroupedEmailWindows } from './sweep-grouped-email-windows' diff --git a/src/jobs/notifications/sweep-grouped-email-windows.test.ts b/src/jobs/notifications/sweep-grouped-email-windows.test.ts new file mode 100644 index 000000000..bcb9e0177 --- /dev/null +++ b/src/jobs/notifications/sweep-grouped-email-windows.test.ts @@ -0,0 +1,52 @@ +const mockQueryRaw = jest.fn() +const mockExecuteRaw = jest.fn() +const mockEnqueue = jest.fn() + +jest.mock('@trigger.dev/sdk/v3', () => ({ + schedules: { task: ({ run }: { run: () => unknown }) => ({ run }) }, + logger: { log: jest.fn(), error: jest.fn(), warn: jest.fn() }, +})) + +jest.mock('./flush-grouped-email', () => ({ + enqueueGroupedEmailFlush: (...args: unknown[]) => mockEnqueue(...args), +})) + +jest.mock('@/lib/db', () => ({ + __esModule: true, + default: { + getInstance: () => ({ + $queryRaw: (...args: unknown[]) => mockQueryRaw(...args), + $executeRaw: (...args: unknown[]) => mockExecuteRaw(...args), + }), + }, +})) + +import { sweepGroupedEmailWindowsRun } from './sweep-grouped-email-windows' + +describe('sweepGroupedEmailWindows', () => { + beforeEach(() => { + jest.clearAllMocks() + mockExecuteRaw.mockResolvedValue(3) + }) + + it('re-enqueues every stale window and reports what it pruned', async () => { + mockQueryRaw.mockResolvedValue([ + { workspaceId: 'ws_1', windowKey: 'client_1:win_1' }, + { workspaceId: 'ws_2', windowKey: 'iu_1:iu:win_2' }, + ]) + + await expect(sweepGroupedEmailWindowsRun()).resolves.toEqual({ requeued: 2, pruned: 3 }) + expect(mockEnqueue.mock.calls.map(([payload]) => payload)).toEqual([ + { workspaceId: 'ws_1', windowKey: 'client_1:win_1' }, + { workspaceId: 'ws_2', windowKey: 'iu_1:iu:win_2' }, + ]) + }) + + it('prunes without enqueueing anything when no window is stale', async () => { + mockQueryRaw.mockResolvedValue([]) + + await expect(sweepGroupedEmailWindowsRun()).resolves.toEqual({ requeued: 0, pruned: 3 }) + expect(mockExecuteRaw).toHaveBeenCalledTimes(1) + expect(mockEnqueue).not.toHaveBeenCalled() + }) +}) diff --git a/src/jobs/notifications/sweep-grouped-email-windows.ts b/src/jobs/notifications/sweep-grouped-email-windows.ts new file mode 100644 index 000000000..23df4221b --- /dev/null +++ b/src/jobs/notifications/sweep-grouped-email-windows.ts @@ -0,0 +1,36 @@ +import 'server-only' + +import DBClient from '@/lib/db' +import { logger, schedules } from '@trigger.dev/sdk/v3' + +import { enqueueGroupedEmailFlush } from './flush-grouped-email' + +type StaleWindow = { workspaceId: string; windowKey: string } + +export const sweepGroupedEmailWindowsRun = async () => { + const db = DBClient.getInstance() + + const pruned = await db.$executeRaw`DELETE FROM "GroupedEmailEvents" WHERE "createdAt" < now() - interval '15 days'` + + // Past 24h a window has already been re-enqueued ~24 times; leave it for the prune rather + // than retrying a poisoned window hourly for a fortnight. + const stale = await db.$queryRaw` + SELECT DISTINCT "workspaceId", "windowKey" FROM "GroupedEmailEvents" + WHERE "sentAt" IS NULL + AND "createdAt" < now() - interval '30 minutes' + AND "createdAt" > now() - interval '24 hours'` + + for (const { workspaceId, windowKey } of stale) { + await enqueueGroupedEmailFlush({ workspaceId, windowKey }) + } + + logger.log('sweep-grouped-email-windows: done', { requeued: stale.length, pruned }) + return { requeued: stale.length, pruned } +} + +export const sweepGroupedEmailWindows = schedules.task({ + id: 'sweep-grouped-email-windows', + cron: '0 * * * *', + maxDuration: 300, + run: sweepGroupedEmailWindowsRun, +}) From ac51552f4c33691ec8e022be839f66e0fb10d246 Mon Sep 17 00:00:00 2001 From: priosshrsth Date: Wed, 26 Aug 2026 05:18:23 +0000 Subject: [PATCH 4/5] OUT-4105 | Note the sweeper's at-least-once delivery contract Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Y4tF84tzJW7Bo1DehsDX22 --- src/jobs/notifications/sweep-grouped-email-windows.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/src/jobs/notifications/sweep-grouped-email-windows.ts b/src/jobs/notifications/sweep-grouped-email-windows.ts index 23df4221b..fd71c426e 100644 --- a/src/jobs/notifications/sweep-grouped-email-windows.ts +++ b/src/jobs/notifications/sweep-grouped-email-windows.ts @@ -20,6 +20,7 @@ export const sweepGroupedEmailWindowsRun = async () => { AND "createdAt" < now() - interval '30 minutes' AND "createdAt" > now() - interval '24 hours'` + // At-least-once by design: duplicate notification mail beats silently dropping it. for (const { workspaceId, windowKey } of stale) { await enqueueGroupedEmailFlush({ workspaceId, windowKey }) } From f65a62ff0c88432d26b62a5635e7579f3f38d869 Mon Sep 17 00:00:00 2001 From: priosshrsth Date: Wed, 26 Aug 2026 05:43:16 +0000 Subject: [PATCH 5/5] OUT-4105 | Claim grouped-email windows atomically so a window can never send twice MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The sweeper can enqueue a window whose original flush is still pending, and two runs reading "sentAt" IS NULL would both send. Replace the read with an UPDATE ... RETURNING that claims the unsent rows via the existing batchId column: the losing run matches nothing and no-ops. A failed attempt releases its claim so the next Trigger.dev retry re-sends, and the sweeper releases any claim older than 30 minutes — well past the job's 60s maxDuration, so a release can never race a live send. Covered by an integration test that runs two flushes concurrently against real Postgres; it fails with two sends if the claim is reduced to a plain SELECT. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Y4tF84tzJW7Bo1DehsDX22 --- .../flush-grouped-email.integration.test.ts | 26 +++++++++ .../notifications/flush-grouped-email.test.ts | 12 ++++- src/jobs/notifications/flush-grouped-email.ts | 54 ++++++++++++++----- .../sweep-grouped-email-windows.test.ts | 10 ++-- .../sweep-grouped-email-windows.ts | 11 ++-- 5 files changed, 90 insertions(+), 23 deletions(-) diff --git a/src/jobs/notifications/flush-grouped-email.integration.test.ts b/src/jobs/notifications/flush-grouped-email.integration.test.ts index 2cd83e1ad..e5330d2e4 100644 --- a/src/jobs/notifications/flush-grouped-email.integration.test.ts +++ b/src/jobs/notifications/flush-grouped-email.integration.test.ts @@ -123,6 +123,32 @@ describe('flush-grouped-email idempotency (real DB)', () => { expect(await totalCount(window)).toBe(0) }) + it('sends once when two runs race the same window', async () => { + const window = 'win_concurrent' + const taskA = await seedTask({ workspaceId: WS, assigneeId: CLIENT_A, assigneeType: 'client', companyId: COMPANY }) + const taskB = await seedTask({ workspaceId: WS, assigneeId: CLIENT_A, assigneeType: 'client', companyId: COMPANY }) + await seedEvent({ + windowKey: window, + taskId: taskA, + recipientClientId: CLIENT_A, + eventType: GroupedEmailEventType.ASSIGNED, + }) + await seedEvent({ + windowKey: window, + taskId: taskB, + recipientClientId: CLIENT_A, + eventType: GroupedEmailEventType.SHARED, + }) + + await Promise.all([ + flushGroupedEmailRun({ workspaceId: WS, windowKey: window }), + flushGroupedEmailRun({ workspaceId: WS, windowKey: window }), + ]) + + expect(mockCreateNotification).toHaveBeenCalledTimes(1) + expect(await totalCount(window)).toBe(0) + }) + it('is idempotent: re-flushing a fully-sent window is a no-op', async () => { const window = 'win_idempotent' const taskA = await seedTask({ workspaceId: WS, assigneeId: CLIENT_A, assigneeType: 'client', companyId: COMPANY }) diff --git a/src/jobs/notifications/flush-grouped-email.test.ts b/src/jobs/notifications/flush-grouped-email.test.ts index a520f79cb..36b74f594 100644 --- a/src/jobs/notifications/flush-grouped-email.test.ts +++ b/src/jobs/notifications/flush-grouped-email.test.ts @@ -304,7 +304,17 @@ describe('flushGroupedEmailRun', () => { mockCreateNotification.mockRejectedValue(new Error('copilot 5xx')) await expect(flushGroupedEmailRun(payload)).rejects.toThrow('copilot 5xx') - expect(mockExecuteRaw).not.toHaveBeenCalled() + const statements = mockExecuteRaw.mock.calls.map(([sql]) => (sql as string[]).join('')) + expect(statements.some((sql) => sql.includes('"sentAt" = now()'))).toBe(false) + expect(statements.some((sql) => sql.includes('"batchId" = NULL'))).toBe(true) + }) + + it('no-ops when a concurrent run already claimed the window', async () => { + mockQueryRaw.mockResolvedValue([]) + + await expect(flushGroupedEmailRun(payload)).resolves.toMatchObject({ sent: 0, skipped: true }) + expect(mockSendGroupedEmail).not.toHaveBeenCalled() + expect(mockCreateNotification).not.toHaveBeenCalled() }) it('throws when a grouped window has no buffered sender and no workspace internal user', async () => { diff --git a/src/jobs/notifications/flush-grouped-email.ts b/src/jobs/notifications/flush-grouped-email.ts index 39001f6c1..926720aea 100644 --- a/src/jobs/notifications/flush-grouped-email.ts +++ b/src/jobs/notifications/flush-grouped-email.ts @@ -41,11 +41,25 @@ type IuRecipientGroup = { const TASK_ID = 'flush-grouped-email' -const readUnsentWindowEvents = (db: ReturnType, windowKey: string) => +type WindowClaim = { db: ReturnType; windowKey: string; batchId: string } + +type ClaimedWindow = { + payload: FlushGroupedEmailPayload + db: ReturnType + batchId: string + rows: BufferedRow[] +} + +const claimUnsentWindowEvents = ({ db, windowKey, batchId }: WindowClaim) => db.$queryRaw` - SELECT "eventType", "taskId", "taskTitleSnapshot", "createdAt", "recipientClientId", "recipientCompanyId", "recipientIuId", "individualEmail" - FROM "GroupedEmailEvents" - WHERE "windowKey" = ${windowKey} AND "sentAt" IS NULL` + UPDATE "GroupedEmailEvents" SET "batchId" = ${batchId}::uuid + WHERE "windowKey" = ${windowKey} AND "sentAt" IS NULL AND "batchId" IS NULL + RETURNING "eventType", "taskId", "taskTitleSnapshot", "createdAt", "recipientClientId", "recipientCompanyId", "recipientIuId", "individualEmail"` + +const releaseClaim = ({ db, windowKey, batchId }: WindowClaim) => + db.$executeRaw` + UPDATE "GroupedEmailEvents" SET "batchId" = NULL + WHERE "windowKey" = ${windowKey} AND "batchId" = ${batchId}::uuid AND "sentAt" IS NULL` const deleteWindowRows = (db: ReturnType, windowKey: string) => db.$executeRaw`DELETE FROM "GroupedEmailEvents" WHERE "windowKey" = ${windowKey} AND "sentAt" IS NOT NULL` @@ -178,17 +192,8 @@ const groupIuRecipients = (rows: BufferedRow[]): IuRecipientGroup[] => { return [...groups.values()] } -export const flushGroupedEmailRun = async (payload: FlushGroupedEmailPayload) => { +const dispatchClaimedWindow = async ({ payload, db, batchId, rows }: ClaimedWindow) => { const { workspaceId, windowKey } = payload - const db = DBClient.getInstance() - const batchId = randomUUID() - - const rows = await readUnsentWindowEvents(db, windowKey) - if (rows.length === 0) { - logger.log('flush-grouped-email: nothing to send', { workspaceId, windowKey }) - return { windowKey, recipients: 0, sent: 0, skipped: true as const } - } - const copilot = new CopilotAPI('', `${workspaceId}/${copilotAPIKey}`) const liveTaskIds = await getLiveTaskIds(db, [...new Set(rows.map((r) => r.taskId))]) @@ -318,6 +323,27 @@ export const flushGroupedEmailRun = async (payload: FlushGroupedEmailPayload) => return { windowKey, recipients: cuGroups.length + iuGroups.length, sent, sentGrouped, sentIndividual } } +export const flushGroupedEmailRun = async (payload: FlushGroupedEmailPayload) => { + const { workspaceId, windowKey } = payload + const db = DBClient.getInstance() + const batchId = randomUUID() + + // Atomic claim: a concurrent run on the same window matches no rows and no-ops, so a + // window can never be sent twice. Claims outliving their run are released by the sweeper. + const rows = await claimUnsentWindowEvents({ db, windowKey, batchId }) + if (rows.length === 0) { + logger.log('flush-grouped-email: nothing to send', { workspaceId, windowKey }) + return { windowKey, recipients: 0, sent: 0, skipped: true as const } + } + + try { + return await dispatchClaimedWindow({ payload, db, batchId, rows }) + } catch (err) { + await releaseClaim({ db, windowKey, batchId }) + throw err + } +} + export const flushGroupedEmailOnFailure = async ({ payload, error }: { payload: unknown; error: unknown }) => { const { workspaceId, windowKey } = payload as FlushGroupedEmailPayload Sentry.captureException(error, { tags: { job: TASK_ID, workspaceId, windowKey } }) diff --git a/src/jobs/notifications/sweep-grouped-email-windows.test.ts b/src/jobs/notifications/sweep-grouped-email-windows.test.ts index bcb9e0177..e37a00d1e 100644 --- a/src/jobs/notifications/sweep-grouped-email-windows.test.ts +++ b/src/jobs/notifications/sweep-grouped-email-windows.test.ts @@ -26,7 +26,7 @@ import { sweepGroupedEmailWindowsRun } from './sweep-grouped-email-windows' describe('sweepGroupedEmailWindows', () => { beforeEach(() => { jest.clearAllMocks() - mockExecuteRaw.mockResolvedValue(3) + mockExecuteRaw.mockResolvedValueOnce(3).mockResolvedValueOnce(1) }) it('re-enqueues every stale window and reports what it pruned', async () => { @@ -35,18 +35,18 @@ describe('sweepGroupedEmailWindows', () => { { workspaceId: 'ws_2', windowKey: 'iu_1:iu:win_2' }, ]) - await expect(sweepGroupedEmailWindowsRun()).resolves.toEqual({ requeued: 2, pruned: 3 }) + await expect(sweepGroupedEmailWindowsRun()).resolves.toEqual({ requeued: 2, released: 1, pruned: 3 }) expect(mockEnqueue.mock.calls.map(([payload]) => payload)).toEqual([ { workspaceId: 'ws_1', windowKey: 'client_1:win_1' }, { workspaceId: 'ws_2', windowKey: 'iu_1:iu:win_2' }, ]) }) - it('prunes without enqueueing anything when no window is stale', async () => { + it('prunes and releases stale claims without enqueueing when no window is stale', async () => { mockQueryRaw.mockResolvedValue([]) - await expect(sweepGroupedEmailWindowsRun()).resolves.toEqual({ requeued: 0, pruned: 3 }) - expect(mockExecuteRaw).toHaveBeenCalledTimes(1) + await expect(sweepGroupedEmailWindowsRun()).resolves.toEqual({ requeued: 0, released: 1, pruned: 3 }) + expect(mockExecuteRaw).toHaveBeenCalledTimes(2) expect(mockEnqueue).not.toHaveBeenCalled() }) }) diff --git a/src/jobs/notifications/sweep-grouped-email-windows.ts b/src/jobs/notifications/sweep-grouped-email-windows.ts index fd71c426e..387662088 100644 --- a/src/jobs/notifications/sweep-grouped-email-windows.ts +++ b/src/jobs/notifications/sweep-grouped-email-windows.ts @@ -12,6 +12,12 @@ export const sweepGroupedEmailWindowsRun = async () => { const pruned = await db.$executeRaw`DELETE FROM "GroupedEmailEvents" WHERE "createdAt" < now() - interval '15 days'` + // Safe because a flush run is capped at maxDuration 60s, so a claim this old cannot belong + // to a live run — releasing it can never race a send. + const released = await db.$executeRaw` + UPDATE "GroupedEmailEvents" SET "batchId" = NULL + WHERE "sentAt" IS NULL AND "batchId" IS NOT NULL AND "createdAt" < now() - interval '30 minutes'` + // Past 24h a window has already been re-enqueued ~24 times; leave it for the prune rather // than retrying a poisoned window hourly for a fortnight. const stale = await db.$queryRaw` @@ -20,13 +26,12 @@ export const sweepGroupedEmailWindowsRun = async () => { AND "createdAt" < now() - interval '30 minutes' AND "createdAt" > now() - interval '24 hours'` - // At-least-once by design: duplicate notification mail beats silently dropping it. for (const { workspaceId, windowKey } of stale) { await enqueueGroupedEmailFlush({ workspaceId, windowKey }) } - logger.log('sweep-grouped-email-windows: done', { requeued: stale.length, pruned }) - return { requeued: stale.length, pruned } + logger.log('sweep-grouped-email-windows: done', { requeued: stale.length, released, pruned }) + return { requeued: stale.length, released, pruned } } export const sweepGroupedEmailWindows = schedules.task({