+1 -1

Dockerfile

reviewed

··· 19 19 COPY --from=deps /app/node_modules ./node_modules 20 20 COPY --from=build /app/dist ./dist 21 21 22 22 - EXPOSE 3000 22 22 + EXPOSE 5175 23 23 24 24 CMD ["sh", "-c", "bun run lib/db/migrate.ts && bun run app/server.ts"]

+3 -3

app/islands/RecordFormBuilder.tsx

reviewed

··· 64 64 if (value === undefined) continue; 65 65 switch (node.type) { 66 66 case "string": 67 67 - state[key] = typeof value === "string" ? value : String(value); 67 67 + state[key] = typeof value === "string" ? value : JSON.stringify(value); 68 68 break; 69 69 case "integer": 70 70 - state[key] = String(value); 70 70 + state[key] = typeof value === "number" ? String(value) : JSON.stringify(value); 71 71 break; 72 72 case "boolean": 73 73 - state[key] = String(value); 73 73 + state[key] = typeof value === "boolean" ? String(value) : JSON.stringify(value); 74 74 break; 75 75 case "object": 76 76 if (typeof value === "object" && value !== null) {

+289

app/routes/api/subscriptions/[rkey].test.ts

reviewed

··· 1 1 + import { describe, it, expect, vi, beforeEach } from "vitest"; 2 2 + import { Hono } from "hono"; 3 3 + 4 4 + vi.mock("@/db/index.js", async () => { 5 5 + const { createTestDb } = await import("@/test/db.js"); 6 6 + return { db: createTestDb() }; 7 7 + }); 8 8 + 9 9 + vi.mock("@/config.js", () => ({ 10 10 + config: { 11 11 + databasePath: ":memory:", 12 12 + publicUrl: "http://localhost:5175", 13 13 + pdsUrl: "", 14 14 + jetstreamUrl: "wss://test/subscribe", 15 15 + cookieSecret: "test", 16 16 + nsidAllowlist: [], 17 17 + nsidBlocklist: [], 18 18 + }, 19 19 + })); 20 20 + 21 21 + vi.mock("@/subscriptions/verify.js", () => ({ 22 22 + verifyCallback: vi.fn(), 23 23 + })); 24 24 + 25 25 + vi.mock("@/subscriptions/pds.js", () => ({ 26 26 + getRecord: vi.fn(), 27 27 + putRecord: vi.fn(), 28 28 + deleteRecord: vi.fn(), 29 29 + })); 30 30 + 31 31 + vi.mock("@/jetstream/consumer.js", () => ({ 32 32 + notifySubscriptionChange: vi.fn(), 33 33 + })); 34 34 + 35 35 + import { GET, PATCH, DELETE } from "./[rkey].js"; 36 36 + import { verifyCallback } from "@/subscriptions/verify.js"; 37 37 + import { getRecord, putRecord, deleteRecord } from "@/subscriptions/pds.js"; 38 38 + import { eq } from "drizzle-orm"; 39 39 + import { db } from "@/db/index.js"; 40 40 + import { subscriptions, deliveryLogs } from "@/db/schema.js"; 41 41 + 42 42 + const mockVerify = vi.mocked(verifyCallback); 43 43 + const mockGetRecord = vi.mocked(getRecord); 44 44 + const mockPutRecord = vi.mocked(putRecord); 45 45 + const mockDeleteRecord = vi.mocked(deleteRecord); 46 46 + 47 47 + const TEST_USER = { did: "did:plc:testuser", handle: "test.bsky.social" }; 48 48 + const TEST_SUB = { 49 49 + uri: "at://did:plc:testuser/app.rglw.subscription/rk1", 50 50 + did: "did:plc:testuser", 51 51 + rkey: "rk1", 52 52 + name: "Test Sub", 53 53 + description: null, 54 54 + lexicon: "app.bsky.feed.like", 55 55 + actions: [ 56 56 + { $type: "webhook" as const, callbackUrl: "https://example.com/hook", secret: "old-secret" }, 57 57 + ], 58 58 + fetches: [] as any[], 59 59 + conditions: [] as any[], 60 60 + active: true, 61 61 + indexedAt: new Date("2024-01-01"), 62 62 + }; 63 63 + 64 64 + function createTestApp() { 65 65 + const app = new Hono<{ Variables: { user: typeof TEST_USER } }>(); 66 66 + app.use("*", async (c, next) => { 67 67 + c.set("user", TEST_USER); 68 68 + return next(); 69 69 + }); 70 70 + app.get("/api/subscriptions/:rkey", ...GET); 71 71 + app.patch("/api/subscriptions/:rkey", ...PATCH); 72 72 + app.delete("/api/subscriptions/:rkey", ...DELETE); 73 73 + return app; 74 74 + } 75 75 + 76 76 + function patchReq(rkey: string, body: Record<string, unknown>) { 77 77 + return new Request(`http://localhost/api/subscriptions/${rkey}`, { 78 78 + method: "PATCH", 79 79 + headers: { "Content-Type": "application/json" }, 80 80 + body: JSON.stringify(body), 81 81 + }); 82 82 + } 83 83 + 84 84 + describe("GET /api/subscriptions/:rkey", () => { 85 85 + let app: ReturnType<typeof createTestApp>; 86 86 + 87 87 + beforeEach(async () => { 88 88 + app = createTestApp(); 89 89 + await db.delete(deliveryLogs); 90 90 + await db.delete(subscriptions); 91 91 + }); 92 92 + 93 93 + it("returns subscription with delivery logs", async () => { 94 94 + await db.insert(subscriptions).values(TEST_SUB); 95 95 + await db.insert(deliveryLogs).values({ 96 96 + subscriptionUri: TEST_SUB.uri, 97 97 + actionIndex: 0, 98 98 + eventTimeUs: 1700000000000000, 99 99 + payload: null, 100 100 + statusCode: 200, 101 101 + error: null, 102 102 + attempt: 1, 103 103 + createdAt: new Date(), 104 104 + }); 105 105 + 106 106 + const res = await app.request("http://localhost/api/subscriptions/rk1"); 107 107 + expect(res.status).toBe(200); 108 108 + const body = await res.json(); 109 109 + expect(body.name).toBe("Test Sub"); 110 110 + expect(body.deliveryLogs).toHaveLength(1); 111 111 + expect(body.deliveryLogs[0].statusCode).toBe(200); 112 112 + }); 113 113 + 114 114 + it("strips webhook secrets from response", async () => { 115 115 + await db.insert(subscriptions).values(TEST_SUB); 116 116 + 117 117 + const res = await app.request("http://localhost/api/subscriptions/rk1"); 118 118 + const body = await res.json(); 119 119 + const action = body.actions[0]; 120 120 + expect(action.callbackUrl).toBe("https://example.com/hook"); 121 121 + expect(action.secret).toBeUndefined(); 122 122 + }); 123 123 + 124 124 + it("returns 404 for non-existent subscription", async () => { 125 125 + const res = await app.request("http://localhost/api/subscriptions/nonexistent"); 126 126 + expect(res.status).toBe(404); 127 127 + }); 128 128 + 129 129 + it("returns 404 for another user's subscription", async () => { 130 130 + await db.insert(subscriptions).values({ ...TEST_SUB, did: "did:plc:other" }); 131 131 + 132 132 + const res = await app.request("http://localhost/api/subscriptions/rk1"); 133 133 + expect(res.status).toBe(404); 134 134 + }); 135 135 + }); 136 136 + 137 137 + describe("PATCH /api/subscriptions/:rkey", () => { 138 138 + let app: ReturnType<typeof createTestApp>; 139 139 + 140 140 + beforeEach(async () => { 141 141 + app = createTestApp(); 142 142 + mockVerify.mockReset().mockResolvedValue({ ok: true }); 143 143 + mockGetRecord.mockReset().mockResolvedValue(null); 144 144 + mockPutRecord.mockReset().mockResolvedValue(undefined); 145 145 + 146 146 + await db.delete(deliveryLogs); 147 147 + await db.delete(subscriptions); 148 148 + await db.insert(subscriptions).values(TEST_SUB); 149 149 + }); 150 150 + 151 151 + it("updates name only", async () => { 152 152 + const res = await app.request(patchReq("rk1", { name: "Updated Name" })); 153 153 + expect(res.status).toBe(200); 154 154 + 155 155 + const sub = await db.query.subscriptions.findFirst(); 156 156 + expect(sub!.name).toBe("Updated Name"); 157 157 + expect(sub!.lexicon).toBe("app.bsky.feed.like"); // unchanged 158 158 + }); 159 159 + 160 160 + it("toggles active to false", async () => { 161 161 + const res = await app.request(patchReq("rk1", { active: false })); 162 162 + expect(res.status).toBe(200); 163 163 + 164 164 + const sub = await db.query.subscriptions.findFirst(); 165 165 + expect(sub!.active).toBe(false); 166 166 + }); 167 167 + 168 168 + it("re-verifies callbacks when reactivating", async () => { 169 169 + // First deactivate 170 170 + await db 171 171 + .update(subscriptions) 172 172 + .set({ active: false }) 173 173 + .where(eq(subscriptions.uri, TEST_SUB.uri)); 174 174 + 175 175 + const res = await app.request(patchReq("rk1", { active: true })); 176 176 + expect(res.status).toBe(200); 177 177 + expect(mockVerify).toHaveBeenCalledWith("https://example.com/hook", "app.bsky.feed.like"); 178 178 + }); 179 179 + 180 180 + it("preserves webhook secret when URL unchanged", async () => { 181 181 + const res = await app.request( 182 182 + patchReq("rk1", { 183 183 + actions: [{ type: "webhook", callbackUrl: "https://example.com/hook" }], 184 184 + }), 185 185 + ); 186 186 + expect(res.status).toBe(200); 187 187 + 188 188 + const sub = await db.query.subscriptions.findFirst(); 189 189 + const action = sub!.actions[0]! as { secret: string }; 190 190 + expect(action.secret).toBe("old-secret"); 191 191 + }); 192 192 + 193 193 + it("generates new secret when URL changes", async () => { 194 194 + const res = await app.request( 195 195 + patchReq("rk1", { 196 196 + actions: [{ type: "webhook", callbackUrl: "https://new.example.com/hook" }], 197 197 + }), 198 198 + ); 199 199 + expect(res.status).toBe(200); 200 200 + 201 201 + const sub = await db.query.subscriptions.findFirst(); 202 202 + const action = sub!.actions[0]! as { secret: string }; 203 203 + expect(action.secret).not.toBe("old-secret"); 204 204 + expect(action.secret).toHaveLength(32); 205 205 + }); 206 206 + 207 207 + it("validates conditions", async () => { 208 208 + const res = await app.request( 209 209 + patchReq("rk1", { 210 210 + conditions: [{ field: "event.did", operator: "badOp", value: "x" }], 211 211 + }), 212 212 + ); 213 213 + expect(res.status).toBe(400); 214 214 + }); 215 215 + 216 216 + it("returns 404 for non-existent subscription", async () => { 217 217 + const res = await app.request(patchReq("nonexistent", { name: "New" })); 218 218 + expect(res.status).toBe(404); 219 219 + }); 220 220 + 221 221 + it("returns 502 on PDS update failure", async () => { 222 222 + mockPutRecord.mockRejectedValueOnce(new Error("PDS down")); 223 223 + 224 224 + const res = await app.request(patchReq("rk1", { name: "New" })); 225 225 + expect(res.status).toBe(502); 226 226 + }); 227 227 + 228 228 + it("returns 400 for empty actions array", async () => { 229 229 + const res = await app.request(patchReq("rk1", { actions: [] })); 230 230 + expect(res.status).toBe(400); 231 231 + }); 232 232 + }); 233 233 + 234 234 + describe("DELETE /api/subscriptions/:rkey", () => { 235 235 + let app: ReturnType<typeof createTestApp>; 236 236 + 237 237 + beforeEach(async () => { 238 238 + app = createTestApp(); 239 239 + mockDeleteRecord.mockReset().mockResolvedValue(undefined); 240 240 + 241 241 + await db.delete(deliveryLogs); 242 242 + await db.delete(subscriptions); 243 243 + await db.insert(subscriptions).values(TEST_SUB); 244 244 + }); 245 245 + 246 246 + it("deletes subscription from DB", async () => { 247 247 + const res = await app.request( 248 248 + new Request("http://localhost/api/subscriptions/rk1", { method: "DELETE" }), 249 249 + ); 250 250 + expect(res.status).toBe(200); 251 251 + 252 252 + const subs = await db.query.subscriptions.findMany(); 253 253 + expect(subs).toHaveLength(0); 254 254 + }); 255 255 + 256 256 + it("cascade-deletes delivery logs", async () => { 257 257 + await db.insert(deliveryLogs).values({ 258 258 + subscriptionUri: TEST_SUB.uri, 259 259 + actionIndex: 0, 260 260 + eventTimeUs: 1700000000000000, 261 261 + payload: null, 262 262 + statusCode: 200, 263 263 + error: null, 264 264 + attempt: 1, 265 265 + createdAt: new Date(), 266 266 + }); 267 267 + 268 268 + await app.request(new Request("http://localhost/api/subscriptions/rk1", { method: "DELETE" })); 269 269 + 270 270 + const logs = await db.query.deliveryLogs.findMany(); 271 271 + expect(logs).toHaveLength(0); 272 272 + }); 273 273 + 274 274 + it("returns 404 for non-existent subscription", async () => { 275 275 + const res = await app.request( 276 276 + new Request("http://localhost/api/subscriptions/nonexistent", { method: "DELETE" }), 277 277 + ); 278 278 + expect(res.status).toBe(404); 279 279 + }); 280 280 + 281 281 + it("returns 502 on PDS deletion failure", async () => { 282 282 + mockDeleteRecord.mockRejectedValueOnce(new Error("PDS down")); 283 283 + 284 284 + const res = await app.request( 285 285 + new Request("http://localhost/api/subscriptions/rk1", { method: "DELETE" }), 286 286 + ); 287 287 + expect(res.status).toBe(502); 288 288 + }); 289 289 + });

+5 -1

app/routes/api/subscriptions/[rkey].ts

reviewed

··· 54 54 name: sub.name, 55 55 description: sub.description, 56 56 lexicon: sub.lexicon, 57 57 - actions: sub.actions, 57 57 + actions: sub.actions.map((a) => 58 58 + a.$type === "webhook" 59 59 + ? { $type: a.$type, callbackUrl: a.callbackUrl, comment: a.comment } 60 60 + : a, 61 61 + ), 58 62 fetches: sub.fetches, 59 63 conditions: sub.conditions, 60 64 active: sub.active,

+361

app/routes/api/subscriptions/index.test.ts

reviewed

··· 1 1 + import { describe, it, expect, vi, beforeEach } from "vitest"; 2 2 + import { Hono } from "hono"; 3 3 + 4 4 + vi.mock("@/db/index.js", async () => { 5 5 + const { createTestDb } = await import("@/test/db.js"); 6 6 + return { db: createTestDb() }; 7 7 + }); 8 8 + 9 9 + vi.mock("@/config.js", () => ({ 10 10 + config: { 11 11 + databasePath: ":memory:", 12 12 + publicUrl: "http://localhost:5175", 13 13 + pdsUrl: "", 14 14 + jetstreamUrl: "wss://test/subscribe", 15 15 + cookieSecret: "test", 16 16 + nsidAllowlist: [], 17 17 + nsidBlocklist: ["blocked.nsid.*"], 18 18 + }, 19 19 + })); 20 20 + 21 21 + vi.mock("@/subscriptions/verify.js", () => ({ 22 22 + verifyCallback: vi.fn(), 23 23 + })); 24 24 + 25 25 + vi.mock("@/subscriptions/pds.js", () => ({ 26 26 + createRecord: vi.fn(), 27 27 + deleteRecord: vi.fn(), 28 28 + })); 29 29 + 30 30 + vi.mock("@/jetstream/consumer.js", () => ({ 31 31 + notifySubscriptionChange: vi.fn(), 32 32 + })); 33 33 + 34 34 + import { GET, POST } from "./index.js"; 35 35 + import { verifyCallback } from "@/subscriptions/verify.js"; 36 36 + import { createRecord, deleteRecord } from "@/subscriptions/pds.js"; 37 37 + import { db } from "@/db/index.js"; 38 38 + import { subscriptions } from "@/db/schema.js"; 39 39 + 40 40 + const mockVerify = vi.mocked(verifyCallback); 41 41 + const mockCreateRecord = vi.mocked(createRecord); 42 42 + const mockDeleteRecord = vi.mocked(deleteRecord); 43 43 + 44 44 + const TEST_USER = { did: "did:plc:testuser", handle: "test.bsky.social" }; 45 45 + 46 46 + function createTestApp() { 47 47 + const app = new Hono<{ Variables: { user: typeof TEST_USER } }>(); 48 48 + app.use("*", async (c, next) => { 49 49 + c.set("user", TEST_USER); 50 50 + return next(); 51 51 + }); 52 52 + app.get("/api/subscriptions", ...GET); 53 53 + app.post("/api/subscriptions", ...POST); 54 54 + return app; 55 55 + } 56 56 + 57 57 + function jsonReq(path: string, body: Record<string, unknown>) { 58 58 + return new Request(`http://localhost${path}`, { 59 59 + method: "POST", 60 60 + headers: { "Content-Type": "application/json" }, 61 61 + body: JSON.stringify(body), 62 62 + }); 63 63 + } 64 64 + 65 65 + describe("POST /api/subscriptions", () => { 66 66 + let app: ReturnType<typeof createTestApp>; 67 67 + 68 68 + beforeEach(async () => { 69 69 + app = createTestApp(); 70 70 + mockVerify.mockReset(); 71 71 + mockCreateRecord.mockReset(); 72 72 + mockDeleteRecord.mockReset(); 73 73 + 74 74 + await db.delete(subscriptions); 75 75 + 76 76 + mockVerify.mockResolvedValue({ ok: true }); 77 77 + mockCreateRecord.mockResolvedValue({ 78 78 + uri: "at://did:plc:testuser/app.rglw.subscription/tid1", 79 79 + rkey: "tid1", 80 80 + }); 81 81 + }); 82 82 + 83 83 + it("creates a webhook subscription and returns 201", async () => { 84 84 + const res = await app.request( 85 85 + jsonReq("/api/subscriptions", { 86 86 + name: "My Sub", 87 87 + lexicon: "app.bsky.feed.like", 88 88 + actions: [{ type: "webhook", callbackUrl: "https://example.com/hook" }], 89 89 + }), 90 90 + ); 91 91 + 92 92 + expect(res.status).toBe(201); 93 93 + const body = await res.json(); 94 94 + expect(body.uri).toContain("app.rglw.subscription"); 95 95 + expect(body.rkey).toBe("tid1"); 96 96 + expect(body.actions[0].type).toBe("webhook"); 97 97 + expect(body.actions[0].secret).toBeDefined(); 98 98 + }); 99 99 + 100 100 + it("creates a record action subscription", async () => { 101 101 + const res = await app.request( 102 102 + jsonReq("/api/subscriptions", { 103 103 + name: "Record Sub", 104 104 + lexicon: "app.bsky.feed.like", 105 105 + actions: [ 106 106 + { 107 107 + type: "record", 108 108 + targetCollection: "app.bsky.feed.post", 109 109 + recordTemplate: '{"text":"{{event.did}}"}', 110 110 + }, 111 111 + ], 112 112 + }), 113 113 + ); 114 114 + 115 115 + expect(res.status).toBe(201); 116 116 + }); 117 117 + 118 118 + it("returns 400 for missing name", async () => { 119 119 + const res = await app.request( 120 120 + jsonReq("/api/subscriptions", { 121 121 + name: "", 122 122 + lexicon: "app.bsky.feed.like", 123 123 + actions: [{ type: "webhook", callbackUrl: "https://example.com/hook" }], 124 124 + }), 125 125 + ); 126 126 + expect(res.status).toBe(400); 127 127 + }); 128 128 + 129 129 + it("returns 400 for name too long", async () => { 130 130 + const res = await app.request( 131 131 + jsonReq("/api/subscriptions", { 132 132 + name: "a".repeat(129), 133 133 + lexicon: "app.bsky.feed.like", 134 134 + actions: [{ type: "webhook", callbackUrl: "https://example.com/hook" }], 135 135 + }), 136 136 + ); 137 137 + expect(res.status).toBe(400); 138 138 + }); 139 139 + 140 140 + it("returns 400 for description too long", async () => { 141 141 + const res = await app.request( 142 142 + jsonReq("/api/subscriptions", { 143 143 + name: "Test", 144 144 + description: "x".repeat(1025), 145 145 + lexicon: "app.bsky.feed.like", 146 146 + actions: [{ type: "webhook", callbackUrl: "https://example.com/hook" }], 147 147 + }), 148 148 + ); 149 149 + expect(res.status).toBe(400); 150 150 + }); 151 151 + 152 152 + it("returns 400 for invalid lexicon NSID", async () => { 153 153 + const res = await app.request( 154 154 + jsonReq("/api/subscriptions", { 155 155 + name: "Test", 156 156 + lexicon: "not-valid", 157 157 + actions: [{ type: "webhook", callbackUrl: "https://example.com/hook" }], 158 158 + }), 159 159 + ); 160 160 + expect(res.status).toBe(400); 161 161 + }); 162 162 + 163 163 + it("returns 403 for blocked lexicon", async () => { 164 164 + const res = await app.request( 165 165 + jsonReq("/api/subscriptions", { 166 166 + name: "Test", 167 167 + lexicon: "blocked.nsid.something", 168 168 + actions: [{ type: "webhook", callbackUrl: "https://example.com/hook" }], 169 169 + }), 170 170 + ); 171 171 + expect(res.status).toBe(403); 172 172 + }); 173 173 + 174 174 + it("returns 400 for no actions", async () => { 175 175 + const res = await app.request( 176 176 + jsonReq("/api/subscriptions", { 177 177 + name: "Test", 178 178 + lexicon: "app.bsky.feed.like", 179 179 + actions: [], 180 180 + }), 181 181 + ); 182 182 + expect(res.status).toBe(400); 183 183 + }); 184 184 + 185 185 + it("returns 400 for too many actions", async () => { 186 186 + const res = await app.request( 187 187 + jsonReq("/api/subscriptions", { 188 188 + name: "Test", 189 189 + lexicon: "app.bsky.feed.like", 190 190 + actions: Array.from({ length: 11 }, () => ({ 191 191 + type: "webhook", 192 192 + callbackUrl: "https://example.com/hook", 193 193 + })), 194 194 + }), 195 195 + ); 196 196 + expect(res.status).toBe(400); 197 197 + }); 198 198 + 199 199 + it("returns 400 for invalid callback URL", async () => { 200 200 + const res = await app.request( 201 201 + jsonReq("/api/subscriptions", { 202 202 + name: "Test", 203 203 + lexicon: "app.bsky.feed.like", 204 204 + actions: [{ type: "webhook", callbackUrl: "not-a-url" }], 205 205 + }), 206 206 + ); 207 207 + expect(res.status).toBe(400); 208 208 + }); 209 209 + 210 210 + it("returns 422 when callback verification fails", async () => { 211 211 + mockVerify.mockResolvedValueOnce({ ok: false, error: "Callback failed" }); 212 212 + 213 213 + const res = await app.request( 214 214 + jsonReq("/api/subscriptions", { 215 215 + name: "Test", 216 216 + lexicon: "app.bsky.feed.like", 217 217 + actions: [{ type: "webhook", callbackUrl: "https://example.com/hook" }], 218 218 + }), 219 219 + ); 220 220 + expect(res.status).toBe(422); 221 221 + }); 222 222 + 223 223 + it("returns 400 for invalid record template", async () => { 224 224 + const res = await app.request( 225 225 + jsonReq("/api/subscriptions", { 226 226 + name: "Test", 227 227 + lexicon: "app.bsky.feed.like", 228 228 + actions: [ 229 229 + { type: "record", targetCollection: "app.bsky.feed.post", recordTemplate: "not json" }, 230 230 + ], 231 231 + }), 232 232 + ); 233 233 + expect(res.status).toBe(400); 234 234 + }); 235 235 + 236 236 + it("returns 400 for invalid condition operator", async () => { 237 237 + const res = await app.request( 238 238 + jsonReq("/api/subscriptions", { 239 239 + name: "Test", 240 240 + lexicon: "app.bsky.feed.like", 241 241 + actions: [{ type: "webhook", callbackUrl: "https://example.com/hook" }], 242 242 + conditions: [{ field: "event.did", operator: "invalid", value: "x" }], 243 243 + }), 244 244 + ); 245 245 + expect(res.status).toBe(400); 246 246 + }); 247 247 + 248 248 + it("returns 400 for too many conditions", async () => { 249 249 + const res = await app.request( 250 250 + jsonReq("/api/subscriptions", { 251 251 + name: "Test", 252 252 + lexicon: "app.bsky.feed.like", 253 253 + actions: [{ type: "webhook", callbackUrl: "https://example.com/hook" }], 254 254 + conditions: Array.from({ length: 21 }, () => ({ 255 255 + field: "event.did", 256 256 + operator: "eq", 257 257 + value: "x", 258 258 + })), 259 259 + }), 260 260 + ); 261 261 + expect(res.status).toBe(400); 262 262 + }); 263 263 + 264 264 + it("returns 400 for too many fetch steps", async () => { 265 265 + const res = await app.request( 266 266 + jsonReq("/api/subscriptions", { 267 267 + name: "Test", 268 268 + lexicon: "app.bsky.feed.like", 269 269 + actions: [{ type: "webhook", callbackUrl: "https://example.com/hook" }], 270 270 + fetches: Array.from({ length: 6 }, (_, i) => ({ 271 271 + name: `step${i}`, 272 272 + uri: `at://did/col/rk${i}`, 273 273 + })), 274 274 + }), 275 275 + ); 276 276 + expect(res.status).toBe(400); 277 277 + }); 278 278 + 279 279 + it("returns 502 on PDS write failure", async () => { 280 280 + mockCreateRecord.mockRejectedValueOnce(new Error("PDS down")); 281 281 + 282 282 + const res = await app.request( 283 283 + jsonReq("/api/subscriptions", { 284 284 + name: "Test", 285 285 + lexicon: "app.bsky.feed.like", 286 286 + actions: [{ type: "webhook", callbackUrl: "https://example.com/hook" }], 287 287 + }), 288 288 + ); 289 289 + expect(res.status).toBe(502); 290 290 + }); 291 291 + 292 292 + it("returns 400 for invalid action type", async () => { 293 293 + const res = await app.request( 294 294 + jsonReq("/api/subscriptions", { 295 295 + name: "Test", 296 296 + lexicon: "app.bsky.feed.like", 297 297 + actions: [{ type: "invalid" }], 298 298 + }), 299 299 + ); 300 300 + expect(res.status).toBe(400); 301 301 + }); 302 302 + }); 303 303 + 304 304 + describe("GET /api/subscriptions", () => { 305 305 + let app: ReturnType<typeof createTestApp>; 306 306 + 307 307 + beforeEach(async () => { 308 308 + app = createTestApp(); 309 309 + await db.delete(subscriptions); 310 310 + }); 311 311 + 312 312 + it("returns user subscriptions", async () => { 313 313 + await db.insert(subscriptions).values({ 314 314 + uri: "at://did:plc:testuser/app.rglw.subscription/rk1", 315 315 + did: "did:plc:testuser", 316 316 + rkey: "rk1", 317 317 + name: "Sub 1", 318 318 + description: null, 319 319 + lexicon: "app.bsky.feed.like", 320 320 + actions: [{ $type: "webhook", callbackUrl: "https://x.com/h", secret: "s" }], 321 321 + fetches: [], 322 322 + conditions: [], 323 323 + active: true, 324 324 + indexedAt: new Date(), 325 325 + }); 326 326 + 327 327 + const res = await app.request("http://localhost/api/subscriptions"); 328 328 + expect(res.status).toBe(200); 329 329 + const body = await res.json(); 330 330 + expect(body).toHaveLength(1); 331 331 + expect(body[0].name).toBe("Sub 1"); 332 332 + expect(body[0].rkey).toBe("rk1"); 333 333 + }); 334 334 + 335 335 + it("returns empty array when no subscriptions", async () => { 336 336 + const res = await app.request("http://localhost/api/subscriptions"); 337 337 + expect(res.status).toBe(200); 338 338 + const body = await res.json(); 339 339 + expect(body).toEqual([]); 340 340 + }); 341 341 + 342 342 + it("does not return other users' subscriptions", async () => { 343 343 + await db.insert(subscriptions).values({ 344 344 + uri: "at://did:plc:other/app.rglw.subscription/rk1", 345 345 + did: "did:plc:other", 346 346 + rkey: "rk1", 347 347 + name: "Other Sub", 348 348 + description: null, 349 349 + lexicon: "app.bsky.feed.like", 350 350 + actions: [], 351 351 + fetches: [], 352 352 + conditions: [], 353 353 + active: true, 354 354 + indexedAt: new Date(), 355 355 + }); 356 356 + 357 357 + const res = await app.request("http://localhost/api/subscriptions"); 358 358 + const body = await res.json(); 359 359 + expect(body).toEqual([]); 360 360 + }); 361 361 + });

+5 -1

app/routes/api/subscriptions/index.ts

reviewed

··· 39 39 name: r.name, 40 40 description: r.description, 41 41 lexicon: r.lexicon, 42 42 - actions: r.actions, 42 42 + actions: r.actions.map((a) => 43 43 + a.$type === "webhook" 44 44 + ? { $type: a.$type, callbackUrl: a.callbackUrl, comment: a.comment } 45 45 + : a, 46 46 + ), 43 47 fetches: r.fetches, 44 48 conditions: r.conditions, 45 49 active: r.active,

+32 -26

app/server.ts

reviewed

··· 1 1 import { createApp } from "honox/server"; 2 2 + import { createMiddleware } from "hono/factory"; 2 3 import { getOAuthClient } from "@/auth/client.js"; 3 4 import { startJetstream } from "@/jetstream/consumer.js"; 4 4 - import { dispatch } from "@/webhooks/dispatcher.js"; 5 5 - import { executeAction } from "@/actions/executor.js"; 6 6 - import { resolveFetches } from "@/actions/fetcher.js"; 5 5 + import { handleMatchedEvent } from "@/jetstream/handler.js"; 6 6 + import { rateLimit } from "@/rate-limit.js"; 7 7 + import { securityHeaders } from "@/security-headers.js"; 8 8 + import { startCleanupJob } from "@/db/cleanup.js"; 7 9 8 10 const app = createApp(); 9 11 10 10 - // Start Jetstream consumer — routes matched events to action handlers 11 11 - startJetstream(async (match) => { 12 12 - // Resolve fetch steps once for all actions 13 13 - let fetchContext: Record<string, { uri: string; cid: string; record: Record<string, unknown> }> = 14 14 - {}; 15 15 - if (match.subscription.fetches.length > 0) { 16 16 - const result = await resolveFetches( 17 17 - match.subscription.fetches, 18 18 - match.event, 19 19 - match.subscription.did, 20 20 - ); 21 21 - fetchContext = result.context; 22 22 - for (const err of result.errors) { 23 23 - console.warn(`Fetch "${err.name}" failed for ${match.subscription.uri}: ${err.error}`); 12 12 + // Security headers on all responses 13 13 + app.use("*", securityHeaders()); 14 14 + 15 15 + // Require application/json Content-Type on API mutation endpoints 16 16 + app.use( 17 17 + "/api/*", 18 18 + createMiddleware(async (c, next) => { 19 19 + if (c.req.method === "POST" || c.req.method === "PATCH" || c.req.method === "PUT") { 20 20 + const ct = c.req.header("content-type"); 21 21 + if (!ct || !ct.includes("application/json")) { 22 22 + return c.json({ error: "Content-Type must be application/json" }, 415); 23 23 + } 24 24 } 25 25 - } 25 25 + return next(); 26 26 + }), 27 27 + ); 28 28 + 29 29 + // Rate limiting on auth and subscription-creation endpoints 30 30 + const authLimiter = rateLimit({ windowMs: 60_000, max: 10 }); 31 31 + const apiLimiter = rateLimit({ windowMs: 60_000, max: 30 }); 32 32 + 33 33 + app.use("/auth/*", authLimiter); 34 34 + app.use("/api/subscriptions", apiLimiter); 26 35 27 27 - for (let i = 0; i < match.subscription.actions.length; i++) { 28 28 - const action = match.subscription.actions[i]!; 29 29 - const handler = action.$type === "record" ? executeAction : dispatch; 30 30 - handler(match, i, fetchContext).catch((err) => { 31 31 - console.error(`Action ${i} (${action.$type}) delivery error:`, err); 32 32 - }); 33 33 - } 34 34 - }); 36 36 + // Start Jetstream consumer — routes matched events to action handlers 37 37 + startJetstream(handleMatchedEvent); 38 38 + 39 39 + // Periodic cleanup: expired OAuth sessions, old delivery logs 40 40 + startCleanupJob(); 35 41 36 42 // OAuth discovery endpoints (production — loopback clients don't need these) 37 43 app.get("/oauth/client-metadata.json", async (c) => {

+4 -1

bun.lock

reviewed

··· 10 10 "@vanilla-extract/sprinkles": "^1.6.5", 11 11 "@vanilla-extract/vite-plugin": "^5.2.2", 12 12 "drizzle-orm": "^0.45.2", 13 13 - "hono": "^4.12.10", 13 13 + "hono": "^4.12.12", 14 14 "honox": "^0.1.55", 15 15 "nanoid": "^5.1.7", 16 16 "vite": "npm:@voidzero-dev/vite-plus-core@latest", 17 17 + "vitest": "^0.1.16", 17 18 }, 18 19 "devDependencies": { 19 20 "@types/better-sqlite3": "^7.6.13", ··· 690 691 "vite-node": ["vite-node@6.0.0", "", { "dependencies": { "cac": "^7.0.0", "es-module-lexer": "^2.0.0", "obug": "^2.1.1", "pathe": "^2.0.3", "vite": "^8.0.0" }, "bin": { "vite-node": "dist/cli.mjs" } }, "sha512-oj4PVrT+pDh6GYf5wfUXkcZyekYS8kKPfLPXVl8qe324Ec6l4K2DUKNadRbZ3LQl0qGcDz+PyOo7ZAh00Y+JjQ=="], 691 692 692 693 "vite-plus": ["vite-plus@0.1.16", "", { "dependencies": { "@oxc-project/types": "=0.123.0", "@voidzero-dev/vite-plus-core": "0.1.16", "@voidzero-dev/vite-plus-test": "0.1.16", "oxfmt": "=0.43.0", "oxlint": "=1.58.0", "oxlint-tsgolint": "=0.20.0" }, "optionalDependencies": { "@voidzero-dev/vite-plus-darwin-arm64": "0.1.16", "@voidzero-dev/vite-plus-darwin-x64": "0.1.16", "@voidzero-dev/vite-plus-linux-arm64-gnu": "0.1.16", "@voidzero-dev/vite-plus-linux-arm64-musl": "0.1.16", "@voidzero-dev/vite-plus-linux-x64-gnu": "0.1.16", "@voidzero-dev/vite-plus-linux-x64-musl": "0.1.16", "@voidzero-dev/vite-plus-win32-arm64-msvc": "0.1.16", "@voidzero-dev/vite-plus-win32-x64-msvc": "0.1.16" }, "bin": { "vp": "bin/vp", "oxfmt": "bin/oxfmt", "oxlint": "bin/oxlint" } }, "sha512-sgYHc5zWLSDInaHb/abvEA7UOwh7sUWuyNt+Slphj55jPvzodT8Dqw115xyKwDARTuRFSpm1eo/t58qZ8/NylQ=="], 694 694 + 695 695 + "vitest": ["@voidzero-dev/vite-plus-test@0.1.16", "", { "dependencies": { "@standard-schema/spec": "^1.1.0", "@types/chai": "^5.2.2", "@voidzero-dev/vite-plus-core": "0.1.16", "es-module-lexer": "^1.7.0", "obug": "^2.1.1", "pixelmatch": "^7.1.0", "pngjs": "^7.0.0", "sirv": "^3.0.2", "std-env": "^4.0.0", "tinybench": "^2.9.0", "tinyexec": "^1.0.2", "tinyglobby": "^0.2.15", "ws": "^8.18.3" }, "peerDependencies": { "@edge-runtime/vm": "*", "@opentelemetry/api": "^1.9.0", "@types/node": "^20.0.0 || ^22.0.0 || >=24.0.0", "@vitest/ui": "4.1.2", "happy-dom": "*", "jsdom": "*", "vite": "^6.0.0 || ^7.0.0 || ^8.0.0" }, "optionalPeers": ["@edge-runtime/vm", "@opentelemetry/api", "@types/node", "@vitest/ui", "happy-dom", "jsdom"] }, "sha512-d/rJPX/heMzoAFdnpZsp04MAa6nw1yH1tA4mVCV4m8goVcE9nAvt69mjLMzE8N/rYIQOSgenf3hDXuQRuD6OKQ=="], 693 696 694 697 "wrappy": ["wrappy@1.0.2", "", {}, "sha512-l4Sp/DRseor9wL6EvV2+TuQn63dMkPjZ/sp9XkghTEbV9KlPS1xUsZ3u7/IQO4wxtcFB4bgpQPRcR3QCvezPcQ=="], 695 698

+142

lib/actions/executor.test.ts

reviewed

··· 1 1 + import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; 2 2 + 3 3 + vi.mock("@/db/index.js", async () => { 4 4 + const { createTestDb } = await import("../test/db.js"); 5 5 + return { db: createTestDb() }; 6 6 + }); 7 7 + 8 8 + vi.mock("@/subscriptions/pds.js", () => ({ 9 9 + createArbitraryRecord: vi.fn(), 10 10 + })); 11 11 + 12 12 + import { executeAction } from "./executor.js"; 13 13 + import { createArbitraryRecord } from "../subscriptions/pds.js"; 14 14 + import { db } from "../db/index.js"; 15 15 + import { subscriptions, deliveryLogs } from "../db/schema.js"; 16 16 + import { makeMatch, makeRecordAction, makeSubscription } from "../test/fixtures.js"; 17 17 + 18 18 + const mockCreateRecord = vi.mocked(createArbitraryRecord); 19 19 + 20 20 + describe("executeAction", () => { 21 21 + beforeEach(async () => { 22 22 + vi.useFakeTimers(); 23 23 + vi.setSystemTime(new Date("2024-06-15T12:00:00.000Z")); 24 24 + mockCreateRecord.mockReset(); 25 25 + 26 26 + await db.delete(deliveryLogs); 27 27 + await db.delete(subscriptions); 28 28 + await db.insert(subscriptions).values(makeSubscription()); 29 29 + }); 30 30 + 31 31 + afterEach(() => { 32 32 + vi.useRealTimers(); 33 33 + }); 34 34 + 35 35 + it("renders template and creates record on PDS", async () => { 36 36 + mockCreateRecord.mockResolvedValueOnce({ uri: "at://x/col/rk", cid: "c" }); 37 37 + 38 38 + const action = makeRecordAction({ 39 39 + targetCollection: "app.bsky.feed.post", 40 40 + recordTemplate: '{"text":"Post by {{event.did}}","createdAt":"{{now}}"}', 41 41 + }); 42 42 + const match = makeMatch({ subscription: { actions: [action] } }); 43 43 + await executeAction(match, 0); 44 44 + 45 45 + expect(mockCreateRecord).toHaveBeenCalledWith(match.subscription.did, "app.bsky.feed.post", { 46 46 + text: "Post by did:plc:testuser123", 47 47 + createdAt: "2024-06-15T12:00:00.000Z", 48 48 + }); 49 49 + 50 50 + const logs = await db.query.deliveryLogs.findMany(); 51 51 + expect(logs).toHaveLength(1); 52 52 + expect(logs[0]!.statusCode).toBe(200); 53 53 + }); 54 54 + 55 55 + it("logs template rendering error with status 0", async () => { 56 56 + const action = makeRecordAction({ 57 57 + recordTemplate: '{"broken":{{event.commit.record.missing}}}', 58 58 + }); 59 59 + const match = makeMatch({ subscription: { actions: [action] } }); 60 60 + await executeAction(match, 0); 61 61 + 62 62 + expect(mockCreateRecord).not.toHaveBeenCalled(); 63 63 + 64 64 + const logs = await db.query.deliveryLogs.findMany(); 65 65 + expect(logs).toHaveLength(1); 66 66 + expect(logs[0]!.statusCode).toBe(0); 67 67 + expect(logs[0]!.error).toContain("Template error"); 68 68 + }); 69 69 + 70 70 + it("extracts status code from PDS error message", async () => { 71 71 + mockCreateRecord.mockRejectedValueOnce( 72 72 + new Error("PDS com.atproto.repo.createRecord failed (400): bad request"), 73 73 + ); 74 74 + 75 75 + const action = makeRecordAction(); 76 76 + const match = makeMatch({ subscription: { actions: [action] } }); 77 77 + await executeAction(match, 0); 78 78 + 79 79 + const logs = await db.query.deliveryLogs.findMany(); 80 80 + expect(logs).toHaveLength(1); 81 81 + expect(logs[0]!.statusCode).toBe(400); 82 82 + }); 83 83 + 84 84 + it("uses status 0 for generic PDS errors", async () => { 85 85 + mockCreateRecord.mockRejectedValueOnce(new Error("Network timeout")); 86 86 + 87 87 + const action = makeRecordAction(); 88 88 + const match = makeMatch({ subscription: { actions: [action] } }); 89 89 + await executeAction(match, 0); 90 90 + 91 91 + const logs = await db.query.deliveryLogs.findMany(); 92 92 + expect(logs).toHaveLength(1); 93 93 + expect(logs[0]!.statusCode).toBe(0); 94 94 + }); 95 95 + 96 96 + it("retries on 5xx PDS errors", async () => { 97 97 + mockCreateRecord 98 98 + .mockRejectedValueOnce(new Error("PDS failed (500): internal")) 99 99 + .mockResolvedValueOnce({ uri: "at://x/col/rk", cid: "c" }); 100 100 + 101 101 + const action = makeRecordAction(); 102 102 + const match = makeMatch({ subscription: { actions: [action] } }); 103 103 + await executeAction(match, 0); 104 104 + 105 105 + expect(mockCreateRecord).toHaveBeenCalledTimes(1); 106 106 + 107 107 + await vi.advanceTimersByTimeAsync(5_000); 108 108 + expect(mockCreateRecord).toHaveBeenCalledTimes(2); 109 109 + 110 110 + const logs = await db.query.deliveryLogs.findMany(); 111 111 + expect(logs).toHaveLength(2); 112 112 + }); 113 113 + 114 114 + it("does not retry on 4xx PDS errors", async () => { 115 115 + mockCreateRecord.mockRejectedValueOnce(new Error("PDS failed (400): bad request")); 116 116 + 117 117 + const action = makeRecordAction(); 118 118 + const match = makeMatch({ subscription: { actions: [action] } }); 119 119 + await executeAction(match, 0); 120 120 + 121 121 + await vi.advanceTimersByTimeAsync(60_000); 122 122 + expect(mockCreateRecord).toHaveBeenCalledTimes(1); 123 123 + }); 124 124 + 125 125 + it("passes fetchContext to template rendering", async () => { 126 126 + mockCreateRecord.mockResolvedValueOnce({ uri: "at://x/col/rk", cid: "c" }); 127 127 + 128 128 + const action = makeRecordAction({ 129 129 + recordTemplate: '{"name":"{{profile.record.displayName}}"}', 130 130 + }); 131 131 + const match = makeMatch({ subscription: { actions: [action] } }); 132 132 + const fetchContext = { 133 133 + profile: { uri: "at://x/col/rk", cid: "c", record: { displayName: "Alice" } }, 134 134 + }; 135 135 + 136 136 + await executeAction(match, 0, fetchContext); 137 137 + 138 138 + expect(mockCreateRecord).toHaveBeenCalledWith(match.subscription.did, "app.bsky.feed.post", { 139 139 + name: "Alice", 140 140 + }); 141 141 + }); 142 142 + });

+136

lib/actions/fetcher.test.ts

reviewed

··· 1 1 + import { describe, it, expect, vi, beforeEach } from "vitest"; 2 2 + import { resolveFetches } from "./fetcher.js"; 3 3 + import { makeEvent } from "../test/fixtures.js"; 4 4 + 5 5 + vi.mock("@/pds/resolver.js", () => ({ 6 6 + fetchRecord: vi.fn(), 7 7 + })); 8 8 + 9 9 + import { fetchRecord } from "../pds/resolver.js"; 10 10 + const mockFetchRecord = vi.mocked(fetchRecord); 11 11 + 12 12 + describe("resolveFetches", () => { 13 13 + const ownerDid = "did:plc:owner"; 14 14 + const event = makeEvent({ did: "did:plc:eventuser" }); 15 15 + 16 16 + beforeEach(() => { 17 17 + mockFetchRecord.mockReset(); 18 18 + }); 19 19 + 20 20 + it("resolves a single fetch step", async () => { 21 21 + mockFetchRecord.mockResolvedValueOnce({ 22 22 + uri: "at://did:plc:eventuser/app.bsky.actor.profile/self", 23 23 + cid: "bafycid", 24 24 + record: { displayName: "Alice" }, 25 25 + }); 26 26 + 27 27 + const result = await resolveFetches( 28 28 + [{ name: "profile", uri: "at://{{event.did}}/app.bsky.actor.profile/self" }], 29 29 + event, 30 30 + ownerDid, 31 31 + ); 32 32 + 33 33 + expect(result.context.profile).toEqual({ 34 34 + uri: "at://did:plc:eventuser/app.bsky.actor.profile/self", 35 35 + cid: "bafycid", 36 36 + record: { displayName: "Alice" }, 37 37 + }); 38 38 + expect(result.errors).toEqual([]); 39 39 + expect(mockFetchRecord).toHaveBeenCalledWith( 40 40 + "at://did:plc:eventuser/app.bsky.actor.profile/self", 41 41 + ); 42 42 + }); 43 43 + 44 44 + it("resolves multiple fetch steps in parallel", async () => { 45 45 + mockFetchRecord 46 46 + .mockResolvedValueOnce({ 47 47 + uri: "at://did1/col/rk1", 48 48 + cid: "c1", 49 49 + record: { a: 1 }, 50 50 + }) 51 51 + .mockResolvedValueOnce({ 52 52 + uri: "at://did2/col/rk2", 53 53 + cid: "c2", 54 54 + record: { b: 2 }, 55 55 + }); 56 56 + 57 57 + const result = await resolveFetches( 58 58 + [ 59 59 + { name: "step1", uri: "at://did1/col/rk1" }, 60 60 + { name: "step2", uri: "at://did2/col/rk2" }, 61 61 + ], 62 62 + event, 63 63 + ownerDid, 64 64 + ); 65 65 + 66 66 + expect(Object.keys(result.context)).toHaveLength(2); 67 67 + expect(result.context.step1?.record).toEqual({ a: 1 }); 68 68 + expect(result.context.step2?.record).toEqual({ b: 2 }); 69 69 + expect(result.errors).toEqual([]); 70 70 + }); 71 71 + 72 72 + it("resolves {{self}} in URI template", async () => { 73 73 + mockFetchRecord.mockResolvedValueOnce({ 74 74 + uri: "at://did:plc:owner/col/rk", 75 75 + cid: "c", 76 76 + record: {}, 77 77 + }); 78 78 + 79 79 + await resolveFetches([{ name: "myRecord", uri: "at://{{self}}/col/rk" }], event, ownerDid); 80 80 + 81 81 + expect(mockFetchRecord).toHaveBeenCalledWith("at://did:plc:owner/col/rk"); 82 82 + }); 83 83 + 84 84 + it("collects errors for non-AT URIs", async () => { 85 85 + const noCommitEvent = makeEvent({ commit: undefined, kind: "identity" }); 86 86 + 87 87 + const result = await resolveFetches( 88 88 + [{ name: "bad", uri: "{{event.commit.rkey}}" }], 89 89 + noCommitEvent, 90 90 + ownerDid, 91 91 + ); 92 92 + 93 93 + expect(result.context).toEqual({}); 94 94 + expect(result.errors).toHaveLength(1); 95 95 + expect(result.errors[0]!.name).toBe("bad"); 96 96 + expect(result.errors[0]!.error).toContain("not a valid AT URI"); 97 97 + }); 98 98 + 99 99 + it("rejects AT URIs with missing segments (e.g. empty DID)", async () => { 100 100 + // at:///col/rk has an empty DID segment — previously slipped through startsWith("at://") 101 101 + const noCommitEvent = makeEvent({ commit: undefined, kind: "identity" }); 102 102 + 103 103 + const result = await resolveFetches( 104 104 + [{ name: "bad", uri: "at://{{event.commit.rkey}}/col/rk" }], 105 105 + noCommitEvent, 106 106 + ownerDid, 107 107 + ); 108 108 + 109 109 + expect(result.context).toEqual({}); 110 110 + expect(result.errors).toHaveLength(1); 111 111 + expect(result.errors[0]!.error).toContain("not a valid AT URI"); 112 112 + }); 113 113 + 114 114 + it("catches fetchRecord errors per step without blocking others", async () => { 115 115 + mockFetchRecord.mockRejectedValueOnce(new Error("PDS unreachable")).mockResolvedValueOnce({ 116 116 + uri: "at://ok/col/rk", 117 117 + cid: "c", 118 118 + record: { ok: true }, 119 119 + }); 120 120 + 121 121 + const result = await resolveFetches( 122 122 + [ 123 123 + { name: "failing", uri: "at://did1/col/rk1" }, 124 124 + { name: "succeeding", uri: "at://did2/col/rk2" }, 125 125 + ], 126 126 + event, 127 127 + ownerDid, 128 128 + ); 129 129 + 130 130 + expect(result.context.succeeding).toBeDefined(); 131 131 + expect(result.context.failing).toBeUndefined(); 132 132 + expect(result.errors).toHaveLength(1); 133 133 + expect(result.errors[0]!.name).toBe("failing"); 134 134 + expect(result.errors[0]!.error).toContain("PDS unreachable"); 135 135 + }); 136 136 + });

+4 -1

lib/actions/fetcher.ts

reviewed

··· 3 3 import type { JetstreamEvent } from "../jetstream/matcher.js"; 4 4 import { resolveEventPlaceholder, PLACEHOLDER_RE, type FetchContext } from "./template.js"; 5 5 6 6 + // AT URI format: at://did/collection/rkey 7 7 + const AT_URI_RE = /^at:\/\/[^/]+\/[^/]+\/[^/]+$/; 8 8 + 6 9 /** Resolve a fetch URI template against event data. Returns empty string on non-string values. */ 7 10 function resolveUri(uriTemplate: string, event: JetstreamEvent, ownerDid: string): string { 8 11 return uriTemplate.replace(PLACEHOLDER_RE, (_, path: string) => { ··· 26 29 steps.map(async (step) => { 27 30 try { 28 31 const resolvedUri = resolveUri(step.uri, event, ownerDid); 29 29 - if (!resolvedUri || !resolvedUri.startsWith("at://")) { 32 32 + if (!resolvedUri || !AT_URI_RE.test(resolvedUri)) { 30 33 errors.push({ 31 34 name: step.name, 32 35 error: `Resolved URI is not a valid AT URI: ${resolvedUri}`,

+247

lib/actions/template.test.ts

reviewed

··· 1 1 + import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; 2 2 + import { 3 3 + validateTemplate, 4 4 + validateFetchStep, 5 5 + validateFetchUri, 6 6 + renderTemplate, 7 7 + resolveEventPlaceholder, 8 8 + } from "./template.js"; 9 9 + import { makeEvent } from "../test/fixtures.js"; 10 10 + 11 11 + describe("validateTemplate", () => { 12 12 + it("returns valid for a JSON object with placeholders", () => { 13 13 + const result = validateTemplate('{"text":"{{event.did}}"}'); 14 14 + expect(result).toEqual({ valid: true, placeholders: ["event.did"] }); 15 15 + }); 16 16 + 17 17 + it("accepts multiple placeholder types", () => { 18 18 + const result = validateTemplate('{"a":"{{now}}","b":"{{self}}","c":"{{event.commit.rkey}}"}'); 19 19 + expect(result).toEqual({ 20 20 + valid: true, 21 21 + placeholders: ["now", "self", "event.commit.rkey"], 22 22 + }); 23 23 + }); 24 24 + 25 25 + it("accepts fetch name placeholders when fetchNames provided", () => { 26 26 + const result = validateTemplate('{"val":"{{profile.record.name}}"}', ["profile"]); 27 27 + expect(result).toEqual({ valid: true, placeholders: ["profile.record.name"] }); 28 28 + }); 29 29 + 30 30 + it("rejects unknown placeholder roots", () => { 31 31 + const result = validateTemplate('{"val":"{{unknown.field}}"}'); 32 32 + expect(result).toEqual({ valid: false, error: "Invalid placeholder: {{unknown.field}}" }); 33 33 + }); 34 34 + 35 35 + it("rejects non-JSON template", () => { 36 36 + const result = validateTemplate("not json {{event.did}}"); 37 37 + expect(result).toEqual({ valid: false, error: "Template is not valid JSON" }); 38 38 + }); 39 39 + 40 40 + it("rejects non-object JSON (array)", () => { 41 41 + const result = validateTemplate('[{"x":"{{event.did}}"}]'); 42 42 + expect(result).toEqual({ valid: false, error: "Template must be a JSON object" }); 43 43 + }); 44 44 + 45 45 + it("rejects template without placeholders", () => { 46 46 + const result = validateTemplate('{"text":"static"}'); 47 47 + expect(result).toEqual({ 48 48 + valid: false, 49 49 + error: "Template must contain at least one {{placeholder}}", 50 50 + }); 51 51 + }); 52 52 + 53 53 + it("handles bare placeholder as JSON value", () => { 54 54 + const result = validateTemplate('{"data":{{event.commit.record}}}'); 55 55 + expect(result).toEqual({ valid: true, placeholders: ["event.commit.record"] }); 56 56 + }); 57 57 + }); 58 58 + 59 59 + describe("validateFetchStep", () => { 60 60 + it("returns valid for a correct name and URI", () => { 61 61 + const result = validateFetchStep( 62 62 + "profile", 63 63 + "at://{{event.did}}/app.bsky.actor.profile/self", 64 64 + new Set(), 65 65 + ); 66 66 + expect(result).toEqual({ valid: true }); 67 67 + }); 68 68 + 69 69 + it("rejects empty name", () => { 70 70 + const result = validateFetchStep("", "at://x/y/z", new Set()); 71 71 + expect(result.valid).toBe(false); 72 72 + }); 73 73 + 74 74 + it("rejects name starting with digit", () => { 75 75 + const result = validateFetchStep("1bad", "at://x/y/z", new Set()); 76 76 + expect(result.valid).toBe(false); 77 77 + }); 78 78 + 79 79 + it("rejects reserved names", () => { 80 80 + for (const name of ["event", "now", "self"]) { 81 81 + const result = validateFetchStep(name, "at://x/y/z", new Set()); 82 82 + expect(result.valid).toBe(false); 83 83 + expect((result as any).error).toContain("reserved"); 84 84 + } 85 85 + }); 86 86 + 87 87 + it("rejects duplicate names", () => { 88 88 + const result = validateFetchStep("profile", "at://x/y/z", new Set(["profile"])); 89 89 + expect(result.valid).toBe(false); 90 90 + expect((result as any).error).toContain("Duplicate"); 91 91 + }); 92 92 + 93 93 + it("propagates URI validation errors", () => { 94 94 + const result = validateFetchStep("myStep", "", new Set()); 95 95 + expect(result.valid).toBe(false); 96 96 + }); 97 97 + }); 98 98 + 99 99 + describe("validateFetchUri", () => { 100 100 + it("accepts literal AT URI", () => { 101 101 + const result = validateFetchUri("at://did:plc:abc/app.bsky.feed.post/xyz"); 102 102 + expect(result).toEqual({ valid: true, placeholders: [] }); 103 103 + }); 104 104 + 105 105 + it("accepts URI with event placeholders", () => { 106 106 + const result = validateFetchUri("at://{{event.did}}/col/{{event.commit.rkey}}"); 107 107 + expect(result).toEqual({ 108 108 + valid: true, 109 109 + placeholders: ["event.did", "event.commit.rkey"], 110 110 + }); 111 111 + }); 112 112 + 113 113 + it("accepts {{self}} and {{now}} placeholders", () => { 114 114 + const result = validateFetchUri("at://{{self}}/col/rkey"); 115 115 + expect(result).toEqual({ valid: true, placeholders: ["self"] }); 116 116 + }); 117 117 + 118 118 + it("rejects empty URI", () => { 119 119 + const result = validateFetchUri(""); 120 120 + expect(result.valid).toBe(false); 121 121 + }); 122 122 + 123 123 + it("rejects non-AT URI without placeholders", () => { 124 124 + const result = validateFetchUri("https://example.com"); 125 125 + expect(result.valid).toBe(false); 126 126 + }); 127 127 + 128 128 + it("rejects fetch context placeholders in URI", () => { 129 129 + const result = validateFetchUri("at://{{profile.record.did}}/col/rkey"); 130 130 + expect(result.valid).toBe(false); 131 131 + expect((result as any).error).toContain("Invalid placeholder"); 132 132 + }); 133 133 + }); 134 134 + 135 135 + describe("renderTemplate", () => { 136 136 + beforeEach(() => { 137 137 + vi.useFakeTimers(); 138 138 + vi.setSystemTime(new Date("2024-06-15T12:00:00.000Z")); 139 139 + }); 140 140 + 141 141 + afterEach(() => { 142 142 + vi.useRealTimers(); 143 143 + }); 144 144 + 145 145 + const event = makeEvent({ 146 146 + did: "did:plc:user1", 147 147 + commit: { 148 148 + rev: "r1", 149 149 + operation: "create", 150 150 + collection: "app.bsky.feed.post", 151 151 + rkey: "rk1", 152 152 + record: { text: "hello", count: 5 }, 153 153 + cid: "bafycid", 154 154 + }, 155 155 + }); 156 156 + 157 157 + it("renders event.did", () => { 158 158 + const result = renderTemplate('{"author":"{{event.did}}"}', event); 159 159 + expect(result).toEqual({ author: "did:plc:user1" }); 160 160 + }); 161 161 + 162 162 + it("renders event.commit fields", () => { 163 163 + const result = renderTemplate( 164 164 + '{"op":"{{event.commit.operation}}","rkey":"{{event.commit.rkey}}"}', 165 165 + event, 166 166 + ); 167 167 + expect(result).toEqual({ op: "create", rkey: "rk1" }); 168 168 + }); 169 169 + 170 170 + it("renders record fields", () => { 171 171 + const result = renderTemplate('{"msg":"{{event.commit.record.text}}"}', event); 172 172 + expect(result).toEqual({ msg: "hello" }); 173 173 + }); 174 174 + 175 175 + it("renders {{now}} as ISO datetime", () => { 176 176 + const result = renderTemplate('{"ts":"{{now}}"}', event); 177 177 + expect(result).toEqual({ ts: "2024-06-15T12:00:00.000Z" }); 178 178 + }); 179 179 + 180 180 + it("renders {{self}} as ownerDid", () => { 181 181 + const result = renderTemplate('{"owner":"{{self}}"}', event, undefined, "did:plc:owner"); 182 182 + expect(result).toEqual({ owner: "did:plc:owner" }); 183 183 + }); 184 184 + 185 185 + it("renders fetch context placeholders", () => { 186 186 + const fetchContext = { 187 187 + profile: { 188 188 + uri: "at://did/col/rkey", 189 189 + cid: "bafycid", 190 190 + record: { displayName: "Alice" }, 191 191 + }, 192 192 + }; 193 193 + const result = renderTemplate('{"name":"{{profile.record.displayName}}"}', event, fetchContext); 194 194 + expect(result).toEqual({ name: "Alice" }); 195 195 + }); 196 196 + 197 197 + it("resolves undefined placeholders to empty string", () => { 198 198 + const result = renderTemplate('{"val":"{{event.commit.record.missing}}"}', event); 199 199 + expect(result).toEqual({ val: "" }); 200 200 + }); 201 201 + 202 202 + it("JSON-stringifies objects in placeholders", () => { 203 203 + const fetchContext = { 204 204 + data: { 205 205 + uri: "at://x", 206 206 + cid: "c", 207 207 + record: { nested: { a: 1 } }, 208 208 + }, 209 209 + }; 210 210 + const result = renderTemplate('{"obj":{{data.record.nested}}}', event, fetchContext); 211 211 + expect(result).toEqual({ obj: { a: 1 } }); 212 212 + }); 213 213 + 214 214 + it("throws on invalid rendered JSON", () => { 215 215 + // A template that might produce invalid JSON after rendering 216 216 + expect(() => { 217 217 + renderTemplate('{"key":{{event.commit.record.missing}}}', event); 218 218 + }).toThrow(); 219 219 + }); 220 220 + }); 221 221 + 222 222 + describe("resolveEventPlaceholder", () => { 223 223 + const event = makeEvent({ did: "did:plc:test" }); 224 224 + 225 225 + it("resolves 'now' to ISO string", () => { 226 226 + vi.useFakeTimers(); 227 227 + vi.setSystemTime(new Date("2024-01-01T00:00:00.000Z")); 228 228 + expect(resolveEventPlaceholder("now", event)).toBe("2024-01-01T00:00:00.000Z"); 229 229 + vi.useRealTimers(); 230 230 + }); 231 231 + 232 232 + it("resolves 'self' to ownerDid", () => { 233 233 + expect(resolveEventPlaceholder("self", event, "did:plc:me")).toBe("did:plc:me"); 234 234 + }); 235 235 + 236 236 + it("resolves event.did", () => { 237 237 + expect(resolveEventPlaceholder("event.did", event)).toBe("did:plc:test"); 238 238 + }); 239 239 + 240 240 + it("returns undefined for unknown paths", () => { 241 241 + expect(resolveEventPlaceholder("unknown.path", event)).toBeUndefined(); 242 242 + }); 243 243 + 244 244 + it("returns undefined for fetch context paths (no context)", () => { 245 245 + expect(resolveEventPlaceholder("profile.record.name", event)).toBeUndefined(); 246 246 + }); 247 247 + });

+2 -2

lib/auth/client.ts

reviewed

··· 30 30 } 31 31 32 32 const key = await JoseKey.generate(["ES256"]); 33 33 - mkdirSync(dirname(KEY_PATH), { recursive: true }); 34 34 - writeFileSync(KEY_PATH, JSON.stringify(key.privateJwk)); 33 33 + mkdirSync(dirname(KEY_PATH), { recursive: true, mode: 0o700 }); 34 34 + writeFileSync(KEY_PATH, JSON.stringify(key.privateJwk), { mode: 0o600 }); 35 35 return key; 36 36 } 37 37

+8 -1

lib/config.ts

reviewed

··· 13 13 return process.env[key] ?? fallback; 14 14 } 15 15 16 16 + const cookieSecret = env("COOKIE_SECRET", ""); 17 17 + if (!cookieSecret || cookieSecret.length < 32) { 18 18 + throw new Error( 19 19 + "COOKIE_SECRET must be at least 32 characters. Generate one with: openssl rand -base64 32", 20 20 + ); 21 21 + } 22 22 + 16 23 export const config = { 17 24 port: Number(env("PORT", "5175")), 18 25 databasePath: env("DATABASE_PATH", "./data/airglow.db"), 19 26 publicUrl: env("PUBLIC_URL", "http://127.0.0.1:5175"), 20 27 pdsUrl: process.env.PDS_URL?.replace(/\/$/, "") || "", 21 28 jetstreamUrl: env("JETSTREAM_URL", "wss://jetstream2.us-east.bsky.network/subscribe"), 22 22 - cookieSecret: env("COOKIE_SECRET", ""), 29 29 + cookieSecret, 23 30 nsidAllowlist: env("NSID_ALLOWLIST", "").split(",").filter(Boolean), 24 31 nsidBlocklist: env("NSID_BLOCKLIST", "").split(",").filter(Boolean), 25 32 } as const;

+26

lib/db/cleanup.ts

reviewed

··· 1 1 + import { lt } from "drizzle-orm"; 2 2 + import { db } from "./index.js"; 3 3 + import { deliveryLogs, oauthSessions, oauthStates } from "./schema.js"; 4 4 + 5 5 + const THIRTY_DAYS_MS = 30 * 24 * 60 * 60 * 1000; 6 6 + const CLEANUP_INTERVAL_MS = 60 * 60 * 1000; // 1 hour 7 7 + 8 8 + function runCleanup() { 9 9 + const cutoff = new Date(Date.now() - THIRTY_DAYS_MS); 10 10 + 11 11 + // Delete delivery logs older than 30 days 12 12 + db.delete(deliveryLogs).where(lt(deliveryLogs.createdAt, cutoff)).run(); 13 13 + 14 14 + // Delete expired OAuth sessions and states 15 15 + const now = new Date(); 16 16 + db.delete(oauthSessions).where(lt(oauthSessions.expiresAt, now)).run(); 17 17 + db.delete(oauthStates).where(lt(oauthStates.expiresAt, now)).run(); 18 18 + } 19 19 + 20 20 + export function startCleanupJob() { 21 21 + // Run once on startup 22 22 + runCleanup(); 23 23 + 24 24 + const interval = setInterval(runCleanup, CLEANUP_INTERVAL_MS); 25 25 + if (typeof interval === "object" && "unref" in interval) interval.unref(); 26 26 + }

+1 -1

lib/db/index.ts

reviewed

··· 6 6 import { mkdirSync } from "node:fs"; 7 7 import { dirname } from "node:path"; 8 8 9 9 - mkdirSync(dirname(config.databasePath), { recursive: true }); 9 9 + mkdirSync(dirname(config.databasePath), { recursive: true, mode: 0o700 }); 10 10 11 11 const sqlite = new Database(config.databasePath); 12 12 sqlite.exec("PRAGMA journal_mode = WAL;");

+247

lib/jetstream/consumer.test.ts

reviewed

··· 1 1 + import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; 2 2 + 3 3 + vi.mock("@/config.js", () => ({ 4 4 + config: { 5 5 + databasePath: ":memory:", 6 6 + jetstreamUrl: "wss://jetstream.test/subscribe", 7 7 + }, 8 8 + })); 9 9 + 10 10 + vi.mock("@/db/index.js", async () => { 11 11 + const { createTestDb } = await import("../test/db.js"); 12 12 + return { db: createTestDb() }; 13 13 + }); 14 14 + 15 15 + import { JetstreamConsumer, type MatchedEvent } from "./consumer.js"; 16 16 + import { db } from "../db/index.js"; 17 17 + import { subscriptions } from "../db/schema.js"; 18 18 + import { makeSubscription, makeEvent } from "../test/fixtures.js"; 19 19 + 20 20 + // Mock WebSocket to prevent real connections 21 21 + class MockWebSocket { 22 22 + static OPEN = 1; 23 23 + readyState = 1; 24 24 + close = vi.fn(); 25 25 + addEventListener = vi.fn(); 26 26 + } 27 27 + vi.stubGlobal("WebSocket", MockWebSocket); 28 28 + 29 29 + describe("JetstreamConsumer", () => { 30 30 + let handler: ReturnType<typeof vi.fn<(match: MatchedEvent) => void>>; 31 31 + let consumer: JetstreamConsumer; 32 32 + 33 33 + beforeEach(async () => { 34 34 + vi.useFakeTimers(); 35 35 + handler = vi.fn(); 36 36 + consumer = new JetstreamConsumer(handler); 37 37 + 38 38 + await db.delete(subscriptions); 39 39 + }); 40 40 + 41 41 + afterEach(() => { 42 42 + consumer.stop(); 43 43 + vi.useRealTimers(); 44 44 + }); 45 45 + 46 46 + describe("refreshSubscriptions", () => { 47 47 + it("loads active subscriptions grouped by collection", async () => { 48 48 + await db.insert(subscriptions).values([ 49 49 + makeSubscription({ 50 50 + uri: "at://u/s/1", 51 51 + rkey: "1", 52 52 + lexicon: "app.bsky.feed.like", 53 53 + active: true, 54 54 + }), 55 55 + makeSubscription({ 56 56 + uri: "at://u/s/2", 57 57 + rkey: "2", 58 58 + lexicon: "app.bsky.feed.like", 59 59 + active: true, 60 60 + }), 61 61 + makeSubscription({ 62 62 + uri: "at://u/s/3", 63 63 + rkey: "3", 64 64 + lexicon: "app.bsky.feed.post", 65 65 + active: true, 66 66 + }), 67 67 + ]); 68 68 + 69 69 + await consumer.refreshSubscriptions(); 70 70 + 71 71 + // The consumer is internal, but we can test it indirectly via processEvent 72 72 + const event = makeEvent({ 73 73 + commit: { 74 74 + rev: "r", 75 75 + operation: "create", 76 76 + collection: "app.bsky.feed.like", 77 77 + rkey: "rk", 78 78 + record: {}, 79 79 + }, 80 80 + }); 81 81 + // Access processEvent via prototype 82 82 + (consumer as any).processEvent(event); 83 83 + 84 84 + // Both like subscriptions should match (empty conditions = match all) 85 85 + expect(handler).toHaveBeenCalledTimes(2); 86 86 + }); 87 87 + 88 88 + it("does not load inactive subscriptions", async () => { 89 89 + await db 90 90 + .insert(subscriptions) 91 91 + .values(makeSubscription({ uri: "at://u/s/1", rkey: "1", active: false })); 92 92 + 93 93 + await consumer.refreshSubscriptions(); 94 94 + 95 95 + const event = makeEvent(); 96 96 + (consumer as any).processEvent(event); 97 97 + 98 98 + expect(handler).not.toHaveBeenCalled(); 99 99 + }); 100 100 + }); 101 101 + 102 102 + describe("processEvent", () => { 103 103 + it("calls handler for matching subscriptions", async () => { 104 104 + await db.insert(subscriptions).values( 105 105 + makeSubscription({ 106 106 + uri: "at://u/s/1", 107 107 + rkey: "1", 108 108 + lexicon: "app.bsky.feed.like", 109 109 + conditions: [], 110 110 + }), 111 111 + ); 112 112 + await consumer.refreshSubscriptions(); 113 113 + 114 114 + const event = makeEvent({ 115 115 + commit: { 116 116 + rev: "r", 117 117 + operation: "create", 118 118 + collection: "app.bsky.feed.like", 119 119 + rkey: "rk", 120 120 + record: {}, 121 121 + }, 122 122 + }); 123 123 + (consumer as any).processEvent(event); 124 124 + 125 125 + expect(handler).toHaveBeenCalledOnce(); 126 126 + const match = handler.mock.calls[0]![0]; 127 127 + expect(match.event).toBe(event); 128 128 + expect(match.subscription.rkey).toBe("1"); 129 129 + }); 130 130 + 131 131 + it("does not call handler for non-matching collection", async () => { 132 132 + await db.insert(subscriptions).values( 133 133 + makeSubscription({ 134 134 + uri: "at://u/s/1", 135 135 + rkey: "1", 136 136 + lexicon: "app.bsky.feed.post", 137 137 + }), 138 138 + ); 139 139 + await consumer.refreshSubscriptions(); 140 140 + 141 141 + const event = makeEvent({ 142 142 + commit: { 143 143 + rev: "r", 144 144 + operation: "create", 145 145 + collection: "app.bsky.feed.like", 146 146 + rkey: "rk", 147 147 + record: {}, 148 148 + }, 149 149 + }); 150 150 + (consumer as any).processEvent(event); 151 151 + 152 152 + expect(handler).not.toHaveBeenCalled(); 153 153 + }); 154 154 + 155 155 + it("does not call handler when conditions do not match", async () => { 156 156 + await db.insert(subscriptions).values( 157 157 + makeSubscription({ 158 158 + uri: "at://u/s/1", 159 159 + rkey: "1", 160 160 + lexicon: "app.bsky.feed.like", 161 161 + conditions: [{ field: "event.did", operator: "eq", value: "did:plc:specific" }], 162 162 + }), 163 163 + ); 164 164 + await consumer.refreshSubscriptions(); 165 165 + 166 166 + const event = makeEvent({ 167 167 + did: "did:plc:other", 168 168 + commit: { 169 169 + rev: "r", 170 170 + operation: "create", 171 171 + collection: "app.bsky.feed.like", 172 172 + rkey: "rk", 173 173 + record: {}, 174 174 + }, 175 175 + }); 176 176 + (consumer as any).processEvent(event); 177 177 + 178 178 + expect(handler).not.toHaveBeenCalled(); 179 179 + }); 180 180 + 181 181 + it("calls handler for multiple matching subscriptions on same collection", async () => { 182 182 + await db 183 183 + .insert(subscriptions) 184 184 + .values([ 185 185 + makeSubscription({ uri: "at://u/s/1", rkey: "1", lexicon: "app.bsky.feed.like" }), 186 186 + makeSubscription({ uri: "at://u/s/2", rkey: "2", lexicon: "app.bsky.feed.like" }), 187 187 + ]); 188 188 + await consumer.refreshSubscriptions(); 189 189 + 190 190 + const event = makeEvent({ 191 191 + commit: { 192 192 + rev: "r", 193 193 + operation: "create", 194 194 + collection: "app.bsky.feed.like", 195 195 + rkey: "rk", 196 196 + record: {}, 197 197 + }, 198 198 + }); 199 199 + (consumer as any).processEvent(event); 200 200 + 201 201 + expect(handler).toHaveBeenCalledTimes(2); 202 202 + }); 203 203 + 204 204 + it("evaluates {{self}} condition against subscription owner DID", async () => { 205 205 + await db.insert(subscriptions).values( 206 206 + makeSubscription({ 207 207 + uri: "at://u/s/1", 208 208 + rkey: "1", 209 209 + did: "did:plc:owner", 210 210 + lexicon: "app.bsky.feed.like", 211 211 + conditions: [{ field: "event.did", operator: "eq", value: "{{self}}" }], 212 212 + }), 213 213 + ); 214 214 + await consumer.refreshSubscriptions(); 215 215 + 216 216 + // Event from the owner 217 217 + const ownerEvent = makeEvent({ 218 218 + did: "did:plc:owner", 219 219 + commit: { 220 220 + rev: "r", 221 221 + operation: "create", 222 222 + collection: "app.bsky.feed.like", 223 223 + rkey: "rk", 224 224 + record: {}, 225 225 + }, 226 226 + }); 227 227 + (consumer as any).processEvent(ownerEvent); 228 228 + expect(handler).toHaveBeenCalledOnce(); 229 229 + 230 230 + handler.mockClear(); 231 231 + 232 232 + // Event from someone else 233 233 + const otherEvent = makeEvent({ 234 234 + did: "did:plc:someone-else", 235 235 + commit: { 236 236 + rev: "r", 237 237 + operation: "create", 238 238 + collection: "app.bsky.feed.like", 239 239 + rkey: "rk2", 240 240 + record: {}, 241 241 + }, 242 242 + }); 243 243 + (consumer as any).processEvent(otherEvent); 244 244 + expect(handler).not.toHaveBeenCalled(); 245 245 + }); 246 246 + }); 247 247 + });

+148

lib/jetstream/handler.test.ts

reviewed

··· 1 1 + import { describe, it, expect, vi, beforeEach } from "vitest"; 2 2 + 3 3 + vi.mock("@/webhooks/dispatcher.js", () => ({ 4 4 + dispatch: vi.fn(), 5 5 + })); 6 6 + 7 7 + vi.mock("@/actions/executor.js", () => ({ 8 8 + executeAction: vi.fn(), 9 9 + })); 10 10 + 11 11 + vi.mock("@/actions/fetcher.js", () => ({ 12 12 + resolveFetches: vi.fn(), 13 13 + })); 14 14 + 15 15 + import { handleMatchedEvent } from "./handler.js"; 16 16 + import { dispatch } from "../webhooks/dispatcher.js"; 17 17 + import { executeAction } from "../actions/executor.js"; 18 18 + import { resolveFetches } from "../actions/fetcher.js"; 19 19 + import { makeMatch, makeWebhookAction, makeRecordAction, makeFetchStep } from "../test/fixtures.js"; 20 20 + 21 21 + const mockDispatch = vi.mocked(dispatch); 22 22 + const mockExecuteAction = vi.mocked(executeAction); 23 23 + const mockResolveFetches = vi.mocked(resolveFetches); 24 24 + 25 25 + describe("handleMatchedEvent", () => { 26 26 + beforeEach(() => { 27 27 + mockDispatch.mockReset().mockResolvedValue(undefined); 28 28 + mockExecuteAction.mockReset().mockResolvedValue(undefined); 29 29 + mockResolveFetches.mockReset(); 30 30 + }); 31 31 + 32 32 + it("dispatches a webhook action", async () => { 33 33 + const match = makeMatch({ 34 34 + subscription: { actions: [makeWebhookAction()], fetches: [] }, 35 35 + }); 36 36 + 37 37 + await handleMatchedEvent(match); 38 38 + 39 39 + expect(mockDispatch).toHaveBeenCalledWith(match, 0, {}); 40 40 + expect(mockExecuteAction).not.toHaveBeenCalled(); 41 41 + expect(mockResolveFetches).not.toHaveBeenCalled(); 42 42 + }); 43 43 + 44 44 + it("executes a record action", async () => { 45 45 + const match = makeMatch({ 46 46 + subscription: { actions: [makeRecordAction()], fetches: [] }, 47 47 + }); 48 48 + 49 49 + await handleMatchedEvent(match); 50 50 + 51 51 + expect(mockExecuteAction).toHaveBeenCalledWith(match, 0, {}); 52 52 + expect(mockDispatch).not.toHaveBeenCalled(); 53 53 + }); 54 54 + 55 55 + it("handles mixed action types with correct routing", async () => { 56 56 + const match = makeMatch({ 57 57 + subscription: { 58 58 + actions: [ 59 59 + makeWebhookAction(), 60 60 + makeRecordAction(), 61 61 + makeWebhookAction({ callbackUrl: "https://other.com/hook" }), 62 62 + ], 63 63 + fetches: [], 64 64 + }, 65 65 + }); 66 66 + 67 67 + await handleMatchedEvent(match); 68 68 + 69 69 + expect(mockDispatch).toHaveBeenCalledTimes(2); 70 70 + expect(mockDispatch).toHaveBeenCalledWith(match, 0, {}); 71 71 + expect(mockDispatch).toHaveBeenCalledWith(match, 2, {}); 72 72 + expect(mockExecuteAction).toHaveBeenCalledOnce(); 73 73 + expect(mockExecuteAction).toHaveBeenCalledWith(match, 1, {}); 74 74 + }); 75 75 + 76 76 + it("resolves fetches and passes context to all actions", async () => { 77 77 + const fetchContext = { 78 78 + profile: { uri: "at://x/col/rk", cid: "c", record: { name: "Alice" } }, 79 79 + }; 80 80 + mockResolveFetches.mockResolvedValueOnce({ context: fetchContext, errors: [] }); 81 81 + 82 82 + const match = makeMatch({ 83 83 + subscription: { 84 84 + actions: [makeWebhookAction(), makeRecordAction()], 85 85 + fetches: [makeFetchStep()], 86 86 + }, 87 87 + }); 88 88 + 89 89 + await handleMatchedEvent(match); 90 90 + 91 91 + expect(mockResolveFetches).toHaveBeenCalledWith( 92 92 + match.subscription.fetches, 93 93 + match.event, 94 94 + match.subscription.did, 95 95 + ); 96 96 + expect(mockDispatch).toHaveBeenCalledWith(match, 0, fetchContext); 97 97 + expect(mockExecuteAction).toHaveBeenCalledWith(match, 1, fetchContext); 98 98 + }); 99 99 + 100 100 + it("skips fetch resolution when no fetch steps", async () => { 101 101 + const match = makeMatch({ 102 102 + subscription: { actions: [makeWebhookAction()], fetches: [] }, 103 103 + }); 104 104 + 105 105 + await handleMatchedEvent(match); 106 106 + 107 107 + expect(mockResolveFetches).not.toHaveBeenCalled(); 108 108 + }); 109 109 + 110 110 + it("passes partial fetch context when some fetches fail", async () => { 111 111 + const partialContext = { 112 112 + profile: { uri: "at://x/col/rk", cid: "c", record: { name: "Alice" } }, 113 113 + }; 114 114 + mockResolveFetches.mockResolvedValueOnce({ 115 115 + context: partialContext, 116 116 + errors: [{ name: "extra", error: "PDS unreachable" }], 117 117 + }); 118 118 + 119 119 + const match = makeMatch({ 120 120 + subscription: { 121 121 + actions: [makeWebhookAction()], 122 122 + fetches: [makeFetchStep(), makeFetchStep({ name: "extra", uri: "at://x/col/rk2" })], 123 123 + }, 124 124 + }); 125 125 + 126 126 + await handleMatchedEvent(match); 127 127 + 128 128 + // Actions still receive the partial context 129 129 + expect(mockDispatch).toHaveBeenCalledWith(match, 0, partialContext); 130 130 + }); 131 131 + 132 132 + it("one action failing does not block others", async () => { 133 133 + mockDispatch.mockRejectedValueOnce(new Error("webhook failed")); 134 134 + 135 135 + const match = makeMatch({ 136 136 + subscription: { 137 137 + actions: [makeWebhookAction(), makeRecordAction()], 138 138 + fetches: [], 139 139 + }, 140 140 + }); 141 141 + 142 142 + await handleMatchedEvent(match); 143 143 + 144 144 + // Both actions were called despite the first one rejecting 145 145 + expect(mockDispatch).toHaveBeenCalledOnce(); 146 146 + expect(mockExecuteAction).toHaveBeenCalledOnce(); 147 147 + }); 148 148 + });

+29

lib/jetstream/handler.ts

reviewed

··· 1 1 + import { dispatch } from "../webhooks/dispatcher.js"; 2 2 + import { executeAction } from "../actions/executor.js"; 3 3 + import { resolveFetches } from "../actions/fetcher.js"; 4 4 + import type { MatchedEvent } from "./consumer.js"; 5 5 + 6 6 + /** Handle a matched Jetstream event: resolve fetches, then dispatch all actions. */ 7 7 + export async function handleMatchedEvent(match: MatchedEvent) { 8 8 + let fetchContext: Record<string, { uri: string; cid: string; record: Record<string, unknown> }> = 9 9 + {}; 10 10 + if (match.subscription.fetches.length > 0) { 11 11 + const result = await resolveFetches( 12 12 + match.subscription.fetches, 13 13 + match.event, 14 14 + match.subscription.did, 15 15 + ); 16 16 + fetchContext = result.context; 17 17 + for (const err of result.errors) { 18 18 + console.warn(`Fetch "${err.name}" failed for ${match.subscription.uri}: ${err.error}`); 19 19 + } 20 20 + } 21 21 + 22 22 + for (let i = 0; i < match.subscription.actions.length; i++) { 23 23 + const action = match.subscription.actions[i]!; 24 24 + const handler = action.$type === "record" ? executeAction : dispatch; 25 25 + handler(match, i, fetchContext).catch((err) => { 26 26 + console.error(`Action ${i} (${action.$type}) delivery error:`, err); 27 27 + }); 28 28 + } 29 29 + }

+241

lib/jetstream/matcher.test.ts

reviewed

··· 1 1 + import { describe, it, expect } from "vitest"; 2 2 + import { matchConditions } from "./matcher.js"; 3 3 + import { makeEvent } from "../test/fixtures.js"; 4 4 + 5 5 + describe("matchConditions", () => { 6 6 + const ownerDid = "did:plc:owner123"; 7 7 + 8 8 + describe("empty conditions", () => { 9 9 + it("matches any event", () => { 10 10 + expect(matchConditions(makeEvent(), [], ownerDid)).toBe(true); 11 11 + }); 12 12 + }); 13 13 + 14 14 + describe("eq operator", () => { 15 15 + it("matches when field equals value", () => { 16 16 + const event = makeEvent({ did: "did:plc:target" }); 17 17 + const conditions = [{ field: "event.did", operator: "eq", value: "did:plc:target" }]; 18 18 + expect(matchConditions(event, conditions, ownerDid)).toBe(true); 19 19 + }); 20 20 + 21 21 + it("returns false when field does not equal value", () => { 22 22 + const event = makeEvent({ did: "did:plc:other" }); 23 23 + const conditions = [{ field: "event.did", operator: "eq", value: "did:plc:target" }]; 24 24 + expect(matchConditions(event, conditions, ownerDid)).toBe(false); 25 25 + }); 26 26 + }); 27 27 + 28 28 + describe("startsWith operator", () => { 29 29 + it("matches when field starts with value", () => { 30 30 + const event = makeEvent({ did: "did:plc:user123" }); 31 31 + const conditions = [{ field: "event.did", operator: "startsWith", value: "did:plc:" }]; 32 32 + expect(matchConditions(event, conditions, ownerDid)).toBe(true); 33 33 + }); 34 34 + 35 35 + it("returns false when prefix does not match", () => { 36 36 + const event = makeEvent({ did: "did:web:user" }); 37 37 + const conditions = [{ field: "event.did", operator: "startsWith", value: "did:plc:" }]; 38 38 + expect(matchConditions(event, conditions, ownerDid)).toBe(false); 39 39 + }); 40 40 + }); 41 41 + 42 42 + describe("endsWith operator", () => { 43 43 + it("matches when field ends with value", () => { 44 44 + const event = makeEvent({ 45 45 + commit: { 46 46 + rev: "r1", 47 47 + operation: "create", 48 48 + collection: "app.bsky.feed.like", 49 49 + rkey: "abc", 50 50 + record: { status: "published" }, 51 51 + }, 52 52 + }); 53 53 + const conditions = [{ field: "status", operator: "endsWith", value: "shed" }]; 54 54 + expect(matchConditions(event, conditions, ownerDid)).toBe(true); 55 55 + }); 56 56 + 57 57 + it("returns false when suffix does not match", () => { 58 58 + const event = makeEvent({ 59 59 + commit: { 60 60 + rev: "r1", 61 61 + operation: "create", 62 62 + collection: "app.bsky.feed.like", 63 63 + rkey: "abc", 64 64 + record: { status: "draft" }, 65 65 + }, 66 66 + }); 67 67 + const conditions = [{ field: "status", operator: "endsWith", value: "shed" }]; 68 68 + expect(matchConditions(event, conditions, ownerDid)).toBe(false); 69 69 + }); 70 70 + }); 71 71 + 72 72 + describe("contains operator", () => { 73 73 + it("matches when field contains value", () => { 74 74 + const event = makeEvent({ 75 75 + commit: { 76 76 + rev: "r1", 77 77 + operation: "create", 78 78 + collection: "app.bsky.feed.like", 79 79 + rkey: "abc", 80 80 + record: { text: "hello world" }, 81 81 + }, 82 82 + }); 83 83 + const conditions = [{ field: "text", operator: "contains", value: "lo wo" }]; 84 84 + expect(matchConditions(event, conditions, ownerDid)).toBe(true); 85 85 + }); 86 86 + }); 87 87 + 88 88 + describe("unknown operator", () => { 89 89 + it("returns false", () => { 90 90 + const conditions = [{ field: "event.did", operator: "notAnOperator", value: "x" }]; 91 91 + expect(matchConditions(makeEvent(), conditions, ownerDid)).toBe(false); 92 92 + }); 93 93 + }); 94 94 + 95 95 + describe("AND logic (multiple conditions)", () => { 96 96 + it("returns true when all conditions match", () => { 97 97 + const event = makeEvent({ 98 98 + did: "did:plc:target", 99 99 + commit: { 100 100 + rev: "r1", 101 101 + operation: "create", 102 102 + collection: "app.bsky.feed.like", 103 103 + rkey: "abc", 104 104 + record: { status: "active" }, 105 105 + }, 106 106 + }); 107 107 + const conditions = [ 108 108 + { field: "event.did", operator: "eq", value: "did:plc:target" }, 109 109 + { field: "status", operator: "eq", value: "active" }, 110 110 + ]; 111 111 + expect(matchConditions(event, conditions, ownerDid)).toBe(true); 112 112 + }); 113 113 + 114 114 + it("returns false when one condition fails", () => { 115 115 + const event = makeEvent({ 116 116 + did: "did:plc:target", 117 117 + commit: { 118 118 + rev: "r1", 119 119 + operation: "create", 120 120 + collection: "app.bsky.feed.like", 121 121 + rkey: "abc", 122 122 + record: { status: "inactive" }, 123 123 + }, 124 124 + }); 125 125 + const conditions = [ 126 126 + { field: "event.did", operator: "eq", value: "did:plc:target" }, 127 127 + { field: "status", operator: "eq", value: "active" }, 128 128 + ]; 129 129 + expect(matchConditions(event, conditions, ownerDid)).toBe(false); 130 130 + }); 131 131 + }); 132 132 + 133 133 + describe("{{self}} placeholder", () => { 134 134 + it("resolves to ownerDid", () => { 135 135 + const event = makeEvent({ did: ownerDid }); 136 136 + const conditions = [{ field: "event.did", operator: "eq", value: "{{self}}" }]; 137 137 + expect(matchConditions(event, conditions, ownerDid)).toBe(true); 138 138 + }); 139 139 + 140 140 + it("does not match when DID differs from owner", () => { 141 141 + const event = makeEvent({ did: "did:plc:someone-else" }); 142 142 + const conditions = [{ field: "event.did", operator: "eq", value: "{{self}}" }]; 143 143 + expect(matchConditions(event, conditions, ownerDid)).toBe(false); 144 144 + }); 145 145 + }); 146 146 + 147 147 + describe("field resolution", () => { 148 148 + it("resolves event.did", () => { 149 149 + const event = makeEvent({ did: "did:plc:abc" }); 150 150 + const conditions = [{ field: "event.did", operator: "eq", value: "did:plc:abc" }]; 151 151 + expect(matchConditions(event, conditions, ownerDid)).toBe(true); 152 152 + }); 153 153 + 154 154 + it("resolves legacy 'repo' alias to event.did", () => { 155 155 + const event = makeEvent({ did: "did:plc:abc" }); 156 156 + const conditions = [{ field: "repo", operator: "eq", value: "did:plc:abc" }]; 157 157 + expect(matchConditions(event, conditions, ownerDid)).toBe(true); 158 158 + }); 159 159 + 160 160 + it("resolves event.commit.operation", () => { 161 161 + const conditions = [{ field: "event.commit.operation", operator: "eq", value: "create" }]; 162 162 + expect(matchConditions(makeEvent(), conditions, ownerDid)).toBe(true); 163 163 + }); 164 164 + 165 165 + it("resolves dotted record path (single level)", () => { 166 166 + const event = makeEvent({ 167 167 + commit: { 168 168 + rev: "r1", 169 169 + operation: "create", 170 170 + collection: "test", 171 171 + rkey: "k", 172 172 + record: { status: "active" }, 173 173 + }, 174 174 + }); 175 175 + const conditions = [{ field: "status", operator: "eq", value: "active" }]; 176 176 + expect(matchConditions(event, conditions, ownerDid)).toBe(true); 177 177 + }); 178 178 + 179 179 + it("resolves nested dotted record path", () => { 180 180 + const event = makeEvent({ 181 181 + commit: { 182 182 + rev: "r1", 183 183 + operation: "create", 184 184 + collection: "test", 185 185 + rkey: "k", 186 186 + record: { subject: { uri: "at://did/col/rkey" } }, 187 187 + }, 188 188 + }); 189 189 + const conditions = [{ field: "subject.uri", operator: "startsWith", value: "at://did" }]; 190 190 + expect(matchConditions(event, conditions, ownerDid)).toBe(true); 191 191 + }); 192 192 + 193 193 + it("returns false for missing field", () => { 194 194 + const event = makeEvent({ 195 195 + commit: { 196 196 + rev: "r1", 197 197 + operation: "create", 198 198 + collection: "test", 199 199 + rkey: "k", 200 200 + record: {}, 201 201 + }, 202 202 + }); 203 203 + const conditions = [{ field: "nonexistent", operator: "eq", value: "x" }]; 204 204 + expect(matchConditions(event, conditions, ownerDid)).toBe(false); 205 205 + }); 206 206 + 207 207 + it("returns false when event has no commit", () => { 208 208 + const event = makeEvent({ commit: undefined, kind: "identity" }); 209 209 + const conditions = [{ field: "status", operator: "eq", value: "active" }]; 210 210 + expect(matchConditions(event, conditions, ownerDid)).toBe(false); 211 211 + }); 212 212 + 213 213 + it("returns false when commit has no record", () => { 214 214 + const event = makeEvent({ 215 215 + commit: { 216 216 + rev: "r1", 217 217 + operation: "delete", 218 218 + collection: "test", 219 219 + rkey: "k", 220 220 + record: undefined, 221 221 + }, 222 222 + }); 223 223 + const conditions = [{ field: "status", operator: "eq", value: "x" }]; 224 224 + expect(matchConditions(event, conditions, ownerDid)).toBe(false); 225 225 + }); 226 226 + 227 227 + it("JSON-stringifies non-string field values", () => { 228 228 + const event = makeEvent({ 229 229 + commit: { 230 230 + rev: "r1", 231 231 + operation: "create", 232 232 + collection: "test", 233 233 + rkey: "k", 234 234 + record: { count: 42 }, 235 235 + }, 236 236 + }); 237 237 + const conditions = [{ field: "count", operator: "eq", value: "42" }]; 238 238 + expect(matchConditions(event, conditions, ownerDid)).toBe(true); 239 239 + }); 240 240 + }); 241 241 + });

+170

lib/lexicons/resolver.test.ts

reviewed

··· 1 1 + import { describe, it, expect } from "vitest"; 2 2 + import { isValidNsid, nsidToAuthority, isNsidAllowed, parseLexicon } from "./resolver.js"; 3 3 + 4 4 + describe("isValidNsid", () => { 5 5 + it("accepts valid 3-segment NSID", () => { 6 6 + expect(isValidNsid("app.bsky.feed")).toBe(true); 7 7 + }); 8 8 + 9 9 + it("accepts valid 4-segment NSID with camelCase name", () => { 10 10 + expect(isValidNsid("site.exosphere.featureRequest.status")).toBe(true); 11 11 + }); 12 12 + 13 13 + it("accepts NSID with hyphens in segments", () => { 14 14 + expect(isValidNsid("com.my-app.feed.post")).toBe(true); 15 15 + }); 16 16 + 17 17 + it("rejects 2-segment NSID", () => { 18 18 + expect(isValidNsid("app.bsky")).toBe(false); 19 19 + }); 20 20 + 21 21 + it("rejects NSID starting with uppercase", () => { 22 22 + expect(isValidNsid("App.bsky.feed")).toBe(false); 23 23 + }); 24 24 + 25 25 + it("rejects empty string", () => { 26 26 + expect(isValidNsid("")).toBe(false); 27 27 + }); 28 28 + 29 29 + it("rejects NSID with spaces", () => { 30 30 + expect(isValidNsid("app.bsky .feed")).toBe(false); 31 31 + }); 32 32 + }); 33 33 + 34 34 + describe("nsidToAuthority", () => { 35 35 + it("reverses authority segments (3-segment NSID)", () => { 36 36 + expect(nsidToAuthority("app.bsky.feed")).toBe("bsky.app"); 37 37 + }); 38 38 + 39 39 + it("reverses authority segments (4-segment NSID)", () => { 40 40 + expect(nsidToAuthority("sh.tangled.feed.star")).toBe("feed.tangled.sh"); 41 41 + }); 42 42 + }); 43 43 + 44 44 + describe("isNsidAllowed", () => { 45 45 + it("allows everything when both lists are empty", () => { 46 46 + expect(isNsidAllowed("app.bsky.feed.like", [], [])).toBe(true); 47 47 + }); 48 48 + 49 49 + it("blocks NSID on blocklist", () => { 50 50 + expect(isNsidAllowed("app.bsky.feed.like", [], ["app.bsky.feed.like"])).toBe(false); 51 51 + }); 52 52 + 53 53 + it("blocks NSID matching blocklist glob", () => { 54 54 + expect(isNsidAllowed("app.bsky.feed.like", [], ["app.bsky.*"])).toBe(false); 55 55 + }); 56 56 + 57 57 + it("allows NSID on allowlist", () => { 58 58 + expect(isNsidAllowed("app.bsky.feed.like", ["app.bsky.feed.like"], [])).toBe(true); 59 59 + }); 60 60 + 61 61 + it("rejects NSID not on non-empty allowlist", () => { 62 62 + expect(isNsidAllowed("app.bsky.feed.like", ["sh.tangled.*"], [])).toBe(false); 63 63 + }); 64 64 + 65 65 + it("allowlist glob matches prefix", () => { 66 66 + expect(isNsidAllowed("app.bsky.feed.like", ["app.bsky.*"], [])).toBe(true); 67 67 + }); 68 68 + 69 69 + it("blocklist takes precedence over allowlist", () => { 70 70 + expect( 71 71 + isNsidAllowed("app.bsky.feed.like", ["app.bsky.feed.like"], ["app.bsky.feed.like"]), 72 72 + ).toBe(false); 73 73 + }); 74 74 + }); 75 75 + 76 76 + describe("parseLexicon", () => { 77 77 + it("extracts primitive fields from record properties", () => { 78 78 + const json = { 79 79 + lexicon: 1, 80 80 + id: "app.test.record", 81 81 + defs: { 82 82 + main: { 83 83 + type: "record", 84 84 + record: { 85 85 + properties: { 86 86 + name: { type: "string", description: "The name" }, 87 87 + count: { type: "integer" }, 88 88 + active: { type: "boolean" }, 89 89 + }, 90 90 + }, 91 91 + }, 92 92 + }, 93 93 + }; 94 94 + const result = parseLexicon("app.test.record", json); 95 95 + expect(result.nsid).toBe("app.test.record"); 96 96 + expect(result.fields).toEqual([ 97 97 + { path: "name", type: "string", description: "The name" }, 98 98 + { path: "count", type: "integer", description: undefined }, 99 99 + { path: "active", type: "boolean", description: undefined }, 100 100 + ]); 101 101 + }); 102 102 + 103 103 + it("extracts nested object fields with dotted paths", () => { 104 104 + const json = { 105 105 + lexicon: 1, 106 106 + id: "app.test.nested", 107 107 + defs: { 108 108 + main: { 109 109 + type: "record", 110 110 + record: { 111 111 + properties: { 112 112 + subject: { 113 113 + type: "object", 114 114 + properties: { 115 115 + uri: { type: "string" }, 116 116 + cid: { type: "string" }, 117 117 + }, 118 118 + }, 119 119 + }, 120 120 + }, 121 121 + }, 122 122 + }, 123 123 + }; 124 124 + const result = parseLexicon("app.test.nested", json); 125 125 + expect(result.fields).toEqual([ 126 126 + { path: "subject.uri", type: "string", description: undefined }, 127 127 + { path: "subject.cid", type: "string", description: undefined }, 128 128 + ]); 129 129 + }); 130 130 + 131 131 + it("resolves local #ref properties", () => { 132 132 + const json = { 133 133 + lexicon: 1, 134 134 + id: "app.test.ref", 135 135 + defs: { 136 136 + main: { 137 137 + type: "record", 138 138 + record: { 139 139 + properties: { 140 140 + condition: { type: "ref", ref: "#conditionDef" }, 141 141 + }, 142 142 + }, 143 143 + }, 144 144 + conditionDef: { 145 145 + type: "object", 146 146 + properties: { 147 147 + field: { type: "string" }, 148 148 + value: { type: "string" }, 149 149 + }, 150 150 + }, 151 151 + }, 152 152 + }; 153 153 + const result = parseLexicon("app.test.ref", json); 154 154 + expect(result.fields).toEqual([ 155 155 + { path: "condition.field", type: "string", description: undefined }, 156 156 + { path: "condition.value", type: "string", description: undefined }, 157 157 + ]); 158 158 + }); 159 159 + 160 160 + it("throws for lexicon without record definition", () => { 161 161 + const json = { 162 162 + lexicon: 1, 163 163 + id: "app.test.token", 164 164 + defs: { 165 165 + main: { type: "token" }, 166 166 + }, 167 167 + }; 168 168 + expect(() => parseLexicon("app.test.token", json)).toThrow("no record definition"); 169 169 + }); 170 170 + });

+143

lib/pds/resolver.test.ts

reviewed

··· 1 1 + import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; 2 2 + import { fetchRecord, resolveDid } from "./resolver.js"; 3 3 + 4 4 + describe("resolveDid", () => { 5 5 + const mockFetch = vi.fn(); 6 6 + 7 7 + beforeEach(() => { 8 8 + vi.stubGlobal("fetch", mockFetch); 9 9 + }); 10 10 + 11 11 + afterEach(() => { 12 12 + vi.unstubAllGlobals(); 13 13 + }); 14 14 + 15 15 + it("returns PDS endpoint from DID document", async () => { 16 16 + mockFetch.mockResolvedValueOnce({ 17 17 + ok: true, 18 18 + json: async () => ({ 19 19 + service: [{ id: "#atproto_pds", serviceEndpoint: "https://pds.example.com" }], 20 20 + }), 21 21 + }); 22 22 + 23 23 + const result = await resolveDid("did:plc:abc"); 24 24 + expect(result).toBe("https://pds.example.com"); 25 25 + }); 26 26 + 27 27 + it("returns null when plc.directory is unreachable", async () => { 28 28 + mockFetch.mockRejectedValueOnce(new Error("network error")); 29 29 + 30 30 + const result = await resolveDid("did:plc:abc"); 31 31 + expect(result).toBeNull(); 32 32 + }); 33 33 + 34 34 + it("returns null when plc.directory returns non-200", async () => { 35 35 + mockFetch.mockResolvedValueOnce({ ok: false, status: 404 }); 36 36 + 37 37 + const result = await resolveDid("did:plc:abc"); 38 38 + expect(result).toBeNull(); 39 39 + }); 40 40 + 41 41 + it("returns null when no #atproto_pds service in document", async () => { 42 42 + mockFetch.mockResolvedValueOnce({ 43 43 + ok: true, 44 44 + json: async () => ({ 45 45 + service: [{ id: "#other", serviceEndpoint: "https://other.com" }], 46 46 + }), 47 47 + }); 48 48 + 49 49 + const result = await resolveDid("did:plc:abc"); 50 50 + expect(result).toBeNull(); 51 51 + }); 52 52 + }); 53 53 + 54 54 + describe("fetchRecord", () => { 55 55 + const mockFetch = vi.fn(); 56 56 + 57 57 + beforeEach(() => { 58 58 + vi.stubGlobal("fetch", mockFetch); 59 59 + }); 60 60 + 61 61 + afterEach(() => { 62 62 + vi.unstubAllGlobals(); 63 63 + }); 64 64 + 65 65 + it("fetches record via DID resolution + PDS getRecord", async () => { 66 66 + // First call: DID resolution 67 67 + mockFetch.mockResolvedValueOnce({ 68 68 + ok: true, 69 69 + json: async () => ({ 70 70 + service: [{ id: "#atproto_pds", serviceEndpoint: "https://pds.example.com" }], 71 71 + }), 72 72 + }); 73 73 + // Second call: getRecord 74 74 + mockFetch.mockResolvedValueOnce({ 75 75 + ok: true, 76 76 + json: async () => ({ 77 77 + uri: "at://did:plc:abc/app.bsky.feed.post/rk1", 78 78 + cid: "bafycid123", 79 79 + value: { text: "hello" }, 80 80 + }), 81 81 + }); 82 82 + 83 83 + const result = await fetchRecord("at://did:plc:abc/app.bsky.feed.post/rk1"); 84 84 + expect(result).toEqual({ 85 85 + uri: "at://did:plc:abc/app.bsky.feed.post/rk1", 86 86 + cid: "bafycid123", 87 87 + record: { text: "hello" }, 88 88 + }); 89 89 + 90 90 + // Verify getRecord URL 91 91 + const getRecordCall = mockFetch.mock.calls[1]!; 92 92 + const url = new URL(getRecordCall[0] as string); 93 93 + expect(url.origin).toBe("https://pds.example.com"); 94 94 + expect(url.pathname).toBe("/xrpc/com.atproto.repo.getRecord"); 95 95 + expect(url.searchParams.get("repo")).toBe("did:plc:abc"); 96 96 + expect(url.searchParams.get("collection")).toBe("app.bsky.feed.post"); 97 97 + expect(url.searchParams.get("rkey")).toBe("rk1"); 98 98 + }); 99 99 + 100 100 + it("throws for invalid AT URI format", async () => { 101 101 + await expect(fetchRecord("https://not-an-at-uri")).rejects.toThrow("Invalid AT URI"); 102 102 + }); 103 103 + 104 104 + it("throws when DID cannot be resolved", async () => { 105 105 + mockFetch.mockResolvedValueOnce({ ok: false, status: 404 }); 106 106 + 107 107 + await expect(fetchRecord("at://did:plc:unknown/col/rk")).rejects.toThrow( 108 108 + "Could not resolve PDS", 109 109 + ); 110 110 + }); 111 111 + 112 112 + it("throws when PDS getRecord fails", async () => { 113 113 + mockFetch 114 114 + .mockResolvedValueOnce({ 115 115 + ok: true, 116 116 + json: async () => ({ 117 117 + service: [{ id: "#atproto_pds", serviceEndpoint: "https://pds.example.com" }], 118 118 + }), 119 119 + }) 120 120 + .mockResolvedValueOnce({ ok: false, status: 404 }); 121 121 + 122 122 + await expect(fetchRecord("at://did:plc:abc/col/rk")).rejects.toThrow("getRecord failed (404)"); 123 123 + }); 124 124 + 125 125 + it("returns defaults for missing cid/value in response", async () => { 126 126 + mockFetch 127 127 + .mockResolvedValueOnce({ 128 128 + ok: true, 129 129 + json: async () => ({ 130 130 + service: [{ id: "#atproto_pds", serviceEndpoint: "https://pds.example.com" }], 131 131 + }), 132 132 + }) 133 133 + .mockResolvedValueOnce({ 134 134 + ok: true, 135 135 + json: async () => ({}), 136 136 + }); 137 137 + 138 138 + const result = await fetchRecord("at://did:plc:abc/col/rk"); 139 139 + expect(result.uri).toBe("at://did:plc:abc/col/rk"); 140 140 + expect(result.cid).toBe(""); 141 141 + expect(result.record).toEqual({}); 142 142 + }); 143 143 + });

+42

lib/rate-limit.test.ts

reviewed

··· 1 1 + import { describe, it, expect } from "vitest"; 2 2 + import { Hono } from "hono"; 3 3 + import { rateLimit } from "./rate-limit.js"; 4 4 + 5 5 + describe("rateLimit", () => { 6 6 + function createApp(max: number) { 7 7 + const app = new Hono(); 8 8 + app.use("*", rateLimit({ windowMs: 60_000, max })); 9 9 + app.get("/", (c) => c.text("ok")); 10 10 + return app; 11 11 + } 12 12 + 13 13 + it("allows requests under the limit", async () => { 14 14 + const app = createApp(3); 15 15 + const req = new Request("http://localhost/", { headers: { "x-forwarded-for": "1.2.3.4" } }); 16 16 + 17 17 + const r1 = await app.request(req); 18 18 + expect(r1.status).toBe(200); 19 19 + expect(r1.headers.get("X-RateLimit-Remaining")).toBe("2"); 20 20 + }); 21 21 + 22 22 + it("returns 429 when limit exceeded", async () => { 23 23 + const app = createApp(2); 24 24 + const req = () => 25 25 + new Request("http://localhost/", { headers: { "x-forwarded-for": "5.6.7.8" } }); 26 26 + 27 27 + await app.request(req()); 28 28 + await app.request(req()); 29 29 + const r3 = await app.request(req()); 30 30 + expect(r3.status).toBe(429); 31 31 + }); 32 32 + 33 33 + it("tracks clients independently", async () => { 34 34 + const app = createApp(1); 35 35 + const req1 = new Request("http://localhost/", { headers: { "x-forwarded-for": "10.0.0.1" } }); 36 36 + const req2 = new Request("http://localhost/", { headers: { "x-forwarded-for": "10.0.0.2" } }); 37 37 + 38 38 + await app.request(req1); // uses up client 1's quota 39 39 + const r2 = await app.request(req2); // client 2 still has quota 40 40 + expect(r2.status).toBe(200); 41 41 + }); 42 42 + });

+56

lib/rate-limit.ts

reviewed

··· 1 1 + import { createMiddleware } from "hono/factory"; 2 2 + 3 3 + interface RateLimitOptions { 4 4 + windowMs: number; 5 5 + max: number; 6 6 + } 7 7 + 8 8 + interface Entry { 9 9 + count: number; 10 10 + resetAt: number; 11 11 + } 12 12 + 13 13 + /** 14 14 + * Simple in-memory fixed-window rate limiter. 15 15 + * Key is derived from the client IP. 16 16 + */ 17 17 + export function rateLimit({ windowMs, max }: RateLimitOptions) { 18 18 + const store = new Map<string, Entry>(); 19 19 + 20 20 + // Periodic cleanup every 60s 21 21 + const interval = setInterval(() => { 22 22 + const now = Date.now(); 23 23 + for (const [key, entry] of store) { 24 24 + if (entry.resetAt <= now) store.delete(key); 25 25 + } 26 26 + }, 60_000); 27 27 + // Don't block process exit 28 28 + if (typeof interval === "object" && "unref" in interval) interval.unref(); 29 29 + 30 30 + return createMiddleware(async (c, next) => { 31 31 + const key = 32 32 + c.req.header("x-forwarded-for")?.split(",")[0]?.trim() || 33 33 + c.req.header("x-real-ip") || 34 34 + "unknown"; 35 35 + 36 36 + const now = Date.now(); 37 37 + let entry = store.get(key); 38 38 + 39 39 + if (!entry || entry.resetAt <= now) { 40 40 + entry = { count: 0, resetAt: now + windowMs }; 41 41 + store.set(key, entry); 42 42 + } 43 43 + 44 44 + entry.count++; 45 45 + 46 46 + c.header("X-RateLimit-Limit", String(max)); 47 47 + c.header("X-RateLimit-Remaining", String(Math.max(0, max - entry.count))); 48 48 + c.header("X-RateLimit-Reset", String(Math.ceil(entry.resetAt / 1000))); 49 49 + 50 50 + if (entry.count > max) { 51 51 + return c.json({ error: "Too many requests, please try again later" }, 429); 52 52 + } 53 53 + 54 54 + return next(); 55 55 + }); 56 56 + }

+19

lib/security-headers.ts

reviewed

··· 1 1 + import { createMiddleware } from "hono/factory"; 2 2 + 3 3 + /** 4 4 + * Set standard security headers on all responses. 5 5 + */ 6 6 + export function securityHeaders() { 7 7 + return createMiddleware(async (c, next) => { 8 8 + await next(); 9 9 + c.header("X-Content-Type-Options", "nosniff"); 10 10 + c.header("X-Frame-Options", "DENY"); 11 11 + c.header("Referrer-Policy", "strict-origin-when-cross-origin"); 12 12 + c.header("Strict-Transport-Security", "max-age=63072000; includeSubDomains"); 13 13 + c.header( 14 14 + "Content-Security-Policy", 15 15 + "default-src 'self'; script-src 'self'; style-src 'self' 'unsafe-inline'; img-src 'self' https:; connect-src 'self'; frame-ancestors 'none'", 16 16 + ); 17 17 + c.header("Permissions-Policy", "camera=(), microphone=(), geolocation=()"); 18 18 + }); 19 19 + }

+28

lib/subscriptions/pds.test.ts

reviewed

··· 1 1 + import { describe, it, expect } from "vitest"; 2 2 + import { generateTid } from "./pds.js"; 3 3 + 4 4 + describe("generateTid", () => { 5 5 + const BASE32_CHARS = "234567abcdefghijklmnopqrstuvwxyz"; 6 6 + 7 7 + it("returns a 13-character string", () => { 8 8 + expect(generateTid()).toHaveLength(13); 9 9 + }); 10 10 + 11 11 + it("uses only base32-sort characters", () => { 12 12 + const tid = generateTid(); 13 13 + for (const char of tid) { 14 14 + expect(BASE32_CHARS).toContain(char); 15 15 + } 16 16 + }); 17 17 + 18 18 + it("generates unique values (monotonic clockId)", () => { 19 19 + const tids = new Set(Array.from({ length: 100 }, () => generateTid())); 20 20 + expect(tids.size).toBe(100); 21 21 + }); 22 22 + 23 23 + it("is monotonically increasing (lexicographic order)", () => { 24 24 + const tids = Array.from({ length: 50 }, () => generateTid()); 25 25 + const sorted = [...tids].sort(); 26 26 + expect(tids).toEqual(sorted); 27 27 + }); 28 28 + });

+11 -2

lib/subscriptions/pds.ts

reviewed

··· 9 9 config.publicUrl.startsWith("http://127.0.0.1"); 10 10 11 11 // AT Protocol TID: base32-sort encoded (timestamp_us << 10 | clock_id) 12 12 + // Uses a monotonic counter instead of random clockId to guarantee uniqueness. 12 13 const S32 = "234567abcdefghijklmnopqrstuvwxyz"; 14 14 + let lastTimestampUs = 0n; 15 15 + let clockId = 0n; 16 16 + 13 17 export function generateTid(): string { 14 18 const timestampUs = BigInt(Date.now()) * 1000n; 15 15 - const clockId = BigInt(Math.floor(Math.random() * 1024)); 16 16 - let tid = (timestampUs << 10n) | clockId; 19 19 + if (timestampUs === lastTimestampUs) { 20 20 + clockId++; 21 21 + } else { 22 22 + lastTimestampUs = timestampUs; 23 23 + clockId = 0n; 24 24 + } 25 25 + let tid = (timestampUs << 10n) | (clockId & 0x3ffn); 17 26 let s = ""; 18 27 while (tid > 0n) { 19 28 s = S32[Number(tid % 32n)] + s;

+103

lib/subscriptions/verify.test.ts

reviewed

··· 1 1 + import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; 2 2 + import { verifyCallback } from "./verify.js"; 3 3 + 4 4 + describe("verifyCallback", () => { 5 5 + const mockFetch = vi.fn(); 6 6 + 7 7 + beforeEach(() => { 8 8 + vi.stubGlobal("fetch", mockFetch); 9 9 + }); 10 10 + 11 11 + afterEach(() => { 12 12 + vi.unstubAllGlobals(); 13 13 + }); 14 14 + 15 15 + it("returns ok when manifest has matching path and lexicon", async () => { 16 16 + mockFetch.mockResolvedValueOnce({ 17 17 + ok: true, 18 18 + json: async () => ({ 19 19 + callbacks: [{ path: "/webhook", lexicons: ["app.bsky.feed.like"] }], 20 20 + }), 21 21 + }); 22 22 + 23 23 + const result = await verifyCallback("https://example.com/webhook", "app.bsky.feed.like"); 24 24 + expect(result).toEqual({ ok: true }); 25 25 + expect(mockFetch).toHaveBeenCalledWith( 26 26 + "https://example.com/.well-known/airglow", 27 27 + expect.objectContaining({ headers: { Accept: "application/json" } }), 28 28 + ); 29 29 + }); 30 30 + 31 31 + it("returns error for invalid callback URL", async () => { 32 32 + const result = await verifyCallback("not-a-url", "app.bsky.feed.like"); 33 33 + expect(result).toEqual({ ok: false, error: "Invalid callback URL" }); 34 34 + }); 35 35 + 36 36 + it("returns error when server is unreachable", async () => { 37 37 + mockFetch.mockRejectedValueOnce(new Error("ECONNREFUSED")); 38 38 + 39 39 + const result = await verifyCallback("https://example.com/webhook", "app.bsky.feed.like"); 40 40 + expect(result.ok).toBe(false); 41 41 + expect((result as any).error).toContain("Could not reach"); 42 42 + }); 43 43 + 44 44 + it("returns error when manifest returns non-200", async () => { 45 45 + mockFetch.mockResolvedValueOnce({ ok: false, status: 404 }); 46 46 + 47 47 + const result = await verifyCallback("https://example.com/webhook", "app.bsky.feed.like"); 48 48 + expect(result.ok).toBe(false); 49 49 + expect((result as any).error).toContain("404"); 50 50 + }); 51 51 + 52 52 + it("returns error when callbacks is not an array", async () => { 53 53 + mockFetch.mockResolvedValueOnce({ 54 54 + ok: true, 55 55 + json: async () => ({ callbacks: "invalid" }), 56 56 + }); 57 57 + 58 58 + const result = await verifyCallback("https://example.com/webhook", "app.bsky.feed.like"); 59 59 + expect(result).toEqual({ ok: false, error: "Invalid manifest: missing callbacks array" }); 60 60 + }); 61 61 + 62 62 + it("returns error when path is not found in manifest", async () => { 63 63 + mockFetch.mockResolvedValueOnce({ 64 64 + ok: true, 65 65 + json: async () => ({ 66 66 + callbacks: [{ path: "/other", lexicons: ["app.bsky.feed.like"] }], 67 67 + }), 68 68 + }); 69 69 + 70 70 + const result = await verifyCallback("https://example.com/webhook", "app.bsky.feed.like"); 71 71 + expect(result.ok).toBe(false); 72 72 + expect((result as any).error).toContain("/webhook"); 73 73 + expect((result as any).error).toContain("not found"); 74 74 + }); 75 75 + 76 76 + it("returns error when lexicon is not accepted at path", async () => { 77 77 + mockFetch.mockResolvedValueOnce({ 78 78 + ok: true, 79 79 + json: async () => ({ 80 80 + callbacks: [{ path: "/webhook", lexicons: ["app.bsky.feed.post"] }], 81 81 + }), 82 82 + }); 83 83 + 84 84 + const result = await verifyCallback("https://example.com/webhook", "app.bsky.feed.like"); 85 85 + expect(result.ok).toBe(false); 86 86 + expect((result as any).error).toContain("does not accept lexicon"); 87 87 + }); 88 88 + 89 89 + it("matches the correct callback among multiple entries", async () => { 90 90 + mockFetch.mockResolvedValueOnce({ 91 91 + ok: true, 92 92 + json: async () => ({ 93 93 + callbacks: [ 94 94 + { path: "/other", lexicons: ["app.bsky.feed.post"] }, 95 95 + { path: "/webhook", lexicons: ["app.bsky.feed.like", "app.bsky.feed.repost"] }, 96 96 + ], 97 97 + }), 98 98 + }); 99 99 + 100 100 + const result = await verifyCallback("https://example.com/webhook", "app.bsky.feed.like"); 101 101 + expect(result).toEqual({ ok: true }); 102 102 + }); 103 103 + });

+5 -2

lib/subscriptions/verify.ts

reviewed

··· 1 1 + import { assertPublicUrl, UrlGuardError } from "../url-guard.js"; 2 2 + 1 3 type AirglowManifest = { 2 4 callbacks: Array<{ 3 5 path: string; ··· 16 18 ): Promise<{ ok: true } | { ok: false; error: string }> { 17 19 let url: URL; 18 20 try { 19 19 - url = new URL(callbackUrl); 20 20 - } catch { 21 21 + url = await assertPublicUrl(callbackUrl); 22 22 + } catch (err) { 23 23 + if (err instanceof UrlGuardError) return { ok: false, error: err.message }; 21 24 return { ok: false, error: "Invalid callback URL" }; 22 25 } 23 26

+22

lib/test/db.ts

reviewed

··· 1 1 + import { readFileSync } from "node:fs"; 2 2 + import { resolve } from "node:path"; 3 3 + import Database from "better-sqlite3"; 4 4 + import { drizzle } from "drizzle-orm/better-sqlite3"; 5 5 + import * as schema from "../db/schema.js"; 6 6 + 7 7 + const migrationSql = readFileSync( 8 8 + resolve(__dirname, "../db/migrations/0000_yellow_silvermane.sql"), 9 9 + "utf-8", 10 10 + ); 11 11 + 12 12 + export function createTestDb() { 13 13 + const sqlite = new Database(":memory:"); 14 14 + sqlite.pragma("foreign_keys = ON"); 15 15 + 16 16 + for (const statement of migrationSql.split("--> statement-breakpoint")) { 17 17 + const sql = statement.trim(); 18 18 + if (sql) sqlite.exec(sql); 19 19 + } 20 20 + 21 21 + return drizzle(sqlite, { schema }); 22 22 + }

+93

lib/test/fixtures.ts

reviewed

··· 1 1 + import type { JetstreamEvent } from "../jetstream/matcher.js"; 2 2 + import type { Action, WebhookAction, RecordAction, FetchStep } from "../db/schema.js"; 3 3 + import type { MatchedEvent } from "../jetstream/consumer.js"; 4 4 + 5 5 + type Subscription = { 6 6 + uri: string; 7 7 + did: string; 8 8 + rkey: string; 9 9 + name: string; 10 10 + description: string | null; 11 11 + lexicon: string; 12 12 + actions: Action[]; 13 13 + fetches: FetchStep[]; 14 14 + conditions: Array<{ field: string; operator: string; value: string; comment?: string }>; 15 15 + active: boolean; 16 16 + indexedAt: Date; 17 17 + }; 18 18 + 19 19 + export function makeEvent(overrides?: Partial<JetstreamEvent>): JetstreamEvent { 20 20 + return { 21 21 + did: "did:plc:testuser123", 22 22 + time_us: 1700000000000000, 23 23 + kind: "commit", 24 24 + commit: { 25 25 + rev: "abc123", 26 26 + operation: "create", 27 27 + collection: "app.bsky.feed.like", 28 28 + rkey: "3k2la7bx", 29 29 + record: { 30 30 + subject: { 31 31 + uri: "at://did:plc:other/app.bsky.feed.post/xyz", 32 32 + cid: "bafyreihash", 33 33 + }, 34 34 + createdAt: "2024-01-01T00:00:00.000Z", 35 35 + }, 36 36 + cid: "bafyreieventcid", 37 37 + }, 38 38 + ...overrides, 39 39 + }; 40 40 + } 41 41 + 42 42 + export function makeWebhookAction(overrides?: Partial<WebhookAction>): WebhookAction { 43 43 + return { 44 44 + $type: "webhook", 45 45 + callbackUrl: "https://example.com/hook", 46 46 + secret: "test-secret-32chars-placeholder00", 47 47 + ...overrides, 48 48 + }; 49 49 + } 50 50 + 51 51 + export function makeRecordAction(overrides?: Partial<RecordAction>): RecordAction { 52 52 + return { 53 53 + $type: "record", 54 54 + targetCollection: "app.bsky.feed.post", 55 55 + recordTemplate: '{"text":"{{event.did}}","createdAt":"{{now}}"}', 56 56 + ...overrides, 57 57 + }; 58 58 + } 59 59 + 60 60 + export function makeFetchStep(overrides?: Partial<FetchStep>): FetchStep { 61 61 + return { 62 62 + name: "profile", 63 63 + uri: "at://{{event.did}}/app.bsky.actor.profile/self", 64 64 + ...overrides, 65 65 + }; 66 66 + } 67 67 + 68 68 + export function makeSubscription(overrides?: Partial<Subscription>): Subscription { 69 69 + return { 70 70 + uri: "at://did:plc:testuser123/app.rglw.subscription/abc123", 71 71 + did: "did:plc:testuser123", 72 72 + rkey: "abc123", 73 73 + name: "Test Subscription", 74 74 + description: null, 75 75 + lexicon: "app.bsky.feed.like", 76 76 + actions: [makeWebhookAction()], 77 77 + fetches: [], 78 78 + conditions: [], 79 79 + active: true, 80 80 + indexedAt: new Date("2024-01-01T00:00:00.000Z"), 81 81 + ...overrides, 82 82 + }; 83 83 + } 84 84 + 85 85 + export function makeMatch(overrides?: { 86 86 + subscription?: Partial<Subscription>; 87 87 + event?: Partial<JetstreamEvent>; 88 88 + }): MatchedEvent { 89 89 + return { 90 90 + subscription: makeSubscription(overrides?.subscription) as any, 91 91 + event: makeEvent(overrides?.event), 92 92 + }; 93 93 + }

+109

lib/url-guard.test.ts

reviewed

··· 1 1 + import { describe, it, expect, vi, beforeEach } from "vitest"; 2 2 + import { assertPublicUrl, UrlGuardError } from "./url-guard.js"; 3 3 + 4 4 + // Mock DNS resolution to control test behavior without network 5 5 + vi.mock("node:dns/promises", () => ({ 6 6 + resolve4: vi.fn(), 7 7 + resolve6: vi.fn(), 8 8 + })); 9 9 + 10 10 + import { resolve4, resolve6 } from "node:dns/promises"; 11 11 + 12 12 + const mockResolve4 = vi.mocked(resolve4); 13 13 + const mockResolve6 = vi.mocked(resolve6); 14 14 + 15 15 + function resolveAs(ipv4: string[] = [], ipv6: string[] = []) { 16 16 + mockResolve4.mockResolvedValue(ipv4); 17 17 + mockResolve6.mockResolvedValue(ipv6); 18 18 + } 19 19 + 20 20 + function resolveNothing() { 21 21 + mockResolve4.mockRejectedValue(new Error("ENOTFOUND")); 22 22 + mockResolve6.mockRejectedValue(new Error("ENOTFOUND")); 23 23 + } 24 24 + 25 25 + describe("assertPublicUrl", () => { 26 26 + beforeEach(() => { 27 27 + mockResolve4.mockReset(); 28 28 + mockResolve6.mockReset(); 29 29 + }); 30 30 + 31 31 + it("accepts a public URL", async () => { 32 32 + resolveAs(["93.184.216.34"]); 33 33 + const url = await assertPublicUrl("https://example.com/hook"); 34 34 + expect(url.hostname).toBe("example.com"); 35 35 + }); 36 36 + 37 37 + it("rejects non-HTTP schemes", async () => { 38 38 + await expect(assertPublicUrl("ftp://example.com")).rejects.toThrow(UrlGuardError); 39 39 + await expect(assertPublicUrl("file:///etc/passwd")).rejects.toThrow(UrlGuardError); 40 40 + }); 41 41 + 42 42 + it("rejects invalid URLs", async () => { 43 43 + await expect(assertPublicUrl("not a url")).rejects.toThrow(); 44 44 + }); 45 45 + 46 46 + it("rejects localhost IP literal", async () => { 47 47 + await expect(assertPublicUrl("http://127.0.0.1/hook")).rejects.toThrow(UrlGuardError); 48 48 + }); 49 49 + 50 50 + it("rejects private IP 10.x.x.x", async () => { 51 51 + await expect(assertPublicUrl("http://10.0.0.1/hook")).rejects.toThrow(UrlGuardError); 52 52 + }); 53 53 + 54 54 + it("rejects private IP 192.168.x.x", async () => { 55 55 + await expect(assertPublicUrl("http://192.168.1.1/hook")).rejects.toThrow(UrlGuardError); 56 56 + }); 57 57 + 58 58 + it("rejects private IP 172.16.x.x", async () => { 59 59 + await expect(assertPublicUrl("http://172.16.0.1/hook")).rejects.toThrow(UrlGuardError); 60 60 + }); 61 61 + 62 62 + it("rejects cloud metadata IP", async () => { 63 63 + await expect(assertPublicUrl("http://169.254.169.254/latest")).rejects.toThrow(UrlGuardError); 64 64 + }); 65 65 + 66 66 + it("rejects localhost hostname", async () => { 67 67 + await expect(assertPublicUrl("http://localhost/hook")).rejects.toThrow(UrlGuardError); 68 68 + }); 69 69 + 70 70 + it("rejects .local hostname", async () => { 71 71 + await expect(assertPublicUrl("http://myhost.local/hook")).rejects.toThrow(UrlGuardError); 72 72 + }); 73 73 + 74 74 + it("rejects when DNS resolves to private IP", async () => { 75 75 + resolveAs(["127.0.0.1"]); 76 76 + await expect(assertPublicUrl("https://evil.com/hook")).rejects.toThrow(UrlGuardError); 77 77 + }); 78 78 + 79 79 + it("rejects when DNS resolves to 10.x.x.x", async () => { 80 80 + resolveAs(["10.0.0.5"]); 81 81 + await expect(assertPublicUrl("https://evil.com/hook")).rejects.toThrow(UrlGuardError); 82 82 + }); 83 83 + 84 84 + it("rejects when any resolved IP is private", async () => { 85 85 + resolveAs(["93.184.216.34", "10.0.0.1"]); 86 86 + await expect(assertPublicUrl("https://evil.com/hook")).rejects.toThrow(UrlGuardError); 87 87 + }); 88 88 + 89 89 + it("rejects unresolvable hostnames", async () => { 90 90 + resolveNothing(); 91 91 + await expect(assertPublicUrl("https://nonexistent.invalid/hook")).rejects.toThrow( 92 92 + UrlGuardError, 93 93 + ); 94 94 + }); 95 95 + 96 96 + it("rejects IPv6 loopback", async () => { 97 97 + await expect(assertPublicUrl("http://[::1]/hook")).rejects.toThrow(UrlGuardError); 98 98 + }); 99 99 + 100 100 + it("rejects when DNS resolves to IPv6 link-local", async () => { 101 101 + resolveAs([], ["fe80::1"]); 102 102 + await expect(assertPublicUrl("https://evil.com/hook")).rejects.toThrow(UrlGuardError); 103 103 + }); 104 104 + 105 105 + it("rejects when DNS resolves to IPv6 unique local", async () => { 106 106 + resolveAs([], ["fd12::1"]); 107 107 + await expect(assertPublicUrl("https://evil.com/hook")).rejects.toThrow(UrlGuardError); 108 108 + }); 109 109 + });

+105

lib/url-guard.ts

reviewed

··· 1 1 + import { resolve4, resolve6 } from "node:dns/promises"; 2 2 + 3 3 + /** 4 4 + * Validate that a URL does not target private/internal networks. 5 5 + * Resolves the hostname to IPs and rejects reserved ranges. 6 6 + */ 7 7 + export async function assertPublicUrl(rawUrl: string): Promise<URL> { 8 8 + const url = new URL(rawUrl); // throws on invalid URL 9 9 + 10 10 + // Block non-HTTP schemes 11 11 + if (url.protocol !== "https:" && url.protocol !== "http:") { 12 12 + throw new UrlGuardError("Only http and https URLs are allowed"); 13 13 + } 14 14 + 15 15 + // Resolve hostname to IPs and check each one 16 16 + const hostname = url.hostname; 17 17 + 18 18 + // Catch IP-literal hostnames directly (e.g. http://127.0.0.1) 19 19 + if (isPrivateIP(hostname)) { 20 20 + throw new UrlGuardError("Callback URL must not target a private network"); 21 21 + } 22 22 + 23 23 + // Resolve DNS and check resolved IPs 24 24 + const ips = await resolveAll(hostname); 25 25 + if (ips.length === 0) { 26 26 + throw new UrlGuardError("Could not resolve callback hostname"); 27 27 + } 28 28 + 29 29 + for (const ip of ips) { 30 30 + if (isPrivateIP(ip)) { 31 31 + throw new UrlGuardError("Callback URL must not target a private network"); 32 32 + } 33 33 + } 34 34 + 35 35 + return url; 36 36 + } 37 37 + 38 38 + export class UrlGuardError extends Error { 39 39 + constructor(message: string) { 40 40 + super(message); 41 41 + this.name = "UrlGuardError"; 42 42 + } 43 43 + } 44 44 + 45 45 + async function resolveAll(hostname: string): Promise<string[]> { 46 46 + const results: string[] = []; 47 47 + try { 48 48 + results.push(...(await resolve4(hostname))); 49 49 + } catch { 50 50 + // no A records 51 51 + } 52 52 + try { 53 53 + results.push(...(await resolve6(hostname))); 54 54 + } catch { 55 55 + // no AAAA records 56 56 + } 57 57 + return results; 58 58 + } 59 59 + 60 60 + /** Check if an IP address or hostname falls within private/reserved ranges. */ 61 61 + function isPrivateIP(ip: string): boolean { 62 62 + // IPv4 63 63 + if (ip.includes(".")) { 64 64 + const parts = ip.split(".").map(Number); 65 65 + if (parts.length !== 4 || parts.some((p) => Number.isNaN(p))) return false; 66 66 + 67 67 + const a = parts[0]!; 68 68 + const b = parts[1]!; 69 69 + 70 70 + return ( 71 71 + a === 0 || // 0.0.0.0/8 — current network 72 72 + a === 10 || // 10.0.0.0/8 73 73 + a === 127 || // 127.0.0.0/8 — loopback 74 74 + (a === 169 && b === 254) || // 169.254.0.0/16 — link-local / cloud metadata 75 75 + (a === 172 && b >= 16 && b <= 31) || // 172.16.0.0/12 76 76 + (a === 192 && b === 168) || // 192.168.0.0/16 77 77 + a >= 224 // 224+ — multicast & reserved 78 78 + ); 79 79 + } 80 80 + 81 81 + // IPv6 82 82 + const normalized = ip.toLowerCase().replace(/^\[|]$/g, ""); 83 83 + if ( 84 84 + normalized === "::1" || // loopback 85 85 + normalized === "::" || // unspecified 86 86 + normalized.startsWith("fc") || // fc00::/7 — unique local 87 87 + normalized.startsWith("fd") || // fc00::/7 88 88 + normalized.startsWith("fe80") || // fe80::/10 — link-local 89 89 + normalized.startsWith("::ffff:") // IPv4-mapped — check the embedded IPv4 90 90 + ) { 91 91 + // For IPv4-mapped IPv6 (::ffff:127.0.0.1), extract and re-check 92 92 + if (normalized.startsWith("::ffff:")) { 93 93 + const v4 = normalized.slice(7); 94 94 + if (v4.includes(".")) return isPrivateIP(v4); 95 95 + } 96 96 + return true; 97 97 + } 98 98 + 99 99 + // Also catch common hostname aliases 100 100 + if (normalized === "localhost" || normalized.endsWith(".local")) { 101 101 + return true; 102 102 + } 103 103 + 104 104 + return false; 105 105 + }

+179

lib/webhooks/dispatcher.test.ts

reviewed

··· 1 1 + import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; 2 2 + 3 3 + vi.mock("@/db/index.js", async () => { 4 4 + const { createTestDb } = await import("../test/db.js"); 5 5 + return { db: createTestDb() }; 6 6 + }); 7 7 + 8 8 + vi.mock("@/url-guard.js", () => ({ 9 9 + assertPublicUrl: vi.fn(async (url: string) => new URL(url)), 10 10 + UrlGuardError: class extends Error {}, 11 11 + })); 12 12 + 13 13 + import { dispatch } from "./dispatcher.js"; 14 14 + import { sign } from "./signer.js"; 15 15 + import { db } from "../db/index.js"; 16 16 + import { subscriptions, deliveryLogs } from "../db/schema.js"; 17 17 + import { makeMatch, makeWebhookAction, makeSubscription } from "../test/fixtures.js"; 18 18 + 19 19 + describe("dispatch", () => { 20 20 + const mockFetch = vi.fn(); 21 21 + 22 22 + beforeEach(async () => { 23 23 + vi.useFakeTimers(); 24 24 + vi.stubGlobal("fetch", mockFetch); 25 25 + mockFetch.mockReset(); 26 26 + 27 27 + // Clean up and re-seed so FK is satisfied 28 28 + await db.delete(deliveryLogs); 29 29 + await db.delete(subscriptions); 30 30 + const sub = makeSubscription(); 31 31 + await db.insert(subscriptions).values(sub); 32 32 + }); 33 33 + 34 34 + afterEach(() => { 35 35 + vi.useRealTimers(); 36 36 + vi.unstubAllGlobals(); 37 37 + }); 38 38 + 39 39 + it("delivers webhook with correct headers and payload", async () => { 40 40 + mockFetch.mockResolvedValueOnce({ status: 200 }); 41 41 + 42 42 + const action = makeWebhookAction({ callbackUrl: "https://example.com/hook", secret: "s3cret" }); 43 43 + const match = makeMatch({ subscription: { actions: [action] } }); 44 44 + 45 45 + await dispatch(match, 0); 46 46 + 47 47 + expect(mockFetch).toHaveBeenCalledOnce(); 48 48 + const [url, options] = mockFetch.mock.calls[0]!; 49 49 + expect(url).toBe("https://example.com/hook"); 50 50 + expect(options.method).toBe("POST"); 51 51 + 52 52 + const headers = options.headers; 53 53 + expect(headers["Content-Type"]).toBe("application/json"); 54 54 + expect(headers["X-Airglow-Subscription"]).toBe(match.subscription.uri); 55 55 + expect(headers["X-Airglow-Timestamp"]).toBeDefined(); 56 56 + 57 57 + // Verify HMAC signature 58 58 + const body = options.body; 59 59 + const expectedSig = sign(body, "s3cret"); 60 60 + expect(headers["X-Airglow-Signature"]).toBe(`sha256=${expectedSig}`); 61 61 + 62 62 + // Verify payload structure 63 63 + const payload = JSON.parse(body); 64 64 + expect(payload.subscription).toBe(match.subscription.uri); 65 65 + expect(payload.lexicon).toBe(match.subscription.lexicon); 66 66 + expect(payload.event.did).toBe(match.event.did); 67 67 + expect(payload.event.commit.operation).toBe("create"); 68 68 + }); 69 69 + 70 70 + it("logs successful delivery without payload", async () => { 71 71 + mockFetch.mockResolvedValueOnce({ status: 200 }); 72 72 + 73 73 + const match = makeMatch(); 74 74 + await dispatch(match, 0); 75 75 + 76 76 + const logs = await db.query.deliveryLogs.findMany(); 77 77 + expect(logs).toHaveLength(1); 78 78 + expect(logs[0]!.statusCode).toBe(200); 79 79 + expect(logs[0]!.payload).toBeNull(); 80 80 + expect(logs[0]!.attempt).toBe(1); 81 81 + }); 82 82 + 83 83 + it("logs failed delivery with payload for 4xx", async () => { 84 84 + mockFetch.mockResolvedValueOnce({ status: 400 }); 85 85 + 86 86 + const match = makeMatch(); 87 87 + await dispatch(match, 0); 88 88 + 89 89 + const logs = await db.query.deliveryLogs.findMany(); 90 90 + expect(logs).toHaveLength(1); 91 91 + expect(logs[0]!.statusCode).toBe(400); 92 92 + expect(logs[0]!.payload).not.toBeNull(); 93 93 + }); 94 94 + 95 95 + it("does not retry on 4xx", async () => { 96 96 + mockFetch.mockResolvedValueOnce({ status: 400 }); 97 97 + 98 98 + const match = makeMatch(); 99 99 + await dispatch(match, 0); 100 100 + 101 101 + // Advance past all retry delays 102 102 + await vi.advanceTimersByTimeAsync(60_000); 103 103 + expect(mockFetch).toHaveBeenCalledOnce(); 104 104 + }); 105 105 + 106 106 + it("retries on 5xx with correct delays", async () => { 107 107 + mockFetch 108 108 + .mockResolvedValueOnce({ status: 500 }) // initial 109 109 + .mockResolvedValueOnce({ status: 500 }) // retry 1 110 110 + .mockResolvedValueOnce({ status: 200 }); // retry 2 111 111 + 112 112 + const match = makeMatch(); 113 113 + await dispatch(match, 0); 114 114 + expect(mockFetch).toHaveBeenCalledTimes(1); 115 115 + 116 116 + // First retry after 5s 117 117 + await vi.advanceTimersByTimeAsync(5_000); 118 118 + expect(mockFetch).toHaveBeenCalledTimes(2); 119 119 + 120 120 + // Second retry after 30s 121 121 + await vi.advanceTimersByTimeAsync(30_000); 122 122 + expect(mockFetch).toHaveBeenCalledTimes(3); 123 123 + 124 124 + const logs = await db.query.deliveryLogs.findMany(); 125 125 + expect(logs).toHaveLength(3); 126 126 + expect(logs.map((l) => l.attempt).sort((a, b) => a - b)).toEqual([1, 2, 3]); 127 127 + }); 128 128 + 129 129 + it("retries on network failure (status 0)", async () => { 130 130 + mockFetch 131 131 + .mockRejectedValueOnce(new Error("ECONNREFUSED")) // initial 132 132 + .mockResolvedValueOnce({ status: 200 }); // retry 1 133 133 + 134 134 + const match = makeMatch(); 135 135 + await dispatch(match, 0); 136 136 + 137 137 + const logsBeforeRetry = await db.query.deliveryLogs.findMany(); 138 138 + expect(logsBeforeRetry[0]!.statusCode).toBe(0); 139 139 + 140 140 + await vi.advanceTimersByTimeAsync(5_000); 141 141 + expect(mockFetch).toHaveBeenCalledTimes(2); 142 142 + }); 143 143 + 144 144 + it("stops retrying after max retries (3 total attempts)", async () => { 145 145 + mockFetch.mockResolvedValue({ status: 500 }); 146 146 + 147 147 + const match = makeMatch(); 148 148 + await dispatch(match, 0); 149 149 + 150 150 + await vi.advanceTimersByTimeAsync(5_000); // retry 1 151 151 + await vi.advanceTimersByTimeAsync(30_000); // retry 2 152 152 + await vi.advanceTimersByTimeAsync(60_000); // no more retries 153 153 + 154 154 + expect(mockFetch).toHaveBeenCalledTimes(3); 155 155 + }); 156 156 + 157 157 + it("includes fetchContext in payload when provided", async () => { 158 158 + mockFetch.mockResolvedValueOnce({ status: 200 }); 159 159 + 160 160 + const fetchContext = { 161 161 + profile: { uri: "at://x/col/rk", cid: "c", record: { name: "Alice" } }, 162 162 + }; 163 163 + const match = makeMatch(); 164 164 + await dispatch(match, 0, fetchContext); 165 165 + 166 166 + const body = JSON.parse(mockFetch.mock.calls[0]![1].body); 167 167 + expect(body.fetches).toEqual(fetchContext); 168 168 + }); 169 169 + 170 170 + it("omits fetches from payload when context is empty", async () => { 171 171 + mockFetch.mockResolvedValueOnce({ status: 200 }); 172 172 + 173 173 + const match = makeMatch(); 174 174 + await dispatch(match, 0, {}); 175 175 + 176 176 + const body = JSON.parse(mockFetch.mock.calls[0]![1].body); 177 177 + expect(body.fetches).toBeUndefined(); 178 178 + }); 179 179 + });

+4

lib/webhooks/dispatcher.ts

reviewed

··· 1 1 import { db } from "../db/index.js"; 2 2 import { deliveryLogs, type WebhookAction } from "../db/schema.js"; 3 3 import { sign } from "./signer.js"; 4 4 + import { assertPublicUrl } from "../url-guard.js"; 4 5 import type { MatchedEvent } from "../jetstream/consumer.js"; 5 6 import type { FetchContext } from "../actions/template.js"; 6 7 ··· 59 60 const signature = sign(body, secret); 60 61 61 62 try { 63 63 + // Re-validate URL at delivery time to prevent DNS rebinding 64 64 + await assertPublicUrl(callbackUrl); 65 65 + 62 66 const res = await fetch(callbackUrl, { 63 67 method: "POST", 64 68 headers: {

+29

lib/webhooks/signer.test.ts

reviewed

··· 1 1 + import { describe, it, expect } from "vitest"; 2 2 + import { createHmac } from "node:crypto"; 3 3 + import { sign } from "./signer.js"; 4 4 + 5 5 + describe("sign", () => { 6 6 + it("returns correct HMAC-SHA256 hex digest", () => { 7 7 + const body = '{"test":"data"}'; 8 8 + const secret = "my-secret"; 9 9 + const expected = createHmac("sha256", secret).update(body).digest("hex"); 10 10 + expect(sign(body, secret)).toBe(expected); 11 11 + }); 12 12 + 13 13 + it("returns consistent results for same inputs", () => { 14 14 + expect(sign("body", "secret")).toBe(sign("body", "secret")); 15 15 + }); 16 16 + 17 17 + it("produces different signatures for different bodies", () => { 18 18 + expect(sign("body-a", "secret")).not.toBe(sign("body-b", "secret")); 19 19 + }); 20 20 + 21 21 + it("produces different signatures for different secrets", () => { 22 22 + expect(sign("body", "secret-a")).not.toBe(sign("body", "secret-b")); 23 23 + }); 24 24 + 25 25 + it("handles empty body and secret", () => { 26 26 + const result = sign("", ""); 27 27 + expect(result).toMatch(/^[a-f0-9]{64}$/); 28 28 + }); 29 29 + });

+2 -1

package.json

reviewed

··· 18 18 "hono": "^4.12.12", 19 19 "honox": "^0.1.55", 20 20 "nanoid": "^5.1.7", 21 21 - "vite": "npm:@voidzero-dev/vite-plus-core@latest" 21 21 + "vite": "npm:@voidzero-dev/vite-plus-core@latest", 22 22 + "vitest": "^0.1.16" 22 23 }, 23 24 "devDependencies": { 24 25 "@types/better-sqlite3": "^7.6.13",

+3

vite.config.ts

reviewed

··· 175 175 } 176 176 177 177 export default defineConfig({ 178 178 + test: { 179 179 + silent: "passed-only", 180 180 + }, 178 181 fmt: {}, 179 182 lint: { options: { typeAware: true, typeCheck: true } }, 180 183 server: {