Something went wrong. Try again.
Automations and webhooks for the AT Protocol airglow.run
automation webhook atproto atprotocol
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778import { startCleanupJob } from "@/db/cleanup.js";import { startLivenessJob } from "@/auth/liveness.js";import { startStatsRefresh } from "@/stats/cache.js";import { startJetstream } from "@/jetstream/consumer.js";import { handleMatchedEvent } from "@/jetstream/handler.js";import { startFilterCountFlush } from "@/jetstream/filter-counts.js";import { startScheduleRunner } from "./schedule-runner.js";import { startSuspendRunner } from "./suspend-runner.js";import { startDryRunExpirationJob } from "./dry-run-expiration.js";import { syncPartitions } from "@/sync/subscriptions.js";import { startSyncRunner } from "@/sync/runner.js";import { runListingDue } from "@/sync/backfill.js";import { startSweepJob } from "@/sync/sweep.js";import type { Handle } from "./shutdown.js";
/** * Boots the Jetstream consumer plus the periodic jobs (DB cleanup, OAuth * liveness probe, stats cache refresh, scheduled-action runner, suspended-run * runner). * * Shutdown: stop periodic jobs → `manager.stop()` → `manager.drain()` with * 80% of the timeout budget → final `flushCursors()` only if the drain * completed cleanly. */export function startWorker(): Handle { const manager = startJetstream(handleMatchedEvent, { partitionSources: [syncPartitions] }); const cleanup = startCleanupJob(); const liveness = startLivenessJob(); const stats = startStatsRefresh(); const scheduleRunner = startScheduleRunner(); const suspendRunner = startSuspendRunner(); const dryRunExpiration = startDryRunExpirationJob(); const syncRunner = startSyncRunner(); const syncListing = startSyncRunner(() => runListingDue(), { registerWake: false }); const syncSweep = startSweepJob(); const filterCounts = startFilterCountFlush();
return { description: "worker (jetstream + cleanup + liveness + stats + schedule-runner + suspend-runner + dry-run-expiration + sync runner, listing and sweep)", async stop(timeoutMs) { cleanup.stop(); liveness.stop(); stats.stop(); dryRunExpiration.stop(); syncSweep.stop(); manager.stop(); // Drain the Jetstream handlers and any in-flight schedule tick in // parallel — they're independent. Both get the same budget; whichever // finishes first lets the other keep using its remaining time. const drainBudget = Math.floor(timeoutMs * 0.8); const [, , drainResult] = await Promise.all([ scheduleRunner.stop(drainBudget), suspendRunner.stop(drainBudget), manager.drain(drainBudget), syncRunner.stop(drainBudget), syncListing.stop(drainBudget), ]); // After the drain, so the counts its handlers noted are written too. await filterCounts.stop(); const { remaining } = drainResult; if (remaining > 0) { // Leave the cursor where it was so Jetstream replays the in-flight // events on restart (dedup via `automation_executions` PK). console.warn( `Worker drained with ${remaining} handler(s) still in flight; skipping final cursor flush`, ); return; } try { await manager.flushCursors(); } catch (err) { console.error("Final cursor flush failed:", err); } }, };}