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 { db } from "../db/index.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 function hasOAuthSession(did: string): boolean { return ( db .select({ key: oauthSessions.key }) .from(oauthSessions) .where(eq(oauthSessions.key, did)) .get() !== 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 function pauseSet(did: string, now: Date, reason: PauseReason): boolean { const paused = db .update(syncSets) .set({ pausedAt: now, updatedAt: now }) .where(and(eq(syncSets.did, did), isNull(syncSets.pausedAt))) .returning({ id: syncSets.id }) .all(); if (paused.length === 0) return false; appendActivity({ did, outcome: "paused", reason, createdAt: now }); return true;}
/** Count a 401 against the user's streak and pause at the threshold. */export function recordAuthFailure(did: string, now: Date): void { const set = getSet(did); if (!set) return; const failures = set.authFailures + 1; db.update(syncSets).set({ authFailures: failures }).where(eq(syncSets.id, set.id)).run(); if (failures >= AUTH_FAILURE_PAUSE_THRESHOLD) pauseSet(did, now, "session_rejected");}
export function resetAuthFailures(did: string): void { db.update(syncSets).set({ authFailures: 0 }).where(eq(syncSets.did, did)).run();}
/** Clear a pause and its streak. Returns whether the set was paused. */export function resumeSet(did: string, now: Date): boolean { const set = getSet(did); if (!set) return false; const wasPaused = set.pausedAt !== null; db.update(syncSets) .set({ pausedAt: null, authFailures: 0, parkedUntil: null, updatedAt: now }) .where(eq(syncSets.id, set.id)) .run(); return wasPaused;}