Something went wrong. Try again.
Automations and webhooks for the AT Protocol airglow.run
automation webhook atproto atprotocol
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596import { describe, it, expect, vi, beforeEach } from "vitest";
vi.mock("@/db/index.js", async () => { const { createTestDb } = await import("../test/db.js"); return { db: createTestDb() };});vi.mock("../automations/pds.js", () => ({}));vi.mock("../automations/profile-status.js", () => ({ profileStatus: vi.fn() }));
import { db } from "../db/index.js";import { syncActivity, syncBackfillTasks, syncSets, users } from "../db/schema.js";import { AUTH_FAILURE_PAUSE_THRESHOLD, pauseSet, recordAuthFailure } from "./pause.js";import { resumeSyncAfterReauth } from "./reauth.js";import { getSet, saveEnabledSet } from "./store.js";import { syncPartitions } from "./subscriptions.js";import { seedUser } from "./test-helpers.js";import { registerSyncWake } from "./wake.js";
vi.mock("../config.js", () => ({ config: { databasePath: ":memory:", nsidAllowlist: [], nsidBlocklist: [], nsidRequireDids: [], syncShardCount: 1, syncJetstreamMaxLookbackUs: 1, },}));
const OWNER = "did:plc:ownerownerownerownerowne";const NOW = new Date(Date.UTC(2026, 8, 14, 12));const wake = vi.fn();
beforeEach(() => { wake.mockReset(); registerSyncWake(wake); db.delete(users).run(); seedUser(OWNER, NOW); saveEnabledSet(OWNER, ["bluesky", "tangled"], NOW);});
describe("pausing", () => { it("pauses at the five-401 threshold with one paused activity row", () => { expect(AUTH_FAILURE_PAUSE_THRESHOLD).toBe(5); for (let i = 0; i < 4; i++) recordAuthFailure(OWNER, NOW); expect(getSet(OWNER)!.pausedAt).toBeNull();
recordAuthFailure(OWNER, NOW); recordAuthFailure(OWNER, NOW);
expect(getSet(OWNER)!.pausedAt).toEqual(NOW); expect(db.select().from(syncActivity).all()).toHaveLength(1); });
it("keeps a paused user in their shard", () => { pauseSet(OWNER, NOW, "session_missing"); expect(syncPartitions()[0]!.wantedDids).toEqual([OWNER]); });});
describe("resumeSyncAfterReauth", () => { it("clears the pause and streak, wakes the runner, and queues a sweep", () => { for (let i = 0; i < 5; i++) recordAuthFailure(OWNER, NOW);
expect(resumeSyncAfterReauth(OWNER, NOW)).toBe(true);
expect(getSet(OWNER)).toMatchObject({ pausedAt: null, authFailures: 0 }); expect(wake).toHaveBeenCalled(); expect( db .select() .from(syncBackfillTasks) .all() .map((t) => t.kind), ).toEqual(["sweep", "sweep"]); });
it("clears a stale pause on a disabled set without sweeping", () => { for (let i = 0; i < 5; i++) recordAuthFailure(OWNER, NOW); db.update(syncSets).set({ enabled: false }).run();
expect(resumeSyncAfterReauth(OWNER, NOW)).toBe(true);
expect(getSet(OWNER)).toMatchObject({ pausedAt: null, authFailures: 0 }); expect(db.select().from(syncBackfillTasks).all()).toEqual([]); expect(wake).not.toHaveBeenCalled(); });
it("does nothing for an unpaused set or a user without sync", () => { expect(resumeSyncAfterReauth(OWNER, NOW)).toBe(false); expect(resumeSyncAfterReauth("did:plc:nobodynobodynobodynobody", NOW)).toBe(false); expect(db.select().from(syncBackfillTasks).all()).toEqual([]); expect(wake).not.toHaveBeenCalled(); });});