feat(notifications): email on schedule auto-disable and 100% usage limit (#6038)

* feat(notifications): email on schedule auto-disable and 100% usage limit

* fix(notifications): keep partial recipients when one lookup fails

Author: TheodoreSpeaks

apps/sim/app/api/emails/preview/route.ts

@@ -11,6 +11,8 @@ import {
   renderPasswordResetEmail,
   renderPaymentFailedEmail,
   renderPlanWelcomeEmail,
+  renderScheduleDisabledEmail,
+  renderUsageLimitReachedEmail,
   renderUsageThresholdEmail,
   renderWelcomeEmail,
   renderWorkspaceInvitationEmail,
@@ -93,6 +95,43 @@ const emailTemplates = {
       billingPortalUrl: 'https://sim.ai/settings/billing',
       failureReason: 'Card declined',
     }),
+  'usage-limit-reached': () =>
+    renderUsageLimitReachedEmail({
+      userName: 'John',
+      planName: 'Pro',
+      scope: 'user',
+      currentUsage: 20,
+      limit: 20,
+      ctaLink: 'https://sim.ai/settings/billing',
+    }),
+  'usage-limit-reached-org': () =>
+    renderUsageLimitReachedEmail({
+      userName: 'John',
+      planName: 'Team',
+      scope: 'organization',
+      currentUsage: 500,
+      limit: 500,
+      ctaLink: 'https://sim.ai/organization/org_123/settings/billing',
+    }),
+
+  // Operational notification emails
+  'schedule-disabled': () =>
+    renderScheduleDisabledEmail({
+      recipientName: 'John',
+      kind: 'workflow',
+      resourceName: 'Daily digest',
+      reason: 'consecutive_failures',
+      failedCount: 100,
+      manageLink: 'https://sim.ai/workspace/ws_123/w/wf_456',
+    }),
+  'schedule-disabled-auth': () =>
+    renderScheduleDisabledEmail({
+      recipientName: 'John',
+      kind: 'job',
+      resourceName: 'Weekly report',
+      reason: 'authentication_error',
+      manageLink: 'https://sim.ai/workspace/ws_123/scheduled-tasks',
+    }),
 } as const
 
 type EmailTemplate = keyof typeof emailTemplates
@@ -122,7 +161,10 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
         'plan-welcome-team',
         'credit-purchase',
         'payment-failed',
+        'usage-limit-reached',
+        'usage-limit-reached-org',
       ],
+      Notifications: ['schedule-disabled', 'schedule-disabled-auth'],
     }
 
     const categoryHtml = Object.entries(categories)

apps/sim/app/api/schedules/execute/route.test.ts

@@ -30,6 +30,8 @@ const {
   mockShouldExecuteInline,
   mockResolveSystemBillingAttribution,
   mockAssertBillingAttributionSnapshot,
+  mockApplyScheduleFailureUpdate,
+  mockNotifyScheduleAutoDisabled,
 } = vi.hoisted(() => ({
   mockVerifyCronAuth: vi.fn().mockReturnValue(null),
   mockExecuteScheduleJob: vi.fn().mockResolvedValue(undefined),
@@ -44,6 +46,8 @@ const {
   mockShouldExecuteInline: vi.fn().mockReturnValue(false),
   mockResolveSystemBillingAttribution: vi.fn(),
   mockAssertBillingAttributionSnapshot: vi.fn(),
+  mockApplyScheduleFailureUpdate: vi.fn().mockResolvedValue({ updated: true, disabled: false }),
+  mockNotifyScheduleAutoDisabled: vi.fn().mockResolvedValue(undefined),
 }))
 
 vi.mock('@/lib/auth/internal', () => ({
@@ -59,15 +63,11 @@ vi.mock('@/background/schedule-execution', () => ({
   executeScheduleJob: mockExecuteScheduleJob,
   executeJobInline: mockExecuteJobInline,
   releaseScheduleLock: mockReleaseScheduleLock,
-  buildScheduleFailureUpdate: (now: Date, nextRunAt: Date | null) => ({
-    updatedAt: now,
-    lastQueuedAt: null,
-    nextRunAt,
-    failedCount: { type: 'sql' },
-    lastFailedAt: now,
-    status: { type: 'sql' },
-    infraRetryCount: 0,
-  }),
+  applyScheduleFailureUpdate: mockApplyScheduleFailureUpdate,
+}))
+
+vi.mock('@/lib/workflows/schedules/disable-notifications', () => ({
+  notifyScheduleAutoDisabled: mockNotifyScheduleAutoDisabled,
 }))
 
 vi.mock('@/lib/core/async-jobs', () => ({
@@ -730,11 +730,11 @@ describe('Scheduled Workflow Execution API Route', () => {
         error: expect.stringContaining('exhausted retry attempts'),
       })
     )
-    expect(dbChainMockFns.set).toHaveBeenCalledWith(
+    expect(mockApplyScheduleFailureUpdate).toHaveBeenCalledWith(
       expect.objectContaining({
-        lastQueuedAt: null,
-        lastFailedAt: expect.any(Date),
-        nextRunAt: expect.any(Date),
+        scheduleId: 'schedule-1',
+        expectedLastQueuedAt: claimedAt,
+        executor: expect.anything(),
       })
     )
   })
@@ -779,12 +779,11 @@ describe('Scheduled Workflow Execution API Route', () => {
 
     await runScheduleTick('test-request-id')
     expect(mockEnqueue).not.toHaveBeenCalled()
-    expect(dbChainMockFns.set).toHaveBeenCalledWith(
+    expect(mockApplyScheduleFailureUpdate).toHaveBeenCalledWith(
       expect.objectContaining({
-        lastQueuedAt: null,
-        lastFailedAt: expect.any(Date),
+        scheduleId: schedule.id,
+        expectedLastQueuedAt: claimedAt,
         nextRunAt: expect.any(Date),
-        infraRetryCount: 0,
       })
     )
     expect(dbChainMockFns.set).not.toHaveBeenCalledWith(
@@ -794,6 +793,44 @@ describe('Scheduled Workflow Execution API Route', () => {
     )
   })
 
+  it('emails the schedule owners when a non-retryable setup failure disables the schedule', async () => {
+    const claimedAt = new Date('2025-01-01T00:00:00.000Z')
+    const schedule = {
+      ...SINGLE_SCHEDULE[0],
+      lastQueuedAt: claimedAt,
+    }
+    mockGetJob.mockRejectedValueOnce(new Error('bad setup invariant'))
+    mockApplyScheduleFailureUpdate.mockResolvedValueOnce({ updated: true, disabled: true })
+    dbChainMockFns.limit
+      .mockResolvedValueOnce(SINGLE_CLAIMED_SCHEDULE_ROWS)
+      .mockResolvedValueOnce([])
+    dbChainMockFns.returning.mockReturnValueOnce([schedule]).mockReturnValueOnce([])
+
+    await runScheduleTick('test-request-id')
+
+    expect(mockNotifyScheduleAutoDisabled).toHaveBeenCalledWith(
+      expect.objectContaining({ scheduleId: schedule.id, reason: 'consecutive_failures' })
+    )
+  })
+
+  it('does not email when the failure update leaves the schedule active', async () => {
+    const claimedAt = new Date('2025-01-01T00:00:00.000Z')
+    const schedule = {
+      ...SINGLE_SCHEDULE[0],
+      lastQueuedAt: claimedAt,
+    }
+    mockGetJob.mockRejectedValueOnce(new Error('bad setup invariant'))
+    mockApplyScheduleFailureUpdate.mockResolvedValueOnce({ updated: true, disabled: false })
+    dbChainMockFns.limit
+      .mockResolvedValueOnce(SINGLE_CLAIMED_SCHEDULE_ROWS)
+      .mockResolvedValueOnce([])
+    dbChainMockFns.returning.mockReturnValueOnce([schedule]).mockReturnValueOnce([])
+
+    await runScheduleTick('test-request-id')
+
+    expect(mockNotifyScheduleAutoDisabled).not.toHaveBeenCalled()
+  })
+
   it('uses one backend mode decision for slot accounting and schedule processing', async () => {
     mockShouldExecuteInline.mockReturnValue(true)
     dbChainMockFns.limit

apps/sim/app/api/schedules/execute/route.ts

@@ -22,6 +22,7 @@ import { getMaxExecutionTimeout } from '@/lib/core/execution-limits'
 import { runDetached } from '@/lib/core/utils/background'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+import { notifyScheduleAutoDisabled } from '@/lib/workflows/schedules/disable-notifications'
 import {
   SCHEDULE_EXECUTION_CONCURRENCY_LIMIT,
   SCHEDULE_EXECUTION_QUEUE_NAME,
@@ -31,7 +32,7 @@ import {
 } from '@/lib/workflows/schedules/execution-limits'
 import { calculateScheduleInfraRetryDelayMs } from '@/lib/workflows/schedules/retry'
 import {
-  buildScheduleFailureUpdate,
+  applyScheduleFailureUpdate,
   executeJobInline,
   executeScheduleJob,
   releaseScheduleLock,
@@ -43,6 +44,12 @@ export const maxDuration = 3600
 
 const logger = createLogger('ScheduledExecuteAPI')
 const WORKFLOW_CHUNK_SIZE = 100
+/**
+ * Recovery sweeps a batch of up to `STALE_SCHEDULE_RECOVERY_BATCH_SIZE` schedules,
+ * each fanning out to every workspace admin. Cap the mail so one tick can't turn
+ * into hundreds of inline sends; the remainder is logged.
+ */
+const STALE_SCHEDULE_RECOVERY_NOTIFY_LIMIT = 25
 const JOB_CHUNK_SIZE = 100
 const MAX_TICK_DURATION_MS = 3 * 60 * 1000
 const STALE_SCHEDULE_CLAIM_MS = getMaxExecutionTimeout()
@@ -394,20 +401,22 @@ async function markClaimedScheduleFailed(
   context: string
 ): Promise<void> {
   const now = new Date()
-  await db
-    .update(workflowSchedule)
-    .set(buildScheduleFailureUpdate(now, getScheduleNextRunAt(schedule, now)))
-    .where(
-      and(
-        eq(workflowSchedule.id, schedule.id),
-        isNull(workflowSchedule.archivedAt),
-        eq(workflowSchedule.lastQueuedAt, expectedLastQueuedAt)
-      )
-    )
-    .catch((error) => {
-      logger.error(`[${requestId}] ${context}`, error)
-      throw error
+  const { disabled } = await applyScheduleFailureUpdate({
+    scheduleId: schedule.id,
+    now,
+    nextRunAt: getScheduleNextRunAt(schedule, now),
+    expectedLastQueuedAt,
+    requestId,
+    context,
+  })
+
+  if (disabled) {
+    await notifyScheduleAutoDisabled({
+      scheduleId: schedule.id,
+      reason: 'consecutive_failures',
+      requestId,
     })
+  }
 }
 
 async function deferClaimedScheduleAfterQueueFailure(
@@ -491,6 +500,13 @@ async function handleClaimedScheduleSetupFailure(
 async function recoverStaleDatabaseScheduleJobs(now: Date): Promise<void> {
   const staleStartedBefore = getStaleScheduleExecutionCutoff(now)
 
+  /**
+   * Collected inside the transaction, flushed after it commits. Emailing inside
+   * would both notify about writes a rollback discards and issue pooled-client
+   * reads while the transaction still holds row locks under the advisory lock.
+   */
+  const disabledScheduleIds: string[] = []
+
   await db.transaction(async (tx) => {
     const [lock] = await tx.execute<{ acquired: boolean }>(
       sql`SELECT pg_try_advisory_xact_lock(hashtextextended(${SCHEDULE_EXECUTION_QUEUE_NAME}, 0)) AS acquired`
@@ -539,16 +555,17 @@ async function recoverStaleDatabaseScheduleJobs(now: Date): Promise<void> {
       const claimedAt = getSchedulePayloadClaimedAt(payload)
       if (!payload || !claimedAt) continue
 
-      await tx
-        .update(workflowSchedule)
-        .set(buildScheduleFailureUpdate(now, getScheduleNextRunAt(payload, now)))
-        .where(
-          and(
-            eq(workflowSchedule.id, payload.scheduleId),
-            isNull(workflowSchedule.archivedAt),
-            eq(workflowSchedule.lastQueuedAt, claimedAt)
-          )
-        )
+      const { disabled } = await applyScheduleFailureUpdate({
+        scheduleId: payload.scheduleId,
+        now,
+        nextRunAt: getScheduleNextRunAt(payload, now),
+        expectedLastQueuedAt: claimedAt,
+        requestId: 'stale-schedule-recovery',
+        context: `Error updating schedule ${payload.scheduleId} after stale lease recovery`,
+        executor: tx,
+      })
+
+      if (disabled) disabledScheduleIds.push(payload.scheduleId)
     }
 
     if (retryableRows.length > 0) {
@@ -568,6 +585,22 @@ async function recoverStaleDatabaseScheduleJobs(now: Date): Promise<void> {
         )
     }
   })
+
+  const notifiable = disabledScheduleIds.slice(0, STALE_SCHEDULE_RECOVERY_NOTIFY_LIMIT)
+  if (disabledScheduleIds.length > notifiable.length) {
+    logger.warn('Capped schedule auto-disable notifications for stale recovery batch', {
+      disabled: disabledScheduleIds.length,
+      notified: notifiable.length,
+    })
+  }
+
+  for (const scheduleId of notifiable) {
+    await notifyScheduleAutoDisabled({
+      scheduleId,
+      reason: 'consecutive_failures',
+      requestId: 'stale-schedule-recovery',
+    })
+  }
 }
 
 function isStaleDatabaseScheduleJob(job: { status: string; startedAt?: Date }): boolean {