konnosDashboard
schedules.ts378 lines · main
1import { newId, NotFoundError, ValidationError } from '@briven/shared';
2import { and, asc, eq, isNull, lte, sql as drizzleSql } from 'drizzle-orm';
3
4import { getDb } from '../db/client.js';
5import {
6 projectSchedules,
7 type ProjectSchedule,
8 type ScheduleRunStatus,
9} from '../db/schema.js';
10
11/**
12 * Cron-triggered function invocations. Scope of this iteration:
13 *
14 * - 5-field UTC cron expressions only (minute hour dom month dow). No
15 * seconds, no L/W/#, no per-project timezones. UTC keeps the dispatcher
16 * dst-immune and avoids time-zone-by-project state.
17 * - Aliases: @hourly @daily @weekly @monthly @yearly @midnight.
18 * - One row per (project, name) among non-deleted rows; soft-deleted
19 * names can be reused.
20 * - The dispatcher (workers/schedule-dispatcher.ts) claims due rows by
21 * bumping next_run_at forward in the same UPDATE that records the
22 * run outcome — optimistic concurrency, no explicit row lock.
23 */
24
25const NAME_RE = /^[a-z0-9][a-z0-9_-]{0,63}$/i;
26const FUNCTION_NAME_RE = /^[a-zA-Z_$][a-zA-Z0-9_$]{0,128}$/;
27const MAX_NEXT_RUN_SEARCH_MIN = 366 * 24 * 60;
28
29interface CronSets {
30 minutes: Set<number>;
31 hours: Set<number>;
32 daysOfMonth: Set<number>;
33 months: Set<number>; // 1-12
34 daysOfWeek: Set<number>; // 0-6, sun=0
35 // dom and dow are OR'd per POSIX cron semantics when both are
36 // non-wildcard. The parser sets these flags so nextRunAfter can pick
37 // the right branch without re-inspecting the source string.
38 domRestricted: boolean;
39 dowRestricted: boolean;
40}
41
42const ALIASES: Record<string, string> = {
43 '@hourly': '0 * * * *',
44 '@daily': '0 0 * * *',
45 '@midnight': '0 0 * * *',
46 '@weekly': '0 0 * * 0',
47 '@monthly': '0 0 1 * *',
48 '@yearly': '0 0 1 1 *',
49 '@annually': '0 0 1 1 *',
50};
51
52function expandField(field: string, min: number, max: number): Set<number> {
53 const out = new Set<number>();
54 for (const part of field.split(',')) {
55 const [rangePart, stepPart] = part.split('/');
56 const step = stepPart === undefined ? 1 : Number(stepPart);
57 if (!Number.isInteger(step) || step < 1) {
58 throw new ValidationError(`invalid step in cron field "${field}"`);
59 }
60 let lo: number;
61 let hi: number;
62 if (rangePart === undefined || rangePart === '*') {
63 lo = min;
64 hi = max;
65 } else if (rangePart.includes('-')) {
66 const [a, b] = rangePart.split('-');
67 lo = Number(a);
68 hi = Number(b);
69 } else {
70 lo = Number(rangePart);
71 hi = lo;
72 }
73 if (
74 !Number.isInteger(lo) ||
75 !Number.isInteger(hi) ||
76 lo < min ||
77 hi > max ||
78 lo > hi
79 ) {
80 throw new ValidationError(`invalid range "${rangePart}" in cron field "${field}"`);
81 }
82 for (let v = lo; v <= hi; v += step) out.add(v);
83 }
84 if (out.size === 0) {
85 throw new ValidationError(`cron field "${field}" matches no values`);
86 }
87 return out;
88}
89
90export function parseCron(expression: string): CronSets {
91 const normalised = expression.trim();
92 const resolved = ALIASES[normalised] ?? normalised;
93 const fields = resolved.split(/\s+/);
94 if (fields.length !== 5) {
95 throw new ValidationError(
96 `cron expression must have 5 fields (minute hour day month dow); got ${fields.length}`,
97 );
98 }
99 const [minute, hour, dom, month, dow] = fields as [string, string, string, string, string];
100 return {
101 minutes: expandField(minute, 0, 59),
102 hours: expandField(hour, 0, 23),
103 daysOfMonth: expandField(dom, 1, 31),
104 months: expandField(month, 1, 12),
105 // dow accepts 0-6 (sun=0). 7 → 0 as a convenience. We expand 7 manually.
106 daysOfWeek: expandField(dow.replace(/\b7\b/g, '0'), 0, 6),
107 domRestricted: dom !== '*',
108 dowRestricted: dow !== '*',
109 };
110}
111
112/**
113 * First UTC minute strictly after `from` that satisfies `sets`. Throws if
114 * no match is found within one year — that means the expression is valid
115 * by field but unreachable (e.g. `0 0 30 2 *` — Feb 30).
116 */
117export function nextRunAfter(sets: CronSets, from: Date): Date {
118 // Start at next minute boundary (drop seconds + ms, then +1 min).
119 const start = new Date(from);
120 start.setUTCSeconds(0, 0);
121 start.setUTCMinutes(start.getUTCMinutes() + 1);
122
123 for (let i = 0; i < MAX_NEXT_RUN_SEARCH_MIN; i += 1) {
124 const candidate = new Date(start.getTime() + i * 60_000);
125 if (!sets.minutes.has(candidate.getUTCMinutes())) continue;
126 if (!sets.hours.has(candidate.getUTCHours())) continue;
127 if (!sets.months.has(candidate.getUTCMonth() + 1)) continue;
128 const domHit = sets.daysOfMonth.has(candidate.getUTCDate());
129 const dowHit = sets.daysOfWeek.has(candidate.getUTCDay());
130 // POSIX cron: if BOTH dom and dow are restricted, fire when EITHER
131 // matches. If only one is restricted, that one must match. If
132 // neither is restricted both sets are full so this resolves the
133 // same way.
134 if (sets.domRestricted && sets.dowRestricted) {
135 if (!domHit && !dowHit) continue;
136 } else if (sets.domRestricted) {
137 if (!domHit) continue;
138 } else if (sets.dowRestricted) {
139 if (!dowHit) continue;
140 } else if (!domHit || !dowHit) {
141 // both wildcard → both sets full, defensive
142 continue;
143 }
144 return candidate;
145 }
146 throw new ValidationError(
147 'cron expression has no valid next run within one year (impossible date?)',
148 );
149}
150
151export interface CreateScheduleInput {
152 projectId: string;
153 name: string;
154 functionName: string;
155 cronExpression: string;
156 args?: Record<string, unknown>;
157 enabled?: boolean;
158 createdBy: string | null;
159}
160
161export interface UpdateScheduleInput {
162 name?: string;
163 functionName?: string;
164 cronExpression?: string;
165 args?: Record<string, unknown>;
166 enabled?: boolean;
167}
168
169function validateName(name: string): void {
170 if (!NAME_RE.test(name)) {
171 throw new ValidationError(
172 'schedule name must be 1-64 chars: alphanumerics, underscore, hyphen; must start with alphanumeric',
173 );
174 }
175}
176
177function validateFunctionName(fn: string): void {
178 if (!FUNCTION_NAME_RE.test(fn)) {
179 throw new ValidationError('function name must be a valid javascript identifier');
180 }
181}
182
183export async function listSchedules(projectId: string): Promise<ProjectSchedule[]> {
184 const db = getDb();
185 return db
186 .select()
187 .from(projectSchedules)
188 .where(
189 and(eq(projectSchedules.projectId, projectId), isNull(projectSchedules.deletedAt)),
190 )
191 .orderBy(asc(projectSchedules.name));
192}
193
194export async function getSchedule(
195 scheduleId: string,
196 projectId: string,
197): Promise<ProjectSchedule> {
198 const db = getDb();
199 const rows = await db
200 .select()
201 .from(projectSchedules)
202 .where(
203 and(
204 eq(projectSchedules.id, scheduleId),
205 eq(projectSchedules.projectId, projectId),
206 isNull(projectSchedules.deletedAt),
207 ),
208 )
209 .limit(1);
210 const row = rows[0];
211 if (!row) throw new NotFoundError('schedule', scheduleId);
212 return row;
213}
214
215export async function createSchedule(input: CreateScheduleInput): Promise<ProjectSchedule> {
216 validateName(input.name);
217 validateFunctionName(input.functionName);
218 const sets = parseCron(input.cronExpression);
219 const nextRunAt = nextRunAfter(sets, new Date());
220
221 const db = getDb();
222 const id = newId('sch');
223 const row = {
224 id,
225 projectId: input.projectId,
226 name: input.name,
227 functionName: input.functionName,
228 cronExpression: input.cronExpression.trim(),
229 args: input.args ?? {},
230 enabled: input.enabled ?? true,
231 nextRunAt,
232 createdBy: input.createdBy,
233 };
234
235 try {
236 const inserted = await db.insert(projectSchedules).values(row).returning();
237 if (!inserted[0]) throw new Error('insert returned no row');
238 return inserted[0];
239 } catch (err) {
240 if (err instanceof Error && /unique|duplicate/i.test(err.message)) {
241 throw new ValidationError(`a schedule named "${input.name}" already exists for this project`);
242 }
243 throw err;
244 }
245}
246
247export async function updateSchedule(
248 scheduleId: string,
249 projectId: string,
250 patch: UpdateScheduleInput,
251): Promise<ProjectSchedule> {
252 // Load first so an attempted edit against a non-existent / cross-project
253 // id 404s instead of silently no-op'ing.
254 const existing = await getSchedule(scheduleId, projectId);
255
256 const updates: Partial<ProjectSchedule> = { updatedAt: new Date() };
257
258 if (patch.name !== undefined && patch.name !== existing.name) {
259 validateName(patch.name);
260 updates.name = patch.name;
261 }
262 if (patch.functionName !== undefined && patch.functionName !== existing.functionName) {
263 validateFunctionName(patch.functionName);
264 updates.functionName = patch.functionName;
265 }
266 if (
267 patch.cronExpression !== undefined &&
268 patch.cronExpression.trim() !== existing.cronExpression
269 ) {
270 const sets = parseCron(patch.cronExpression);
271 updates.cronExpression = patch.cronExpression.trim();
272 updates.nextRunAt = nextRunAfter(sets, new Date());
273 }
274 if (patch.args !== undefined) updates.args = patch.args;
275 if (patch.enabled !== undefined && patch.enabled !== existing.enabled) {
276 updates.enabled = patch.enabled;
277 // Re-enabling a schedule whose next_run_at is in the past would fire
278 // on the next dispatcher tick — fine when intentional (catch-up). To
279 // avoid surprise back-fills, push next_run_at forward to the next
280 // valid slot from now.
281 if (patch.enabled) {
282 const sets = parseCron(updates.cronExpression ?? existing.cronExpression);
283 updates.nextRunAt = nextRunAfter(sets, new Date());
284 }
285 }
286
287 const db = getDb();
288 const result = await db
289 .update(projectSchedules)
290 .set(updates)
291 .where(and(eq(projectSchedules.id, scheduleId), eq(projectSchedules.projectId, projectId)))
292 .returning();
293 if (!result[0]) throw new NotFoundError('schedule', scheduleId);
294 return result[0];
295}
296
297export async function deleteSchedule(scheduleId: string, projectId: string): Promise<void> {
298 const db = getDb();
299 const result = await db
300 .update(projectSchedules)
301 .set({ deletedAt: new Date(), enabled: false, updatedAt: new Date() })
302 .where(
303 and(
304 eq(projectSchedules.id, scheduleId),
305 eq(projectSchedules.projectId, projectId),
306 isNull(projectSchedules.deletedAt),
307 ),
308 )
309 .returning({ id: projectSchedules.id });
310 if (!result[0]) throw new NotFoundError('schedule', scheduleId);
311}
312
313/**
314 * Dispatcher claim path. Selects up to `limit` schedules whose next_run_at
315 * is in the past, returning them with their pre-claim values. The caller
316 * is expected to compute the new next_run_at locally, invoke the function,
317 * and call recordScheduleResult — which performs an UPDATE guarded by the
318 * pre-claim next_run_at so two concurrent dispatchers can't double-fire.
319 */
320export async function claimDueSchedules(
321 now: Date,
322 limit: number,
323): Promise<ProjectSchedule[]> {
324 const db = getDb();
325 return db
326 .select()
327 .from(projectSchedules)
328 .where(
329 and(
330 eq(projectSchedules.enabled, true),
331 isNull(projectSchedules.deletedAt),
332 lte(projectSchedules.nextRunAt, now),
333 ),
334 )
335 .orderBy(asc(projectSchedules.nextRunAt))
336 .limit(limit);
337}
338
339/**
340 * Records a run outcome and advances next_run_at. The WHERE clause checks
341 * the pre-claim next_run_at so a concurrent dispatcher that already moved
342 * the row forward will get rowsAffected=0 and skip recording.
343 *
344 * Returns whether the update was applied — callers can log a "lost race"
345 * counter when this returns false.
346 */
347export async function recordScheduleResult(input: {
348 scheduleId: string;
349 claimedNextRunAt: Date;
350 newNextRunAt: Date;
351 ranAt: Date;
352 status: ScheduleRunStatus;
353 errorMessage?: string;
354}): Promise<boolean> {
355 const db = getDb();
356 const result = await db
357 .update(projectSchedules)
358 .set({
359 nextRunAt: input.newNextRunAt,
360 lastRunAt: input.ranAt,
361 lastRunStatus: input.status,
362 lastRunError: input.errorMessage ?? null,
363 updatedAt: new Date(),
364 })
365 .where(
366 and(
367 eq(projectSchedules.id, input.scheduleId),
368 eq(projectSchedules.nextRunAt, input.claimedNextRunAt),
369 ),
370 )
371 .returning({ id: projectSchedules.id });
372 return result.length > 0;
373}
374
375// Re-export drizzle's sql tag for callers that need raw expressions on
376// the schedules table. Kept here so route + worker files don't have to
377// reach into drizzle-orm directly.
378export { drizzleSql };