Function claimJobSchedules

apps/sim/app/api/schedules/execute/route.ts:174–208

(queuedAt: Date, limit: number)

}

async function claimJobSchedules(queuedAt: Date, limit: number) {
  if (limit <= 0) return []

  return db.transaction(async (tx) => {
    const rows = await tx
      .select({ id: workflowSchedule.id })
      .from(workflowSchedule)
      .where(jobScheduleFilter(queuedAt))
      .for('update', { skipLocked: true })
      .limit(limit)

    if (rows.length === 0) return []

    return tx
      .update(workflowSchedule)
      .set({ lastQueuedAt: queuedAt, updatedAt: queuedAt })
      .where(
        and(
          jobScheduleFilter(queuedAt),
          inArray(
            workflowSchedule.id,
            rows.map((row) => row.id)
          )
        )
      )
      .returning({
        id: workflowSchedule.id,
        cronExpression: workflowSchedule.cronExpression,
        timezone: workflowSchedule.timezone,
        failedCount: workflowSchedule.failedCount,
        lastQueuedAt: workflowSchedule.lastQueuedAt,
        sourceType: workflowSchedule.sourceType,
      })
  })
}

type ClaimedSchedule = Awaited<ReturnType<typeof claimWorkflowSchedules>>[number]
type ClaimedJob = Awaited<ReturnType<typeof claimJobSchedules>>[number]