Something went wrong. Try again.
Automations and webhooks for the AT Protocol airglow.run
automation webhook atproto atprotocol
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465// Pausing sync when a user's OAuth session is gone (R17, KTD-10). A paused// user stays subscribed and keeps queuing; only the runner stops claiming// their lanes. The OAuth callback resumes them.
import { and, eq, isNull } from "drizzle-orm";import { conn, first } from "../db/tx.js";import { oauthSessions, syncSets } from "../db/schema.js";import { appendActivity, getSet } from "./store.js";
/** Consecutive 401s before a set pauses. Same tolerance as automation session * loss (SESSION_LOSS_THRESHOLD in lib/jetstream/session-loss.ts): one * transient token-refresh hiccup must not park sync until the user signs in. */export const AUTH_FAILURE_PAUSE_THRESHOLD = 5;
export async function hasOAuthSession(did: string): Promise<boolean> { return ( (await conn() .select({ key: oauthSessions.key }) .from(oauthSessions) .where(eq(oauthSessions.key, did)) .limit(1) .then(first)) !== undefined );}
/** Pause the user's set if it is not paused yet. Records one paused activity * row per incident; returns whether this call paused it. */export type PauseReason = "session_missing" | "session_rejected";
export async function pauseSet(did: string, now: Date, reason: PauseReason): Promise<boolean> { const paused = await conn() .update(syncSets) .set({ pausedAt: now, updatedAt: now }) .where(and(eq(syncSets.did, did), isNull(syncSets.pausedAt))) .returning({ id: syncSets.id }); if (paused.length === 0) return false; await appendActivity({ did, outcome: "paused", reason, createdAt: now }); return true;}
/** Count a 401 against the user's streak and pause at the threshold. */export async function recordAuthFailure(did: string, now: Date): Promise<void> { const set = await getSet(did); if (!set) return; const failures = set.authFailures + 1; await conn().update(syncSets).set({ authFailures: failures }).where(eq(syncSets.id, set.id)); if (failures >= AUTH_FAILURE_PAUSE_THRESHOLD) await pauseSet(did, now, "session_rejected");}
export async function resetAuthFailures(did: string): Promise<void> { await conn().update(syncSets).set({ authFailures: 0 }).where(eq(syncSets.did, did));}
/** Clear a pause and its streak. Returns whether the set was paused. */export async function resumeSet(did: string, now: Date): Promise<boolean> { const set = await getSet(did); if (!set) return false; const wasPaused = set.pausedAt !== null; await conn() .update(syncSets) .set({ pausedAt: null, authFailures: 0, parkedUntil: null, updatedAt: now }) .where(eq(syncSets.id, set.id)); return wasPaused;}