Something went wrong. Try again.
The IoT platform Cloudflare didn't build. Telemetry, realtime dashboards, and automations on Workers + D1 + R2 + Durable Objects — in your… nodrix.live
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238// Scheduling math for the SchedulerDO: compute the next fire time for schedule +// sunset/sunrise automations, and run a planned set when due.//// The DO alarm fires at `computeNextScheduled().fireAt`, then runs exactly the// `due` (automation, node) pairs planned for that instant — so it's robust to// wake-up drift (we fire what was due at the planned time, not the clock).
import type { Env } from '../../env';import type { AutomationContext, AutomationRow, ScheduleTriggerConfig, SolarTriggerConfig,} from './types';import { sunTimes } from './solar';import { runAutomation } from './run';import { toGraph, triggerNodes, nodesById } from './graph';import { chunk, inClause, MAX_BOUND_PARAMS } from '../lib/sql';
// Column projection the engine reads from `automations`.const AUTOMATION_COLS = 'id, project_id, name, enabled, trigger_type, graph, last_run_at';
// A due fire is a specific schedule/solar trigger node within an automation.export type ScheduleDue = { id: string; nodeId: string };export type SchedulePlan = { fireAt: number | null; due: ScheduleDue[] };
// Soonest upcoming fire across every schedule/solar trigger node of all enabled// automations, plus the (automation, node) pairs that share that instant.export async function computeNextScheduled(env: Env): Promise<SchedulePlan> { const rows = await env.DB .prepare( `SELECT ${AUTOMATION_COLS} FROM automations WHERE enabled = 1 AND (trigger_kinds LIKE '%,schedule,%' OR trigger_kinds LIKE '%,sunset_sunrise,%')` ) .all<AutomationRow>();
const now = Date.now(); let best: number | null = null; const entries: { id: string; nodeId: string; fireAt: number }[] = [];
for (const a of rows.results) { for (const node of triggerNodes(toGraph(a))) { let next: number | null = null; try { if (node.kind === 'schedule') next = nextScheduleFireAt(node.config as ScheduleTriggerConfig, now); else if (node.kind === 'sunset_sunrise') next = nextSolarFireAt(node.config as SolarTriggerConfig, now); } catch { next = null; } if (next == null) continue; entries.push({ id: a.id, nodeId: node.id, fireAt: next }); if (best == null || next < best) best = next; } }
if (best == null) return { fireAt: null, due: [] }; const due = entries.filter((e) => e.fireAt - best! < 1000).map(({ id, nodeId }) => ({ id, nodeId })); return { fireAt: best, due };}
// Run the planned fires. `fireAtSec` dedups against last_run_at so a duplicate// scheduler instance (or replay) can't double-fire the same instant; firedIds// keeps an automation to one fire per instant even with several due trigger nodes.export async function runScheduledDue(env: Env, due: ScheduleDue[], fireAtSec: number): Promise<number> { if (due.length === 0) return 0; const automations = await loadAutomationsByIds(env, due.map((d) => d.id));
let ran = 0; const firedIds = new Set<string>(); for (const { id, nodeId } of due) { const a = automations.get(id); if (!a) continue; // deleted or disabled if (firedIds.has(id)) continue; // already fired this instant if (a.last_run_at != null && a.last_run_at >= fireAtSec) continue;
const node = nodesById(toGraph(a)).get(nodeId); const ctx: AutomationContext = { source: node?.kind === 'sunset_sunrise' ? 'sunset_sunrise' : 'schedule', projectId: a.project_id, ts: Math.floor(Date.now() / 1000), depth: 0, entryNodeId: nodeId, }; try { await runAutomation(env, a, ctx); firedIds.add(id); ran++; } catch (e) { console.error('[scheduler] run failed', id, e); } } return ran;}
// One indexed read for a batch of automation ids (PK IN-list), chunked under the// SQLite bound-param cap. Disabled/deleted rows are simply absent from the map.async function loadAutomationsByIds(env: Env, ids: string[]): Promise<Map<string, AutomationRow>> { const map = new Map<string, AutomationRow>(); for (const part of chunk([...new Set(ids)], MAX_BOUND_PARAMS)) { const rows = await env.DB .prepare(`SELECT ${AUTOMATION_COLS} FROM automations WHERE enabled = 1 AND id IN ${inClause(part.length)}`) .bind(...part) .all<AutomationRow>(); for (const a of rows.results) map.set(a.id, a); } return map;}
// ── delay continuations ──────────────────────────────────────────────────────
// Soonest pending delay resume (ms epoch), or null when none are queued.export async function nextDelayFireAt(env: Env): Promise<number | null> { const row = await env.DB .prepare(`SELECT MIN(fire_at) AS m FROM automation_delays`) .first<{ m: number | null }>(); return row?.m ?? null;}
// Resume every delay whose fire_at has passed. Each row is deleted before its run// so it can't double-fire; re-entry at resume_node_id makes the delay pass through.export async function runDueDelays(env: Env, nowMs: number): Promise<number> { const rows = await env.DB .prepare( `SELECT id, automation_id, resume_node_id, ctx FROM automation_delays WHERE fire_at <= ? ORDER BY fire_at` ) .bind(nowMs) .all<{ id: string; automation_id: string; resume_node_id: string; ctx: string }>();
if (rows.results.length === 0) return 0; // One batched read for all referenced automations instead of one per delay row. const automations = await loadAutomationsByIds(env, rows.results.map((r) => r.automation_id));
let ran = 0; for (const row of rows.results) { await env.DB.prepare(`DELETE FROM automation_delays WHERE id = ?`).bind(row.id).run();
const a = automations.get(row.automation_id); if (!a) continue; // automation deleted or disabled
let ctx: AutomationContext; try { ctx = JSON.parse(row.ctx) as AutomationContext; } catch { continue; } ctx.entryNodeId = row.resume_node_id; ctx.resumeNodeId = row.resume_node_id; ctx.ts = Math.floor(Date.now() / 1000);
try { await runAutomation(env, a, ctx); ran++; } catch (e) { console.error('[scheduler] delay resume failed', row.id, e); } } return ran;}
// Next 'HH:MM' (tz-local, optional weekdays) strictly after `from` (ms epoch).export function nextScheduleFireAt(cfg: ScheduleTriggerConfig, from: number): number | null { if (!cfg || typeof cfg.time !== 'string' || !/^\d{1,2}:\d{2}$/.test(cfg.time)) return null; const parts = cfg.time.split(':'); const h = Number(parts[0]); const m = Number(parts[1]); const tz = cfg.tz || 'UTC'; const days = Array.isArray(cfg.days) && cfg.days.length > 0 ? cfg.days : null;
const base = ymdInTz(from, tz); for (let off = 0; off <= 14; off++) { const d = addDays(base, off); if (days && !days.includes(weekdayOf(d))) continue; const inst = zonedWallToUtc(d.y, d.mo, d.d, h, m, tz); if (inst > from) return inst; } return null;}
// Next sunrise/sunset (+offset) strictly after `from` (ms epoch).export function nextSolarFireAt(cfg: SolarTriggerConfig, from: number): number | null { if (!cfg || typeof cfg.lat !== 'number' || typeof cfg.lng !== 'number') return null; const offMs = (cfg.offset_minutes || 0) * 60000; for (let off = 0; off <= 4; off++) { const t = sunTimes(new Date(from + off * 86400000), cfg.lat, cfg.lng); const target = cfg.event === 'sunset' ? t.sunset : t.sunrise; if (!target) continue; const inst = target.getTime() + offMs; if (inst > from) return inst; } return null;}
// ── timezone helpers (no external deps) ──────────────────────────────────────
type Ymd = { y: number; mo: number; d: number };
function ymdInTz(ms: number, tz: string): Ymd { const p = new Intl.DateTimeFormat('en-US', { timeZone: tz, year: 'numeric', month: '2-digit', day: '2-digit', }).formatToParts(new Date(ms)); const get = (t: string) => Number(p.find((x) => x.type === t)?.value); return { y: get('year'), mo: get('month'), d: get('day') };}
function addDays(b: Ymd, n: number): Ymd { const dt = new Date(Date.UTC(b.y, b.mo - 1, b.d)); dt.setUTCDate(dt.getUTCDate() + n); return { y: dt.getUTCFullYear(), mo: dt.getUTCMonth() + 1, d: dt.getUTCDate() };}
function weekdayOf(d: Ymd): number { return new Date(Date.UTC(d.y, d.mo - 1, d.d)).getUTCDay(); // 0=Sun..6=Sat}
// Wall-clock time in `tz` -> UTC epoch ms. Two passes settle the DST offset.function zonedWallToUtc(y: number, mo: number, d: number, h: number, mi: number, tz: string): number { const wallUtc = Date.UTC(y, mo - 1, d, h, mi); let guess = wallUtc; for (let k = 0; k < 2; k++) { const corrected = wallUtc - tzOffsetMs(guess, tz); if (corrected === guess) break; guess = corrected; } return guess;}
function tzOffsetMs(utcMs: number, tz: string): number { const p = new Intl.DateTimeFormat('en-US', { timeZone: tz, hourCycle: 'h23', year: 'numeric', month: '2-digit', day: '2-digit', hour: '2-digit', minute: '2-digit', second: '2-digit', }).formatToParts(new Date(utcMs)); const get = (t: string) => Number(p.find((x) => x.type === t)?.value); const asUtc = Date.UTC(get('year'), get('month') - 1, get('day'), get('hour'), get('minute'), get('second')); return asUtc - utcMs;}