Something went wrong. Try again.
Automations and webhooks for the AT Protocol airglow.run
automation webhook atproto atprotocol
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149// The Jetstream handler for social graph sync. Touches only the database: the mirror, the waiting list and// the write queue. Every PDS call happens later, in the runner.
import type { JetstreamEvent } from "../jetstream/matcher.js";import { FOLLOW_TARGETS, type FollowTarget } from "../automations/follow-targets.js";import { cancelPendingFollows, enqueue, getSet, hasQueuedUnfollow, hasRunningListing, inTransaction, lockOwner, mirrorHasRkey, mirrorRowsForSubject, ownersWaitingFor, purgeWaitingForSubject, removeWaiting, takeMirrorByRkey, touchMirror, upsertMirror, type EnqueueRow,} from "./store.js";import { DID_RE, appForFollowCollection, appForProfileCollection } from "./targets.js";import { addTombstone } from "./tombstones.js";import { wakeSync } from "./wake.js";
export async function handleSyncEvent(event: JetstreamEvent, now = new Date()): Promise<void> { const commit = event.commit; if (!commit) return; const followApp = appForFollowCollection(commit.collection); if (followApp) { if (commit.operation === "create") await handleFollowCreate(event, followApp, now); else if (commit.operation === "delete") await handleFollowDelete(event, followApp, now); return; } const profileApp = appForProfileCollection(commit.collection); if (profileApp && commit.operation === "create") await handleProfileCreate(event, profileApp, now);}
async function enabledSetWith(did: string, app: FollowTarget) { const set = await getSet(did); return set && set.enabled && set.apps.includes(app) ? set : undefined;}
async function handleFollowCreate( event: JetstreamEvent, app: FollowTarget, now: Date,): Promise<void> { const did = event.did; const { rkey, record } = event.commit!; const subject = record?.subject; if (typeof subject !== "string" || !DID_RE.test(subject) || subject === did) return;
let enqueued = false; await inTransaction(async () => { await lockOwner(did); const set = await enabledSetWith(did, app); if (!set) return; // An rkey the mirror already holds is sync's own write (the runner inserts // the row before calling the PDS) or a redelivery. Either way, nothing to // propagate. if (await mirrorHasRkey(did, app, rkey)) { await touchMirror(did, app, rkey, now); return; } await upsertMirror({ did, app, rkey, subject, seenAt: now }); // The owner followed on this app themselves, which settles any wait here. await removeWaiting(did, app, subject); const rows: EnqueueRow[] = set.apps .filter((other) => other !== app) .map((other) => ({ did, app: other, kind: "follow", subject, source: "live" })); await enqueue(rows, now); enqueued = rows.length > 0; }); if (enqueued) wakeSync();}
async function handleFollowDelete( event: JetstreamEvent, app: FollowTarget, now: Date,): Promise<void> { const did = event.did; const { rkey } = event.commit!;
let enqueued = false; await inTransaction(async () => { await lockOwner(did); const set = await enabledSetWith(did, app); if (!set) return; // Resolve and remove in one statement: a replayed delete finds nothing. const subject = await takeMirrorByRkey(did, app, rkey); if (subject === undefined) { if (await hasRunningListing(did, app)) addTombstone(did, app, rkey, now.getTime()); return; } // The echo of sync's own unfollow, possibly seen before the runner hears // back from the PDS (or while a timed-out delete waits to retry). The fan // out already happened when the owner unfollowed; doing it again could // unfollow a follow they made since. if (await hasQueuedUnfollow(did, app, rkey)) return; const remaining = await mirrorRowsForSubject(did, subject); // Deleting one of two duplicate records is cleanup, not an unfollow: the // owner still follows the subject on this app. if (remaining.some((row) => row.app === app)) return;
await cancelPendingFollows(did, subject, now); await purgeWaitingForSubject(did, subject); const rows: EnqueueRow[] = remaining .filter((row) => row.app !== app && set.apps.includes(row.app)) .map((row) => ({ did, app: row.app, kind: "unfollow", subject, rkey: row.rkey, source: "live", })); await enqueue(rows, now); enqueued = rows.length > 0; }); if (enqueued) wakeSync();}
async function handleProfileCreate( event: JetstreamEvent, app: FollowTarget, now: Date,): Promise<void> { if (event.commit!.rkey !== FOLLOW_TARGETS[app].profileRkey) return; const subject = event.did;
let enqueued = false; await inTransaction(async () => { const owners = await ownersWaitingFor(app, subject); const rows: EnqueueRow[] = []; for (const did of owners) { if (!(await enabledSetWith(did, app))) continue; rows.push({ did, app, kind: "follow", subject, source: "waiting" }); } await enqueue(rows, now); enqueued = rows.length > 0; }); if (enqueued) wakeSync();}