Something went wrong. Try again.
Automations and webhooks for the AT Protocol airglow.run
automation webhook atproto atprotocol
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538import { describe, it, expect, vi, beforeEach } from "vitest";import { eq } from "drizzle-orm";
vi.mock("@/config.js", () => ({ config: { databasePath: ":memory:", jetstreamUrl: "wss://jetstream.test/subscribe", publicUrl: "https://airglow.test", nsidAllowlist: [], nsidBlocklist: [], nsidRequireDids: [], },}));
vi.mock("@/db/index.js", async () => { const { createTestDb } = await import("../test/db.js"); return { db: await createTestDb() };});
// Hoisted so vi.mock factories (which run before module-level code) can see// these handles. `mockExecute` is a controllable executor stub; the Proxy// makes every action type resolve to the same fake def so the runner can// dispatch a stored action regardless of its $type. `maybeDisableForSessionLoss`// is the real function's stand-in — the real one has its own tests; stubbing// it here lets us assert the runner hands off cleanly without the// SESSION_LOSS_THRESHOLD log history setup.const { mockExecute, notifyAutomationChange, maybeDisableForSessionLoss } = vi.hoisted(() => ({ mockExecute: vi.fn(), notifyAutomationChange: vi.fn(), maybeDisableForSessionLoss: vi.fn().mockResolvedValue(false),}));
vi.mock("@/jetstream/consumer.js", () => ({ notifyAutomationChange }));vi.mock("@/jetstream/session-loss.js", () => ({ maybeDisableForSessionLoss }));vi.mock("@/atmo/notify-system.js", () => ({ notifySystem: vi.fn(async () => "sent"),}));vi.mock("@/automations/urls.js", () => ({ automationPageUrl: vi.fn(async () => "https://airglow.test/u/tester/sr"),}));vi.mock("@/actions/registry.js", () => ({ ACTION_REGISTRY: new Proxy( {}, { get: () => ({ execute: mockExecute }), }, ), isRecordProducingAction: () => false, ACTION_TYPES: [],}));
import { runDue } from "./schedule-runner.js";import { notifySystem } from "../atmo/notify-system.js";import { db } from "../db/index.js";import { automations, scheduledActions, type Action } from "../db/schema.js";import { makeAutomation, makeEvent } from "../test/fixtures.js";
const mockNotifySystem = vi.mocked(notifySystem);
const DID = "did:plc:runnertest";const URI = "at://did:plc:runnertest/run.airglow.automation/sr";
async function seedAutomation(overrides?: Parameters<typeof makeAutomation>[0]) { await db.insert(automations).values( makeAutomation({ uri: URI, did: DID, rkey: "sr", ...overrides, }), );}
type SeedRowOpts = { id?: string; dueAt?: Date; status?: "pending" | "running" | "done" | "failed"; attempts?: number; claimedAt?: Date | null; identifier?: string; action?: Action;};
async function seedRow(opts: SeedRowOpts = {}) { const id = opts.id ?? `row-${Math.random().toString(36).slice(2, 10)}`; await db.insert(scheduledActions).values({ id, ownerDid: DID, automationUri: URI, sourceActionIndex: 0, identifier: opts.identifier ?? `${DID}::test-id`, dueAt: opts.dueAt ?? new Date(Date.now() - 1_000), status: opts.status ?? "pending", attempts: opts.attempts ?? 0, claimedAt: opts.claimedAt ?? null, action: opts.action ?? { $type: "webhook", callbackUrl: "https://example.com/hook", secret: "test-secret-32chars-placeholder00", }, snapshot: { event: makeEvent(), fetchContext: {} }, createdAt: new Date(), }); return id;}
async function getRow(id: string) { const row = await db.query.scheduledActions.findFirst({ where: eq(scheduledActions.id, id), }); if (!row) throw new Error(`row ${id} not found`); return row;}
beforeEach(async () => { await db.delete(scheduledActions); await db.delete(automations); await seedAutomation(); mockExecute.mockReset(); notifyAutomationChange.mockReset(); maybeDisableForSessionLoss.mockReset().mockResolvedValue(false); mockNotifySystem.mockReset().mockResolvedValue("sent");});
describe("runDue: claim & dispatch", () => { it("marks a successful row as done after dispatching to the executor", async () => { mockExecute.mockResolvedValueOnce({ statusCode: 200 }); const id = await seedRow();
await runDue();
expect(mockExecute).toHaveBeenCalledTimes(1); const row = await getRow(id); expect(row.status).toBe("done"); expect(row.attempts).toBe(1); expect(row.executedAt).toBeInstanceOf(Date); expect(row.error).toBeNull(); });
it("skips rows whose dueAt is in the future", async () => { mockExecute.mockResolvedValue({ statusCode: 200 }); const id = await seedRow({ dueAt: new Date(Date.now() + 60_000) });
await runDue();
expect(mockExecute).not.toHaveBeenCalled(); const row = await getRow(id); expect(row.status).toBe("pending"); expect(row.attempts).toBe(0); });
it("processes multiple due rows in one tick", async () => { mockExecute.mockResolvedValue({ statusCode: 200 }); const a = await seedRow({ id: "a" }); const b = await seedRow({ id: "b" }); const c = await seedRow({ id: "c" });
await runDue();
expect(mockExecute).toHaveBeenCalledTimes(3); for (const id of [a, b, c]) { const row = await getRow(id); expect(row.status).toBe("done"); } });
it("sets claimedAt when transitioning to running", async () => { // Hold the executor open so we can observe the in-flight row state. let release: (v: { statusCode: number }) => void; const pending = new Promise<{ statusCode: number }>((resolve) => { release = resolve; }); mockExecute.mockReturnValueOnce(pending); const id = await seedRow();
const inflight = runDue(); // The executor runs only after the claim UPDATE lands. await vi.waitFor(() => expect(mockExecute).toHaveBeenCalled());
const claimed = await getRow(id); expect(claimed.status).toBe("running"); expect(claimed.claimedAt).toBeInstanceOf(Date); expect(claimed.attempts).toBe(1);
release!({ statusCode: 200 }); await inflight;
const done = await getRow(id); expect(done.status).toBe("done"); });});
describe("runDue: orphan recovery", () => { it("resets a row stuck in running past the orphan timeout", async () => { mockExecute.mockResolvedValueOnce({ statusCode: 200 }); const orphan = new Date(Date.now() - 30 * 60_000); // 30min ago, well past 10min cap const id = await seedRow({ status: "running", attempts: 1, claimedAt: orphan, });
await runDue();
// Recovered + claimed + executed in one tick. expect(mockExecute).toHaveBeenCalledTimes(1); const row = await getRow(id); expect(row.status).toBe("done"); expect(row.attempts).toBe(2); });
it("recovers a running row with NULL claimedAt (pre-migration / malformed state)", async () => { // Rows that existed before the claimedAt migration carry claimedAt=NULL. // SQLite's `NULL < cutoff` is NULL (falsy), so without the explicit isNull // branch these rows would be invisible to the orphan sweep AND to claim // (status='running' ≠ 'pending'). Confirm they get reset. mockExecute.mockResolvedValueOnce({ statusCode: 200 }); const id = await seedRow({ status: "running", attempts: 1, claimedAt: null, });
await runDue();
expect(mockExecute).toHaveBeenCalledTimes(1); const row = await getRow(id); expect(row.status).toBe("done"); expect(row.attempts).toBe(2); });
it("leaves a fresh running row alone (multi-process safety)", async () => { mockExecute.mockResolvedValue({ statusCode: 200 }); // A claim by another worker only seconds ago. This runner must NOT // disturb it. const fresh = new Date(Date.now() - 5_000); const id = await seedRow({ status: "running", attempts: 1, claimedAt: fresh, });
await runDue();
expect(mockExecute).not.toHaveBeenCalled(); const row = await getRow(id); expect(row.status).toBe("running"); expect(row.attempts).toBe(1); expect(row.claimedAt?.getTime()).toBe(fresh.getTime()); });});
describe("runOne: failure & retry", () => { it("reschedules with backoff on a retryable failure under MAX_ATTEMPTS", async () => { mockExecute.mockResolvedValueOnce({ statusCode: 503, error: "upstream down" }); const id = await seedRow();
const before = Date.now(); await runDue(); const row = await getRow(id);
expect(row.status).toBe("pending"); expect(row.attempts).toBe(1); expect(row.error).toBe("upstream down"); // dueAt pushed forward by attempts * RETRY_BACKOFF_MS (60s × 1). expect(row.dueAt.getTime()).toBeGreaterThanOrEqual(before + 59_000); });
it("marks failed after the third failed attempt", async () => { mockExecute.mockResolvedValue({ statusCode: 503, error: "still down" }); // Seed at attempts=2; claim bumps to 3 = MAX_ATTEMPTS. The check // `row.attempts < MAX_ATTEMPTS` is then false → mark failed. const id = await seedRow({ attempts: 2 });
await runDue(); const row = await getRow(id);
expect(row.status).toBe("failed"); expect(row.attempts).toBe(3); expect(row.error).toBe("still down"); expect(row.executedAt).toBeInstanceOf(Date); });
it("marks failed terminally on a non-retryable failure (e.g. 400)", async () => { mockExecute.mockResolvedValueOnce({ statusCode: 400, error: "bad payload" }); const id = await seedRow();
await runDue(); const row = await getRow(id);
expect(row.status).toBe("failed"); expect(row.attempts).toBe(1); expect(row.error).toBe("bad payload"); });
it("treats an executor throw as a retryable failure (status 0)", async () => { mockExecute.mockRejectedValueOnce(new Error("boom")); const id = await seedRow();
await runDue(); const row = await getRow(id);
// Thrown error → result.statusCode=0; isRetryable(0)=true; reschedules. expect(row.status).toBe("pending"); expect(row.attempts).toBe(1); expect(row.error).toBe("boom"); });
it("invokes the session-loss handler + notify on 401", async () => { mockExecute.mockResolvedValueOnce({ statusCode: 401, error: "session lost" }); maybeDisableForSessionLoss.mockResolvedValueOnce(true); const id = await seedRow();
await runDue();
expect(maybeDisableForSessionLoss).toHaveBeenCalledWith(URI, expect.any(Number)); expect(notifyAutomationChange).toHaveBeenCalledTimes(1); // 401 is non-retryable → marked failed. const row = await getRow(id); expect(row.status).toBe("failed"); });
it("does not notify when session-loss handler reports no transition", async () => { mockExecute.mockResolvedValueOnce({ statusCode: 401, error: "session lost" }); // Default mock returns false (no flip). await seedRow();
await runDue();
expect(maybeDisableForSessionLoss).toHaveBeenCalled(); expect(notifyAutomationChange).not.toHaveBeenCalled(); });});
describe("runOne: healthy-to-failing finality", () => { async function failingMarker() { const row = await db.query.automations.findFirst({ where: eq(automations.uri, URI) }); return row?.failingNotifiedAt ?? null; }
it("does not set the marker or notify on a rescheduled retryable attempt", async () => { mockExecute.mockResolvedValueOnce({ statusCode: 503, error: "upstream down" }); const id = await seedRow();
await runDue();
expect((await getRow(id)).status).toBe("pending"); expect(mockNotifySystem).not.toHaveBeenCalled(); expect(await failingMarker()).toBeNull(); });
it("sets the marker and notifies once when retries are exhausted (post-execute markFailed)", async () => { mockExecute.mockResolvedValue({ statusCode: 503, error: "still down" }); const id = await seedRow({ attempts: 2 }); // claim bumps to MAX_ATTEMPTS
await runDue();
expect((await getRow(id)).status).toBe("failed"); expect(mockNotifySystem).toHaveBeenCalledTimes(1); expect(mockNotifySystem).toHaveBeenCalledWith( DID, "automation-errors", expect.objectContaining({ body: expect.stringContaining("503") }), ); expect(await failingMarker()).not.toBeNull(); });
it("sets the marker and notifies once on a non-retryable failure", async () => { mockExecute.mockResolvedValueOnce({ statusCode: 400, error: "bad payload" }); await seedRow();
await runDue();
expect(mockNotifySystem).toHaveBeenCalledTimes(1); expect(await failingMarker()).not.toBeNull(); });
it("the pre-execute deleted/disabled guards do not notify", async () => { await db.update(automations).set({ active: false }).where(eq(automations.uri, URI)); const disabledRow = await seedRow({ id: "guard-disabled" });
await runDue();
expect((await getRow(disabledRow)).status).toBe("failed"); expect(mockNotifySystem).not.toHaveBeenCalled(); expect(await failingMarker()).toBeNull(); });
it("a 402 terminal failure marks the row failed without notifying (budget carve-out)", async () => { mockExecute.mockResolvedValueOnce({ statusCode: 402, error: "budget exhausted" }); const id = await seedRow();
await runDue();
expect((await getRow(id)).status).toBe("failed"); expect(mockNotifySystem).not.toHaveBeenCalled(); expect(await failingMarker()).toBeNull(); });});
describe("runOne: automation lifecycle", () => { it("marks failed when the originating automation is disabled", async () => { await db.update(automations).set({ active: false }).where(eq(automations.uri, URI)); const id = await seedRow();
await runDue();
const row = await getRow(id); expect(row.status).toBe("failed"); expect(row.error).toBe("originating automation is disabled"); expect(mockExecute).not.toHaveBeenCalled(); });
it("marks failed when snapshot deserialization fails", async () => { // Seed a row with a snapshot missing required fields. Inserted via raw // overwrite so the type system doesn't reject it. const id = await seedRow(); await db .update(scheduledActions) .set({ snapshot: { broken: true } as never }) .where(eq(scheduledActions.id, id));
await runDue();
const row = await getRow(id); expect(row.status).toBe("failed"); expect(row.error).toMatch(/snapshot deserialization failed/); expect(mockExecute).not.toHaveBeenCalled(); });});
// ---------------------------------------------------------------------------// Steps-model replay (U8): source_step_id lookup, deleted-step refusal, and// the rename-vs-snapshot guard.// ---------------------------------------------------------------------------
describe("steps-model scheduled rows", () => { const STEPS = [ { $type: "variable" as const, id: "v1", name: "tag", value: "x", }, { $type: "webhook" as const, id: "w1", name: "hook", callbackUrl: "https://example.com/hook", secret: "test-secret-32chars-placeholder00", }, ];
async function seedStepsRow(opts: { id?: string; stepId?: string | null; action?: Record<string, unknown>; snapshotContext?: Record<string, unknown>; }) { const id = opts.id ?? `srow-${Math.random().toString(36).slice(2, 10)}`; await db.insert(scheduledActions).values({ id, ownerDid: DID, automationUri: URI, sourceActionIndex: -1, sourceStepId: opts.stepId === undefined ? "w1" : opts.stepId, identifier: `${DID}::steps-id`, dueAt: new Date(Date.now() - 1_000), status: "pending", attempts: 0, claimedAt: null, action: (opts.action ?? { $type: "webhook", id: "w1", name: "hook", callbackUrl: "https://example.com/hook", secret: "test-secret-32chars-placeholder00", }) as never, snapshot: { event: makeEvent(), fetchContext: opts.snapshotContext ?? {} }, createdAt: new Date(), }); return id; }
it("replays at the step's current flat index with runMeta stamped", async () => { await db.delete(automations); await seedAutomation({ steps: STEPS } as never); mockExecute.mockResolvedValue({ statusCode: 200 }); const id = await seedStepsRow({}); await runDue(); expect((await getRow(id)).status).toBe("done"); expect(mockExecute).toHaveBeenCalledTimes(1); const [match, index] = mockExecute.mock.calls[0]!; expect(index).toBe(1); // w1's flat index in the current tree expect((match as { runMeta?: { stepId: string } }).runMeta?.stepId).toBe("w1"); });
it("fails the row cleanly when the source step no longer exists", async () => { await db.delete(automations); await seedAutomation({ steps: [STEPS[0]] } as never); // w1 removed const id = await seedStepsRow({}); await runDue(); const row = await getRow(id); expect(row.status).toBe("failed"); expect(row.error).toContain("no longer exists"); expect(mockExecute).not.toHaveBeenCalled(); });
it("fails the row instead of rendering blanks when templates reference names missing from the snapshot", async () => { await db.delete(automations); await seedAutomation({ steps: STEPS } as never); const id = await seedStepsRow({ action: { $type: "webhook", id: "w1", name: "hook", callbackUrl: "https://example.com/hook?u={{parentPost.uri}}", secret: "test-secret-32chars-placeholder00", }, snapshotContext: {}, // parentPost was renamed/never captured }); await runDue(); const row = await getRow(id); expect(row.status).toBe("failed"); expect(row.error).toContain("parentPost"); expect(mockExecute).not.toHaveBeenCalled(); });
it("legacy rows (no sourceStepId) keep the original index path", async () => { mockExecute.mockResolvedValue({ statusCode: 200 }); const id = await seedRow({}); await runDue(); expect((await getRow(id)).status).toBe("done"); const [, index] = mockExecute.mock.calls[0]!; expect(index).toBe(0); });});