OUT-4105 | Sweep stale grouped-email windows instead of deleting them on failure

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))

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([])
+  })
+})

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<IAssigneeCombined[]> {
   if (typeof window === 'undefined') return []
 
   try {
-    if (!(await document.hasStorageAccess())) {
-      console.info('Browswer has no storage access')
-      await document.requestStorageAccess()
-    }
-
     return (await localforage.getItem<IAssigneeCombined[]>(`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)
   }
 }

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 })

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()
   })
 })

src/jobs/notifications/flush-grouped-email.ts

@@ -41,11 +41,25 @@ type IuRecipientGroup = {
 
 const TASK_ID = 'flush-grouped-email'
 
-const readUnsentWindowEvents = (db: ReturnType<typeof DBClient.getInstance>, windowKey: string) =>
+type WindowClaim = { db: ReturnType<typeof DBClient.getInstance>; windowKey: string; batchId: string }
+
+type ClaimedWindow = {
+  payload: FlushGroupedEmailPayload
+  db: ReturnType<typeof DBClient.getInstance>
+  batchId: string
+  rows: BufferedRow[]
+}
+
+const claimUnsentWindowEvents = ({ db, windowKey, batchId }: WindowClaim) =>
   db.$queryRaw<BufferedRow[]>`
-    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<typeof DBClient.getInstance>, 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,
 })

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'

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()
+  })
+})

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<StaleWindow[]>`
+    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,
+})