mirror of
https://github.com/simstudioai/sim.git
synced 2026-09-24 15:45:35 +08:00
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
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -0,0 +1,101 @@
|
||||
/**
|
||||
* Covers the `active -> disabled` edge detection that drives the auto-disable
|
||||
* email. `applyScheduleFailureUpdate` is the exported seam over the shared
|
||||
* guarded UPDATE, so asserting on it also pins the behavior of the trigger.dev
|
||||
* call sites that go through the same helper.
|
||||
*
|
||||
* @vitest-environment node
|
||||
*/
|
||||
import { databaseMock, dbChainMockFns, resetDbChainMock, schemaMock } from '@sim/testing'
|
||||
import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
const { notifyMock } = vi.hoisted(() => ({
|
||||
notifyMock: vi.fn(() => Promise.resolve()),
|
||||
}))
|
||||
|
||||
vi.mock('@/lib/workflows/schedules/disable-notifications', () => ({
|
||||
notifyScheduleAutoDisabled: notifyMock,
|
||||
}))
|
||||
|
||||
// This module imports its tables from `@sim/db` directly, which the global mock
|
||||
// does not re-export. Widen it rather than rewriting the source's imports.
|
||||
vi.mock('@sim/db', () => ({ ...databaseMock, ...schemaMock }))
|
||||
|
||||
import { applyScheduleFailureUpdate, releaseScheduleLock } from '@/background/schedule-execution'
|
||||
|
||||
const BASE = {
|
||||
scheduleId: 'schedule-1',
|
||||
now: new Date('2025-01-01T00:00:00.000Z'),
|
||||
nextRunAt: new Date('2025-01-01T01:00:00.000Z'),
|
||||
expectedLastQueuedAt: new Date('2024-12-31T23:00:00.000Z'),
|
||||
requestId: 'req-1',
|
||||
context: 'test context',
|
||||
}
|
||||
|
||||
afterAll(() => {
|
||||
resetDbChainMock()
|
||||
})
|
||||
|
||||
describe('applyScheduleFailureUpdate', () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks()
|
||||
resetDbChainMock()
|
||||
})
|
||||
|
||||
it('reports disabled when the write returns a disabled row', async () => {
|
||||
dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'schedule-1', status: 'disabled' }])
|
||||
|
||||
const result = await applyScheduleFailureUpdate(BASE)
|
||||
|
||||
expect(result).toEqual({ updated: true, disabled: true })
|
||||
})
|
||||
|
||||
it('reports not disabled below the threshold (the 99-of-100 case)', async () => {
|
||||
dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'schedule-1', status: 'active' }])
|
||||
|
||||
const result = await applyScheduleFailureUpdate(BASE)
|
||||
|
||||
expect(result).toEqual({ updated: true, disabled: false })
|
||||
})
|
||||
|
||||
it('reports neither updated nor disabled when the claim guard matches no row', async () => {
|
||||
dbChainMockFns.returning.mockResolvedValueOnce([])
|
||||
|
||||
const result = await applyScheduleFailureUpdate(BASE)
|
||||
|
||||
expect(result).toEqual({ updated: false, disabled: false })
|
||||
})
|
||||
|
||||
it('defers the email so an in-transaction caller can send it after commit', async () => {
|
||||
dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'schedule-1', status: 'disabled' }])
|
||||
|
||||
await applyScheduleFailureUpdate(BASE)
|
||||
|
||||
expect(notifyMock).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('excludes already-disabled rows so a disable is reported at most once', async () => {
|
||||
dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'schedule-1', status: 'disabled' }])
|
||||
|
||||
await applyScheduleFailureUpdate(BASE)
|
||||
|
||||
const condition = dbChainMockFns.where.mock.calls.at(-1)?.[0]
|
||||
expect(JSON.stringify(condition)).toContain('disabled')
|
||||
})
|
||||
})
|
||||
|
||||
describe('releaseScheduleLock', () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks()
|
||||
resetDbChainMock()
|
||||
})
|
||||
|
||||
it('never notifies, even when the row it releases is already disabled', async () => {
|
||||
dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'schedule-1', status: 'disabled' }])
|
||||
|
||||
const released = await releaseScheduleLock('schedule-1', 'req-1', new Date(), 'release')
|
||||
|
||||
expect(released).toBe(true)
|
||||
expect(notifyMock).not.toHaveBeenCalled()
|
||||
})
|
||||
})
|
||||
@@ -30,6 +30,7 @@ import {
|
||||
getExecutionTimeout,
|
||||
getTimeoutErrorMessage,
|
||||
} from '@/lib/core/execution-limits'
|
||||
import type { DbOrTx } from '@/lib/db/types'
|
||||
import { preprocessExecution } from '@/lib/execution/preprocessing'
|
||||
import { LoggingSession } from '@/lib/logs/execution/logging-session'
|
||||
import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans'
|
||||
@@ -40,6 +41,8 @@ import {
|
||||
} from '@/lib/workflows/executor/execution-core'
|
||||
import { handlePostExecutionPauseState } from '@/lib/workflows/executor/pause-persistence'
|
||||
import { loadDeployedWorkflowState } from '@/lib/workflows/persistence/utils'
|
||||
import { notifyScheduleAutoDisabled } from '@/lib/workflows/schedules/disable-notifications'
|
||||
import type { ScheduleDisableReason } from '@/lib/workflows/schedules/disable-reasons'
|
||||
import {
|
||||
SCHEDULE_EXECUTION_CONCURRENCY_LIMIT,
|
||||
SCHEDULE_EXECUTION_QUEUE_NAME,
|
||||
@@ -70,6 +73,12 @@ type WorkflowScheduleUpdate = Partial<Omit<WorkflowScheduleInsert, 'failedCount'
|
||||
}
|
||||
type ExecutionCoreResult = Awaited<ReturnType<typeof executeWorkflowCore>>
|
||||
|
||||
/** Result of a guarded schedule UPDATE. `status` is the row's value after the write. */
|
||||
type ScheduleUpdateOutcome = {
|
||||
updated: boolean
|
||||
status: string | null
|
||||
}
|
||||
|
||||
function incrementScheduleFailedCount(): SQL {
|
||||
return sql`COALESCE(${workflowSchedule.failedCount}, 0) + 1`
|
||||
}
|
||||
@@ -141,8 +150,23 @@ async function applyScheduleUpdate(
|
||||
updates: WorkflowScheduleUpdate,
|
||||
requestId: string,
|
||||
context: string,
|
||||
options: { expectedLastQueuedAt?: Date | null; allowCompleted?: boolean } = {}
|
||||
): Promise<boolean> {
|
||||
options: {
|
||||
expectedLastQueuedAt?: Date | null
|
||||
allowCompleted?: boolean
|
||||
/**
|
||||
* Set at call sites that can transition the row to `disabled`. Presence both
|
||||
* opts the site into the auto-disable email and adds a `status <> 'disabled'`
|
||||
* guard, so the transition fires exactly once per disable.
|
||||
*/
|
||||
disableReason?: ScheduleDisableReason
|
||||
/** Required inside a transaction, where mail must wait for commit. */
|
||||
deferNotification?: boolean
|
||||
/** Join a caller's transaction instead of using the pooled client. */
|
||||
executor?: DbOrTx
|
||||
} = {}
|
||||
): Promise<ScheduleUpdateOutcome> {
|
||||
let outcome: ScheduleUpdateOutcome
|
||||
|
||||
try {
|
||||
const claimGuard =
|
||||
options.expectedLastQueuedAt === undefined
|
||||
@@ -162,7 +186,17 @@ async function applyScheduleUpdate(
|
||||
? undefined
|
||||
: ne(workflowSchedule.status, 'completed')
|
||||
|
||||
const updatedRows = await db
|
||||
/**
|
||||
* `RETURNING` yields the NEW row, so `status === 'disabled'` alone only means
|
||||
* "is disabled". Excluding rows that were already disabled makes a returned
|
||||
* row a true `active -> disabled` edge. Scoped to disable-capable call sites
|
||||
* so lock releases on already-disabled rows still work.
|
||||
*/
|
||||
const notAlreadyDisabled = options.disableReason
|
||||
? ne(workflowSchedule.status, 'disabled')
|
||||
: undefined
|
||||
|
||||
const updatedRows = await (options.executor ?? db)
|
||||
.update(workflowSchedule)
|
||||
.set(updates)
|
||||
.where(
|
||||
@@ -170,16 +204,29 @@ async function applyScheduleUpdate(
|
||||
eq(workflowSchedule.id, scheduleId),
|
||||
isNull(workflowSchedule.archivedAt),
|
||||
claimGuard,
|
||||
notCompletedGuard
|
||||
notCompletedGuard,
|
||||
notAlreadyDisabled
|
||||
)
|
||||
)
|
||||
.returning({ id: workflowSchedule.id })
|
||||
.returning({ id: workflowSchedule.id, status: workflowSchedule.status })
|
||||
|
||||
return updatedRows.length > 0
|
||||
const row = updatedRows[0]
|
||||
outcome = { updated: Boolean(row), status: row?.status ?? null }
|
||||
} catch (error) {
|
||||
logger.error(`[${requestId}] ${context}`, error, { cause: describeError(error) })
|
||||
throw error
|
||||
}
|
||||
|
||||
// Outside the try: a mail failure must never surface as a schedule-tick fault.
|
||||
if (options.disableReason && !options.deferNotification && outcome.status === 'disabled') {
|
||||
await notifyScheduleAutoDisabled({
|
||||
scheduleId,
|
||||
reason: options.disableReason,
|
||||
requestId,
|
||||
})
|
||||
}
|
||||
|
||||
return outcome
|
||||
}
|
||||
|
||||
export async function releaseScheduleLock(
|
||||
@@ -199,7 +246,41 @@ export async function releaseScheduleLock(
|
||||
updates.nextRunAt = nextRunAt
|
||||
}
|
||||
|
||||
return applyScheduleUpdate(scheduleId, updates, requestId, context, options)
|
||||
const outcome = await applyScheduleUpdate(scheduleId, updates, requestId, context, options)
|
||||
return outcome.updated
|
||||
}
|
||||
|
||||
/**
|
||||
* Applies {@link buildScheduleFailureUpdate} through the same guarded write the
|
||||
* trigger.dev path uses, and reports whether the row just transitioned to
|
||||
* `disabled`. Callers own the notification so an in-transaction caller can defer
|
||||
* it until after commit.
|
||||
*/
|
||||
export async function applyScheduleFailureUpdate(params: {
|
||||
scheduleId: string
|
||||
now: Date
|
||||
nextRunAt: Date | null
|
||||
expectedLastQueuedAt: Date
|
||||
requestId: string
|
||||
context: string
|
||||
executor?: DbOrTx
|
||||
}): Promise<{ updated: boolean; disabled: boolean }> {
|
||||
const { scheduleId, now, nextRunAt, expectedLastQueuedAt, requestId, context, executor } = params
|
||||
|
||||
const outcome = await applyScheduleUpdate(
|
||||
scheduleId,
|
||||
buildScheduleFailureUpdate(now, nextRunAt),
|
||||
requestId,
|
||||
context,
|
||||
{
|
||||
expectedLastQueuedAt,
|
||||
disableReason: 'consecutive_failures',
|
||||
deferNotification: true,
|
||||
executor,
|
||||
}
|
||||
)
|
||||
|
||||
return { updated: outcome.updated, disabled: outcome.status === 'disabled' }
|
||||
}
|
||||
|
||||
function getScheduleClaimedAt(payload: ScheduleExecutionPayload): Date | null {
|
||||
@@ -239,7 +320,7 @@ async function retryScheduleAfterInfraFailure({
|
||||
buildScheduleFailureUpdate(now, nextRunAt),
|
||||
requestId,
|
||||
`Error updating schedule ${payload.scheduleId} after exhausted infrastructure retries`,
|
||||
{ expectedLastQueuedAt: claimedAt }
|
||||
{ expectedLastQueuedAt: claimedAt, disableReason: 'consecutive_failures' }
|
||||
)
|
||||
return
|
||||
}
|
||||
@@ -664,10 +745,12 @@ export async function executeScheduleJob(payload: ScheduleExecutionPayload) {
|
||||
|
||||
const updateClaimedSchedule = (
|
||||
updates: WorkflowScheduleUpdate,
|
||||
context: string
|
||||
): Promise<boolean> =>
|
||||
context: string,
|
||||
disableReason?: ScheduleDisableReason
|
||||
): Promise<ScheduleUpdateOutcome> =>
|
||||
applyScheduleUpdate(payload.scheduleId, updates, requestId, context, {
|
||||
expectedLastQueuedAt: claimedAt,
|
||||
disableReason,
|
||||
})
|
||||
|
||||
try {
|
||||
@@ -793,7 +876,8 @@ export async function executeScheduleJob(payload: ScheduleExecutionPayload) {
|
||||
status: 'disabled',
|
||||
...resetScheduleInfraRetryCount(),
|
||||
},
|
||||
`Failed to disable schedule ${payload.scheduleId} after authentication error`
|
||||
`Failed to disable schedule ${payload.scheduleId} after authentication error`,
|
||||
'authentication_error'
|
||||
)
|
||||
return
|
||||
}
|
||||
@@ -810,7 +894,8 @@ export async function executeScheduleJob(payload: ScheduleExecutionPayload) {
|
||||
status: 'disabled',
|
||||
...resetScheduleInfraRetryCount(),
|
||||
},
|
||||
`Failed to disable schedule ${payload.scheduleId} after authorization error`
|
||||
`Failed to disable schedule ${payload.scheduleId} after authorization error`,
|
||||
'authorization_error'
|
||||
)
|
||||
return
|
||||
}
|
||||
@@ -824,7 +909,8 @@ export async function executeScheduleJob(payload: ScheduleExecutionPayload) {
|
||||
status: 'disabled',
|
||||
...resetScheduleInfraRetryCount(),
|
||||
},
|
||||
`Failed to disable schedule ${payload.scheduleId} after missing workflow`
|
||||
`Failed to disable schedule ${payload.scheduleId} after missing workflow`,
|
||||
'workflow_not_found'
|
||||
)
|
||||
return
|
||||
}
|
||||
@@ -863,7 +949,8 @@ export async function executeScheduleJob(payload: ScheduleExecutionPayload) {
|
||||
})
|
||||
await updateClaimedSchedule(
|
||||
buildScheduleFailureUpdate(now, nextRunAt),
|
||||
`Error updating schedule ${payload.scheduleId} after usage limit check`
|
||||
`Error updating schedule ${payload.scheduleId} after usage limit check`,
|
||||
'consecutive_failures'
|
||||
)
|
||||
return
|
||||
}
|
||||
@@ -885,7 +972,8 @@ export async function executeScheduleJob(payload: ScheduleExecutionPayload) {
|
||||
|
||||
await updateClaimedSchedule(
|
||||
buildScheduleFailureUpdate(now, nextRunAt),
|
||||
`Error updating schedule ${payload.scheduleId} after preprocessing failure`
|
||||
`Error updating schedule ${payload.scheduleId} after preprocessing failure`,
|
||||
'consecutive_failures'
|
||||
)
|
||||
return
|
||||
}
|
||||
@@ -953,7 +1041,8 @@ export async function executeScheduleJob(payload: ScheduleExecutionPayload) {
|
||||
nextRunAt: null,
|
||||
...resetScheduleInfraRetryCount(),
|
||||
},
|
||||
`Failed to disable schedule ${payload.scheduleId} after skip`
|
||||
`Failed to disable schedule ${payload.scheduleId} after skip`,
|
||||
'invalid_schedule'
|
||||
)
|
||||
return
|
||||
}
|
||||
@@ -983,7 +1072,8 @@ export async function executeScheduleJob(payload: ScheduleExecutionPayload) {
|
||||
|
||||
await updateClaimedSchedule(
|
||||
buildScheduleFailureUpdate(now, nextRunAt),
|
||||
`Error updating schedule ${payload.scheduleId} after failure`
|
||||
`Error updating schedule ${payload.scheduleId} after failure`,
|
||||
'consecutive_failures'
|
||||
)
|
||||
} catch (error: unknown) {
|
||||
logger.error(
|
||||
@@ -995,7 +1085,8 @@ export async function executeScheduleJob(payload: ScheduleExecutionPayload) {
|
||||
|
||||
await updateClaimedSchedule(
|
||||
buildScheduleFailureUpdate(now, nextRunAt),
|
||||
`Error updating schedule ${payload.scheduleId} after execution error`
|
||||
`Error updating schedule ${payload.scheduleId} after execution error`,
|
||||
'consecutive_failures'
|
||||
)
|
||||
}
|
||||
} catch (error: unknown) {
|
||||
@@ -1468,7 +1559,10 @@ export async function executeJobInline(payload: JobExecutionPayload) {
|
||||
},
|
||||
requestId,
|
||||
`Error updating job ${payload.scheduleId} after failure`,
|
||||
{ expectedLastQueuedAt: now }
|
||||
{
|
||||
expectedLastQueuedAt: now,
|
||||
disableReason: shouldDisable ? 'consecutive_failures' : undefined,
|
||||
}
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -6,4 +6,5 @@ export { FreeTierUpgradeEmail } from './free-tier-upgrade-email'
|
||||
export { LimitThresholdEmail } from './limit-threshold-email'
|
||||
export { PaymentFailedEmail } from './payment-failed-email'
|
||||
export { PlanWelcomeEmail } from './plan-welcome-email'
|
||||
export { UsageLimitReachedEmail } from './usage-limit-reached-email'
|
||||
export { UsageThresholdEmail } from './usage-threshold-email'
|
||||
|
||||
@@ -0,0 +1,81 @@
|
||||
import { Link, Section, Text } from '@react-email/components'
|
||||
import { baseStyles } from '@/components/emails/_styles'
|
||||
import { EmailLayout } from '@/components/emails/components'
|
||||
import { dollarsToCredits } from '@/lib/billing/credits/conversion'
|
||||
import { getBrandConfig } from '@/ee/whitelabeling'
|
||||
|
||||
interface UsageLimitReachedEmailProps {
|
||||
userName?: string
|
||||
planName: string
|
||||
/** Drives the remedy copy and CTA label — org admins raise the org limit. */
|
||||
scope: 'user' | 'organization'
|
||||
currentUsage: number
|
||||
limit: number
|
||||
ctaLink: string
|
||||
}
|
||||
|
||||
/**
|
||||
* Sent at 100% of the usage limit to paid and organization accounts. The
|
||||
* free-tier equivalent is `CreditsExhaustedEmail`, whose remedy is an upgrade
|
||||
* rather than a limit change.
|
||||
*/
|
||||
export function UsageLimitReachedEmail({
|
||||
userName,
|
||||
planName,
|
||||
scope,
|
||||
currentUsage,
|
||||
limit,
|
||||
ctaLink,
|
||||
}: UsageLimitReachedEmailProps) {
|
||||
const brand = getBrandConfig()
|
||||
const isOrganization = scope === 'organization'
|
||||
const previewText = `${brand.name}: You've reached your ${planName} usage limit`
|
||||
|
||||
return (
|
||||
<EmailLayout preview={previewText} showUnsubscribe={true}>
|
||||
<Text style={{ ...baseStyles.paragraph, marginTop: 0 }}>
|
||||
{userName ? `Hi ${userName},` : 'Hi,'}
|
||||
</Text>
|
||||
|
||||
<Text style={baseStyles.paragraph}>
|
||||
{isOrganization
|
||||
? `Your organization has reached its monthly usage limit on the ${planName} plan.`
|
||||
: `You've reached your monthly usage limit on the ${planName} plan.`}{' '}
|
||||
Workflow runs, knowledge base search, document uploads, and Chat are paused until the limit
|
||||
goes up.
|
||||
</Text>
|
||||
|
||||
<Section style={baseStyles.infoBox}>
|
||||
<Text style={baseStyles.infoBoxTitle}>Usage</Text>
|
||||
<Text style={baseStyles.infoBoxList}>
|
||||
{dollarsToCredits(currentUsage).toLocaleString()} of{' '}
|
||||
{dollarsToCredits(limit).toLocaleString()} credits used
|
||||
</Text>
|
||||
</Section>
|
||||
|
||||
{/* Divider */}
|
||||
<div style={baseStyles.divider} />
|
||||
|
||||
<Text style={baseStyles.paragraph}>
|
||||
{isOrganization
|
||||
? 'Raise the organization usage limit in billing settings to resume.'
|
||||
: 'Raise your usage limit in billing settings, or upgrade your plan, to resume.'}
|
||||
</Text>
|
||||
|
||||
<Link href={ctaLink} style={{ textDecoration: 'none' }}>
|
||||
<Text style={baseStyles.button}>
|
||||
{isOrganization ? 'Raise Organization Limit' : 'Raise Usage Limit'}
|
||||
</Text>
|
||||
</Link>
|
||||
|
||||
{/* Divider */}
|
||||
<div style={baseStyles.divider} />
|
||||
|
||||
<Text style={{ ...baseStyles.footerText, textAlign: 'left' }}>
|
||||
Sent to the people who manage billing for this account.
|
||||
</Text>
|
||||
</EmailLayout>
|
||||
)
|
||||
}
|
||||
|
||||
export default UsageLimitReachedEmail
|
||||
@@ -8,6 +8,8 @@ export * from './billing'
|
||||
export * from './components'
|
||||
// Invitation emails
|
||||
export * from './invitations'
|
||||
// Operational notification emails
|
||||
export * from './notifications'
|
||||
// Render functions and subjects
|
||||
export * from './render'
|
||||
export * from './subjects'
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
export { ScheduleDisabledEmail } from './schedule-disabled-email'
|
||||
@@ -0,0 +1,83 @@
|
||||
import { Link, Section, Text } from '@react-email/components'
|
||||
import { baseStyles } from '@/components/emails/_styles'
|
||||
import { EmailLayout } from '@/components/emails/components'
|
||||
import {
|
||||
SCHEDULE_DISABLE_REASON_COPY,
|
||||
type ScheduleDisableReason,
|
||||
} from '@/lib/workflows/schedules/disable-reasons'
|
||||
import { getBrandConfig } from '@/ee/whitelabeling'
|
||||
|
||||
interface ScheduleDisabledEmailProps {
|
||||
recipientName?: string
|
||||
/** Drives the noun and the CTA label. */
|
||||
kind: 'workflow' | 'job'
|
||||
/** Workflow name or job title. Absent when the source row could not be read. */
|
||||
resourceName?: string
|
||||
reason: ScheduleDisableReason
|
||||
/** Consecutive failures at disable time. Rendered only for `consecutive_failures`. */
|
||||
failedCount?: number
|
||||
/** Deep link to the workflow or scheduled tasks. Absent when the workspace is unknown. */
|
||||
manageLink?: string
|
||||
}
|
||||
|
||||
/**
|
||||
* Sent when Sim turns a schedule off on its own — either after repeated
|
||||
* failures or on a terminal error that cannot resolve itself.
|
||||
*/
|
||||
export function ScheduleDisabledEmail({
|
||||
recipientName,
|
||||
kind,
|
||||
resourceName,
|
||||
reason,
|
||||
failedCount,
|
||||
manageLink,
|
||||
}: ScheduleDisabledEmailProps) {
|
||||
const brand = getBrandConfig()
|
||||
const resourceLabel =
|
||||
resourceName ?? (kind === 'job' ? 'a scheduled task' : 'a scheduled workflow')
|
||||
const reasonCopy =
|
||||
reason === 'consecutive_failures' && failedCount
|
||||
? `It failed ${failedCount.toLocaleString()} times in a row.`
|
||||
: SCHEDULE_DISABLE_REASON_COPY[reason]
|
||||
const previewText = `${brand.name} turned off the schedule for ${resourceLabel}`
|
||||
|
||||
return (
|
||||
<EmailLayout preview={previewText} showUnsubscribe={true}>
|
||||
<Text style={{ ...baseStyles.paragraph, marginTop: 0 }}>
|
||||
{recipientName ? `Hi ${recipientName},` : 'Hi,'}
|
||||
</Text>
|
||||
|
||||
<Text style={baseStyles.paragraph}>
|
||||
{brand.name} turned off the schedule for {resourceLabel}. It will not run again until you
|
||||
turn it back on.
|
||||
</Text>
|
||||
|
||||
<Section style={baseStyles.infoBox}>
|
||||
<Text style={baseStyles.infoBoxTitle}>Reason</Text>
|
||||
<Text style={baseStyles.infoBoxList}>{reasonCopy}</Text>
|
||||
</Section>
|
||||
|
||||
{/* Divider */}
|
||||
<div style={baseStyles.divider} />
|
||||
|
||||
<Text style={baseStyles.paragraph}>Fix the problem, then turn the schedule back on.</Text>
|
||||
|
||||
{manageLink ? (
|
||||
<Link href={manageLink} style={{ textDecoration: 'none' }}>
|
||||
<Text style={baseStyles.button}>
|
||||
{kind === 'job' ? 'Open scheduled tasks' : 'Open workflow'}
|
||||
</Text>
|
||||
</Link>
|
||||
) : null}
|
||||
|
||||
{/* Divider */}
|
||||
<div style={baseStyles.divider} />
|
||||
|
||||
<Text style={{ ...baseStyles.footerText, textAlign: 'left' }}>
|
||||
Sent to workspace admins and the person who created this schedule.
|
||||
</Text>
|
||||
</EmailLayout>
|
||||
)
|
||||
}
|
||||
|
||||
export default ScheduleDisabledEmail
|
||||
@@ -0,0 +1,98 @@
|
||||
/**
|
||||
* Executes the notification templates for real. Every other suite mocks the
|
||||
* render functions, so without this nothing would catch a template that throws
|
||||
* or silently drops its key copy.
|
||||
*
|
||||
* @vitest-environment node
|
||||
*/
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import {
|
||||
renderScheduleDisabledEmail,
|
||||
renderUsageLimitReachedEmail,
|
||||
} from '@/components/emails/render'
|
||||
|
||||
describe('renderScheduleDisabledEmail', () => {
|
||||
it('renders the failure count for a threshold disable', async () => {
|
||||
const html = await renderScheduleDisabledEmail({
|
||||
recipientName: 'John',
|
||||
kind: 'workflow',
|
||||
resourceName: 'Daily digest',
|
||||
reason: 'consecutive_failures',
|
||||
failedCount: 100,
|
||||
manageLink: 'https://sim.ai/workspace/ws_123/w/wf_456',
|
||||
})
|
||||
|
||||
expect(html).toContain('Daily digest')
|
||||
expect(html).toContain('It failed 100 times in a row.')
|
||||
expect(html).toContain('Open workflow')
|
||||
expect(html).toContain('https://sim.ai/workspace/ws_123/w/wf_456')
|
||||
})
|
||||
|
||||
it('renders reason copy instead of a count for a single-strike disable', async () => {
|
||||
const html = await renderScheduleDisabledEmail({
|
||||
kind: 'job',
|
||||
resourceName: 'Weekly report',
|
||||
reason: 'authentication_error',
|
||||
manageLink: 'https://sim.ai/workspace/ws_123/scheduled-tasks',
|
||||
})
|
||||
|
||||
expect(html).toContain('could not be authenticated')
|
||||
expect(html).not.toContain('times in a row')
|
||||
expect(html).toContain('Open scheduled tasks')
|
||||
})
|
||||
|
||||
it('falls back to a generic noun and omits the CTA when the source is unknown', async () => {
|
||||
const html = await renderScheduleDisabledEmail({
|
||||
kind: 'workflow',
|
||||
reason: 'workflow_not_found',
|
||||
})
|
||||
|
||||
expect(html).toContain('a scheduled workflow')
|
||||
expect(html).not.toContain('Open workflow')
|
||||
})
|
||||
})
|
||||
|
||||
describe('renderUsageLimitReachedEmail', () => {
|
||||
it('tells a personal payer to raise their limit, never to upgrade to Pro', async () => {
|
||||
const html = await renderUsageLimitReachedEmail({
|
||||
userName: 'John',
|
||||
planName: 'Pro',
|
||||
scope: 'user',
|
||||
currentUsage: 20,
|
||||
limit: 20,
|
||||
ctaLink: 'https://sim.ai/account/settings/billing',
|
||||
})
|
||||
|
||||
expect(html).toContain('Raise Usage Limit')
|
||||
expect(html).toContain('upgrade your plan')
|
||||
expect(html).not.toContain('Upgrade to Pro')
|
||||
expect(html).not.toContain('free credits')
|
||||
})
|
||||
|
||||
it('points an organization at the org limit', async () => {
|
||||
const html = await renderUsageLimitReachedEmail({
|
||||
planName: 'Team',
|
||||
scope: 'organization',
|
||||
currentUsage: 500,
|
||||
limit: 500,
|
||||
ctaLink: 'https://sim.ai/organization/org_1/settings/billing',
|
||||
})
|
||||
|
||||
expect(html).toContain('Raise Organization Limit')
|
||||
expect(html).toContain('organization usage limit')
|
||||
expect(html).not.toContain('upgrade your plan')
|
||||
})
|
||||
|
||||
it('renders credits, not dollars', async () => {
|
||||
const html = await renderUsageLimitReachedEmail({
|
||||
planName: 'Pro',
|
||||
scope: 'user',
|
||||
currentUsage: 20,
|
||||
limit: 20,
|
||||
ctaLink: 'https://sim.ai/account/settings/billing',
|
||||
})
|
||||
|
||||
expect(html).toContain('credits used')
|
||||
expect(html).not.toContain('$20')
|
||||
})
|
||||
})
|
||||
@@ -15,6 +15,7 @@ import {
|
||||
LimitThresholdEmail,
|
||||
PaymentFailedEmail,
|
||||
PlanWelcomeEmail,
|
||||
UsageLimitReachedEmail,
|
||||
UsageThresholdEmail,
|
||||
} from '@/components/emails/billing'
|
||||
import {
|
||||
@@ -23,9 +24,11 @@ import {
|
||||
WorkspaceAddedEmail,
|
||||
WorkspaceInvitationEmail,
|
||||
} from '@/components/emails/invitations'
|
||||
import { ScheduleDisabledEmail } from '@/components/emails/notifications'
|
||||
import { HelpConfirmationEmail } from '@/components/emails/support'
|
||||
import type { UpgradeReason } from '@/lib/billing/upgrade-reasons'
|
||||
import { getBaseUrl } from '@/lib/core/utils/urls'
|
||||
import type { ScheduleDisableReason } from '@/lib/workflows/schedules/disable-reasons'
|
||||
|
||||
export { getEmailSubject, getLimitEmailSubject } from './subjects'
|
||||
|
||||
@@ -136,6 +139,28 @@ export async function renderUsageThresholdEmail(params: {
|
||||
)
|
||||
}
|
||||
|
||||
export async function renderUsageLimitReachedEmail(params: {
|
||||
userName?: string
|
||||
planName: string
|
||||
scope: 'user' | 'organization'
|
||||
currentUsage: number
|
||||
limit: number
|
||||
ctaLink: string
|
||||
}): Promise<string> {
|
||||
return await render(UsageLimitReachedEmail(params))
|
||||
}
|
||||
|
||||
export async function renderScheduleDisabledEmail(params: {
|
||||
recipientName?: string
|
||||
kind: 'workflow' | 'job'
|
||||
resourceName?: string
|
||||
reason: ScheduleDisableReason
|
||||
failedCount?: number
|
||||
manageLink?: string
|
||||
}): Promise<string> {
|
||||
return await render(ScheduleDisabledEmail(params))
|
||||
}
|
||||
|
||||
export async function renderFreeTierUpgradeEmail(params: {
|
||||
userName?: string
|
||||
percentUsed: number
|
||||
|
||||
@@ -21,6 +21,7 @@ export type EmailSubjectType =
|
||||
| 'credit-purchase'
|
||||
| 'abandoned-checkout'
|
||||
| 'free-tier-exhausted'
|
||||
| 'schedule-disabled'
|
||||
| 'onboarding-followup'
|
||||
| 'welcome'
|
||||
|
||||
@@ -69,6 +70,8 @@ export function getEmailSubject(type: EmailSubjectType): string {
|
||||
return `Quick question`
|
||||
case 'free-tier-exhausted':
|
||||
return `You've run out of free credits on ${brandName}`
|
||||
case 'schedule-disabled':
|
||||
return `A schedule was turned off on ${brandName}`
|
||||
case 'onboarding-followup':
|
||||
return `Quick question about ${brandName}`
|
||||
case 'welcome':
|
||||
|
||||
@@ -8,7 +8,14 @@
|
||||
*
|
||||
* @vitest-environment node
|
||||
*/
|
||||
import { dbChainMockFns, resetDbChainMock } from '@sim/testing'
|
||||
import {
|
||||
dbChainMockFns,
|
||||
queueTableRows,
|
||||
resetDbChainMock,
|
||||
resetEnvFlagsMock,
|
||||
schemaMock,
|
||||
setEnvFlags,
|
||||
} from '@sim/testing'
|
||||
import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
afterAll(() => {
|
||||
@@ -56,22 +63,52 @@ vi.mock('@/lib/billing/credits/daily-refresh', () => ({
|
||||
getOrgMemberRefreshBounds: vi.fn(),
|
||||
}))
|
||||
|
||||
const {
|
||||
mockGetEmailSubject,
|
||||
mockGetLimitEmailSubject,
|
||||
mockRenderCreditsExhausted,
|
||||
mockRenderFreeTierUpgrade,
|
||||
mockRenderUsageLimitReached,
|
||||
mockRenderUsageThreshold,
|
||||
mockSendEmail,
|
||||
mockGetEmailPreferences,
|
||||
mockIsOrgAdminRole,
|
||||
} = vi.hoisted(() => ({
|
||||
mockGetEmailSubject: vi.fn(() => 'Subject'),
|
||||
mockGetLimitEmailSubject: vi.fn(() => 'Limit subject'),
|
||||
mockRenderCreditsExhausted: vi.fn(() => Promise.resolve('<html>free</html>')),
|
||||
mockRenderFreeTierUpgrade: vi.fn(() => Promise.resolve('<html>nudge</html>')),
|
||||
mockRenderUsageLimitReached: vi.fn(() => Promise.resolve('<html>reached</html>')),
|
||||
mockRenderUsageThreshold: vi.fn(() => Promise.resolve('<html>warning</html>')),
|
||||
mockSendEmail: vi.fn(() => Promise.resolve({ success: true })),
|
||||
mockGetEmailPreferences: vi.fn(() => Promise.resolve(null as unknown)),
|
||||
mockIsOrgAdminRole: vi.fn(() => true),
|
||||
}))
|
||||
|
||||
vi.mock('@/components/emails', () => ({
|
||||
getEmailSubject: vi.fn(),
|
||||
renderCreditsExhaustedEmail: vi.fn(),
|
||||
renderFreeTierUpgradeEmail: vi.fn(),
|
||||
renderUsageThresholdEmail: vi.fn(),
|
||||
getEmailSubject: mockGetEmailSubject,
|
||||
getLimitEmailSubject: mockGetLimitEmailSubject,
|
||||
renderCreditsExhaustedEmail: mockRenderCreditsExhausted,
|
||||
renderFreeTierUpgradeEmail: mockRenderFreeTierUpgrade,
|
||||
renderUsageLimitReachedEmail: mockRenderUsageLimitReached,
|
||||
renderUsageThresholdEmail: mockRenderUsageThreshold,
|
||||
}))
|
||||
|
||||
vi.mock('@/lib/messaging/email/mailer', () => ({
|
||||
sendEmail: vi.fn(),
|
||||
sendEmail: mockSendEmail,
|
||||
}))
|
||||
|
||||
vi.mock('@/lib/messaging/email/unsubscribe', () => ({
|
||||
getEmailPreferences: vi.fn(),
|
||||
getEmailPreferences: mockGetEmailPreferences,
|
||||
}))
|
||||
|
||||
import { getUserUsageLimit, syncUsageLimitsFromSubscription } from '@/lib/billing/core/usage'
|
||||
vi.mock('@sim/platform-authz/workspace', () => ({ isOrgAdminRole: mockIsOrgAdminRole }))
|
||||
|
||||
import {
|
||||
getUserUsageLimit,
|
||||
maybeSendUsageThresholdEmail,
|
||||
syncUsageLimitsFromSubscription,
|
||||
} from '@/lib/billing/core/usage'
|
||||
|
||||
const PRO_SUBSCRIPTION = {
|
||||
id: 'sub-1',
|
||||
@@ -232,3 +269,159 @@ describe('syncUsageLimitsFromSubscription', () => {
|
||||
expect(expression).toContain('creditBalance')
|
||||
})
|
||||
})
|
||||
|
||||
describe('maybeSendUsageThresholdEmail', () => {
|
||||
const paidUser = {
|
||||
scope: 'user' as const,
|
||||
planName: 'Pro',
|
||||
userId: 'user-1',
|
||||
userEmail: 'user-1@example.com',
|
||||
userName: 'Ada',
|
||||
workspaceId: 'ws-1',
|
||||
limit: 20,
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks()
|
||||
resetDbChainMock()
|
||||
setEnvFlags({ isBillingEnabled: true })
|
||||
mockGetEmailPreferences.mockResolvedValue(null)
|
||||
mockIsOrgAdminRole.mockReturnValue(true)
|
||||
})
|
||||
|
||||
afterAll(() => {
|
||||
resetEnvFlagsMock()
|
||||
})
|
||||
|
||||
it('emails a paid personal account at 100% with the raise-your-limit template', async () => {
|
||||
await maybeSendUsageThresholdEmail({
|
||||
...paidUser,
|
||||
percentBefore: 90,
|
||||
percentAfter: 100,
|
||||
currentUsageAfter: 20,
|
||||
})
|
||||
|
||||
expect(mockRenderUsageLimitReached).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ scope: 'user', planName: 'Pro' })
|
||||
)
|
||||
expect(mockRenderCreditsExhausted).not.toHaveBeenCalled()
|
||||
expect(mockGetLimitEmailSubject).toHaveBeenCalledWith('credits', 'reached')
|
||||
expect(mockSendEmail).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('links a paid personal account to billing settings, not the upgrade page', async () => {
|
||||
await maybeSendUsageThresholdEmail({
|
||||
...paidUser,
|
||||
percentBefore: 90,
|
||||
percentAfter: 100,
|
||||
currentUsageAfter: 20,
|
||||
})
|
||||
|
||||
const props = mockRenderUsageLimitReached.mock.calls[0]?.[0] as { ctaLink: string }
|
||||
expect(props.ctaLink).toContain('/account/settings/billing')
|
||||
})
|
||||
|
||||
it('fans out to org admins at 100% and skips non-admin members', async () => {
|
||||
mockIsOrgAdminRole.mockImplementation((role: unknown) => role === 'admin')
|
||||
queueTableRows(schemaMock.member, [
|
||||
{ email: 'admin@example.com', name: 'Admin', enabled: null, role: 'admin' },
|
||||
{ email: 'member@example.com', name: 'Member', enabled: null, role: 'member' },
|
||||
])
|
||||
|
||||
await maybeSendUsageThresholdEmail({
|
||||
scope: 'organization',
|
||||
planName: 'Team',
|
||||
organizationId: 'org-1',
|
||||
workspaceId: 'ws-1',
|
||||
percentBefore: 95,
|
||||
percentAfter: 100,
|
||||
currentUsageAfter: 500,
|
||||
limit: 500,
|
||||
})
|
||||
|
||||
expect(mockSendEmail).toHaveBeenCalledTimes(1)
|
||||
expect(mockSendEmail).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ to: 'admin@example.com', emailType: 'notifications' })
|
||||
)
|
||||
expect(mockRenderUsageLimitReached).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ scope: 'organization' })
|
||||
)
|
||||
})
|
||||
|
||||
it('still sends the free-tier template to a free personal account at 100%', async () => {
|
||||
await maybeSendUsageThresholdEmail({
|
||||
...paidUser,
|
||||
planName: 'Free',
|
||||
percentBefore: 90,
|
||||
percentAfter: 100,
|
||||
currentUsageAfter: 10,
|
||||
limit: 10,
|
||||
})
|
||||
|
||||
expect(mockRenderCreditsExhausted).toHaveBeenCalledTimes(1)
|
||||
expect(mockRenderUsageLimitReached).not.toHaveBeenCalled()
|
||||
expect(mockGetEmailSubject).toHaveBeenCalledWith('free-tier-exhausted')
|
||||
})
|
||||
|
||||
it('sends only the reached email when one execution crosses 80 and 100 together', async () => {
|
||||
await maybeSendUsageThresholdEmail({
|
||||
...paidUser,
|
||||
percentBefore: 70,
|
||||
percentAfter: 100,
|
||||
currentUsageAfter: 20,
|
||||
})
|
||||
|
||||
expect(mockRenderUsageThreshold).not.toHaveBeenCalled()
|
||||
expect(mockRenderUsageLimitReached).toHaveBeenCalledTimes(1)
|
||||
expect(mockSendEmail).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('still sends the 80% warning to a paid account that has not reached 100%', async () => {
|
||||
await maybeSendUsageThresholdEmail({
|
||||
...paidUser,
|
||||
percentBefore: 70,
|
||||
percentAfter: 85,
|
||||
currentUsageAfter: 17,
|
||||
})
|
||||
|
||||
expect(mockRenderUsageThreshold).toHaveBeenCalledTimes(1)
|
||||
expect(mockRenderUsageLimitReached).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('does not send when the recipient unsubscribed from notifications', async () => {
|
||||
mockGetEmailPreferences.mockResolvedValue({ unsubscribeNotifications: true })
|
||||
|
||||
await maybeSendUsageThresholdEmail({
|
||||
...paidUser,
|
||||
percentBefore: 90,
|
||||
percentAfter: 100,
|
||||
currentUsageAfter: 20,
|
||||
})
|
||||
|
||||
expect(mockSendEmail).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('does not send when the per-user billing toggle is off', async () => {
|
||||
queueTableRows(schemaMock.settings, [{ enabled: false }])
|
||||
|
||||
await maybeSendUsageThresholdEmail({
|
||||
...paidUser,
|
||||
percentBefore: 90,
|
||||
percentAfter: 100,
|
||||
currentUsageAfter: 20,
|
||||
})
|
||||
|
||||
expect(mockSendEmail).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('does nothing when no threshold is crossed', async () => {
|
||||
await maybeSendUsageThresholdEmail({
|
||||
...paidUser,
|
||||
percentBefore: 50,
|
||||
percentAfter: 60,
|
||||
currentUsageAfter: 12,
|
||||
})
|
||||
|
||||
expect(mockSendEmail).not.toHaveBeenCalled()
|
||||
})
|
||||
})
|
||||
|
||||
@@ -6,8 +6,10 @@ import { generateId } from '@sim/utils/id'
|
||||
import { and, eq, isNull, sql } from 'drizzle-orm'
|
||||
import {
|
||||
getEmailSubject,
|
||||
getLimitEmailSubject,
|
||||
renderCreditsExhaustedEmail,
|
||||
renderFreeTierUpgradeEmail,
|
||||
renderUsageLimitReachedEmail,
|
||||
renderUsageThresholdEmail,
|
||||
} from '@/components/emails'
|
||||
import { getEffectiveBillingStatus } from '@/lib/billing/core/access'
|
||||
@@ -863,16 +865,55 @@ export async function maybeSendUsageThresholdEmail(params: {
|
||||
|
||||
// Check for 80% threshold crossing — used for paid users (budget warning) and free users (upgrade nudge)
|
||||
const crosses80 = params.percentBefore < 80 && params.percentAfter >= 80
|
||||
// Check for 100% threshold (free users only — credits exhausted)
|
||||
// Check for 100% threshold — every plan and scope (usage limit reached)
|
||||
const crosses100 = params.percentBefore < 100 && params.percentAfter >= 100
|
||||
|
||||
// Skip if no thresholds crossed
|
||||
if (!crosses80 && !crosses100) return
|
||||
|
||||
// For 80% threshold email (paid users only)
|
||||
if (crosses80 && !isFreeUser) {
|
||||
/**
|
||||
* Delivers to the account's notification recipients: the payer for personal
|
||||
* scope, every org admin/owner for organization scope. Honors the per-user
|
||||
* billing-notification toggle in both.
|
||||
*/
|
||||
const deliverToScope = async (send: (email: string, name?: string) => Promise<void>) => {
|
||||
if (params.scope === 'user' && params.userId && params.userEmail) {
|
||||
const rows = await db
|
||||
.select({ enabled: settings.billingUsageNotificationsEnabled })
|
||||
.from(settings)
|
||||
.where(eq(settings.userId, params.userId))
|
||||
.limit(1)
|
||||
if (rows.length > 0 && rows[0].enabled === false) return
|
||||
await send(params.userEmail, params.userName)
|
||||
return
|
||||
}
|
||||
|
||||
if (params.scope === 'organization' && params.organizationId) {
|
||||
const admins = await db
|
||||
.select({
|
||||
email: user.email,
|
||||
name: user.name,
|
||||
enabled: settings.billingUsageNotificationsEnabled,
|
||||
role: member.role,
|
||||
})
|
||||
.from(member)
|
||||
.innerJoin(user, eq(member.userId, user.id))
|
||||
.leftJoin(settings, eq(settings.userId, member.userId))
|
||||
.where(eq(member.organizationId, params.organizationId))
|
||||
|
||||
for (const a of admins) {
|
||||
if (!isOrgAdminRole(a.role)) continue
|
||||
if (a.enabled === false) continue
|
||||
if (!a.email) continue
|
||||
await send(a.email, a.name || undefined)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// !crosses100: one "reached" email, not a "nearing" and a "reached" in the same moment
|
||||
if (crosses80 && !isFreeUser && !crosses100) {
|
||||
const ctaLink = billingSettingsLink
|
||||
const sendTo = async (email: string, name?: string) => {
|
||||
await deliverToScope(async (email, name) => {
|
||||
const prefs = await getEmailPreferences(email)
|
||||
if (prefs?.unsubscribeAll || prefs?.unsubscribeNotifications) return
|
||||
|
||||
@@ -891,43 +932,13 @@ export async function maybeSendUsageThresholdEmail(params: {
|
||||
html,
|
||||
emailType: 'notifications',
|
||||
})
|
||||
}
|
||||
|
||||
if (params.scope === 'user' && params.userId && params.userEmail) {
|
||||
const rows = await db
|
||||
.select({ enabled: settings.billingUsageNotificationsEnabled })
|
||||
.from(settings)
|
||||
.where(eq(settings.userId, params.userId))
|
||||
.limit(1)
|
||||
if (rows.length > 0 && rows[0].enabled === false) return
|
||||
await sendTo(params.userEmail, params.userName)
|
||||
} else if (params.scope === 'organization' && params.organizationId) {
|
||||
const admins = await db
|
||||
.select({
|
||||
email: user.email,
|
||||
name: user.name,
|
||||
enabled: settings.billingUsageNotificationsEnabled,
|
||||
role: member.role,
|
||||
})
|
||||
.from(member)
|
||||
.innerJoin(user, eq(member.userId, user.id))
|
||||
.leftJoin(settings, eq(settings.userId, member.userId))
|
||||
.where(eq(member.organizationId, params.organizationId))
|
||||
|
||||
for (const a of admins) {
|
||||
const isAdmin = isOrgAdminRole(a.role)
|
||||
if (!isAdmin) continue
|
||||
if (a.enabled === false) continue
|
||||
if (!a.email) continue
|
||||
await sendTo(a.email, a.name || undefined)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// For 80% threshold email (free users only — skip if they also crossed 100% in same call)
|
||||
if (crosses80 && isFreeUser && !crosses100) {
|
||||
const upgradeLink = upgradeCreditsLink
|
||||
const sendFreeTierEmail = async (email: string, name?: string) => {
|
||||
await deliverToScope(async (email, name) => {
|
||||
const prefs = await getEmailPreferences(email)
|
||||
if (prefs?.unsubscribeAll || prefs?.unsubscribeNotifications) return
|
||||
|
||||
@@ -952,56 +963,49 @@ export async function maybeSendUsageThresholdEmail(params: {
|
||||
currentUsage: params.currentUsageAfter,
|
||||
limit: params.limit,
|
||||
})
|
||||
}
|
||||
|
||||
// Free users are always individual scope (not organization)
|
||||
if (params.scope === 'user' && params.userId && params.userEmail) {
|
||||
const rows = await db
|
||||
.select({ enabled: settings.billingUsageNotificationsEnabled })
|
||||
.from(settings)
|
||||
.where(eq(settings.userId, params.userId))
|
||||
.limit(1)
|
||||
if (rows.length > 0 && rows[0].enabled === false) return
|
||||
await sendFreeTierEmail(params.userEmail, params.userName)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// For 100% threshold email (free users only — credits exhausted)
|
||||
if (crosses100 && isFreeUser) {
|
||||
const upgradeLink = upgradeCreditsLink
|
||||
const sendExhaustedEmail = async (email: string, name?: string) => {
|
||||
// Paid and org accounts get raise-your-limit copy — upgrading is not their remedy
|
||||
if (crosses100) {
|
||||
const useFreeCopy = isFreeUser && params.scope === 'user'
|
||||
|
||||
await deliverToScope(async (email, name) => {
|
||||
const prefs = await getEmailPreferences(email)
|
||||
if (prefs?.unsubscribeAll || prefs?.unsubscribeNotifications) return
|
||||
|
||||
const html = await renderCreditsExhaustedEmail({
|
||||
userName: name,
|
||||
limit: params.limit,
|
||||
upgradeLink,
|
||||
})
|
||||
const html = useFreeCopy
|
||||
? await renderCreditsExhaustedEmail({
|
||||
userName: name,
|
||||
limit: params.limit,
|
||||
upgradeLink: upgradeCreditsLink,
|
||||
})
|
||||
: await renderUsageLimitReachedEmail({
|
||||
userName: name,
|
||||
planName: params.planName,
|
||||
scope: params.scope,
|
||||
currentUsage: params.currentUsageAfter,
|
||||
limit: params.limit,
|
||||
ctaLink: billingSettingsLink,
|
||||
})
|
||||
|
||||
await sendEmail({
|
||||
to: email,
|
||||
subject: getEmailSubject('free-tier-exhausted'),
|
||||
subject: useFreeCopy
|
||||
? getEmailSubject('free-tier-exhausted')
|
||||
: getLimitEmailSubject('credits', 'reached'),
|
||||
html,
|
||||
emailType: 'notifications',
|
||||
})
|
||||
|
||||
logger.info('Free tier credits exhausted email sent', {
|
||||
logger.info('Usage limit reached email sent', {
|
||||
email,
|
||||
scope: params.scope,
|
||||
planName: params.planName,
|
||||
currentUsage: params.currentUsageAfter,
|
||||
limit: params.limit,
|
||||
})
|
||||
}
|
||||
|
||||
if (params.scope === 'user' && params.userId && params.userEmail) {
|
||||
const rows = await db
|
||||
.select({ enabled: settings.billingUsageNotificationsEnabled })
|
||||
.from(settings)
|
||||
.where(eq(settings.userId, params.userId))
|
||||
.limit(1)
|
||||
if (rows.length > 0 && rows[0].enabled === false) return
|
||||
await sendExhaustedEmail(params.userEmail, params.userName)
|
||||
}
|
||||
})
|
||||
}
|
||||
} catch (error) {
|
||||
logger.error('Failed to send usage threshold email', {
|
||||
|
||||
@@ -0,0 +1,227 @@
|
||||
/**
|
||||
* @vitest-environment node
|
||||
*/
|
||||
import {
|
||||
dbChainMockFns,
|
||||
queueTableRows,
|
||||
resetDbChainMock,
|
||||
resetUrlsMock,
|
||||
schemaMock,
|
||||
urlsMockFns,
|
||||
} from '@sim/testing'
|
||||
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
const { sendEmailSpy, renderMock, subjectMock, getUsersWithPermissionsMock } = vi.hoisted(() => ({
|
||||
sendEmailSpy: vi.fn(() => Promise.resolve({ success: true })),
|
||||
renderMock: vi.fn(() => Promise.resolve('<html></html>')),
|
||||
subjectMock: vi.fn(() => 'A schedule was turned off'),
|
||||
getUsersWithPermissionsMock: vi.fn(() => Promise.resolve([] as unknown[])),
|
||||
}))
|
||||
|
||||
vi.mock('@/lib/messaging/email/mailer', () => ({ sendEmail: sendEmailSpy }))
|
||||
vi.mock('@/components/emails', () => ({
|
||||
renderScheduleDisabledEmail: renderMock,
|
||||
getEmailSubject: subjectMock,
|
||||
}))
|
||||
vi.mock('@/lib/workspaces/permissions/utils', () => ({
|
||||
getUsersWithPermissions: getUsersWithPermissionsMock,
|
||||
}))
|
||||
|
||||
import { notifyScheduleAutoDisabled } from '@/lib/workflows/schedules/disable-notifications'
|
||||
|
||||
const WORKFLOW_SCHEDULE_ROW = {
|
||||
sourceType: 'workflow',
|
||||
jobTitle: null,
|
||||
failedCount: 100,
|
||||
sourceUserId: null,
|
||||
sourceWorkspaceId: null,
|
||||
workflowId: 'wf-1',
|
||||
workflowName: 'Daily digest',
|
||||
workflowUserId: 'creator-1',
|
||||
workflowWorkspaceId: 'ws-1',
|
||||
}
|
||||
|
||||
const CREATOR = { email: 'creator@example.com', name: 'Ada' }
|
||||
|
||||
function admin(email: string, name = 'Admin') {
|
||||
return { userId: `u-${email}`, email, name, permissionType: 'admin' }
|
||||
}
|
||||
|
||||
beforeAll(() => {
|
||||
urlsMockFns.mockGetBaseUrl.mockReturnValue('https://app.sim.ai')
|
||||
})
|
||||
|
||||
afterAll(() => {
|
||||
resetDbChainMock()
|
||||
resetUrlsMock()
|
||||
})
|
||||
|
||||
describe('notifyScheduleAutoDisabled', () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks()
|
||||
resetDbChainMock()
|
||||
getUsersWithPermissionsMock.mockResolvedValue([])
|
||||
})
|
||||
|
||||
it('emails the creator and the workspace admins, one send per address', async () => {
|
||||
queueTableRows(schemaMock.workflowSchedule, [WORKFLOW_SCHEDULE_ROW])
|
||||
queueTableRows(schemaMock.user, [CREATOR])
|
||||
getUsersWithPermissionsMock.mockResolvedValue([
|
||||
admin('admin-a@example.com'),
|
||||
admin('admin-b@example.com'),
|
||||
])
|
||||
|
||||
await notifyScheduleAutoDisabled({ scheduleId: 's-1', reason: 'consecutive_failures' })
|
||||
|
||||
expect(sendEmailSpy).toHaveBeenCalledTimes(3)
|
||||
for (const call of sendEmailSpy.mock.calls) {
|
||||
// Never a `to` array — prepare.ts only checks to[0] for unsubscribe.
|
||||
expect(typeof (call[0] as { to: unknown }).to).toBe('string')
|
||||
expect(call[0]).toMatchObject({ emailType: 'notifications' })
|
||||
}
|
||||
})
|
||||
|
||||
it('sends once when the creator is also a workspace admin', async () => {
|
||||
queueTableRows(schemaMock.workflowSchedule, [WORKFLOW_SCHEDULE_ROW])
|
||||
queueTableRows(schemaMock.user, [CREATOR])
|
||||
getUsersWithPermissionsMock.mockResolvedValue([admin('CREATOR@example.com')])
|
||||
|
||||
await notifyScheduleAutoDisabled({ scheduleId: 's-1', reason: 'consecutive_failures' })
|
||||
|
||||
expect(sendEmailSpy).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('skips non-admin workspace members', async () => {
|
||||
queueTableRows(schemaMock.workflowSchedule, [WORKFLOW_SCHEDULE_ROW])
|
||||
queueTableRows(schemaMock.user, [CREATOR])
|
||||
getUsersWithPermissionsMock.mockResolvedValue([
|
||||
{ userId: 'u-2', email: 'writer@example.com', name: 'W', permissionType: 'write' },
|
||||
])
|
||||
|
||||
await notifyScheduleAutoDisabled({ scheduleId: 's-1', reason: 'consecutive_failures' })
|
||||
|
||||
expect(sendEmailSpy).toHaveBeenCalledTimes(1)
|
||||
expect(sendEmailSpy).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ to: 'creator@example.com' })
|
||||
)
|
||||
})
|
||||
|
||||
it('resolves a job schedule through sourceUserId and links to scheduled tasks', async () => {
|
||||
queueTableRows(schemaMock.workflowSchedule, [
|
||||
{
|
||||
...WORKFLOW_SCHEDULE_ROW,
|
||||
sourceType: 'job',
|
||||
jobTitle: 'Weekly report',
|
||||
sourceUserId: 'job-owner',
|
||||
sourceWorkspaceId: 'ws-9',
|
||||
workflowName: null,
|
||||
workflowUserId: null,
|
||||
workflowWorkspaceId: null,
|
||||
},
|
||||
])
|
||||
queueTableRows(schemaMock.user, [CREATOR])
|
||||
|
||||
await notifyScheduleAutoDisabled({ scheduleId: 's-1', reason: 'authentication_error' })
|
||||
|
||||
expect(renderMock).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
kind: 'job',
|
||||
resourceName: 'Weekly report',
|
||||
reason: 'authentication_error',
|
||||
manageLink: 'https://app.sim.ai/workspace/ws-9/scheduled-tasks',
|
||||
})
|
||||
)
|
||||
})
|
||||
|
||||
it('falls back to the creator alone when the workflow has no workspace', async () => {
|
||||
queueTableRows(schemaMock.workflowSchedule, [
|
||||
{ ...WORKFLOW_SCHEDULE_ROW, workflowWorkspaceId: null },
|
||||
])
|
||||
queueTableRows(schemaMock.user, [CREATOR])
|
||||
|
||||
await notifyScheduleAutoDisabled({ scheduleId: 's-1', reason: 'consecutive_failures' })
|
||||
|
||||
expect(getUsersWithPermissionsMock).not.toHaveBeenCalled()
|
||||
expect(sendEmailSpy).toHaveBeenCalledTimes(1)
|
||||
expect(renderMock).toHaveBeenCalledWith(expect.objectContaining({ manageLink: undefined }))
|
||||
})
|
||||
|
||||
it('sends nothing when the workflow row is gone (404 disable)', async () => {
|
||||
queueTableRows(schemaMock.workflowSchedule, [
|
||||
{
|
||||
...WORKFLOW_SCHEDULE_ROW,
|
||||
workflowName: null,
|
||||
workflowUserId: null,
|
||||
workflowWorkspaceId: null,
|
||||
},
|
||||
])
|
||||
|
||||
await notifyScheduleAutoDisabled({ scheduleId: 's-1', reason: 'workflow_not_found' })
|
||||
|
||||
expect(sendEmailSpy).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('sends nothing when the schedule row cannot be read', async () => {
|
||||
queueTableRows(schemaMock.workflowSchedule, [])
|
||||
|
||||
await notifyScheduleAutoDisabled({ scheduleId: 's-1', reason: 'consecutive_failures' })
|
||||
|
||||
expect(sendEmailSpy).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('caps the fan-out at 20 recipients', async () => {
|
||||
queueTableRows(schemaMock.workflowSchedule, [WORKFLOW_SCHEDULE_ROW])
|
||||
queueTableRows(schemaMock.user, [CREATOR])
|
||||
getUsersWithPermissionsMock.mockResolvedValue(
|
||||
Array.from({ length: 40 }, (_, i) => admin(`admin-${i}@example.com`))
|
||||
)
|
||||
|
||||
await notifyScheduleAutoDisabled({ scheduleId: 's-1', reason: 'consecutive_failures' })
|
||||
|
||||
expect(sendEmailSpy).toHaveBeenCalledTimes(20)
|
||||
expect(sendEmailSpy).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ to: 'creator@example.com' })
|
||||
)
|
||||
})
|
||||
|
||||
it('keeps sending after one recipient fails', async () => {
|
||||
queueTableRows(schemaMock.workflowSchedule, [WORKFLOW_SCHEDULE_ROW])
|
||||
queueTableRows(schemaMock.user, [CREATOR])
|
||||
getUsersWithPermissionsMock.mockResolvedValue([admin('admin-a@example.com')])
|
||||
sendEmailSpy.mockRejectedValueOnce(new Error('smtp down'))
|
||||
|
||||
await notifyScheduleAutoDisabled({ scheduleId: 's-1', reason: 'consecutive_failures' })
|
||||
|
||||
expect(sendEmailSpy).toHaveBeenCalledTimes(2)
|
||||
})
|
||||
|
||||
it('still emails the creator when the admin lookup fails', async () => {
|
||||
queueTableRows(schemaMock.workflowSchedule, [WORKFLOW_SCHEDULE_ROW])
|
||||
queueTableRows(schemaMock.user, [CREATOR])
|
||||
getUsersWithPermissionsMock.mockRejectedValue(new Error('db down'))
|
||||
|
||||
await expect(
|
||||
notifyScheduleAutoDisabled({ scheduleId: 's-1', reason: 'consecutive_failures' })
|
||||
).resolves.toBeUndefined()
|
||||
|
||||
expect(sendEmailSpy).toHaveBeenCalledTimes(1)
|
||||
expect(sendEmailSpy).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ to: 'creator@example.com' })
|
||||
)
|
||||
})
|
||||
|
||||
it('still emails the admins when the creator lookup fails', async () => {
|
||||
// First .limit() is the schedule row, second is the creator lookup.
|
||||
dbChainMockFns.limit
|
||||
.mockResolvedValueOnce([WORKFLOW_SCHEDULE_ROW])
|
||||
.mockRejectedValueOnce(new Error('db down'))
|
||||
getUsersWithPermissionsMock.mockResolvedValue([admin('admin-a@example.com')])
|
||||
|
||||
await notifyScheduleAutoDisabled({ scheduleId: 's-1', reason: 'consecutive_failures' })
|
||||
|
||||
expect(sendEmailSpy).toHaveBeenCalledTimes(1)
|
||||
expect(sendEmailSpy).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ to: 'admin-a@example.com' })
|
||||
)
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,175 @@
|
||||
import { db } from '@sim/db'
|
||||
import { user, workflow, workflowSchedule } from '@sim/db/schema'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { eq } from 'drizzle-orm'
|
||||
import { getEmailSubject, renderScheduleDisabledEmail } from '@/components/emails'
|
||||
import { getBaseUrl } from '@/lib/core/utils/urls'
|
||||
import { sendEmail } from '@/lib/messaging/email/mailer'
|
||||
import type { ScheduleDisableReason } from '@/lib/workflows/schedules/disable-reasons'
|
||||
import { getUsersWithPermissions } from '@/lib/workspaces/permissions/utils'
|
||||
|
||||
const logger = createLogger('ScheduleDisableNotifications')
|
||||
|
||||
/**
|
||||
* Fan-out cap so one tick over a large workspace can't turn into hundreds of
|
||||
* inline sends. The schedule's creator is always kept.
|
||||
*/
|
||||
const MAX_SCHEDULE_DISABLE_RECIPIENTS = 20
|
||||
|
||||
interface Recipient {
|
||||
email: string
|
||||
name: string | null
|
||||
}
|
||||
|
||||
/**
|
||||
* Emails the schedule's creator and the workspace admins after Sim turns a
|
||||
* schedule off on its own.
|
||||
*
|
||||
* Best-effort by contract: this never throws, because a mail failure must not
|
||||
* fault the schedule tick that called it.
|
||||
*/
|
||||
export async function notifyScheduleAutoDisabled(params: {
|
||||
scheduleId: string
|
||||
reason: ScheduleDisableReason
|
||||
requestId?: string
|
||||
}): Promise<void> {
|
||||
const { scheduleId, reason, requestId } = params
|
||||
|
||||
try {
|
||||
const rows = await db
|
||||
.select({
|
||||
sourceType: workflowSchedule.sourceType,
|
||||
jobTitle: workflowSchedule.jobTitle,
|
||||
failedCount: workflowSchedule.failedCount,
|
||||
sourceUserId: workflowSchedule.sourceUserId,
|
||||
sourceWorkspaceId: workflowSchedule.sourceWorkspaceId,
|
||||
workflowId: workflowSchedule.workflowId,
|
||||
workflowName: workflow.name,
|
||||
workflowUserId: workflow.userId,
|
||||
workflowWorkspaceId: workflow.workspaceId,
|
||||
})
|
||||
.from(workflowSchedule)
|
||||
.leftJoin(workflow, eq(workflow.id, workflowSchedule.workflowId))
|
||||
.where(eq(workflowSchedule.id, scheduleId))
|
||||
.limit(1)
|
||||
|
||||
const row = rows[0]
|
||||
if (!row) {
|
||||
logger.warn('Schedule auto-disabled but the row could not be read', { scheduleId, reason })
|
||||
return
|
||||
}
|
||||
|
||||
const isJob = row.sourceType === 'job'
|
||||
const kind = isJob ? 'job' : 'workflow'
|
||||
const ownerUserId = isJob ? row.sourceUserId : row.workflowUserId
|
||||
const workspaceId = isJob ? row.sourceWorkspaceId : row.workflowWorkspaceId
|
||||
const resourceName = (isJob ? row.jobTitle : row.workflowName) ?? undefined
|
||||
|
||||
const recipients = await resolveRecipients(ownerUserId, workspaceId)
|
||||
if (recipients.length === 0) {
|
||||
logger.warn('Schedule auto-disabled but no recipients could be resolved', {
|
||||
scheduleId,
|
||||
reason,
|
||||
workspaceId,
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
const manageLink = buildManageLink(workspaceId, isJob ? null : row.workflowId)
|
||||
const subject = getEmailSubject('schedule-disabled')
|
||||
|
||||
for (const recipient of recipients) {
|
||||
try {
|
||||
const html = await renderScheduleDisabledEmail({
|
||||
recipientName: recipient.name ?? undefined,
|
||||
kind,
|
||||
resourceName,
|
||||
reason,
|
||||
failedCount: row.failedCount,
|
||||
manageLink,
|
||||
})
|
||||
|
||||
await sendEmail({
|
||||
to: recipient.email,
|
||||
subject,
|
||||
html,
|
||||
emailType: 'notifications',
|
||||
})
|
||||
} catch (error) {
|
||||
logger.error('Failed to send schedule-disabled email', error, {
|
||||
scheduleId,
|
||||
requestId,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
logger.info('Sent schedule-disabled notification', {
|
||||
scheduleId,
|
||||
reason,
|
||||
recipientCount: recipients.length,
|
||||
requestId,
|
||||
})
|
||||
} catch (error) {
|
||||
logger.error('Failed to notify schedule auto-disable', error, { scheduleId, reason, requestId })
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Creator first, then workspace admins. Deduped on lowercased email so someone
|
||||
* who is both only receives one message.
|
||||
*
|
||||
* The two lookups are independent and each failure is contained: losing one
|
||||
* must not zero out the other, or a blip in either turns an auto-disable back
|
||||
* into the silent event this notification exists to prevent.
|
||||
*/
|
||||
async function resolveRecipients(
|
||||
ownerUserId: string | null,
|
||||
workspaceId: string | null
|
||||
): Promise<Recipient[]> {
|
||||
const recipients: Recipient[] = []
|
||||
const seen = new Set<string>()
|
||||
|
||||
const add = (email: string | null, name: string | null) => {
|
||||
if (!email) return
|
||||
const key = email.toLowerCase()
|
||||
if (seen.has(key)) return
|
||||
seen.add(key)
|
||||
recipients.push({ email, name })
|
||||
}
|
||||
|
||||
if (ownerUserId) {
|
||||
try {
|
||||
const ownerRows = await db
|
||||
.select({ email: user.email, name: user.name })
|
||||
.from(user)
|
||||
.where(eq(user.id, ownerUserId))
|
||||
.limit(1)
|
||||
add(ownerRows[0]?.email ?? null, ownerRows[0]?.name ?? null)
|
||||
} catch (error) {
|
||||
logger.error('Failed to resolve schedule creator for disable email', error, { ownerUserId })
|
||||
}
|
||||
}
|
||||
|
||||
if (workspaceId) {
|
||||
try {
|
||||
const members = await getUsersWithPermissions(workspaceId)
|
||||
for (const member of members) {
|
||||
if (member.permissionType !== 'admin') continue
|
||||
add(member.email, member.name)
|
||||
}
|
||||
} catch (error) {
|
||||
logger.error('Failed to resolve workspace admins for disable email', error, { workspaceId })
|
||||
}
|
||||
}
|
||||
|
||||
return recipients.slice(0, MAX_SCHEDULE_DISABLE_RECIPIENTS)
|
||||
}
|
||||
|
||||
function buildManageLink(
|
||||
workspaceId: string | null,
|
||||
workflowId: string | null
|
||||
): string | undefined {
|
||||
if (!workspaceId) return undefined
|
||||
const base = `${getBaseUrl()}/workspace/${workspaceId}`
|
||||
return workflowId ? `${base}/w/${workflowId}` : `${base}/scheduled-tasks`
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
/**
|
||||
* Why Sim turned a schedule off. Threaded from the disable call site into the
|
||||
* notification, because `workflow_schedule` has no column that records it.
|
||||
*/
|
||||
export type ScheduleDisableReason =
|
||||
| 'consecutive_failures'
|
||||
| 'authentication_error'
|
||||
| 'authorization_error'
|
||||
| 'workflow_not_found'
|
||||
| 'invalid_schedule'
|
||||
|
||||
/**
|
||||
* User-facing sentence for each reason. Fixed copy rather than the underlying
|
||||
* error string: the threshold paths carry arbitrary workflow-execution errors
|
||||
* that can contain credentials or customer data.
|
||||
*/
|
||||
export const SCHEDULE_DISABLE_REASON_COPY: Record<ScheduleDisableReason, string> = {
|
||||
consecutive_failures: 'It failed too many times in a row.',
|
||||
authentication_error: 'The run could not be authenticated.',
|
||||
authorization_error: 'The run was not authorized.',
|
||||
workflow_not_found: 'The workflow could not be found.',
|
||||
invalid_schedule: 'The schedule is no longer valid.',
|
||||
}
|
||||
Reference in New Issue
Block a user