| 1 | import { cronMatches } from "@diffinery/workflow-yaml"; |
| 2 | import type { Env } from "../env.js"; |
| 3 | |
| 4 | /** |
| 5 | * Timing of workflow `schedule` triggers, kept free of platform imports so it can be unit tested. |
| 6 | * The scheduler (src/scheduler.ts) ticks every few minutes; each tick fires the schedules due in |
| 7 | * the minutes since the last tick that finished, so a late or missed tick is caught up by the next. |
| 8 | */ |
| 9 | |
| 10 | /** The one scheduler object (src/scheduler.ts). */ |
| 11 | export const schedulerStub = (env: Env) => env.SCHEDULER.get(env.SCHEDULER.idFromName("global")); |
| 12 | |
| 13 | /** GitHub's shortest schedule interval, and the default tick. */ |
| 14 | export const DEFAULT_TICK_MINUTES = 5; |
| 15 | /** A tick looks back at most this far: schedules missed for longer are dropped, as on GitHub. */ |
| 16 | export const CATCH_UP_MINUTES = 60; |
| 17 | |
| 18 | const MINUTE = 60_000; |
| 19 | |
| 20 | /** Minutes between ticks: SCHEDULE_TICK_MINUTES if it's a whole number from 1 to 60, else the default. */ |
| 21 | export function tickMinutes(value: string | undefined): number { |
| 22 | const n = Number(value); |
| 23 | return Number.isInteger(n) && n >= 1 && n <= 60 ? n : DEFAULT_TICK_MINUTES; |
| 24 | } |
| 25 | |
| 26 | /** When the tick after `now` is due: the next multiple of the tick (on the hour, :05, :10 ... for 5). */ |
| 27 | export function nextTick(now: number, tick: number): number { |
| 28 | const step = tick * MINUTE; |
| 29 | return (Math.floor(now / step) + 1) * step; |
| 30 | } |
| 31 | |
| 32 | /** |
| 33 | * The whole minutes (UTC) a tick at `now` covers: those after `last` (the previous finished tick) |
| 34 | * up to now's minute, at most CATCH_UP_MINUTES of them. Without a previous tick, the tick's own window. |
| 35 | */ |
| 36 | export function tickWindow(now: number, last: number | undefined, tick: number): Date[] { |
| 37 | const end = Math.floor(now / MINUTE) * MINUTE; |
| 38 | const from = Math.max(last !== undefined ? Math.floor(last / MINUTE) * MINUTE + MINUTE : end - (tick - 1) * MINUTE, end - (CATCH_UP_MINUTES - 1) * MINUTE); |
| 39 | const minutes: Date[] = []; |
| 40 | for (let t = from; t <= end; t += MINUTE) minutes.push(new Date(t)); |
| 41 | return minutes; |
| 42 | } |
| 43 | |
| 44 | /** The schedules due in any of `minutes`, each once however many of them match. Invalid crons are skipped. */ |
| 45 | export function dueSchedules<T extends { cron: string }>(rows: T[], minutes: Date[]): T[] { |
| 46 | return rows.filter((s) => { |
| 47 | try { |
| 48 | return minutes.some((m) => cronMatches(s.cron, m)); |
| 49 | } catch { |
| 50 | return false; |
| 51 | } |
| 52 | }); |
| 53 | } |
| 54 | |
| 55 | /** Runs `fn` over `items`, at most `limit` at a time. Each call's failure is its own (`fn` should catch). */ |
| 56 | export async function eachLimited<T>(items: T[], limit: number, fn: (item: T) => Promise<void>): Promise<void> { |
| 57 | let next = 0; |
| 58 | const worker = async () => { |
| 59 | while (next < items.length) await fn(items[next++]!); |
| 60 | }; |
| 61 | await Promise.all(Array.from({ length: Math.min(limit, items.length) }, worker)); |
| 62 | } |