| 1 | import { afterEach, describe, expect, it, vi } from "vitest"; |
| 2 | // @ts-expect-error Node 22+ provides node:sqlite; Worker production code does not import it. |
| 3 | import { DatabaseSync } from "node:sqlite"; |
| 4 | import worker, { drainFirebaseCrashOutbox } from "./index"; |
| 5 | import type { Env } from "./env"; |
| 6 | import freshSchemaSQL from "../schema.sql?raw"; |
| 7 | import { resetFirebaseAuthForTests } from "./firebase_rtdb"; |
| 8 | import { |
| 9 | FIREBASE_ACTIVE_RESERVATION_BYTES, |
| 10 | FIREBASE_STORAGE_BUDGET_BYTES, |
| 11 | acquireFirebaseGroupLease, |
| 12 | dueFirebaseCrashes, |
| 13 | purgeFirebaseDeliveryState, |
| 14 | releaseFirebaseGroupLease, |
| 15 | reserveFirebaseGroup, |
| 16 | } from "./crash_delivery"; |
| 17 | |
| 18 | const oauthURL = "https://oauth2.googleapis.com/token"; |
| 19 | |
| 20 | type SQLiteD1Statement = D1PreparedStatement & { execute(): D1Result }; |
| 21 | |
| 22 | function sqliteD1(db: DatabaseSync): D1Database { |
| 23 | return { |
| 24 | prepare(sql: string) { |
| 25 | let binds: unknown[] = []; |
| 26 | const statement = { |
| 27 | bind(...values: unknown[]) { binds = values; return statement; }, |
| 28 | async first<T>() { return (db.prepare(sql).get(...binds) ?? null) as T | null; }, |
| 29 | async all<T>() { return { success: true, results: db.prepare(sql).all(...binds) as T[], meta: {} }; }, |
| 30 | async run() { return statement.execute(); }, |
| 31 | execute() { |
| 32 | const result = db.prepare(sql).run(...binds); |
| 33 | return { success: true, results: [], meta: { changes: Number(result.changes) } } as unknown as D1Result; |
| 34 | }, |
| 35 | raw() { return Promise.resolve([]); }, |
| 36 | } as unknown as SQLiteD1Statement; |
| 37 | return statement; |
| 38 | }, |
| 39 | async batch(statements: D1PreparedStatement[]) { |
| 40 | db.exec("BEGIN IMMEDIATE"); |
| 41 | try { |
| 42 | const results = statements.map((statement) => (statement as SQLiteD1Statement).execute()); |
| 43 | db.exec("COMMIT"); |
| 44 | return results; |
| 45 | } catch (error) { |
| 46 | db.exec("ROLLBACK"); |
| 47 | throw error; |
| 48 | } |
| 49 | }, |
| 50 | } as unknown as D1Database; |
| 51 | } |
| 52 | |
| 53 | async function privateKeyPEM(): Promise<string> { |
| 54 | const pair = await crypto.subtle.generateKey( |
| 55 | { name: "RSASSA-PKCS1-v1_5", modulusLength: 2048, publicExponent: new Uint8Array([1, 0, 1]), hash: "SHA-256" }, |
| 56 | true, |
| 57 | ["sign", "verify"], |
| 58 | ) as CryptoKeyPair; |
| 59 | const bytes = new Uint8Array(await crypto.subtle.exportKey("pkcs8", pair.privateKey) as ArrayBuffer); |
| 60 | let binary = ""; |
| 61 | for (const byte of bytes) binary += String.fromCharCode(byte); |
| 62 | return `-----BEGIN PRIVATE KEY-----\n${btoa(binary).match(/.{1,64}/g)?.join("\n") ?? ""}\n-----END PRIVATE KEY-----`; |
| 63 | } |
| 64 | |
| 65 | async function integrationEnv(db: DatabaseSync): Promise<Env> { |
| 66 | return { |
| 67 | DB: sqliteD1(db), RATE_LIMITER: { async limit() { return { success: true }; } }, |
| 68 | CRASH_STORAGE_MODE: "firebase", |
| 69 | FIREBASE_DATABASE_URL: "https://reasonix-test.asia-southeast1.firebasedatabase.app", |
| 70 | FIREBASE_CLIENT_EMAIL: "crash-writer@example.iam.gserviceaccount.com", |
| 71 | FIREBASE_PRIVATE_KEY: await privateKeyPEM(), |
| 72 | } as unknown as Env; |
| 73 | } |
| 74 | |
| 75 | function reportRequest(eventId = "a".repeat(32)): Request { |
| 76 | const body = JSON.stringify({ |
| 77 | eventId, dedupKey: "b".repeat(64), installId: "c".repeat(32), kind: "crash", |
| 78 | version: "v1.25.0", os: "linux", arch: "amd64", |
| 79 | message: "panic at /home/alice/project/main.go:12", source: "go", label: "panic", |
| 80 | errorType: "runtime.error", topFrame: "main.go:12", |
| 81 | }); |
| 82 | return new Request("https://crash.reasonix.io/v1/report", { |
| 83 | method: "POST", |
| 84 | headers: { |
| 85 | "content-type": "application/json", "content-length": String(new TextEncoder().encode(body).byteLength), |
| 86 | "cf-connecting-ip": "127.0.0.1", |
| 87 | }, |
| 88 | body, |
| 89 | }); |
| 90 | } |
| 91 | |
| 92 | function firebaseStub() { |
| 93 | const values = new Map<string, unknown>(); |
| 94 | const etags = new Map<string, number>(); |
| 95 | const puts: Array<{ path: string; body: string }> = []; |
| 96 | let unavailable = false; |
| 97 | let beforeFirstPut: (() => Promise<void>) | undefined; |
| 98 | const pathOf = (url: string) => new URL(url).pathname.replace(/^\//, "").replace(/\.json$/, ""); |
| 99 | const fetcher = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => { |
| 100 | const url = String(input); |
| 101 | if (url === oauthURL) return Response.json({ access_token: "token", expires_in: 3600 }); |
| 102 | if (unavailable) return new Response("unavailable", { status: 503 }); |
| 103 | const path = pathOf(url); |
| 104 | const version = etags.get(path) ?? 1; |
| 105 | if ((init?.method ?? "GET") === "GET") { |
| 106 | return Response.json(values.get(path) ?? null, { headers: { etag: `"${version}"` } }); |
| 107 | } |
| 108 | if (init?.method === "PUT") { |
| 109 | if (beforeFirstPut) { const wait = beforeFirstPut; beforeFirstPut = undefined; await wait(); } |
| 110 | const match = new Headers(init.headers).get("If-Match"); |
| 111 | if (match !== `"${version}"`) return new Response(null, { status: 412 }); |
| 112 | const body = String(init.body); |
| 113 | puts.push({ path, body }); |
| 114 | values.set(path, JSON.parse(body) as unknown); |
| 115 | etags.set(path, version + 1); |
| 116 | return new Response(null, { status: 204 }); |
| 117 | } |
| 118 | if (init?.method === "DELETE") { values.delete(path); return new Response(null, { status: 204 }); } |
| 119 | return new Response(null, { status: 400 }); |
| 120 | }); |
| 121 | return { |
| 122 | values, puts, fetcher, |
| 123 | setUnavailable(value: boolean) { unavailable = value; }, |
| 124 | blockFirstPut(waiter: () => Promise<void>) { beforeFirstPut = waiter; }, |
| 125 | }; |
| 126 | } |
| 127 | |
| 128 | afterEach(() => { vi.unstubAllGlobals(); resetFirebaseAuthForTests(); }); |
| 129 | |
| 130 | describe("Firebase-primary crash ingest", () => { |
| 131 | it("fences stale lease release with an increasing generation", async () => { |
| 132 | const db = new DatabaseSync(":memory:"); |
| 133 | db.exec(freshSchemaSQL); |
| 134 | const env = await integrationEnv(db); |
| 135 | const fingerprint = "9".repeat(64); |
| 136 | db.prepare(`INSERT INTO firebase_crash_group_state ( |
| 137 | fingerprint, reserved_bytes, last_seen |
| 138 | ) VALUES (?, ?, ?)`).run(fingerprint, FIREBASE_ACTIVE_RESERVATION_BYTES, "2026-08-25T00:00:00Z"); |
| 139 | try { |
| 140 | const base = new Date("2026-08-25T00:00:00Z").getTime(); |
| 141 | for (let iteration = 0; iteration < 100; iteration++) { |
| 142 | const started = new Date(base + iteration * 180_000); |
| 143 | const first = await acquireFirebaseGroupLease(env, fingerprint, started); |
| 144 | expect(first?.generation).toBe(iteration * 2 + 1); |
| 145 | expect(await acquireFirebaseGroupLease(env, fingerprint, new Date(started.getTime() + 30_000))).toBeNull(); |
| 146 | const replacement = await acquireFirebaseGroupLease(env, fingerprint, new Date(started.getTime() + 61_000)); |
| 147 | expect(replacement?.generation).toBe(iteration * 2 + 2); |
| 148 | await releaseFirebaseGroupLease(env, fingerprint, first!); |
| 149 | expect(db.prepare("SELECT lease_owner FROM firebase_crash_group_state").get()) |
| 150 | .toEqual({ lease_owner: replacement!.owner }); |
| 151 | await releaseFirebaseGroupLease(env, fingerprint, replacement!); |
| 152 | } |
| 153 | expect(db.prepare("SELECT lease_owner, lease_generation FROM firebase_crash_group_state").get()) |
| 154 | .toEqual({ lease_owner: "", lease_generation: 200 }); |
| 155 | } finally { db.close(); } |
| 156 | }); |
| 157 | |
| 158 | it("reclaims old delivery rows and expired generation leases", async () => { |
| 159 | const db = new DatabaseSync(":memory:"); |
| 160 | db.exec(freshSchemaSQL); |
| 161 | const env = await integrationEnv(db); |
| 162 | try { |
| 163 | db.prepare(`INSERT INTO firebase_crash_outbox ( |
| 164 | event_id, fingerprint, payload, state, attempts, next_attempt_at, created_at, updated_at |
| 165 | ) VALUES (?, ?, '{}', 'processing', 0, ?, ?, ?)`).run( |
| 166 | "8".repeat(32), "7".repeat(64), "2000-01-01T00:00:00Z", "2000-01-01T00:00:00Z", "2000-01-01T00:00:00Z", |
| 167 | ); |
| 168 | db.prepare(`INSERT INTO firebase_crash_receipts ( |
| 169 | event_id, projected_at, group_count, latest_slot, first_sample |
| 170 | ) VALUES (?, ?, 1, 0, 1)`).run("6".repeat(32), "2000-01-01T00:00:00Z"); |
| 171 | db.prepare(`INSERT INTO firebase_crash_group_state ( |
| 172 | fingerprint, reserved_bytes, last_seen, lease_owner, lease_generation, lease_expires_at |
| 173 | ) VALUES (?, ?, ?, 'expired', 4, ?)`).run( |
| 174 | "5".repeat(64), FIREBASE_ACTIVE_RESERVATION_BYTES, "2000-01-01T00:00:00Z", "2000-01-01T00:00:00Z", |
| 175 | ); |
| 176 | expect(await dueFirebaseCrashes(env)).toHaveLength(1); |
| 177 | await purgeFirebaseDeliveryState(env); |
| 178 | expect(db.prepare("SELECT COUNT(*) AS count FROM firebase_crash_outbox").get()).toEqual({ count: 0 }); |
| 179 | expect(db.prepare("SELECT COUNT(*) AS count FROM firebase_crash_receipts").get()).toEqual({ count: 0 }); |
| 180 | expect(db.prepare("SELECT lease_owner, lease_generation FROM firebase_crash_group_state").get()) |
| 181 | .toEqual({ lease_owner: "", lease_generation: 4 }); |
| 182 | } finally { db.close(); } |
| 183 | }); |
| 184 | |
| 185 | it("stores no long-lived D1 sample and deduplicates a repeated eventId", async () => { |
| 186 | const db = new DatabaseSync(":memory:"); |
| 187 | db.exec(freshSchemaSQL); |
| 188 | const env = await integrationEnv(db); |
| 189 | const firebase = firebaseStub(); |
| 190 | vi.stubGlobal("fetch", firebase.fetcher); |
| 191 | try { |
| 192 | expect((await worker.fetch(reportRequest(), env)).status).toBe(202); |
| 193 | expect((await worker.fetch(reportRequest(), env)).status).toBe(202); |
| 194 | expect(db.prepare("SELECT count FROM groups").get()).toEqual({ count: 1 }); |
| 195 | expect(db.prepare("SELECT COUNT(*) AS count FROM reports").get()).toEqual({ count: 0 }); |
| 196 | expect(db.prepare("SELECT events FROM report_daily").get()).toEqual({ events: 1 }); |
| 197 | expect(db.prepare("SELECT COUNT(*) AS count FROM firebase_crash_outbox").get()).toEqual({ count: 0 }); |
| 198 | expect(db.prepare("SELECT group_count, latest_slot, first_sample FROM firebase_crash_receipts").get()) |
| 199 | .toEqual({ group_count: 1, latest_slot: 0, first_sample: 1 }); |
| 200 | expect(firebase.puts).toHaveLength(3); |
| 201 | const bodies = firebase.puts.map((write) => write.body).join("\n"); |
| 202 | expect(bodies).toContain("/home/_/project/main.go:12"); |
| 203 | expect(bodies).not.toContain("installId"); |
| 204 | expect(bodies).not.toContain("alice"); |
| 205 | } finally { db.close(); } |
| 206 | }); |
| 207 | |
| 208 | it("buffers Firebase failure and retries without double-counting D1", async () => { |
| 209 | const db = new DatabaseSync(":memory:"); |
| 210 | db.exec(freshSchemaSQL); |
| 211 | const env = await integrationEnv(db); |
| 212 | const firebase = firebaseStub(); |
| 213 | firebase.setUnavailable(true); |
| 214 | vi.stubGlobal("fetch", firebase.fetcher); |
| 215 | try { |
| 216 | expect((await worker.fetch(reportRequest("d".repeat(32)), env)).status).toBe(202); |
| 217 | expect(db.prepare("SELECT state FROM firebase_crash_outbox").get()).toEqual({ state: "projected" }); |
| 218 | firebase.setUnavailable(false); |
| 219 | db.prepare("UPDATE firebase_crash_outbox SET next_attempt_at = '2000-01-01T00:00:00Z'").run(); |
| 220 | await drainFirebaseCrashOutbox(env); |
| 221 | expect(db.prepare("SELECT COUNT(*) AS count FROM firebase_crash_outbox").get()).toEqual({ count: 0 }); |
| 222 | expect(db.prepare("SELECT count FROM groups").get()).toEqual({ count: 1 }); |
| 223 | } finally { db.close(); } |
| 224 | }); |
| 225 | |
| 226 | it("queues a same-group report while the current generation is writing", async () => { |
| 227 | const db = new DatabaseSync(":memory:"); |
| 228 | db.exec(freshSchemaSQL); |
| 229 | const env = await integrationEnv(db); |
| 230 | const firebase = firebaseStub(); |
| 231 | let release!: () => void; |
| 232 | const released = new Promise<void>((resolve) => { release = resolve; }); |
| 233 | let started!: () => void; |
| 234 | const firstPut = new Promise<void>((resolve) => { started = resolve; }); |
| 235 | firebase.blockFirstPut(async () => { started(); await released; }); |
| 236 | vi.stubGlobal("fetch", firebase.fetcher); |
| 237 | try { |
| 238 | const first = worker.fetch(reportRequest("1".repeat(32)), env); |
| 239 | await firstPut; |
| 240 | expect((await worker.fetch(reportRequest("2".repeat(32)), env)).status).toBe(202); |
| 241 | expect(db.prepare("SELECT state FROM firebase_crash_outbox WHERE event_id = ?").get("2".repeat(32))) |
| 242 | .toEqual({ state: "queued" }); |
| 243 | release(); |
| 244 | expect((await first).status).toBe(202); |
| 245 | await drainFirebaseCrashOutbox(env); |
| 246 | expect(db.prepare("SELECT count FROM groups").get()).toEqual({ count: 2 }); |
| 247 | expect(db.prepare("SELECT COUNT(*) AS count FROM firebase_crash_outbox").get()).toEqual({ count: 0 }); |
| 248 | } finally { release(); db.close(); } |
| 249 | }); |
| 250 | |
| 251 | it("returns 503 at the outbox cap and reclaims only the unused new reservation", async () => { |
| 252 | const db = new DatabaseSync(":memory:"); |
| 253 | db.exec(freshSchemaSQL); |
| 254 | db.exec(`WITH RECURSIVE n(value) AS ( |
| 255 | SELECT 1 UNION ALL SELECT value + 1 FROM n WHERE value < 5000 |
| 256 | ) INSERT INTO firebase_crash_outbox ( |
| 257 | event_id, fingerprint, payload, state, attempts, next_attempt_at, created_at, updated_at |
| 258 | ) SELECT printf('%032x', value), '${"e".repeat(64)}', '{}', 'queued', 0, |
| 259 | '2026-08-25T00:00:00Z', '2026-08-25T00:00:00Z', '2026-08-25T00:00:00Z' FROM n`); |
| 260 | const env = await integrationEnv(db); |
| 261 | try { |
| 262 | expect((await worker.fetch(reportRequest("f".repeat(32)), env)).status).toBe(503); |
| 263 | expect(db.prepare("SELECT COUNT(*) AS count FROM firebase_crash_group_state").get()).toEqual({ count: 0 }); |
| 264 | } finally { db.close(); } |
| 265 | }); |
| 266 | |
| 267 | it("enforces the 700 MiB reservation in the atomic INSERT/UPDATE statement", async () => { |
| 268 | const db = new DatabaseSync(":memory:"); |
| 269 | db.exec(freshSchemaSQL); |
| 270 | const env = await integrationEnv(db); |
| 271 | const padding = FIREBASE_STORAGE_BUDGET_BYTES - FIREBASE_ACTIVE_RESERVATION_BYTES + 1; |
| 272 | db.prepare(`INSERT INTO firebase_crash_group_state ( |
| 273 | fingerprint, reserved_bytes, last_seen |
| 274 | ) VALUES (?, ?, ?)`).run("1".repeat(64), padding, "2026-08-25T00:00:00Z"); |
| 275 | try { |
| 276 | expect(await reserveFirebaseGroup(env, "2".repeat(64), "2026-08-25T00:00:00Z")).toBe("full"); |
| 277 | expect(db.prepare("SELECT SUM(reserved_bytes) AS total FROM firebase_crash_group_state").get()) |
| 278 | .toEqual({ total: padding }); |
| 279 | db.prepare(`INSERT INTO groups ( |
| 280 | fingerprint, kind, count, first_seen, last_seen, last_version |
| 281 | ) VALUES (?, 'crash', 1, ?, ?, 'v1')`).run( |
| 282 | "3".repeat(64), "2026-08-25T00:00:00Z", "2026-08-25T00:00:00Z", |
| 283 | ); |
| 284 | expect(await reserveFirebaseGroup(env, "3".repeat(64), "2026-08-25T00:00:00Z")).toBe("full"); |
| 285 | } finally { db.close(); } |
| 286 | }); |
| 287 | }); |
| 288 |