Function claimWorkflowSchedules

apps/sim/app/api/schedules/execute/route.ts:122–172

(queuedAt: Date, limit: number)

}

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

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

    if (rows.length === 0) return []
    const workspaceIdsByScheduleId = new Map(rows.map((row) => [row.id, row.workspaceId]))

    const claimedRows = await tx
      .update(workflowSchedule)
      .set({ lastQueuedAt: queuedAt, updatedAt: queuedAt })
      .where(
        and(
          workflowScheduleFilter(queuedAt),
          inArray(
            workflowSchedule.id,
            rows.map((row) => row.id)
          )
        )
      )
      .returning({
        id: workflowSchedule.id,
        workflowId: workflowSchedule.workflowId,
        blockId: workflowSchedule.blockId,
        cronExpression: workflowSchedule.cronExpression,
        lastRanAt: workflowSchedule.lastRanAt,
        failedCount: workflowSchedule.failedCount,
        infraRetryCount: workflowSchedule.infraRetryCount,
        nextRunAt: workflowSchedule.nextRunAt,
        lastQueuedAt: workflowSchedule.lastQueuedAt,
        timezone: workflowSchedule.timezone,
        deploymentVersionId: workflowSchedule.deploymentVersionId,
        sourceType: workflowSchedule.sourceType,
      })

    return claimedRows.map((row) => ({
      ...row,
      workspaceId: workspaceIdsByScheduleId.get(row.id) ?? null,
    }))
  })
}

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