feat: events with idempotency and fan-out, delivery reads, and CORS (steps 3.4–3.6)

api/package.json

@@ -14,6 +14,7 @@
     "db:down": "bash scripts/test-db.sh down"
   },
   "dependencies": {
+    "@fastify/cors": "11.3.1",
     "dbmate": "2.37.0",
     "fastify": "5.12.7",
     "ipaddr.js": "2.5.0",

api/src/app.ts

@@ -2,13 +2,16 @@
 // It never listens on a port itself. server.ts does that, so tests can call
 // buildApp() and send fake requests with app.inject().
 import Fastify, { LogController, type FastifyInstance } from "fastify";
+import cors from "@fastify/cors";
 import { realLookup, type Lookup } from "./ssrf.ts";
 import type { Sql } from "./db.ts";
 import { v1Routes } from "./routes/v1.ts";
 import type { Config } from "./config.ts";
 import { errorHandler, notFoundHandler } from "./error-handler.ts";
 import { healthRoutes } from "./routes/health.ts";
 
+const DASHBOARD_ORIGINS = ["https://cdn.myrelay.tech", "http://localhost:4173"];
+
 export function buildApp(
   config: Config,
   deps: { sql: Sql; lookup?: Lookup },
@@ -41,7 +44,18 @@ export function buildApp(
   app.setErrorHandler(errorHandler);
   app.setNotFoundHandler(notFoundHandler);
 
-  // Step 2: only the v1 plugin inherits the authentication hook.
+  // Step 2: preflight runs at the root before the authenticated v1 routes.
+  app.register(cors, {
+    origin: DASHBOARD_ORIGINS,
+    methods: ["GET", "POST", "DELETE"],
+    allowedHeaders: ["authorization", "content-type", "idempotency-key"],
+    exposedHeaders: ["x-request-id", "retry-after", "idempotent-replayed"],
+    maxAge: 600,
+    credentials: false,
+    // Otherwise OPTIONS without preflight headers gets the plugin's plain-text 400.
+    strictPreflight: false,
+  });
+  // Step 3: only the v1 plugin inherits the authentication hook.
   app.register(healthRoutes, { config });
   app.register(v1Routes, {
     prefix: "/v1", sql: deps.sql, lookup: deps.lookup ?? realLookup, allowOrigins: config.SSRF_ALLOW_ORIGINS,

api/src/cors.test.ts

@@ -0,0 +1,52 @@
+// Checks browser preflights at the root without needing a database connection.
+import { afterAll, expect, test } from "bun:test";
+import { buildApp } from "./app.ts";
+import { loadConfig } from "./config.ts";
+import { createDb } from "./db.ts";
+
+const config = loadConfig({ DATABASE_URL: "postgres://relay:x@127.0.0.1:1/relay" });
+const sql = createDb(config);
+const app = buildApp(config, { sql }, { logger: false });
+afterAll(async () => { await app.close(); await sql.end({ timeout: 5 }); });
+const origin = "https://cdn.myrelay.tech";
+// Step 1: a browser's unauthenticated preflight finishes before the v1 auth hook.
+for (const allowed of [origin, "http://localhost:4173"]) {
+  test(`preflight permits ${allowed} and the retry-key header`, async () => {
+    const res = await app.inject({ method: "OPTIONS", url: "/v1/events", headers: {
+      origin: allowed, "access-control-request-method": "POST",
+      "access-control-request-headers": "authorization,content-type,idempotency-key",
+    } });
+    expect(res.statusCode).toBe(204);
+    expect(res.headers["access-control-allow-origin"]).toBe(allowed);
+    expect(res.headers["access-control-allow-headers"]).toContain("idempotency-key");
+    expect(res.headers["access-control-max-age"]).toBe("600");
+    expect(res.headers["access-control-allow-credentials"]).toBeUndefined();
+  });
+}
+test("foreign preflight has no allow-origin header", async () => {
+  const res = await app.inject({ method: "OPTIONS", url: "/v1/events", headers: {
+    origin: "https://evil.example", "access-control-request-method": "POST",
+  } });
+  expect(res.headers["access-control-allow-origin"]).toBeUndefined();
+});
+test("OPTIONS without Origin is 204 without an allow-origin header", async () => {
+  const res = await app.inject({ method: "OPTIONS", url: "/v1/events" });
+  expect(res.statusCode).toBe(204);
+  expect(res.headers["access-control-allow-origin"]).toBeUndefined();
+});
+// Step 2: browser code can read even an authentication failure.
+test("401 includes the permitted origin and exposed response headers", async () => {
+  const res = await app.inject({ url: "/v1/me", headers: { origin } });
+  expect(res.statusCode).toBe(401);
+  expect(res.headers["access-control-allow-origin"]).toBe(origin);
+  for (const header of ["x-request-id", "retry-after", "idempotent-replayed"]) {
+    expect(res.headers["access-control-expose-headers"]).toContain(header);
+  }
+});
+test("health probes still answer without authentication", async () => {
+  for (const url of ["/health/live", "/health/ready"]) {
+    const res = await app.inject({ url });
+    expect(res.statusCode).toBe(200);
+    expect(res.json()).toEqual({ status: "ok" });
+  }
+});

api/src/repositories/deliveries.ts

@@ -0,0 +1,70 @@
+// Fans events out and reads deliveries and attempts within their owner's boundary.
+import type postgres from "postgres";
+import type { Sql } from "../db.ts";
+
+export type DeliveryRow = {
+  id: string; status: "pending" | "delivered" | "dead"; attempts: number;
+  next_attempt_at: Date | null; last_status_code: number | null; last_error: string | null;
+  delivered_at: Date | null; created_at: Date;
+  event_id: string; event_type: string; endpoint_id: string; endpoint_name: string;
+};
+export type DetailRow = DeliveryRow & {
+  payload: Record<string, unknown>; event_created_at: Date; endpoint_url: string;
+};
+export type AttemptRow = {
+  id: string; number: number; started_at: Date; duration_ms: number;
+  status_code: number | null; error: string | null; response_excerpt: string | null;
+};
+export type DeliveryFilters = {
+  status?: DeliveryRow["status"]; endpointId?: string; cursor?: string; limit: number;
+};
+
+// Step 1: column defaults create pending deliveries with no attempts.
+export async function fanOut(tx: postgres.TransactionSql, ownerId: string, eventId: string, type: string) {
+  const rows = await tx`
+    INSERT INTO deliveries (owner_id, event_id, endpoint_id)
+    SELECT ${ownerId}, ${eventId}, e.id FROM endpoints e
+    WHERE e.owner_id = ${ownerId} AND e.deleted_at IS NULL
+      AND (${type} = ANY(e.event_types) OR '*' = ANY(e.event_types))`;
+  return rows.count;
+}
+
+// Step 2: the column is never NULL; during a lease it holds the lease's expiry.
+// The API shows it only for pending deliveries. Both read queries share this rule.
+const nextAttempt = (db: Sql) => db`CASE WHEN d.status = 'pending' THEN d.next_attempt_at END AS next_attempt_at`;
+const columns = (db: Sql) => db`
+  d.id, d.status, d.attempts, ${nextAttempt(db)}, d.last_status_code, d.last_error,
+  d.delivered_at, d.created_at, ev.id AS event_id, ev.type AS event_type,
+  e.id AS endpoint_id, e.name AS endpoint_name`;
+
+export async function listDeliveries(db: Sql, ownerId: string, filters: DeliveryFilters): Promise<DeliveryRow[]> {
+  const { status, endpointId, cursor, limit } = filters;
+  return db<DeliveryRow[]>`
+    SELECT ${columns(db)} FROM deliveries d
+    JOIN events ev ON ev.id = d.event_id AND ev.owner_id = d.owner_id
+    JOIN endpoints e ON e.id = d.endpoint_id AND e.owner_id = d.owner_id
+    WHERE d.owner_id = ${ownerId}
+      ${status ? db`AND d.status = ${status}` : db``}
+      ${endpointId ? db`AND d.endpoint_id = ${endpointId}` : db``}
+      ${cursor ? db`AND d.id < ${cursor}` : db``}
+    ORDER BY d.id DESC LIMIT ${limit + 1}`;
+}
+
+// Step 3: deleted endpoints stay available in historical delivery details.
+export async function getDelivery(db: Sql, ownerId: string, id: string): Promise<DetailRow | null> {
+  const [row] = await db<DetailRow[]>`
+    SELECT ${columns(db)}, ev.payload, ev.created_at AS event_created_at, e.url AS endpoint_url
+    FROM deliveries d
+    JOIN events ev ON ev.id = d.event_id AND ev.owner_id = d.owner_id
+    JOIN endpoints e ON e.id = d.endpoint_id AND e.owner_id = d.owner_id
+    WHERE d.owner_id = ${ownerId} AND d.id = ${id}`;
+  return row ?? null;
+}
+
+export async function listAttempts(db: Sql, ownerId: string, deliveryId: string): Promise<AttemptRow[]> {
+  return db<AttemptRow[]>`
+    SELECT a.id, row_number() OVER (ORDER BY a.started_at, a.id)::int AS number,
+      a.started_at, a.duration_ms, a.status_code, a.error, a.response_excerpt
+    FROM attempts a JOIN deliveries d ON d.id = a.delivery_id AND d.owner_id = ${ownerId}
+    WHERE a.delivery_id = ${deliveryId} ORDER BY a.started_at, a.id`;
+}

api/src/repositories/events.ts

@@ -0,0 +1,36 @@
+// Stores events and retrieves the original answer for an owner's retry key.
+import type postgres from "postgres";
+import { AppError } from "../errors.ts";
+import type { EventInput } from "../schemas/events.ts";
+
+type Tx = postgres.TransactionSql;
+type EventRow = { id: string; type: string; created_at: Date };
+
+// Step 1: the unique constraint chooses the winner, including concurrent requests.
+export async function insertEvent(tx: Tx, ownerId: string, input: EventInput, key?: string): Promise<EventRow | null> {
+  try {
+    const [row] = await tx<EventRow[]>`
+      INSERT INTO events (owner_id, type, payload, idempotency_key)
+      VALUES (${ownerId}, ${input.type}, ${tx.json(input.payload as postgres.JSONValue)}, ${key ?? null})
+      ON CONFLICT (owner_id, idempotency_key) DO NOTHING RETURNING id, type, created_at`;
+    return row ?? null;
+  } catch (error) {
+    const code = (error as { code?: unknown } | null)?.code;
+    if (code === "22P05" || code === "22P02") {
+      throw new AppError(422, "validation_failed", "Request validation failed", [
+        { path: "payload", message: "Payload contains characters that can't be stored" },
+      ]);
+    }
+    throw error;
+  }
+}
+
+// Step 2: a separate statement sees the winning transaction under READ COMMITTED.
+export async function findEventByKey(tx: Tx, ownerId: string, key: string) {
+  const [row] = await tx<(EventRow & { deliveries: number })[]>`
+    SELECT e.id, e.type, e.created_at, count(d.id)::int AS deliveries
+    FROM events e LEFT JOIN deliveries d ON d.event_id = e.id AND d.owner_id = e.owner_id
+    WHERE e.owner_id = ${ownerId} AND e.idempotency_key = ${key}
+    GROUP BY e.id, e.type, e.created_at`;
+  return row ?? null;
+}

api/src/routes/deliveries.ts

@@ -0,0 +1,24 @@
+// Lists deliveries and reads one delivery after checking filters and ids.
+import type { FastifyPluginAsync } from "fastify";
+import type { Sql } from "../db.ts";
+import { notFound, unauthorized } from "../errors.ts";
+import { DeliveryId, DeliveryListQuery } from "../schemas/deliveries.ts";
+import { getDelivery, listDeliveries } from "../services/deliveries.ts";
+
+export const deliveryRoutes: FastifyPluginAsync<{ sql: Sql }> = async (app, { sql }) => {
+  // Step 1: query input uses public names; repository filters use local camelCase.
+  app.get("/deliveries", async (request) => {
+    if (!request.owner) throw unauthorized();
+    const query = DeliveryListQuery.parse(request.query);
+    return listDeliveries(sql, request.owner.id, {
+      status: query.status, endpointId: query.endpoint_id, cursor: query.cursor, limit: query.limit,
+    });
+  });
+  // Step 2: malformed ids have the same 404 as missing and foreign ids.
+  app.get<{ Params: { id: string } }>("/deliveries/:id", async (request) => {
+    if (!request.owner) throw unauthorized();
+    const id = DeliveryId.safeParse(request.params.id);
+    if (!id.success) throw notFound("Delivery");
+    return getDelivery(sql, request.owner.id, id.data);
+  });
+};

api/src/routes/events.ts

@@ -0,0 +1,20 @@
+// Accepts validated event requests and marks answers returned from a retry key.
+import type { FastifyPluginAsync } from "fastify";
+import { z } from "zod";
+import type { Sql } from "../db.ts";
+import { unauthorized } from "../errors.ts";
+import { EventCreate, IdempotencyKey } from "../schemas/events.ts";
+import { createEvent } from "../services/events.ts";
+
+// Step 1: parse the header in an object so validation names its exact public path.
+const Headers = z.object({ "idempotency-key": IdempotencyKey });
+export const eventRoutes: FastifyPluginAsync<{ sql: Sql }> = async (app, { sql }) => {
+  app.post("/events", async (request, reply) => {
+    if (!request.owner) throw unauthorized();
+    const input = EventCreate.parse(request.body);
+    const headers = Headers.parse({ "idempotency-key": request.headers["idempotency-key"] });
+    const { event, deliveries, replayed } = await createEvent(sql, request.owner.id, input, headers["idempotency-key"]);
+    if (replayed) reply.header("idempotent-replayed", "true");
+    return reply.code(202).send({ id: event.id, type: event.type, created_at: event.created_at.toISOString(), deliveries });
+  });
+};

api/src/routes/v1.ts

@@ -5,10 +5,14 @@ import type { Sql } from "../db.ts";
 import type { Lookup } from "../ssrf.ts";
 import { endpointRoutes } from "./endpoints.ts";
 import { meRoutes } from "./me.ts";
+import { eventRoutes } from "./events.ts";
+import { deliveryRoutes } from "./deliveries.ts";
 
 // Step 1: child routes inherit this hook through Fastify encapsulation.
 export const v1Routes: FastifyPluginAsync<{ sql: Sql; lookup: Lookup; allowOrigins: string[] }> = async (app, { sql, lookup, allowOrigins }) => {
   app.addHook("onRequest", authenticate(sql));
   app.register(meRoutes);
   app.register(endpointRoutes, { sql, lookup, allowOrigins });
+  app.register(eventRoutes, { sql });
+  app.register(deliveryRoutes, { sql });
 };

api/src/schemas/deliveries.ts

@@ -0,0 +1,12 @@
+// Checks delivery filters and cursor input without silently accepting numeric formats.
+import { z } from "zod";
+
+// Step 1: only digit text becomes a page size, with a bounded default.
+export const DeliveryListQuery = z.object({
+  status: z.enum(["pending", "delivered", "dead"]).optional(),
+  endpoint_id: z.uuid().optional(),
+  limit: z.string().regex(/^[0-9]{1,3}$/).transform(Number)
+    .pipe(z.number().int().min(1).max(100)).default(50),
+  cursor: z.uuid().optional(),
+}).strict();
+export const DeliveryId = z.uuid();

api/src/schemas/events.ts

@@ -0,0 +1,10 @@
+// Checks event envelopes and optional retry keys before they reach the service.
+import { z } from "zod";
+
+// Step 1: payloads must be objects; unknown envelope fields are rejected.
+export const EventCreate = z.object({
+  type: z.string().regex(/^[a-z0-9][a-z0-9_.-]{0,99}$/),
+  payload: z.record(z.string(), z.unknown()),
+}).strict();
+export const IdempotencyKey = z.string().min(1).max(255).regex(/^[^\p{Cc}]+$/u).optional();
+export type EventInput = z.infer<typeof EventCreate>;

api/src/services/deliveries.ts

@@ -0,0 +1,37 @@
+// Maps database rows to public delivery pages and delivery history.
+import type { Sql } from "../db.ts";
+import { notFound } from "../errors.ts";
+import * as repository from "../repositories/deliveries.ts";
+
+// Step 1: explicitly map public fields, including nullable values and ISO timestamps.
+function delivery(row: repository.DeliveryRow) {
+  return {
+    id: row.id, status: row.status, attempts: row.attempts,
+    next_attempt_at: row.next_attempt_at?.toISOString() ?? null,
+    last_status_code: row.last_status_code, last_error: row.last_error,
+    delivered_at: row.delivered_at?.toISOString() ?? null, created_at: row.created_at.toISOString(),
+    event: { id: row.event_id, type: row.event_type },
+    endpoint: { id: row.endpoint_id, name: row.endpoint_name },
+  };
+}
+export async function listDeliveries(sql: Sql, ownerId: string, filters: repository.DeliveryFilters) {
+  const rows = await repository.listDeliveries(sql, ownerId, filters);
+  const data = rows.slice(0, filters.limit).map(delivery);
+  return { data, next_cursor: rows.length > filters.limit ? data.at(-1)!.id : null };
+}
+
+// Step 2: a missing or foreign delivery has one answer; history never reveals secret fields.
+export async function getDelivery(sql: Sql, ownerId: string, id: string) {
+  const row = await repository.getDelivery(sql, ownerId, id);
+  if (!row) throw notFound("Delivery");
+  const attempts = await repository.listAttempts(sql, ownerId, id);
+  return {
+    ...delivery(row),
+    event: { id: row.event_id, type: row.event_type, payload: row.payload, created_at: row.event_created_at.toISOString() },
+    endpoint: { id: row.endpoint_id, name: row.endpoint_name, url: row.endpoint_url },
+    attempt_log: attempts.map((a) => ({
+      id: a.id, number: a.number, started_at: a.started_at.toISOString(), duration_ms: a.duration_ms,
+      status_code: a.status_code, error: a.error, response_excerpt: a.response_excerpt,
+    })),
+  };
+}

api/src/services/events.ts

@@ -0,0 +1,23 @@
+// Saves an event and its deliveries atomically, or returns the original retry answer.
+import type { Sql } from "../db.ts";
+import { fanOut as repositoryFanOut } from "../repositories/deliveries.ts";
+import { findEventByKey, insertEvent } from "../repositories/events.ts";
+import type { EventInput } from "../schemas/events.ts";
+
+// Step 1: an injected fan-out lets tests prove a failure rolls back the whole transaction.
+export async function createEvent(
+  sql: Sql, ownerId: string, input: EventInput, key?: string,
+  { fanOut = repositoryFanOut }: { fanOut?: typeof repositoryFanOut } = {},
+) {
+  return sql.begin(async (tx) => {
+    const event = await insertEvent(tx, ownerId, input, key);
+    if (event) {
+      const deliveries = await fanOut(tx, ownerId, event.id, event.type);
+      return { event, deliveries, replayed: false };
+    }
+    // Step 2: the key wins, even with a different body; comparing bodies needs a new contract error code.
+    const original = key === undefined ? null : await findEventByKey(tx, ownerId, key);
+    if (!original) throw new Error("Conflicting event could not be found");
+    return { event: original, deliveries: original.deliveries, replayed: true };
+  });
+}