Something went wrong. Try again.
Automations and webhooks for the AT Protocol airglow.run
automation webhook atproto atprotocol
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449import { describe, it, expect, vi, beforeAll, beforeEach, afterEach } from "vitest";import { formatMultikey } from "@atproto/crypto";import { eq } from "drizzle-orm";import { Hono } from "hono";import { SignJWT, exportJWK, generateKeyPair } from "jose";
vi.mock("@/config.js", () => ({ config: { databasePath: ":memory:", publicUrl: "https://airglow.example", atmoRelayUrl: "https://relay.test", atmoRelayDid: "did:web:relay.test", },}));
vi.mock("@/db/index.js", async () => { const { createTestDb } = await import("../test/db.js"); return { db: await createTestDb() };});
import { SUBSCRIBER_CHANGED_NSID, SUBSCRIBER_CHANGED_PATH, clearRelayKeyCache, prewarmRelayKey, subscriberChangedApp,} from "./callback.js";import { markPending } from "./store.js";import { db } from "../db/index.js";import { notificationSettings, users } from "../db/schema.js";
const RELAY_DID = "did:web:relay.test";const APP_DID = "did:web:airglow.example";const DID = "did:plc:alice";
// One relay keypair (and one impostor) for the whole suite; generation is// the slow part.let relayKey: CryptoKey;let relayDidDoc: Record<string, unknown>;let impostorKey: CryptoKey;
async function multikeyFor(publicKey: CryptoKey): Promise<string> { const jwk = await exportJWK(publicKey); const uncompressed = new Uint8Array(65); uncompressed[0] = 0x04; uncompressed.set(Buffer.from(jwk.x!, "base64url"), 1); uncompressed.set(Buffer.from(jwk.y!, "base64url"), 33); return formatMultikey("ES256", uncompressed);}
beforeAll(async () => { const relay = await generateKeyPair("ES256", { extractable: true }); relayKey = relay.privateKey as CryptoKey; relayDidDoc = { id: RELAY_DID, verificationMethod: [ { id: `${RELAY_DID}#atproto`, type: "Multikey", controller: RELAY_DID, publicKeyMultibase: await multikeyFor(relay.publicKey as CryptoKey), }, ], }; const impostor = await generateKeyPair("ES256", { extractable: true }); impostorKey = impostor.privateKey as CryptoKey;});
type TokenOverrides = { iss?: string; aud?: string; lxm?: string; iat?: number; exp?: number | null; key?: CryptoKey | Uint8Array; alg?: string;};
async function relayToken(overrides: TokenOverrides = {}): Promise<string> { const now = Math.floor(Date.now() / 1000); const jwt = new SignJWT({ lxm: overrides.lxm ?? SUBSCRIBER_CHANGED_NSID }) .setProtectedHeader({ alg: overrides.alg ?? "ES256" }) .setIssuer(overrides.iss ?? RELAY_DID) .setAudience(overrides.aud ?? APP_DID) .setIssuedAt(overrides.iat ?? now); if (overrides.exp !== null) jwt.setExpirationTime(overrides.exp ?? now + 60); return jwt.sign(overrides.key ?? relayKey);}
// Mirror the lib/runtime/web.ts wiring: the POST-only sub-app mounted at the// XRPC path on a host app (whose 404 catches other verbs).const app = new Hono().route(SUBSCRIBER_CHANGED_PATH, subscriberChangedApp);
async function post(token: string | null, body: unknown): Promise<Response> { return app.request(SUBSCRIBER_CHANGED_PATH, { method: "POST", headers: { "content-type": "application/json", ...(token ? { authorization: `Bearer ${token}` } : {}), }, body: typeof body === "string" ? body : JSON.stringify(body), });}
function callbackBody(overrides: Record<string, unknown> = {}) { return { recipient: DID, enabled: true, changedAt: new Date().toISOString(), ...overrides, };}
async function loadSettings(did: string) { return db.query.notificationSettings.findFirst({ where: eq(notificationSettings.did, did), });}
const fetchMock = vi.fn();
describe("subscriberChanged callback", () => { beforeEach(async () => { await db.delete(notificationSettings); await db.delete(users); await db.insert(users).values({ did: DID, handle: "alice.test", createdAt: new Date() }); clearRelayKeyCache(); fetchMock.mockReset(); fetchMock.mockResolvedValue({ ok: true, status: 200, json: async () => relayDidDoc }); vi.stubGlobal("fetch", fetchMock); vi.spyOn(console, "debug").mockImplementation(() => {}); });
afterEach(() => { vi.unstubAllGlobals(); vi.restoreAllMocks(); });
it("applies a valid relay-signed grant and resolves the relay did.json", async () => { const changedAt = new Date().toISOString(); const res = await post(await relayToken(), callbackBody({ changedAt })); expect(res.status).toBe(200); expect(await res.json()).toEqual({ ok: true });
const row = await loadSettings(DID); expect(row?.status).toBe("granted"); expect(row?.grantedAt?.toISOString()).toBe(changedAt); expect(fetchMock).toHaveBeenCalledTimes(1); expect(fetchMock).toHaveBeenCalledWith("https://relay.test/.well-known/did.json", { headers: { Accept: "application/json" }, signal: expect.any(AbortSignal), }); });
it("lands revoked from pending: the relay's word bypasses transition guards", async () => { await markPending(DID, Date.now() - 60_000); const res = await post(await relayToken(), callbackBody({ enabled: false })); expect(res.status).toBe(200); expect((await loadSettings(DID))?.status).toBe("revoked"); });
it("caches the resolved relay key across requests", async () => { await post(await relayToken(), callbackBody()); const res = await post(await relayToken(), callbackBody()); expect(res.status).toBe(200); expect(fetchMock).toHaveBeenCalledTimes(1); });
it("prewarms the relay key so the first callback does no request-path fetch", async () => { await prewarmRelayKey(); expect(fetchMock).toHaveBeenCalledTimes(1);
const res = await post(await relayToken(), callbackBody()); expect(res.status).toBe(200); // Still one fetch total: the callback was served from the warmed cache. expect(fetchMock).toHaveBeenCalledTimes(1); });
it("coalesces a cold-cache flood into a single relay did.json fetch", async () => { // Gate the fetch so all concurrent callbacks are in flight before it // resolves, proving they share one resolution rather than racing N fetches. let release!: () => void; const gate = new Promise<void>((resolve) => { release = resolve; }); fetchMock.mockReturnValueOnce( gate.then(() => ({ ok: true, status: 200, json: async () => relayDidDoc })), );
const tokens = await Promise.all([relayToken(), relayToken(), relayToken()]); const inFlight = Promise.all(tokens.map((t) => post(t, callbackBody()))); // Let every callback reach the shared in-flight fetch, then release it. await new Promise((resolve) => setTimeout(resolve, 0)); release();
const results = await inFlight; for (const res of results) expect(res.status).toBe(200); expect(fetchMock).toHaveBeenCalledTimes(1); });
it("discards a resolution whose cache was cleared mid-flight", async () => { // Gate the first fetch so it's still in flight when we clear the cache. let release!: () => void; const gate = new Promise<void>((resolve) => { release = resolve; }); fetchMock.mockReturnValueOnce( gate.then(() => ({ ok: true, status: 200, json: async () => relayDidDoc })), );
// Start a resolution and let it reach the gated fetch, then clear the cache // mid-flight (bumps the generation) and release the stale fetch. const stale = prewarmRelayKey(); await new Promise((resolve) => setTimeout(resolve, 0)); clearRelayKeyCache(); release(); await stale;
// The stale resolution must NOT have populated the cache: a fresh callback // issues its own fetch instead of reading the discarded result. Without the // generation guard the stale .then would have cached the key and this second // fetch would never happen. const res = await post(await relayToken(), callbackBody()); expect(res.status).toBe(200); expect(fetchMock).toHaveBeenCalledTimes(2); });
it("negatively caches a failed relay resolution so a flood does not refetch", async () => { // Relay did.json is failing: every resolution returns null. fetchMock.mockReset(); fetchMock.mockResolvedValue({ ok: false, status: 503 });
// Three validly-claimed callbacks arrive while the relay is down. The first // fetches and fails; the failure is cached for FAILURE_CACHE_TTL_MS, so the // next two short-circuit instead of each issuing their own outbound fetch. const a = await post(await relayToken(), callbackBody()); const b = await post(await relayToken(), callbackBody()); const c = await post(await relayToken(), callbackBody()); expect([a.status, b.status, c.status]).toEqual([401, 401, 401]); expect(fetchMock).toHaveBeenCalledTimes(1); });
it("swallows a failed prewarm and negatively caches it", async () => { fetchMock.mockReset(); fetchMock.mockResolvedValue({ ok: false, status: 503 });
// A prewarm against an unreachable relay must not throw... await expect(prewarmRelayKey()).resolves.toBeUndefined(); expect(fetchMock).toHaveBeenCalledTimes(1);
// ...and the negatively cached failure carries to the request path: the // first callback is rejected without a second outbound fetch. const res = await post(await relayToken(), callbackBody()); expect(res.status).toBe(401); expect(fetchMock).toHaveBeenCalledTimes(1); });
it("does NOT negatively cache a reachable-but-malformed did.json (recovers next request)", async () => { // Relay is UP (200) but serving a document with no usable #atproto key. fetchMock.mockReset(); fetchMock.mockResolvedValue({ ok: true, status: 200, json: async () => ({ id: RELAY_DID, verificationMethod: [] }), });
// Each callback re-fetches: a reachable-but-bad document is not cached, so a // transient bad doc does not reject valid callbacks for a full failure // window (contrast the 503 case above, which IS cached). expect((await post(await relayToken(), callbackBody())).status).toBe(401); expect((await post(await relayToken(), callbackBody())).status).toBe(401); expect(fetchMock).toHaveBeenCalledTimes(2);
// Once the relay serves a good document, the next callback succeeds. fetchMock.mockResolvedValue({ ok: true, status: 200, json: async () => relayDidDoc }); expect((await post(await relayToken(), callbackBody())).status).toBe(200); expect(fetchMock).toHaveBeenCalledTimes(3); });
it("does NOT negatively cache a 200 with an unparseable body (recovers next request)", async () => { // Relay is UP (200) but the body is not valid JSON, so res.json() rejects // with a SyntaxError. That is reachable-but-malformed, not an outage: it must // not be negatively cached (contrast the 503 case, which IS cached). fetchMock.mockReset(); fetchMock.mockResolvedValue({ ok: true, status: 200, json: async () => { throw new SyntaxError("Unexpected token < in JSON at position 0"); }, });
// Each callback re-fetches rather than short-circuiting on a cached failure. expect((await post(await relayToken(), callbackBody())).status).toBe(401); expect((await post(await relayToken(), callbackBody())).status).toBe(401); expect(fetchMock).toHaveBeenCalledTimes(2);
// Once the relay serves a good document, the next callback succeeds. fetchMock.mockResolvedValue({ ok: true, status: 200, json: async () => relayDidDoc }); expect((await post(await relayToken(), callbackBody())).status).toBe(200); expect(fetchMock).toHaveBeenCalledTimes(3); });
it("negatively caches a 200 whose body read fails (aborted stream is an outage)", async () => { // A non-SyntaxError from res.json() (an aborted or stalled body read) is an // outage, not a malformed document: unlike the SyntaxError case above it stays // unreachable and IS negatively cached, so a flood does not refetch. Guards the // else-branch of the body-parse catch against being collapsed into the // recover-next-request path. fetchMock.mockReset(); fetchMock.mockResolvedValue({ ok: true, status: 200, json: async () => { throw new TypeError("The operation was aborted"); }, });
const a = await post(await relayToken(), callbackBody()); const b = await post(await relayToken(), callbackBody()); expect([a.status, b.status]).toEqual([401, 401]); expect(fetchMock).toHaveBeenCalledTimes(1); });
it("negatively caches a fetch-level failure (network error / timeout)", async () => { // fetch() itself rejecting (DNS failure, connection refused, AbortSignal // timeout before any response) is an outage: unreachable and negatively // cached, like a non-2xx. Guards the outer catch that this split isolated. fetchMock.mockReset(); fetchMock.mockRejectedValue(new TypeError("Failed to fetch"));
const a = await post(await relayToken(), callbackBody()); const b = await post(await relayToken(), callbackBody()); expect([a.status, b.status]).toEqual([401, 401]); expect(fetchMock).toHaveBeenCalledTimes(1); });
it("rejects a wrong issuer without resolving any DID", async () => { const res = await post(await relayToken({ iss: "did:web:evil.test" }), callbackBody()); expect(res.status).toBe(401); expect(fetchMock).not.toHaveBeenCalled(); expect(await loadSettings(DID)).toBeUndefined(); });
it("rejects wrong aud, wrong lxm, expired exp, missing exp, future iat without resolution", async () => { const now = Math.floor(Date.now() / 1000); const bad = [ await relayToken({ aud: "did:web:other.example" }), await relayToken({ lxm: "pub.atmo.notify.send" }), await relayToken({ exp: now - 10 }), await relayToken({ exp: null }), await relayToken({ iat: now + 600, exp: now + 700 }), ]; for (const token of bad) { const res = await post(token, callbackBody()); expect(res.status).toBe(401); } expect(fetchMock).not.toHaveBeenCalled(); });
it("rejects HS256-signed tokens and missing or malformed tokens", async () => { const hs256 = await relayToken({ alg: "HS256", key: new TextEncoder().encode("0123456789abcdef0123456789abcdef"), }); expect((await post(hs256, callbackBody())).status).toBe(401); expect((await post(null, callbackBody())).status).toBe(401); expect((await post("not-a-jwt", callbackBody())).status).toBe(401); expect(fetchMock).not.toHaveBeenCalled(); });
it("rejects a token signed with the wrong key even with correct claims", async () => { const res = await post(await relayToken({ key: impostorKey }), callbackBody()); expect(res.status).toBe(401); expect(fetchMock).toHaveBeenCalledTimes(1); expect(await loadSettings(DID)).toBeUndefined(); });
it("treats changedAt at or before the stored state as stale (replay safe)", async () => { const grantedAt = new Date().toISOString(); await post(await relayToken(), callbackBody({ changedAt: grantedAt }));
// Equal timestamp: ignored. const equal = await post( await relayToken(), callbackBody({ enabled: false, changedAt: grantedAt }), ); expect(equal.status).toBe(200); expect((await loadSettings(DID))?.status).toBe("granted");
// Older timestamp: ignored. const older = await post( await relayToken(), callbackBody({ enabled: false, changedAt: new Date(Date.parse(grantedAt) - 60_000).toISOString(), }), ); expect(older.status).toBe(200); expect((await loadSettings(DID))?.status).toBe("granted"); });
it("is idempotent under duplicate delivery", async () => { const body = callbackBody({ enabled: false }); expect((await post(await relayToken(), body)).status).toBe(200); expect((await post(await relayToken(), body)).status).toBe(200); const row = await loadSettings(DID); expect(row?.status).toBe("revoked"); expect(row?.revokedAt?.toISOString()).toBe(body.changedAt); });
it("rejects a far-future changedAt with 400", async () => { const res = await post( await relayToken(), callbackBody({ changedAt: new Date(Date.now() + 10 * 60_000).toISOString() }), ); expect(res.status).toBe(400); expect(await loadSettings(DID)).toBeUndefined(); });
it("answers ok for an unknown recipient without creating a row", async () => { const res = await post(await relayToken(), callbackBody({ recipient: "did:plc:stranger" })); expect(res.status).toBe(200); expect(await res.json()).toEqual({ ok: true }); expect(await db.select().from(notificationSettings)).toEqual([]); });
it("rejects malformed bodies with 400", async () => { const token = await relayToken(); expect((await post(token, "not json")).status).toBe(400); expect((await post(token, { recipient: DID })).status).toBe(400); expect((await post(token, callbackBody({ recipient: "alice.test" }))).status).toBe(400); expect((await post(token, callbackBody({ enabled: "yes" }))).status).toBe(400); expect((await post(token, callbackBody({ changedAt: "not-a-date" }))).status).toBe(400); });
it("rejects oversized bodies with 413", async () => { const res = await post(await relayToken(), callbackBody({ padding: "x".repeat(8192) })); expect(res.status).toBe(413); expect(await loadSettings(DID)).toBeUndefined(); });
it("does not handle GET (or other verbs) on the path", async () => { expect((await app.request(SUBSCRIBER_CHANGED_PATH)).status).toBe(404); expect((await app.request(SUBSCRIBER_CHANGED_PATH, { method: "PUT" })).status).toBe(404); expect(fetchMock).not.toHaveBeenCalled(); });});