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) } } 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 e05b33a31..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 () => { @@ -348,9 +358,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..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,30 +323,41 @@ 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 } }) - 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..e37a00d1e --- /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.mockResolvedValueOnce(3).mockResolvedValueOnce(1) + }) + + 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, 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 and releases stale claims without enqueueing when no window is stale', async () => { + mockQueryRaw.mockResolvedValue([]) + + 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 new file mode 100644 index 000000000..387662088 --- /dev/null +++ b/src/jobs/notifications/sweep-grouped-email-windows.ts @@ -0,0 +1,42 @@ +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'` + + // 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` + 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, released, pruned }) + return { requeued: stale.length, released, pruned } +} + +export const sweepGroupedEmailWindows = schedules.task({ + id: 'sweep-grouped-email-windows', + cron: '0 * * * *', + maxDuration: 300, + run: sweepGroupedEmailWindowsRun, +})