Function processScheduleItem

apps/sim/app/api/schedules/execute/route.ts:733–1031

(
  schedule: ClaimedSchedule,
  queuedAt: Date,
  requestId: string,
  jobQueue: JobQueue,
  useDatabaseFallback: boolean
)

}

async function processScheduleItem(
  schedule: ClaimedSchedule,
  queuedAt: Date,
  requestId: string,
  jobQueue: JobQueue,
  useDatabaseFallback: boolean
) {
  const queueTime = schedule.lastQueuedAt ?? queuedAt
  const executionId = generateId()
  const correlation = {
    executionId,
    requestId,
    source: 'schedule' as const,
    workflowId: schedule.workflowId!,
    scheduleId: schedule.id,
    triggerType: 'schedule',
    scheduledFor: schedule.nextRunAt?.toISOString(),
  }

  const payload = {
    scheduleId: schedule.id,
    workflowId: schedule.workflowId!,
    executionId,
    requestId,
    correlation,
    blockId: schedule.blockId || undefined,
    workspaceId: schedule.workspaceId || undefined,
    deploymentVersionId: schedule.deploymentVersionId || undefined,
    cronExpression: schedule.cronExpression || undefined,
    timezone: schedule.timezone || undefined,
    lastRanAt: schedule.lastRanAt?.toISOString(),
    failedCount: schedule.failedCount || 0,
    infraRetryCount: schedule.infraRetryCount || 0,
    now: queueTime.toISOString(),
    scheduledFor: schedule.nextRunAt?.toISOString(),
  }

  let enqueuedJobId: string | null = null

  try {
    const delayMs = Math.floor(Math.random() * SCHEDULE_JITTER_MAX_MS)

    const scheduleJobId = buildScheduleExecutionJobId(schedule)
    const existingJob = await jobQueue.getJob(scheduleJobId)
    if (existingJob && ['pending', 'processing'].includes(existingJob.status)) {
      const activeJobPayload = getSchedulePayloadFromJob(existingJob)
      const activeJobClaim = getSchedulePayloadClaimedAt(activeJobPayload)

      if (useDatabaseFallback && isStaleDatabaseScheduleJob(existingJob)) {
        await recoverStaleDatabaseScheduleJobs(new Date())
        logger.info(`[${requestId}] Recovered stale database schedule execution jobs`, {
          scheduleId: schedule.id,
          jobId: scheduleJobId,
        })
      }

      const databaseJob = useDatabaseFallback ? await jobQueue.getJob(scheduleJobId) : existingJob
      const databaseJobPayload = databaseJob ? getSchedulePayloadFromJob(databaseJob) : null