Blame · Line-by-line history

`scheduled-workflows.ts`

Each line is annotated with the commit that last touched it. Click any SHA to jump to that commit and see the surrounding change.

scheduled-workflows.tsBlame257 lines · 1 contributor

Normal view Raw

665c8bfClaude	1	/**
	2	* Scheduled workflows — fires `on: schedule` triggers from the autopilot
	3	* tick.
	4	*
	5	* Pipeline (per tick, default every 5 minutes via src/lib/autopilot.ts):
	6	*
	7	* 1. Select all non-disabled workflows whose serialised `parsed` JSON
	8	* includes a non-empty `schedules` array. (Cheap LIKE filter — DB
	9	* doesn't natively know about JSON keys here, and we want to avoid
	10	* pulling all workflows on every tick.)
	11	* 2. For each workflow:
	12	* - Look up the latest schedule-triggered run (event="schedule").
	13	* That row's queuedAt is the `since` boundary; absent → use
	14	* (now - 6 minutes), so a freshly-imported workflow doesn't
	15	* back-fire for hours of crons.
	16	* - For each cron string, parse via src/lib/cron.ts and ask
	17	* `cronFiredBetween(since, now)`. The first cron that fired
	18	* wins — we enqueue exactly one run per workflow per tick.
	19	* 3. enqueueRun(...) with event="schedule", ref=defaultBranch,
	20	* commitSha=resolved-default-branch-HEAD. The existing runner
	21	* (src/lib/workflow-runner.ts) picks it up exactly like a manual
	22	* run.
	23	*
	24	* Fail-open: every step swallows DB errors and returns a result object
	25	* so the autopilot ticker never wedges. Returns a per-call summary so
	26	* callers (e.g. the admin dashboard) can show "last tick fired N runs."
	27	*
	28	* Safety guard: caps each tick at MAX_RUNS_PER_TICK so a misconfigured
	29	* cron and an empty schedule-runs table cannot stampede the queue.
	30	*/
	31
	32	import { and, desc, eq, isNull, sql } from "drizzle-orm";
	33	import { db } from "../db";
	34	import {
	35	workflowRuns,
	36	workflows,
	37	repositories,
	38	} from "../db/schema";
	39	import { parseCron, cronFiredBetween } from "./cron";
	40	import { enqueueRun } from "./workflow-runner";
	41	import { resolveRef, getDefaultBranch } from "../git/repository";
	42
	43	export const MAX_RUNS_PER_TICK = 50;
	44	const SINCE_FALLBACK_MS = 6 * 60_000; // 6 min — slightly > default 5-min tick
	45
	46	export type ScheduledTickResult = {
	47	considered: number;
	48	fired: number;
	49	errors: number;
	50	};
	51
	52	type WorkflowRow = {
	53	id: string;
	54	repositoryId: string;
	55	parsed: string;
	56	};
	57
	58	/**
	59	* Parse the `parsed` JSON column and return the cron expressions, or [].
	60	* Defensive — never throws.
	61	*/
	62	export function schedulesFromParsedJson(parsedJson: string): string[] {
	63	try {
	64	const obj = JSON.parse(parsedJson \|\| "{}");
	65	const out = Array.isArray(obj?.schedules) ? obj.schedules : [];
	66	return out.filter((s: unknown): s is string => typeof s === "string" && s.trim().length > 0);
	67	} catch {
	68	return [];
	69	}
	70	}
	71
	72	/**
	73	* Pure decision helper — given a workflow's schedules + last fire wall +
	74	* current wall, return the first cron that should fire (or null).
	75	* Exposed for unit tests so the cron→fire wiring is verifiable without
	76	* a DB.
	77	*/
	78	export function firstCronToFire(
	79	schedules: string[],
	80	since: Date,
	81	until: Date
	82	): string \| null {
	83	for (const expr of schedules) {
	84	const parsed = parseCron(expr);
	85	if (!parsed.ok) continue;
	86	if (cronFiredBetween(parsed.cron, since, until)) return expr;
	87	}
	88	return null;
	89	}
	90
	91	async function lastScheduleRunQueuedAt(
	92	workflowId: string
	93	): Promise<Date \| null> {
	94	try {
	95	const [row] = await db
	96	.select({ queuedAt: workflowRuns.queuedAt })
	97	.from(workflowRuns)
	98	.where(
	99	and(
	100	eq(workflowRuns.workflowId, workflowId),
	101	eq(workflowRuns.event, "schedule")
	102	)
	103	)
	104	.orderBy(desc(workflowRuns.queuedAt))
	105	.limit(1);
	106	return row ? new Date(row.queuedAt) : null;
	107	} catch {
	108	return null;
	109	}
	110	}
	111
	112	async function loadOwnerAndRepoName(
	113	repositoryId: string
	114	): Promise<{ ownerName: string; repoName: string; defaultBranch: string } \| null> {
	115	try {
	116	const [row] = await db
	117	.select({
	118	repoName: repositories.name,
	119	ownerId: repositories.ownerId,
	120	defaultBranch: repositories.defaultBranch,
	121	})
	122	.from(repositories)
	123	.where(eq(repositories.id, repositoryId))
	124	.limit(1);
	125	if (!row) return null;
	126	// Owner username lookup via the standard repositories.owner_id → users
	127	// join. Performed lazily to avoid a join on the hot list query.
	128	const { users } = await import("../db/schema");
	129	const [owner] = await db
	130	.select({ username: users.username })
	131	.from(users)
	132	.where(eq(users.id, row.ownerId))
	133	.limit(1);
	134	if (!owner) return null;
	135	return {
	136	ownerName: owner.username,
	137	repoName: row.repoName,
	138	defaultBranch: row.defaultBranch \|\| "main",
	139	};
	140	} catch {
	141	return null;
	142	}
	143	}
	144
	145	/**
	146	* Walk every non-disabled workflow whose parsed JSON could include a
	147	* schedules field, decide if any cron fired since the last schedule-run,
	148	* and enqueue at most one run per workflow per tick.
	149	*/
	150	export async function runScheduledWorkflowsTick(
	151	now: Date = new Date()
	152	): Promise<ScheduledTickResult> {
	153	const result: ScheduledTickResult = { considered: 0, fired: 0, errors: 0 };
	154
	155	let candidates: WorkflowRow[] = [];
	156	try {
	157	candidates = await db
	158	.select({
	159	id: workflows.id,
	160	repositoryId: workflows.repositoryId,
	161	parsed: workflows.parsed,
	162	})
	163	.from(workflows)
	164	.where(
	165	and(
	166	eq(workflows.disabled, false),
	167	// Cheap pre-filter: only workflows whose parsed JSON contains
	168	// the literal token "schedules" (presence implies non-empty
	169	// array via the parser contract). This is intentionally a
	170	// string-LIKE — JSON-aware operators are nice-to-have.
	171	sql`${workflows.parsed} LIKE '%"schedules"%'`
	172	)
	173	);
	174	} catch {
	175	candidates = [];
	176	result.errors += 1;
	177	}
	178
	179	for (const w of candidates) {
	180	if (result.fired >= MAX_RUNS_PER_TICK) break;
	181	result.considered += 1;
	182
	183	const schedules = schedulesFromParsedJson(w.parsed);
	184	if (schedules.length === 0) continue;
	185
	186	const lastQ = await lastScheduleRunQueuedAt(w.id);
	187	const since = lastQ
	188	? lastQ
	189	: new Date(now.getTime() - SINCE_FALLBACK_MS);
	190
	191	const expr = firstCronToFire(schedules, since, now);
	192	if (!expr) continue;
	193
	194	const repoMeta = await loadOwnerAndRepoName(w.repositoryId);
	195	if (!repoMeta) {
	196	result.errors += 1;
	197	continue;
	198	}
	199
	200	let commitSha: string \| null = null;
	201	try {
	202	commitSha = await resolveRef(
	203	repoMeta.ownerName,
	204	repoMeta.repoName,
	205	repoMeta.defaultBranch
	206	);
	207	} catch {
	208	commitSha = null;
	209	}
	210	if (!commitSha) {
	211	// Try to recover the default branch via the on-disk repo if the DB
	212	// value is stale — best-effort. If still unknown, skip.
	213	try {
	214	const def = await getDefaultBranch(
	215	repoMeta.ownerName,
	216	repoMeta.repoName
	217	);
	218	if (def) {
	219	commitSha = await resolveRef(
	220	repoMeta.ownerName,
	221	repoMeta.repoName,
	222	def
	223	);
	224	}
	225	} catch {
	226	commitSha = null;
	227	}
	228	}
	229	if (!commitSha) {
	230	result.errors += 1;
	231	continue;
	232	}
	233
	234	try {
	235	await enqueueRun({
	236	workflowId: w.id,
	237	repositoryId: w.repositoryId,
	238	event: "schedule",
	239	ref: `refs/heads/${repoMeta.defaultBranch}`,
	240	commitSha,
	241	triggeredBy: null,
	242	});
	243	result.fired += 1;
	244	} catch {
	245	result.errors += 1;
	246	}
	247	}
	248
	249	return result;
	250	}
	251
	252	/** Test-only exposed internals so DB-less test cases can pin behaviour. */
	253	export const __test = {
	254	schedulesFromParsedJson,
	255	firstCronToFire,
	256	SINCE_FALLBACK_MS,
	257	};