fix(newsletters): handle review recovery cases

Author: Bill Leoutsakos

apps/sim/components/settings/account-settings-renderer.tsx

@@ -49,5 +49,6 @@ export function AccountSettingsRenderer({ section }: AccountSettingsRendererProp
   if (section === 'api-keys') return <ApiKeys scope='personal' />
   if (section === 'admin') return <Admin />
   if (section === 'mothership') return <Mothership />
-  return <Newsletters />
+  if (section === 'newsletters') return <Newsletters />
+  return null
 }

apps/sim/lib/newsletters/push-resend.test.ts

@@ -15,6 +15,7 @@ const mocks = vi.hoisted(() => ({
   isAsyncJobEnqueueError: vi.fn(),
   markFailed: vi.fn(),
   markPushed: vi.fn(),
+  queueCancel: vi.fn(),
   queueEnqueue: vi.fn(),
   queueGetJob: vi.fn(),
   requireAttempt: vi.fn(),
@@ -31,6 +32,8 @@ vi.mock('@/lib/core/async-jobs', () => ({
   JOB_STATUS: {
     COMPLETED: 'completed',
     FAILED: 'failed',
+    PENDING: 'pending',
+    PROCESSING: 'processing',
   },
 }))
 
@@ -70,6 +73,7 @@ describe('newsletter Resend queueing', () => {
     vi.clearAllMocks()
     mocks.getAsyncBackendType.mockReturnValue('trigger-dev')
     mocks.getJobQueue.mockResolvedValue({
+      cancelJob: mocks.queueCancel,
       enqueue: mocks.queueEnqueue,
       getJob: mocks.queueGetJob,
     })
@@ -142,23 +146,151 @@ describe('newsletter Resend queueing', () => {
     )
   })
 
-  it('moves a newsletter run to failed when its persisted database job failed', async () => {
-    mocks.getAsyncBackendType.mockReturnValue('database')
-    mocks.claimAttempt.mockResolvedValue({
-      attempt: 2,
-      jobId: 'newsletter_resend_run-1_2',
-      run,
-      shouldEnqueue: false,
-    })
+  it('starts a new attempt when a persisted Trigger.dev job failed', async () => {
+    mocks.claimAttempt
+      .mockResolvedValueOnce({
+        attempt: 2,
+        jobId: 'trigger-run-failed',
+        run,
+        shouldEnqueue: false,
+      })
+      .mockResolvedValueOnce({
+        attempt: 3,
+        jobId: null,
+        run: { ...run, resendSyncJobId: null },
+        shouldEnqueue: true,
+      })
     mocks.queueGetJob.mockResolvedValue({
-      id: 'newsletter_resend_run-1_2',
+      id: 'trigger-run-failed',
       status: 'failed',
       error: 'worker stopped',
     })
+    mocks.queueEnqueue.mockResolvedValue('trigger-run-retry')
+    mocks.setJob.mockResolvedValue({ ...run, resendSyncJobId: 'trigger-run-retry' })
+
+    const result = await enqueueNewsletterResendSync('run-1', 'admin-1')
+
+    expect(mocks.markFailed).toHaveBeenCalledWith('run-1', 2, expect.any(Error))
+    expect(mocks.queueEnqueue).toHaveBeenCalledWith(
+      'newsletter-resend-sync',
+      { runId: 'run-1', attempt: 3, requestedById: 'admin-1' },
+      expect.objectContaining({ jobId: 'newsletter_resend_run-1_3' })
+    )
+    expect(result.jobId).toBe('trigger-run-retry')
+  })
+
+  it('starts a new attempt when a completed job did not finalize the newsletter run', async () => {
+    mocks.claimAttempt
+      .mockResolvedValueOnce({
+        attempt: 2,
+        jobId: 'trigger-run-completed',
+        run,
+        shouldEnqueue: false,
+      })
+      .mockResolvedValueOnce({
+        attempt: 3,
+        jobId: null,
+        run: { ...run, resendSyncJobId: null },
+        shouldEnqueue: true,
+      })
+    mocks.queueGetJob.mockResolvedValue({
+      id: 'trigger-run-completed',
+      status: 'completed',
+    })
+    mocks.queueEnqueue.mockResolvedValue('trigger-run-retry')
+    mocks.setJob.mockResolvedValue({ ...run, resendSyncJobId: 'trigger-run-retry' })
+
+    const result = await enqueueNewsletterResendSync('run-1', 'admin-1')
+
+    expect(mocks.markFailed).toHaveBeenCalledWith('run-1', 2, expect.any(Error))
+    expect(mocks.queueEnqueue).toHaveBeenCalledWith(
+      'newsletter-resend-sync',
+      { runId: 'run-1', attempt: 3, requestedById: 'admin-1' },
+      expect.objectContaining({ jobId: 'newsletter_resend_run-1_3' })
+    )
+    expect(result.jobId).toBe('trigger-run-retry')
+  })
+
+  it.each([
+    ['completed', { id: 'trigger-run-existing', status: 'completed' }],
+    ['processing', { id: 'trigger-run-existing', status: 'processing' }],
+    ['missing', null],
+  ])(
+    'keeps the stored job when the newsletter run is pushed and the provider job is %s',
+    async (_providerState, providerJob) => {
+      const pushedRun = { ...run, status: 'pushed' }
+      mocks.claimAttempt.mockResolvedValue({
+        attempt: 2,
+        jobId: 'trigger-run-existing',
+        run: pushedRun,
+        shouldEnqueue: false,
+      })
+      mocks.queueGetJob.mockResolvedValue(providerJob)
+
+      const result = await enqueueNewsletterResendSync('run-1', 'admin-1')
+
+      expect(mocks.queueGetJob).not.toHaveBeenCalled()
+      expect(mocks.markFailed).not.toHaveBeenCalled()
+      expect(mocks.queueEnqueue).not.toHaveBeenCalled()
+      expect(result).toEqual({ run: pushedRun, jobId: 'trigger-run-existing' })
+    }
+  )
+
+  it('re-enqueues when a stored Trigger.dev run no longer exists', async () => {
+    mocks.claimAttempt
+      .mockResolvedValueOnce({
+        attempt: 2,
+        jobId: 'trigger-run-missing',
+        run,
+        shouldEnqueue: false,
+      })
+      .mockResolvedValueOnce({
+        attempt: 3,
+        jobId: null,
+        run,
+        shouldEnqueue: true,
+      })
+    mocks.queueGetJob.mockResolvedValue(null)
+
+    await enqueueNewsletterResendSync('run-1', 'admin-1')
+
+    expect(mocks.markFailed).toHaveBeenCalledWith('run-1', 2, expect.any(Error))
+    expect(mocks.queueEnqueue).toHaveBeenCalledWith(
+      'newsletter-resend-sync',
+      { runId: 'run-1', attempt: 3, requestedById: 'admin-1' },
+      expect.objectContaining({ jobId: 'newsletter_resend_run-1_3' })
+    )
+  })
+
+  it('cancels and replaces an active Trigger.dev run when an admin resumes it', async () => {
+    mocks.claimAttempt
+      .mockResolvedValueOnce({
+        attempt: 2,
+        jobId: 'trigger-run-active',
+        run,
+        shouldEnqueue: false,
+      })
+      .mockResolvedValueOnce({
+        attempt: 3,
+        jobId: null,
+        run,
+        shouldEnqueue: true,
+      })
+    mocks.queueGetJob.mockResolvedValue({
+      id: 'trigger-run-active',
+      status: 'processing',
+    })
 
-    await expect(enqueueNewsletterResendSync('run-1', 'admin-1')).rejects.toThrow('worker stopped')
+    const result = await enqueueNewsletterResendSync('run-1', 'admin-1')
+
+    expect(mocks.queueCancel).toHaveBeenCalledWith('trigger-run-active')
     expect(mocks.markFailed).toHaveBeenCalledWith('run-1', 2, expect.any(Error))
-    expect(mocks.queueEnqueue).not.toHaveBeenCalled()
+    expect(mocks.queueEnqueue).toHaveBeenCalledWith(
+      'newsletter-resend-sync',
+      { runId: 'run-1', attempt: 3, requestedById: 'admin-1' },
+      expect.objectContaining({ jobId: 'newsletter_resend_run-1_3' })
+    )
+    expect(result.jobId).toBe('trigger-run-123')
   })
 
   it('resets failed recipients before a task retry', async () => {

apps/sim/lib/newsletters/push-resend.ts

@@ -174,25 +174,41 @@ export async function runNewsletterResendSync(
 }
 
 export async function enqueueNewsletterResendSync(runId: string, requestedById: string) {
-  const claim = await claimNewsletterRunResendAttempt(runId)
+  let claim = await claimNewsletterRunResendAttempt(runId)
   const queue = await getJobQueue()
+  const backendType = getAsyncBackendType()
   if (!claim.shouldEnqueue && claim.jobId) {
-    if (getAsyncBackendType() !== 'database') {
+    if (claim.run.status === 'pushed') {
       return { run: claim.run, jobId: claim.jobId }
     }
-
     const persistedJob = await queue.getJob(claim.jobId)
     if (persistedJob?.status === JOB_STATUS.COMPLETED) {
-      return { run: claim.run, jobId: claim.jobId }
-    }
-    if (persistedJob?.status === JOB_STATUS.FAILED) {
-      const error = new Error(persistedJob.error ?? 'Newsletter sync database job failed')
+      const error = new Error('Newsletter sync job completed without finalizing the newsletter run')
+      await markNewsletterRunPushFailed(runId, claim.attempt, error)
+      claim = await claimNewsletterRunResendAttempt(runId)
+    } else if (backendType !== 'database') {
+      if (
+        persistedJob?.status === JOB_STATUS.PENDING ||
+        persistedJob?.status === JOB_STATUS.PROCESSING
+      ) {
+        await queue.cancelJob(claim.jobId)
+      }
+      const error = new Error(
+        persistedJob?.error ?? 'Newsletter sync was resumed with a fresh background job'
+      )
+      await markNewsletterRunPushFailed(runId, claim.attempt, error)
+      claim = await claimNewsletterRunResendAttempt(runId)
+    } else if (persistedJob?.status === JOB_STATUS.FAILED) {
+      const error = new Error(persistedJob.error ?? 'Newsletter sync job failed')
       await markNewsletterRunPushFailed(runId, claim.attempt, error)
-      throw error
+      claim = await claimNewsletterRunResendAttempt(runId)
     }
   }
 
-  const enqueueKey = claim.jobId ?? `newsletter_resend_${runId}_${claim.attempt}`
+  const enqueueKey =
+    backendType === 'database' && claim.jobId
+      ? claim.jobId
+      : `newsletter_resend_${runId}_${claim.attempt}`
   let jobId: string
   try {
     await resetFailedNewsletterRecipients(runId)

apps/sim/lib/newsletters/resend.test.ts

@@ -127,21 +127,53 @@ describe('newsletter Resend service', () => {
   it('normalizes suppressed email addresses', async () => {
     fetchMock.mockResolvedValueOnce(
       jsonResponse({
-        data: [{ email: ' First@Example.com ' }, { email: 'second@example.com' }],
+        data: [
+          { id: 'suppression-1', email: ' First@Example.com ' },
+          { id: 'suppression-2', email: 'second@example.com' },
+        ],
         has_more: false,
       })
     )
 
     const emails = await getResendSuppressedEmails()
 
     expect(emails).toEqual(new Set(['first@example.com', 'second@example.com']))
+    expect(fetchMock).toHaveBeenCalledWith(
+      'https://api.resend.com/suppressions?limit=100',
+      expect.objectContaining({ method: 'GET' })
+    )
+  })
+
+  it('paginates through all suppressed email addresses', async () => {
+    fetchMock
+      .mockResolvedValueOnce(
+        jsonResponse({
+          data: [{ id: 'suppression-1', email: 'first@example.com' }],
+          has_more: true,
+        })
+      )
+      .mockResolvedValueOnce(
+        jsonResponse({
+          data: [{ id: 'suppression-2', email: 'second@example.com' }],
+          has_more: false,
+        })
+      )
+
+    const emails = await getResendSuppressedEmails()
+
+    expect(emails).toEqual(new Set(['first@example.com', 'second@example.com']))
+    expect(fetchMock).toHaveBeenNthCalledWith(
+      2,
+      'https://api.resend.com/suppressions?limit=100&after=suppression-1',
+      expect.objectContaining({ method: 'GET' })
+    )
   })
 
   it('combines suppressions with globally unsubscribed contacts', async () => {
     fetchMock
       .mockResolvedValueOnce(
         jsonResponse({
-          data: [{ email: 'suppressed@example.com' }],
+          data: [{ id: 'suppression-1', email: 'suppressed@example.com' }],
           has_more: false,
         })
       )

apps/sim/lib/newsletters/resend.ts

@@ -13,6 +13,8 @@ const RESEND_CONTACT_PROPERTY_PAGE_LIMIT = 100
 const RESEND_CONTACT_PROPERTY_MAX_PAGES = 10
 const RESEND_CONTACT_PAGE_LIMIT = 100
 const RESEND_CONTACT_MAX_PAGES = 1000
+const RESEND_SUPPRESSION_PAGE_LIMIT = 100
+const RESEND_SUPPRESSION_MAX_PAGES = 1000
 const NEWSLETTER_CONTACT_PROPERTY_KEYS = ['sim_user_id', 'newsletter_run_id'] as const
 
 interface ResendErrorBody {
@@ -43,7 +45,7 @@ interface ResendRequestOptions {
 }
 
 const resendSuppressionListSchema = z.object({
-  data: z.array(z.object({ email: z.string().min(1) })),
+  data: z.array(z.object({ id: z.string().min(1), email: z.string().min(1) })),
   has_more: z.boolean(),
 })
 
@@ -116,22 +118,37 @@ async function resendRequest<T>(path: string, options: ResendRequestOptions = {}
 }
 
 export async function getResendSuppressedEmails(options?: { required?: boolean }) {
-  const rawResponse = await resendRequest<unknown>('/suppressions', {
-    required: options?.required ?? true,
-  })
-  if (rawResponse === undefined) return new Set<string>()
-  const parsedResponse = resendSuppressionListSchema.safeParse(rawResponse)
-  if (!parsedResponse.success) {
-    throw new Error('Resend suppression list response was malformed')
-  }
-  const response = parsedResponse.data
   const emails = new Set<string>()
-  if (response?.has_more) {
-    throw new Error('Resend suppression list was incomplete')
+  let after: string | null = null
+  let hasMore = false
+
+  for (let page = 0; page < RESEND_SUPPRESSION_MAX_PAGES; page++) {
+    const query = new URLSearchParams({ limit: String(RESEND_SUPPRESSION_PAGE_LIMIT) })
+    if (after) query.set('after', after)
+    const rawResponse = await resendRequest<unknown>(`/suppressions?${query.toString()}`, {
+      required: options?.required ?? true,
+    })
+    if (rawResponse === undefined) return emails
+    const parsedResponse = resendSuppressionListSchema.safeParse(rawResponse)
+    if (!parsedResponse.success) {
+      throw new Error('Resend suppression list response was malformed')
+    }
+    const response = parsedResponse.data
+
+    for (const suppression of response.data) {
+      emails.add(normalizeEmail(suppression.email))
+    }
+
+    hasMore = response.has_more
+    if (!hasMore) break
+    after = response.data.at(-1)?.id ?? null
+    if (!after) throw new Error('Resend suppression pagination returned no cursor')
   }
-  for (const suppression of response.data) {
-    emails.add(normalizeEmail(suppression.email))
+
+  if (hasMore) {
+    throw new Error('Resend suppression list exceeded the pagination safety limit')
   }
+
   return emails
 }