Function tryStartDatabaseScheduleJob

apps/sim/app/api/schedules/execute/route.ts:525–558

(jobId: string)

}

async function tryStartDatabaseScheduleJob(jobId: string): Promise<DatabaseScheduleStartResult> {
  const now = new Date()

  return 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`
    )
    if (!lock?.acquired) return 'capacity_full'

    const [row] = await tx
      .select({
        count: sql<number>`count(*)`,
      })
      .from(asyncJobs)
      .where(activeScheduleExecutionJobsFilter())

    if (Number(row?.count ?? 0) >= SCHEDULE_EXECUTION_CONCURRENCY_LIMIT) {
      return 'capacity_full'
    }

    const [startedJob] = await tx
      .update(asyncJobs)
      .set({
        status: JOB_STATUS.PROCESSING,
        startedAt: now,
        attempts: sql`${asyncJobs.attempts} + 1`,
        updatedAt: now,
      })
      .where(and(eq(asyncJobs.id, jobId), eq(asyncJobs.status, JOB_STATUS.PENDING)))
      .returning({ id: asyncJobs.id })

    return startedJob ? 'started' : 'not_pending'
  })
}

async function executeDatabaseScheduleJob(
  jobQueue: JobQueue,