1import { Database } from "bun:sqlite";2import { createHmac, timingSafeEqual } from "node:crypto";3import { WEBHOOK } from "openiap-commerce-protocol";4import { validate } from "./contract.mjs";5import { createErasureLedger } from "./erasure.mjs";6 7export function sign(secret, timestamp, body) {8 return (9 WEBHOOK.signaturePrefix +10 createHmac("sha256", secret)11 .update(`${timestamp}.`)12 .update(body)13 .digest("hex")14 );15}16 17export function authentic(secrets, timestamp, body, signatures, nowSeconds) {18 if (!/^\d+$/.test(timestamp ?? "")) return false;19 if (20 !Number.isSafeInteger(Number(timestamp)) ||21 Math.abs(nowSeconds - Number(timestamp)) > WEBHOOK.toleranceSeconds22 )23 return false;24 return (signatures ?? "").split(",").some((raw) => {25 const candidate = raw.trim();26 if (!/^v1=[a-f0-9]{64}$/.test(candidate)) return false;27 return secrets.some((secret) =>28 timingSafeEqual(29 Buffer.from(candidate),30 Buffer.from(sign(secret, timestamp, body)),31 ),32 );33 });34}35 36export function createReceiver(path, secret, now) {37 const emitters =38 typeof secret === "string" ? [{ name: "default", secret }] : secret;39 if (40 !Array.isArray(emitters) ||41 !emitters.length ||42 emitters.some(43 (entry) =>44 !entry.name ||45 !entry.secret ||46 (typeof secret !== "string" && !entry.projectId),47 )48 )49 throw new Error(50 "Configure each emitter with a name, project ID, and signing secret",51 );52 if (new Set(emitters.map((entry) => entry.name)).size !== emitters.length)53 throw new Error("Emitter names must be unique");54 const db = new Database(path, { create: true });55 db.exec(56 "CREATE TABLE IF NOT EXISTS inbox (event_id TEXT PRIMARY KEY, body TEXT NOT NULL)",57 );58 const erasures = createErasureLedger(db);59 // Upgrade old single-emitter event IDs without losing durable deduplication.60 db.transaction(() => {61 for (const row of db.query("SELECT event_id, body FROM inbox").all()) {62 const event = JSON.parse(row.body);63 if (row.event_id !== event.eventId) continue;64 const matching = emitters.filter(65 (entry) => entry.projectId === event.projectId,66 );67 const name = matching.length === 1 ? matching[0].name : "default";68 const identity = JSON.stringify([name, event.projectId, event.eventId]);69 db.query("INSERT OR IGNORE INTO inbox VALUES (?, ?)").run(70 identity,71 row.body,72 );73 db.query("DELETE FROM inbox WHERE event_id = ?").run(row.event_id);74 }75 })();76 77 async function fetch(request) {78 const bytes = new Uint8Array(await request.arrayBuffer());79 const authenticated = emitters.filter((emitter) =>80 authentic(81 [emitter.secret],82 request.headers.get(WEBHOOK.timestampHeader),83 bytes,84 request.headers.get(WEBHOOK.signatureHeader),85 Math.floor(now() / 1000),86 ),87 );88 if (!authenticated.length) {89 return new Response("Invalid signature", { status: 401 });90 }91 let body, event;92 try {93 body = new TextDecoder("utf-8", { fatal: true }).decode(bytes);94 event = JSON.parse(body);95 } catch {96 return new Response("Invalid JSON", { status: 400 });97 }98 if (!validate("#/$defs/CommerceEvent", event))99 return new Response("Invalid event", { status: 400 });100 if (request.headers.get(WEBHOOK.eventIdHeader) !== event.eventId)101 return new Response("Event ID mismatch", { status: 400 });102 const emitter = authenticated.find(103 (entry) => !entry.projectId || entry.projectId === event.projectId,104 );105 if (!emitter)106 return new Response("Unexpected emitter project", { status: 401 });107 if (event.userId && erasures.has(event.userId))108 return Response.json({ accepted: true, discarded: "erased-user" });109 // Inbox insertion is the durable effect; downstream jobs can consume it later.110 const result = db111 .query("INSERT OR IGNORE INTO inbox VALUES (?, ?)")112 .run(113 JSON.stringify([emitter.name, event.projectId, event.eventId]),114 body,115 );116 return Response.json({ accepted: true, duplicate: result.changes === 0 });117 }118 119 return {120 fetch,121 eraseUser(userId) {122 return db.transaction(() => {123 erasures.remember(userId);124 return db125 .query("DELETE FROM inbox WHERE json_extract(body, '$.userId') = ?")126 .run(userId).changes;127 })();128 },129 inspect: () =>130 db131 .query("SELECT body FROM inbox ORDER BY rowid")132 .all()133 .map((row) => JSON.parse(row.body)),134 count: () => db.query("SELECT count(*) AS count FROM inbox").get().count,135 close: () => db.close(),136 };137}138 139// The caller injects a loopback test transport. This is not a public HTTPS worker.140export async function deliver(provider, secret, now, post) {141 const results = [];142 const rows = provider.db143 .query(144 "SELECT * FROM outbox WHERE status = 'pending' AND next_at <= ? ORDER BY rowid",145 )146 .all(now());147 for (const candidate of rows) {148 const row = provider.db149 .query("SELECT * FROM outbox WHERE event_id = ? AND status = 'pending'")150 .get(candidate.event_id);151 if (!row) continue;152 const timestamp = Math.floor(now() / 1000).toString();153 const headers = {154 "content-type": WEBHOOK.contentType,155 [WEBHOOK.timestampHeader]: timestamp,156 [WEBHOOK.signatureHeader]: sign(secret, timestamp, row.body),157 [WEBHOOK.eventIdHeader]: row.event_id,158 [WEBHOOK.deliveryIdHeader]: row.delivery_id,159 };160 let status;161 try {162 status = (await post({ method: "POST", headers, body: row.body })).status;163 } catch {164 status = 503;165 }166 const attempts = row.attempts + 1;167 const retryable = status === 408 || status === 429 || status >= 500;168 const next =169 status >= 200 && status < 300170 ? "delivered"171 : retryable && attempts < 3172 ? "pending"173 : "dead-letter";174 provider.db175 .query(176 "UPDATE outbox SET attempts = ?, status = ?, next_at = ? WHERE event_id = ?",177 )178 .run(attempts, next, now() + 30_000 * 2 ** (attempts - 1), row.event_id);179 results.push({180 eventId: row.event_id,181 deliveryId: row.delivery_id,182 httpStatus: status,183 attempt: attempts,184 status: next,185 });186 }187 return results;188}189