Something went wrong. Try again.
Automations and webhooks for the AT Protocol airglow.run
automation webhook atproto atprotocol
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278import { and, eq, isNull, lt, lte, or } from "drizzle-orm";import { db } from "../db/index.js";import { automations, scheduledActions, type Action } from "../db/schema.js";import { ACTION_REGISTRY } from "../actions/registry.js";import { isSuccess, isRetryable, type ActionResult } from "../actions/delivery.js";import { onDeliveryFinalFailure } from "../actions/failure-hooks.js";import { deserializeSnapshot, type ScheduleSnapshot } from "../actions/schedule.js";import { notifyAutomationChange, type MatchedEvent } from "../jetstream/consumer.js";import { maybeDisableForSessionLoss } from "../jetstream/session-loss.js";import { flattenSteps } from "../automations/steps.js";import { placeholderRootRefs, shallowStrings } from "../actions/placeholders.js";import { startIntervalRunner, type IntervalRunnerHandle } from "./interval-runner.js";
const TICK_MS = 30_000;const BATCH_SIZE = 50;const MAX_ATTEMPTS = 3;const RETRY_BACKOFF_MS = 60_000;/** How many due rows the runner executes in parallel within a single tick. * Most actions are I/O-bound (PDS / webhook); a moderate fan-out lets a slow * row stop stalling the others. Bounded so a 50-row batch can't blow the * database write throughput or open hundreds of simultaneous fetches. */const CONCURRENCY = 8;/** A row stuck in `status='running'` past this window is considered orphaned * (crashed mid-execute) and is reset to `pending` for retry. Must be larger * than any expected action runtime — webhook + PDS calls are ~tens of seconds * worst-case; 10 minutes is a comfortable margin. With a `claimedAt` filter * this scheme is safe under multiple runner processes: only stale claims get * reset, so a healthy in-flight claim in another worker isn't disturbed. */const ORPHAN_TIMEOUT_MS = 10 * 60_000;
/** Exported for tests. In production, only `startScheduleRunner` should drive * ticks — the exported handle does not surface this. */export async function runDue() { const now = new Date(); // Orphan recovery: a row in `running` whose `claimedAt` is older than the // orphan timeout is from a previous process that crashed or a runner that // hung. Reset to `pending` so the next claim picks it up. Time-based filter // (vs. "any running row") keeps this safe under multiple runner processes — // an in-flight claim in another worker has a fresh `claimedAt` and is left // alone. NULL claimedAt is recovered too: SQL's `NULL < x` evaluates to // NULL (falsy), so without the explicit isNull branch, any row stuck in // `running` with no recorded claim — e.g. a row migrated from before the // claimedAt column existed — would never be reset and become a permanent // zombie (no claim ever lands on it since status='running' ≠ 'pending'). const orphanCutoff = new Date(Date.now() - ORPHAN_TIMEOUT_MS); await db .update(scheduledActions) .set({ status: "pending" }) .where( and( eq(scheduledActions.status, "running"), or(lt(scheduledActions.claimedAt, orphanCutoff), isNull(scheduledActions.claimedAt)), ), );
const due = await db .select() .from(scheduledActions) .where(and(eq(scheduledActions.status, "pending"), lte(scheduledActions.dueAt, now))) .orderBy(scheduledActions.dueAt) .limit(BATCH_SIZE);
if (due.length === 0) return;
// Bounded-concurrency worker pool. Each worker pops rows off a shared queue // and runs them sequentially; CONCURRENCY workers run in parallel. A slow // row only ties up one worker — the others keep draining the batch. const queue = [...due]; const workerCount = Math.min(CONCURRENCY, queue.length); const workers = Array.from({ length: workerCount }, async () => { while (queue.length > 0) { const row = queue.shift(); if (!row) break; // Per-row atomic claim. The WHERE status='pending' makes the claim a // no-op if a parallel runner (this tick's other workers, or another // process) already took it; we skip silently. Note: `attempts` is // incremented at claim time, not at success/failure — that means a // process crash mid-execute still consumes one attempt. This is // intentional poison-pill protection: a row that crashes the runner // three runs in a row gets marked `failed` rather than looping forever. const claimedAt = new Date(); const claim = await db .update(scheduledActions) .set({ status: "running", attempts: row.attempts + 1, claimedAt }) .where(and(eq(scheduledActions.id, row.id), eq(scheduledActions.status, "pending"))) .returning({ id: scheduledActions.id }); if (claim.length === 0) continue; try { await runOne({ ...row, attempts: row.attempts + 1, claimedAt }); } catch (err) { // Runtime should never throw out of runOne (it catches its own // executor errors), but if something does escape, log and continue so // a single bad row can't take the whole tick down with it. console.error(`[schedule-runner] runOne crashed for ${row.id}:`, err); } } }); await Promise.all(workers);}
type ScheduledRow = typeof scheduledActions.$inferSelect;
async function markFailed(rowId: string, error: string) { await db .update(scheduledActions) .set({ status: "failed", executedAt: new Date(), error }) .where(eq(scheduledActions.id, rowId));}
async function markDone(rowId: string) { await db .update(scheduledActions) .set({ status: "done", executedAt: new Date(), error: null }) .where(eq(scheduledActions.id, rowId));}
async function rescheduleForRetry(rowId: string, attempts: number, error: string) { await db .update(scheduledActions) .set({ status: "pending", dueAt: new Date(Date.now() + attempts * RETRY_BACKOFF_MS), error, }) .where(eq(scheduledActions.id, rowId));}
async function runOne(row: ScheduledRow) { // Re-fetch the automation. If it's gone (FK cascade deleted the row already // pre-execution) or disabled, fail the scheduled row so it doesn't loop on // retry. Match the delete-via-cascade behavior expected when the user // archives an automation: any in-flight scheduled rows shouldn't fire after. const auto = await db.query.automations.findFirst({ where: eq(automations.uri, row.automationUri), }); if (!auto) { await markFailed(row.id, "originating automation no longer exists"); return; } if (!auto.active) { await markFailed(row.id, "originating automation is disabled"); return; }
let snapshot: ScheduleSnapshot; try { // snapshot is stored as JSON via drizzle, comes back as parsed object. // Re-stringify so deserializeSnapshot can do its structural check on a // freshly parsed value (avoids prototype pollution on the raw DB output). const raw = typeof row.snapshot === "string" ? row.snapshot : JSON.stringify(row.snapshot); snapshot = deserializeSnapshot(raw); } catch (err) { await markFailed( row.id, `snapshot deserialization failed: ${err instanceof Error ? err.message : String(err)}`, ); return; }
const storedAction = row.action as Action; const def = ACTION_REGISTRY[storedAction.$type]; if (!def) { await markFailed(row.id, `unknown action type "${storedAction.$type}"`); return; }
// Steps-engine rows (source_step_id set): the step must still exist on the // automation - a row whose source step was deleted between schedule and // replay fails cleanly rather than executing detached work. Its templates // must also still resolve against the frozen snapshot: a step RENAMED // between schedule and replay would otherwise render blanks (the snapshot // keys carry the old names), which is worse than an explicit failure (U8). let stepFlatIndex = -1; if (row.sourceStepId) { stepFlatIndex = auto.steps ? flattenSteps(auto.steps).findIndex((s) => s.id === row.sourceStepId) : -1; if (stepFlatIndex === -1) { await markFailed(row.id, `source step ${row.sourceStepId} no longer exists`); return; } const missing = findMissingSnapshotRoot(storedAction, snapshot.fetchContext); if (missing) { await markFailed( row.id, `template references "${missing}", which is not in this row's snapshot (renamed after scheduling?)`, ); return; } }
// Synthetic MatchedEvent: a sparse `actions` array with the storedAction // (schedule-stripped) at the replay index. Legacy rows use their original // `sourceActionIndex`; steps rows carry the sentinel -1 and replay at the // step's current flat index so the delivery row lands next to the user's // other rows. runMeta stamps the step identity onto the delivery row. const flatIndex = row.sourceStepId ? stepFlatIndex : Math.max(row.sourceActionIndex, 0); const actions: Action[] = []; actions[flatIndex] = storedAction; const synthetic: MatchedEvent = { automation: { ...auto, actions }, event: snapshot.event, // Tell wrapWithDelivery to skip its in-memory setTimeout retry path — // this runner already retries via reschedule + backoff, persisted in the // scheduled_actions row. Without this flag, a retryable failure fires the // inner action 2-3× per scheduled row. scheduledReplay: true, ...(row.sourceStepId ? { runMeta: { runId: `sched-${row.id}`, stepId: row.sourceStepId } } : {}), };
let result: ActionResult; try { result = await def.execute(synthetic, flatIndex, snapshot.fetchContext, snapshot.item); } catch (err) { result = { statusCode: 0, error: err instanceof Error ? err.message : String(err), }; }
if (isSuccess(result.statusCode)) { await markDone(row.id); return; }
// 401 = OAuth session is gone. Mirror the live handler: after the threshold // of consecutive 401s the automation auto-disables, so a user who revoked // Airglow doesn't keep accumulating quiet failures on their scheduled rows. // notifyAutomationChange() pokes the Jetstream subscription so it drops the // automation from the active set in the same tick. if (result.statusCode === 401) { const justDisabled = await maybeDisableForSessionLoss( row.automationUri, snapshot.event.time_us, ); if (justDisabled) notifyAutomationChange(); }
if (row.attempts < MAX_ATTEMPTS && isRetryable(result.statusCode)) { await rescheduleForRetry(row.id, row.attempts, result.error ?? "retryable failure"); return; } await markFailed(row.id, result.error ?? `terminal failure (status ${result.statusCode})`); // Finality for scheduled work is decided HERE, not in the wrapper: the // scheduledReplay flag suppresses the wrapper's final-failure hook since a // wrapper-level failure may still be rescheduled above. Only the // post-execute markFailed (out of attempts or non-retryable) is final; // the pre-execute guards never ran the action and never notify. await onDeliveryFinalFailure(synthetic, result.statusCode, result.error);}
export function startScheduleRunner(): IntervalRunnerHandle { return startIntervalRunner({ name: "schedule-runner", tickMs: TICK_MS, run: () => runDue() });}
/** Scan a stored step/action's own template strings for a placeholder root * that is neither a builtin, the item alias, nor a key in the frozen * snapshot context. Returns the first missing root, or null. */export function findMissingSnapshotRoot( stored: Record<string, unknown>, fetchContext: Record<string, unknown>,): string | null { const builtin = new Set(["event", "now", "self", "automation", "item"]); for (const [key, value] of Object.entries(stored)) { if (key === "id" || key === "name" || key === "comment" || key === "$type") continue; for (const s of shallowStrings(value)) { for (const { root } of placeholderRootRefs(s)) { if (builtin.has(root)) continue; if (root in fetchContext) continue; return root; } } } return null;}