Something went wrong. Try again.
Automations and webhooks for the AT Protocol airglow.run
automation webhook atproto atprotocol
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405// Listing tasks (U6): walk an owner's follow records on one app and seed the// follow mirror. Backfill runs end with one guarded union step that queues the// follows needed to merge the graphs; sweep runs refresh the mirror and queue// anything a missed event left out. Neither ever queues an unfollow (KTD-6).
import { randomUUID } from "node:crypto";import { and, asc, eq, inArray, isNull, lt, or } from "drizzle-orm";import { conn, first } from "../db/tx.js";import { syncBackfillTasks, syncFollowMirror, syncQueue, syncSets, syncWaiting,} from "../db/schema.js";import type { FollowTarget } from "../db/schema.js";import { listRecordsPages } from "../pds/list-records.js";import { enqueue, getSet, inTransaction, isAppSeeded, lockOwner, mirroredRkeys, touchMirrors, upsertMirrors, type EnqueueRow,} from "./store.js";import { DID_RE, followCollection } from "./targets.js";import { isTombstoned } from "./tombstones.js";import { wakeSync } from "./wake.js";
type Task = typeof syncBackfillTasks.$inferSelect;
/** Pages walked per task per tick. A task yields between bursts, keeping its * cursor, so one large account cannot starve the others. */const PAGES_PER_TICK = 10;const TASKS_PER_TICK = 4;const TASK_ORPHAN_MS = 10 * 60 * 1000;const TASK_RETRY_BASE_MS = 30_000;const TASK_RETRY_MAX_MS = 60 * 60 * 1000;
/** Start a backfill for `apps` (on enable, or when apps are added). Apps whose * earlier listing never finished are listed again in the same run, and older * unfinished tasks are cancelled, so a run always covers every unseeded app * and its union step sees the whole set. Returns the run id, or null when * nothing needs listing. */export async function startBackfill( did: string, apps: FollowTarget[], now: Date,): Promise<string | null> { return inTransaction(async () => { await lockOwner(did); const set = await getSet(did); if (!set || !set.enabled) return null; const toList: FollowTarget[] = []; for (const app of set.apps) { if (apps.includes(app) || !(await isAppSeeded(did, app))) toList.push(app); } if (toList.length === 0) return null; const runId = randomUUID(); await conn() .update(syncBackfillTasks) .set({ status: "cancelled", finishedAt: now }) .where( and( eq(syncBackfillTasks.did, did), eq(syncBackfillTasks.kind, "backfill"), inArray(syncBackfillTasks.status, ["pending", "running"]), ), ); await conn() .insert(syncBackfillTasks) .values( toList.map((app) => ({ did, app, kind: "backfill" as const, runId, createdAt: now })), ); await conn() .update(syncSets) .set({ backfillId: runId, backfillTotal: null, updatedAt: now }) .where(eq(syncSets.id, set.id)); return runId; });}
/** Queue a sweep for one owner unless one is already in progress. */export async function startSweepListing(did: string, now: Date): Promise<string | null> { return inTransaction(async () => { await lockOwner(did); const set = await getSet(did); if (!set || !set.enabled) return null; const inProgress = await conn() .select({ id: syncBackfillTasks.id }) .from(syncBackfillTasks) .where( and( eq(syncBackfillTasks.did, did), eq(syncBackfillTasks.kind, "sweep"), inArray(syncBackfillTasks.status, ["pending", "running"]), ), ) .limit(1) .then(first); if (inProgress) return null; const runId = `sweep-${randomUUID()}`; await conn() .insert(syncBackfillTasks) .values(set.apps.map((app) => ({ did, app, kind: "sweep" as const, runId, createdAt: now }))); return runId; });}
async function resetOrphanTasks(now: Date): Promise<void> { await conn() .update(syncBackfillTasks) .set({ status: "pending" }) .where( and( eq(syncBackfillTasks.status, "running"), or( lt(syncBackfillTasks.claimedAt, new Date(now.getTime() - TASK_ORPHAN_MS)), isNull(syncBackfillTasks.claimedAt), ), ), );}
async function claimTasks(now: Date): Promise<Task[]> { const candidates = await conn() .select() .from(syncBackfillTasks) .where(eq(syncBackfillTasks.status, "pending")) .orderBy(asc(syncBackfillTasks.id)); const claimed: Task[] = []; for (const task of candidates) { if (claimed.length >= TASKS_PER_TICK) break; // Back off after a failed burst: 30s, 60s, ... capped at an hour. if (task.attempts > 0 && task.finishedAt) { const wait = Math.min(TASK_RETRY_BASE_MS * 2 ** (task.attempts - 1), TASK_RETRY_MAX_MS); if (now.getTime() - task.finishedAt.getTime() < wait) continue; } const listingStartedAt = task.listingStartedAt ?? now; const ok = await conn() .update(syncBackfillTasks) .set({ status: "running", claimedAt: now, listingStartedAt }) .where(and(eq(syncBackfillTasks.id, task.id), eq(syncBackfillTasks.status, "pending"))) .returning({ id: syncBackfillTasks.id }); if (ok.length > 0) claimed.push({ ...task, status: "running", claimedAt: now, listingStartedAt }); } return claimed;}
async function taskStillRunning(id: number): Promise<boolean> { return ( ( await conn() .select({ status: syncBackfillTasks.status }) .from(syncBackfillTasks) .where(eq(syncBackfillTasks.id, id)) .limit(1) .then(first) )?.status === "running" );}
/** Upsert one page of listed follow records into the mirror. Returns the * subjects that were not mirrored before (a sweep propagates those). */async function mirrorPage( task: Task, records: Array<{ rkey: string; value: Record<string, unknown> }>, now: Date,): Promise<string[]> { const follows = new Map<string, string>(); for (const record of records) { const subject = record.value.subject; if (typeof subject !== "string" || !DID_RE.test(subject) || subject === task.did) continue; // Deleted while this listing ran: the page predates the delete. if (isTombstoned(task.did, task.app, record.rkey, now.getTime())) continue; if (!follows.has(record.rkey)) follows.set(record.rkey, subject); } const known = await mirroredRkeys(task.did, task.app, [...follows.keys()]); const fresh = [...follows].filter(([rkey]) => !known.has(rkey)); await touchMirrors(task.did, task.app, [...known], now); await upsertMirrors( fresh.map(([rkey, subject]) => ({ did: task.did, app: task.app, rkey, subject, seenAt: now })), ); return fresh.map(([, subject]) => subject);}
async function runTaskBurst(task: Task, clock: () => Date): Promise<void> { let cursor = task.cursor; let pages = 0; const discovered: string[] = []; try { for await (const page of listRecordsPages(task.did, followCollection(task.app), { startCursor: cursor, })) { const now = clock(); const stillRunning = await inTransaction(async () => { // Serialize with the live follow handlers so a delete cannot land // between the tombstone check and the mirror write (KTD-7). await lockOwner(task.did); if (!(await taskStillRunning(task.id))) return false; discovered.push(...(await mirrorPage(task, page.records, now))); await conn() .update(syncBackfillTasks) .set({ cursor: page.cursor }) .where(eq(syncBackfillTasks.id, task.id)); return true; }); if (!stillRunning) return; cursor = page.cursor; pages++; if (!cursor) break; if (pages >= PAGES_PER_TICK) { const enqueued = await inTransaction(async () => { await lockOwner(task.did); const queued = task.kind === "sweep" && (await propagateDiscovered(task, discovered, now)); await conn() .update(syncBackfillTasks) .set({ status: "pending", claimedAt: null }) .where(and(eq(syncBackfillTasks.id, task.id), eq(syncBackfillTasks.status, "running"))); return queued; }); if (enqueued) wakeSync(); return; } } } catch (err) { const now = clock(); await conn() .update(syncBackfillTasks) .set({ status: "pending", claimedAt: null, attempts: task.attempts + 1, finishedAt: now, lastError: err instanceof Error ? err.message : String(err), }) .where(and(eq(syncBackfillTasks.id, task.id), eq(syncBackfillTasks.status, "running"))); if (task.kind === "sweep") { const enqueued = await inTransaction(async () => { await lockOwner(task.did); return propagateDiscovered(task, discovered, now); }); if (enqueued) wakeSync(); } return; }
const now = clock(); const enqueued = await inTransaction(async () => { await lockOwner(task.did); const done = await conn() .update(syncBackfillTasks) .set({ status: "done", finishedAt: now, lastError: null }) .where(and(eq(syncBackfillTasks.id, task.id), eq(syncBackfillTasks.status, "running"))) .returning({ id: syncBackfillTasks.id }); if (done.length === 0) return false; if (task.kind === "sweep") { const queued = await propagateDiscovered(task, discovered, now); await dropUnseenMirrorRows(task); return queued; } return completeBackfillIfDone(task.did, task.runId, now); }); // Woken only after commit: the runner must see the queued rows and must not // run inside this transaction. if (enqueued) wakeSync();}
/** One tick of listing work. Exported for tests; production drives it * through a sync runner loop. */export async function runListingDue(clock: () => Date = () => new Date()): Promise<number> { await resetOrphanTasks(clock()); const tasks = await claimTasks(clock()); await Promise.all(tasks.map((task) => runTaskBurst(task, clock))); return tasks.length;}
/** The guarded union step: once every task of the owner's current backfill * is done, queue one follow per (subject, app) pair the mirror lacks. The * total is written only while still null, so it runs at most once per run. * Returns whether it queued anything; the caller wakes the runner once its * transaction commits. */export async function completeBackfillIfDone( did: string, runId: string, now: Date,): Promise<boolean> { const set = await getSet(did); if (!set || !set.enabled || set.backfillId !== runId || set.backfillTotal !== null) return false; const unfinished = await conn() .select({ id: syncBackfillTasks.id }) .from(syncBackfillTasks) .where( and( eq(syncBackfillTasks.runId, runId), inArray(syncBackfillTasks.status, ["pending", "running"]), ), ) .limit(1) .then(first); if (unfinished) return false;
const mirror = await conn() .select({ app: syncFollowMirror.app, subject: syncFollowMirror.subject }) .from(syncFollowMirror) .where(and(eq(syncFollowMirror.did, did), inArray(syncFollowMirror.app, set.apps))); const appsBySubject = new Map<string, Set<FollowTarget>>(); for (const { app, subject } of mirror) { const apps = appsBySubject.get(subject) ?? new Set(); apps.add(app); appsBySubject.set(subject, apps); } const skip = await skipKeys(did); const rows: EnqueueRow[] = []; for (const [subject, apps] of appsBySubject) { for (const app of set.apps) { if (apps.has(app) || skip.has(`${app}\0${subject}`)) continue; rows.push({ did, app, kind: "follow", subject, source: "backfill", backfillId: runId }); } } await enqueue(rows, now); await conn() .update(syncSets) .set({ backfillTotal: rows.length, updatedAt: now }) .where( and(eq(syncSets.id, set.id), eq(syncSets.backfillId, runId), isNull(syncSets.backfillTotal)), ); return rows.length > 0;}
/** Pairs to leave alone: already waiting for a profile, or already queued. */async function skipKeys(did: string): Promise<Set<string>> { const keys = new Set<string>(); for (const w of await conn() .select({ app: syncWaiting.app, subject: syncWaiting.subject }) .from(syncWaiting) .where(eq(syncWaiting.did, did))) { keys.add(`${w.app}\0${w.subject}`); } for (const q of await conn() .select({ app: syncQueue.app, subject: syncQueue.subject }) .from(syncQueue) .where( and( eq(syncQueue.did, did), eq(syncQueue.kind, "follow"), inArray(syncQueue.status, ["pending", "running"]), ), )) { keys.add(`${q.app}\0${q.subject}`); } return keys;}
/** A sweep found follow records the mirror missed (an event lost to * downtime): propagate them like a live follow. Returns whether it queued * anything; the caller wakes the runner once its transaction commits. */async function propagateDiscovered(task: Task, subjects: string[], now: Date): Promise<boolean> { if (subjects.length === 0) return false; const set = await getSet(task.did); if (!set || !set.enabled) return false; const skip = await skipKeys(task.did); const unique = [...new Set(subjects)]; const mirrored = new Set( ( await conn() .select({ app: syncFollowMirror.app, subject: syncFollowMirror.subject }) .from(syncFollowMirror) .where(and(eq(syncFollowMirror.did, task.did), inArray(syncFollowMirror.subject, unique))) ).map((r) => `${r.app}\0${r.subject}`), ); const rows: EnqueueRow[] = []; for (const subject of unique) { for (const app of set.apps) { const key = `${app}\0${subject}`; if (mirrored.has(key) || skip.has(key)) continue; rows.push({ did: task.did, app, kind: "follow", subject, source: "sweep" }); } } subjects.length = 0; await enqueue(rows, now); return rows.length > 0;}
/** After a complete sweep listing, a mirror row not seen since the listing * started names a record that no longer exists. Drop the row only: the * unfollow that removed it was missed, and deletes come only from events. */async function dropUnseenMirrorRows(task: Task): Promise<void> { if (!task.listingStartedAt) return; await conn() .delete(syncFollowMirror) .where( and( eq(syncFollowMirror.did, task.did), eq(syncFollowMirror.app, task.app), lt(syncFollowMirror.seenAt, task.listingStartedAt), ), );}