Something went wrong. Try again.
Automations and webhooks for the AT Protocol airglow.run
automation webhook atproto atprotocol
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";import { eq } from "drizzle-orm";
const { mockConfig } = vi.hoisted(() => ({ mockConfig: { syncHourlyPoints: 1500, syncDailyPoints: 15000 },}));vi.mock("../config.js", () => ({ config: mockConfig }));vi.mock("@/config.js", () => ({ config: mockConfig }));vi.mock("@/db/index.js", async () => { const { createTestDb } = await import("../test/db.js"); return { db: await createTestDb() };});vi.mock("../automations/pds.js", () => ({ createArbitraryRecord: vi.fn(), deleteArbitraryRecord: vi.fn(), generateTid: vi.fn(),}));vi.mock("../automations/profile-status.js", () => ({ profileStatus: vi.fn() }));
import { createArbitraryRecord, deleteArbitraryRecord, generateTid } from "../automations/pds.js";import { PdsHttpError } from "../automations/errors.js";import { profileStatus } from "../automations/profile-status.js";import { db } from "../db/index.js";import { conn, inTransaction } from "../db/tx.js";import { oauthSessions, syncActivity, syncFollowMirror, syncQueue, syncSets, users,} from "../db/schema.js";import { MAX_ATTEMPTS, ORPHAN_TIMEOUT_MS, runSyncQueue, startSyncRunner } from "./runner.js";import { enqueue, getSet, mirrorRowsForSubject, saveEnabledSet, upsertMirror, waitingCounts,} from "./store.js";import { markSeeded, seedUser } from "./test-helpers.js";import { wakeSync } from "./wake.js";
const OWNER = "did:plc:ownerownerownerownerowne";const BOB = "did:plc:bobbobbobbobbobbobbobbob";const T0 = Date.UTC(2026, 8, 14, 12);let clockMs = T0;const clock = () => new Date(clockMs);
const mockCreate = vi.mocked(createArbitraryRecord);const mockDelete = vi.mocked(deleteArbitraryRecord);const mockTid = vi.mocked(generateTid);const mockProfile = vi.mocked(profileStatus);
function rows() { return db.select().from(syncQueue).orderBy(syncQueue.id);}async function activity() { return (await db.select().from(syncActivity).orderBy(syncActivity.id)).map((a) => ({ outcome: a.outcome, reason: a.reason, app: a.app, source: a.source, points: a.points, }));}function http(status: number, body = "", reset: number | null = null) { return new PdsHttpError("com.atproto.repo.createRecord", status, body, reset);}
let tidCounter = 0;
beforeEach(async () => { vi.clearAllMocks(); clockMs = T0; tidCounter = 0; mockTid.mockImplementation(() => `tid${++tidCounter}`); mockCreate.mockResolvedValue({ uri: "at://x", cid: "c" }); mockDelete.mockResolvedValue(undefined); mockProfile.mockResolvedValue("found"); mockConfig.syncHourlyPoints = 1500; mockConfig.syncDailyPoints = 15000; await db.delete(users); await db.delete(oauthSessions); await seedUser(OWNER, clock()); await db.insert(oauthSessions).values({ key: OWNER, value: "{}" }); await saveEnabledSet(OWNER, ["bluesky", "tangled", "sifa"], clock()); await markSeeded(OWNER, ["bluesky", "tangled", "sifa"], clock()); // The owner follows Bob on Bluesky, so propagating Bob is still wanted. await upsertMirror({ did: OWNER, app: "bluesky", rkey: "b1", subject: BOB, seenAt: clock() });});
async function follow( app: "tangled" | "sifa" | "bluesky", source: "live" | "backfill" | "waiting" = "live",) { await enqueue([{ did: OWNER, app, kind: "follow", subject: BOB, source }], clock());}
describe("runSyncQueue: follow rows", () => { it("writes the follow with a pre-generated rkey and records three points", async () => { await follow("tangled");
expect(await runSyncQueue(clock)).toBe(1);
expect(mockCreate).toHaveBeenCalledWith( OWNER, "sh.tangled.graph.follow", { subject: BOB, createdAt: new Date(T0).toISOString() }, "tid1", ); expect((await rows())[0]).toMatchObject({ status: "done", rkey: "tid1" }); expect((await mirrorRowsForSubject(OWNER, BOB)).find((m) => m.app === "tangled")).toMatchObject( { rkey: "tid1", createdBySync: true, }, ); expect(await activity()).toEqual([ { outcome: "written", reason: null, app: "tangled", source: "live", points: 3 }, ]); });
it("lands the mirror row before the PDS call", async () => { await follow("tangled"); mockCreate.mockImplementationOnce(async () => { expect((await mirrorRowsForSubject(OWNER, BOB)).some((m) => m.rkey === "tid1")).toBe(true); return { uri: "at://x", cid: "c" }; }); await runSyncQueue(clock); expect(mockCreate).toHaveBeenCalledTimes(1); });
it("skips a subject the owner no longer follows anywhere in the set", async () => { await db.delete(syncActivity); await follow("tangled"); await db.update(syncSets).set({ apps: ["tangled", "sifa"] });
await runSyncQueue(clock);
expect(mockCreate).not.toHaveBeenCalled(); expect(await activity()).toEqual([ { outcome: "skipped", reason: "no_longer_followed", app: "tangled", source: "live", points: 0, }, ]); });
it("writes once when the same follow is queued twice", async () => { await follow("tangled"); await follow("tangled");
await runSyncQueue(clock);
expect(mockCreate).toHaveBeenCalledTimes(1); expect((await rows()).map((r) => r.status)).toEqual(["done", "skipped"]); expect((await activity()).map((a) => a.reason)).toEqual([null, "already_following"]); });
it("puts a subject with no profile on the waiting list", async () => { mockProfile.mockResolvedValueOnce("not-found"); await follow("tangled");
await runSyncQueue(clock);
expect(mockCreate).not.toHaveBeenCalled(); expect((await rows())[0]!.status).toBe("waiting"); expect(await waitingCounts(OWNER)).toEqual({ tangled: 1 }); expect((await activity())[0]).toMatchObject({ outcome: "skipped", reason: "no_profile" }); });
it("retries a row whose profile lookup failed instead of writing", async () => { mockProfile.mockResolvedValueOnce("unknown"); await follow("tangled");
await runSyncQueue(clock);
expect(mockCreate).not.toHaveBeenCalled(); expect((await rows())[0]).toMatchObject({ status: "pending", attempts: 1 }); });
it("does not write when the owner unfollows while the profile lookup is in flight", async () => { await follow("tangled"); mockProfile.mockImplementationOnce(async () => { // The owner's Bluesky unfollow lands while the lookup awaits the network. await db.delete(syncFollowMirror).where(eq(syncFollowMirror.rkey, "b1")); return "found"; });
await runSyncQueue(clock);
expect(mockCreate).not.toHaveBeenCalled(); expect((await rows())[0]).toMatchObject({ status: "skipped" }); expect(await mirrorRowsForSubject(OWNER, BOB)).toEqual([]); });
it("reuses the rkey when a timed-out write is retried", async () => { await follow("tangled"); mockCreate.mockRejectedValueOnce(new Error("PDS createRecord fetch failed: timeout"));
await runSyncQueue(clock); expect((await rows())[0]).toMatchObject({ status: "pending", rkey: "tid1" });
clockMs += 60_000; mockCreate.mockRejectedValueOnce(http(400, '{"message":"Record already exists"}')); await runSyncQueue(clock);
expect(mockCreate).toHaveBeenCalledTimes(2); expect(mockCreate.mock.calls[1]![3]).toBe("tid1"); expect((await rows())[0]!.status).toBe("done"); expect( (await mirrorRowsForSubject(OWNER, BOB)).filter((m) => m.app === "tangled"), ).toHaveLength(1); });
it("does not claim follow rows for an app whose listing is not done", async () => { await db.delete(users); await seedUser(OWNER, clock()); await db.insert(oauthSessions).values({ key: OWNER, value: "{}" }).onConflictDoNothing(); await saveEnabledSet(OWNER, ["bluesky", "tangled", "sifa"], clock()); await markSeeded(OWNER, ["bluesky", "sifa"], clock()); await upsertMirror({ did: OWNER, app: "bluesky", rkey: "b1", subject: BOB, seenAt: clock() }); await follow("tangled"); await follow("sifa");
await runSyncQueue(clock);
expect((await rows()).map((r) => [r.app, r.status])).toEqual([ ["tangled", "pending"], ["sifa", "done"], ]); });});
describe("runSyncQueue: unfollow rows and ordering", () => { it("deletes by rkey, drops the mirror row, and records one point", async () => { await upsertMirror({ did: OWNER, app: "tangled", rkey: "t1", subject: BOB, seenAt: clock() }); await enqueue( [{ did: OWNER, app: "tangled", kind: "unfollow", subject: BOB, rkey: "t1", source: "live" }], clock(), );
await runSyncQueue(clock);
expect(mockDelete).toHaveBeenCalledWith(OWNER, "sh.tangled.graph.follow", "t1"); expect((await mirrorRowsForSubject(OWNER, BOB)).some((m) => m.rkey === "t1")).toBe(false); expect(await activity()).toEqual([ { outcome: "deleted", reason: null, app: "tangled", source: "live", points: 1 }, ]); });
it("runs an unfollow before a later follow on the same lane, even when the delete is slow", async () => { await upsertMirror({ did: OWNER, app: "tangled", rkey: "t1", subject: BOB, seenAt: clock() }); await enqueue( [{ did: OWNER, app: "tangled", kind: "unfollow", subject: BOB, rkey: "t1", source: "live" }], clock(), ); await follow("tangled"); const order: string[] = []; mockDelete.mockImplementationOnce(async () => { await new Promise((r) => setTimeout(r, 20)); order.push("delete"); }); mockCreate.mockImplementationOnce(async () => { order.push("create"); return { uri: "at://x", cid: "c" }; });
await runSyncQueue(clock);
expect(order).toEqual(["delete", "create"]); expect((await rows()).map((r) => r.status)).toEqual(["done", "done"]); });
it("blocks a lane behind a head that is waiting on a retry", async () => { await follow("tangled"); await enqueue( [{ did: OWNER, app: "tangled", kind: "unfollow", subject: BOB, rkey: "t9", source: "live" }], clock(), ); mockCreate.mockRejectedValueOnce(http(503, "down"));
await runSyncQueue(clock);
expect((await rows()).map((r) => r.status)).toEqual(["pending", "pending"]); expect(mockDelete).not.toHaveBeenCalled(); });});
describe("runSyncQueue: failures, pacing and sessions", () => { it("backs off on a 5xx and fails the row after the attempt cap", async () => { await follow("tangled"); mockCreate.mockRejectedValue(http(503, "down"));
for (let i = 0; i < MAX_ATTEMPTS; i++) { await runSyncQueue(clock); clockMs += 60 * 60 * 1000; }
expect((await rows())[0]).toMatchObject({ status: "failed", attempts: MAX_ATTEMPTS }); expect((await activity()).at(-1)).toMatchObject({ outcome: "failed" }); });
it("parks the account until the reported reset on a 429, without spending an attempt", async () => { await follow("tangled"); await follow("sifa"); const resetSeconds = Math.floor(T0 / 1000) + 600; mockCreate.mockRejectedValueOnce(http(429, "RateLimitExceeded", resetSeconds)); mockCreate.mockRejectedValueOnce(http(429, "RateLimitExceeded", resetSeconds));
await runSyncQueue(clock);
expect((await rows()).map((r) => [r.status, r.attempts])).toEqual([ ["pending", 0], ["pending", 0], ]); expect((await getSet(OWNER))!.parkedUntil!.getTime()).toBe(resetSeconds * 1000);
clockMs += 5 * 60 * 1000; await runSyncQueue(clock); expect(mockCreate).toHaveBeenCalledTimes(2);
clockMs += 6 * 60 * 1000; await runSyncQueue(clock); expect((await rows()).map((r) => r.status)).toEqual(["done", "done"]); });
it("does not claim beyond the hourly point budget", async () => { mockConfig.syncHourlyPoints = 4; await follow("tangled"); await follow("sifa");
await runSyncQueue(clock);
expect((await rows()).map((r) => r.status)).toEqual(["done", "pending"]); });
it("pauses a set whose OAuth session row is gone and claims nothing", async () => { await db.delete(oauthSessions); await follow("tangled");
await runSyncQueue(clock);
expect(mockCreate).not.toHaveBeenCalled(); expect((await getSet(OWNER))!.pausedAt).not.toBeNull(); expect(await activity()).toEqual([ { outcome: "paused", reason: "session_missing", app: null, source: null, points: 0 }, ]); });
it("parks on a single 401 and pauses only after five in a row", async () => { await follow("tangled"); mockCreate.mockRejectedValue(http(401, "expired"));
await runSyncQueue(clock); expect(await getSet(OWNER)).toMatchObject({ authFailures: 1, pausedAt: null }); expect((await rows())[0]).toMatchObject({ status: "pending", attempts: 0 });
for (let i = 0; i < 4; i++) { clockMs += 2 * 60 * 1000; await runSyncQueue(clock); } expect((await getSet(OWNER))!.pausedAt).not.toBeNull(); expect((await activity()).filter((a) => a.outcome === "paused")).toHaveLength(1);
clockMs += 2 * 60 * 1000; await runSyncQueue(clock); expect(mockCreate).toHaveBeenCalledTimes(5); });
it("resets the 401 streak after a success", async () => { await follow("tangled"); await follow("sifa"); mockCreate.mockRejectedValueOnce(http(401, "expired"));
await runSyncQueue(clock); expect((await getSet(OWNER))!.authFailures).toBe(0); });
it("re-queues a row orphaned past the recovery window", async () => { await follow("tangled"); await db .update(syncQueue) .set({ status: "running", claimedAt: new Date(T0 - ORPHAN_TIMEOUT_MS - 1) });
await runSyncQueue(clock);
expect((await rows())[0]!.status).toBe("done"); });
it("cancels a row for an app no longer in the set", async () => { await follow("tangled"); await db .update(syncSets) .set({ apps: ["bluesky", "sifa"] }) .where(eq(syncSets.did, OWNER));
await runSyncQueue(clock);
expect((await rows())[0]!.status).toBe("cancelled"); });});
describe("startSyncRunner", () => { afterEach(() => { vi.useRealTimers(); });
it("runs on a wake before the fallback tick and coalesces wakes during a tick", async () => { vi.useFakeTimers(); let release: () => void = () => {}; const work = vi.fn( () => new Promise<void>((resolve) => { release = resolve; }), ); const handle = startSyncRunner(work); expect(work).toHaveBeenCalledTimes(1);
wakeSync(); wakeSync(); expect(work).toHaveBeenCalledTimes(1);
release(); await vi.advanceTimersByTimeAsync(0); expect(work).toHaveBeenCalledTimes(2);
release(); await handle.stop(10); });
it("runs a tick woken from inside a transaction on the pool, not in that transaction", async () => { let release: () => void = () => {}; const onPool: boolean[] = []; const work = vi.fn(async () => { onPool.push(conn() === db); await new Promise<void>((resolve) => { release = resolve; }); }); const handle = startSyncRunner(work); release(); await new Promise((r) => setTimeout(r, 0)); // let the startup tick settle
await inTransaction(async () => { wakeSync(); }); expect(work).toHaveBeenCalledTimes(2); expect(onPool).toEqual([true, true]);
release(); await handle.stop(10); });
it("waits for the in-flight tick on stop", async () => { vi.useFakeTimers(); let finished = false; const handle = startSyncRunner(async () => { await new Promise((r) => setTimeout(r, 500)); finished = true; }); const stopping = handle.stop(5_000); await vi.advanceTimersByTimeAsync(600); await stopping; expect(finished).toBe(true); });});