diff --git a/apps/server/openapi.json b/apps/server/openapi.json index 5393a08..bc434ee 100644 --- a/apps/server/openapi.json +++ b/apps/server/openapi.json @@ -120,27 +120,6 @@ "updatedAt" ] }, - "WebhookSubscription": { - "type": "object", - "properties": { - "id": { - "type": "string" - }, - "ledgerId": { - "type": "string" - }, - "url": { - "type": "string" - }, - "createdAt": { - "type": "string" - }, - "updatedAt": { - "type": "string" - } - }, - "required": ["id", "ledgerId", "url", "createdAt", "updatedAt"] - }, "AvailabilityItem": { "type": "object", "properties": { @@ -3384,397 +3363,6 @@ } } }, - "/v1/ledgers/{ledgerId}/webhooks": { - "get": { - "tags": ["Webhooks"], - "summary": "List webhook subscriptions", - "parameters": [ - { - "schema": { - "type": "string" - }, - "required": true, - "name": "ledgerId", - "in": "path" - } - ], - "responses": { - "200": { - "description": "List of webhook subscriptions", - "content": { - "application/json": { - "schema": { - "type": "object", - "properties": { - "data": { - "type": "array", - "items": { - "type": "object", - "properties": { - "id": { - "type": "string" - }, - "ledgerId": { - "type": "string" - }, - "url": { - "type": "string" - }, - "createdAt": { - "type": "string" - }, - "updatedAt": { - "type": "string" - } - }, - "required": ["id", "ledgerId", "url", "createdAt", "updatedAt"] - } - } - }, - "required": ["data"] - } - } - } - } - } - }, - "post": { - "tags": ["Webhooks"], - "summary": "Create a webhook subscription", - "description": "Creates a new webhook subscription. The secret is only returned once at creation time.", - "parameters": [ - { - "schema": { - "type": "string" - }, - "required": true, - "name": "ledgerId", - "in": "path" - } - ], - "requestBody": { - "content": { - "application/json": { - "schema": { - "type": "object", - "properties": { - "url": { - "type": "string", - "format": "uri", - "example": "https://example.com/webhook" - } - }, - "required": ["url"] - } - } - } - }, - "responses": { - "201": { - "description": "Webhook subscription created (includes secret)", - "content": { - "application/json": { - "schema": { - "type": "object", - "properties": { - "data": { - "type": "object", - "properties": { - "id": { - "type": "string" - }, - "ledgerId": { - "type": "string" - }, - "url": { - "type": "string" - }, - "createdAt": { - "type": "string" - }, - "updatedAt": { - "type": "string" - }, - "secret": { - "type": "string" - } - }, - "required": ["id", "ledgerId", "url", "createdAt", "updatedAt", "secret"] - } - }, - "required": ["data"] - } - } - } - } - } - } - }, - "/v1/ledgers/{ledgerId}/webhooks/{id}": { - "patch": { - "tags": ["Webhooks"], - "summary": "Update a webhook subscription", - "parameters": [ - { - "schema": { - "type": "string" - }, - "required": true, - "name": "ledgerId", - "in": "path" - }, - { - "schema": { - "type": "string" - }, - "required": true, - "name": "id", - "in": "path" - } - ], - "requestBody": { - "content": { - "application/json": { - "schema": { - "type": "object", - "properties": { - "url": { - "type": "string", - "format": "uri" - } - } - } - } - } - }, - "responses": { - "200": { - "description": "Webhook subscription updated", - "content": { - "application/json": { - "schema": { - "type": "object", - "properties": { - "data": { - "type": "object", - "properties": { - "id": { - "type": "string" - }, - "ledgerId": { - "type": "string" - }, - "url": { - "type": "string" - }, - "createdAt": { - "type": "string" - }, - "updatedAt": { - "type": "string" - } - }, - "required": ["id", "ledgerId", "url", "createdAt", "updatedAt"] - } - }, - "required": ["data"] - } - } - } - }, - "404": { - "description": "Webhook subscription not found", - "content": { - "application/json": { - "schema": { - "type": "object", - "properties": { - "error": { - "type": "object", - "properties": { - "code": { - "type": "string" - }, - "message": { - "type": "string" - }, - "details": { - "type": "object", - "additionalProperties": {} - }, - "issues": { - "type": "array", - "items": {} - } - }, - "required": ["code", "message"] - } - }, - "required": ["error"] - } - } - } - } - } - }, - "delete": { - "tags": ["Webhooks"], - "summary": "Delete a webhook subscription", - "parameters": [ - { - "schema": { - "type": "string" - }, - "required": true, - "name": "ledgerId", - "in": "path" - }, - { - "schema": { - "type": "string" - }, - "required": true, - "name": "id", - "in": "path" - } - ], - "responses": { - "204": { - "description": "Webhook subscription deleted" - }, - "404": { - "description": "Webhook subscription not found", - "content": { - "application/json": { - "schema": { - "type": "object", - "properties": { - "error": { - "type": "object", - "properties": { - "code": { - "type": "string" - }, - "message": { - "type": "string" - }, - "details": { - "type": "object", - "additionalProperties": {} - }, - "issues": { - "type": "array", - "items": {} - } - }, - "required": ["code", "message"] - } - }, - "required": ["error"] - } - } - } - } - } - } - }, - "/v1/ledgers/{ledgerId}/webhooks/{id}/rotate-secret": { - "post": { - "tags": ["Webhooks"], - "summary": "Rotate webhook secret", - "description": "Generates a new secret for the webhook subscription. The old secret is invalidated immediately.", - "parameters": [ - { - "schema": { - "type": "string" - }, - "required": true, - "name": "ledgerId", - "in": "path" - }, - { - "schema": { - "type": "string" - }, - "required": true, - "name": "id", - "in": "path" - } - ], - "responses": { - "200": { - "description": "New secret generated (includes secret)", - "content": { - "application/json": { - "schema": { - "type": "object", - "properties": { - "data": { - "type": "object", - "properties": { - "id": { - "type": "string" - }, - "ledgerId": { - "type": "string" - }, - "url": { - "type": "string" - }, - "createdAt": { - "type": "string" - }, - "updatedAt": { - "type": "string" - }, - "secret": { - "type": "string" - } - }, - "required": ["id", "ledgerId", "url", "createdAt", "updatedAt", "secret"] - } - }, - "required": ["data"] - } - } - } - }, - "404": { - "description": "Webhook subscription not found", - "content": { - "application/json": { - "schema": { - "type": "object", - "properties": { - "error": { - "type": "object", - "properties": { - "code": { - "type": "string" - }, - "message": { - "type": "string" - }, - "details": { - "type": "object", - "additionalProperties": {} - }, - "issues": { - "type": "array", - "items": {} - } - }, - "required": ["code", "message"] - } - }, - "required": ["error"] - } - } - } - } - } - } - }, "/v1/ledgers/{ledgerId}/policies": { "get": { "tags": ["Policies"], diff --git a/apps/server/src/config/index.ts b/apps/server/src/config/index.ts index 33fbd48..80edc26 100644 --- a/apps/server/src/config/index.ts +++ b/apps/server/src/config/index.ts @@ -5,6 +5,8 @@ const schema = z.object({ PORT: z.coerce.number().default(4000), DATABASE_URL: z.url(), FLOYD_API_KEY: z.string().min(1).optional(), + FLOYD_EVENT_INGEST_URL: z.url().optional(), + FLOYD_ENGINE_SECRET: z.string().min(1).optional(), }); export const config = schema.parse(process.env); diff --git a/apps/server/src/database/schema.ts b/apps/server/src/database/schema.ts index 91cd242..9834eb2 100644 --- a/apps/server/src/database/schema.ts +++ b/apps/server/src/database/schema.ts @@ -1,9 +1,5 @@ import type { Generated, Insertable, Selectable, Updateable } from "kysely"; -import type { - BookingStatus, - IdempotencyStatus, - WebhookDeliveryStatus, -} from "@floyd-run/schema/types"; +import type { BookingStatus, IdempotencyStatus } from "@floyd-run/schema/types"; export interface LedgersTable { id: string; @@ -75,27 +71,17 @@ export interface IdempotencyKeysTable { createdAt: Generated; } -export interface WebhookSubscriptionsTable { +export interface OutboxEventsTable { id: string; ledgerId: string; - url: string; - secret: string; - createdAt: Generated; - updatedAt: Generated; -} - -export interface WebhookDeliveriesTable { - id: string; - subscriptionId: string; eventType: string; + schemaVersion: number; payload: Record; - status: WebhookDeliveryStatus; - attempts: number; - maxAttempts: number; - nextAttemptAt: Date | null; - lastError: string | null; - lastStatusCode: number | null; createdAt: Generated; + publishedAt: Date | null; + publishAttempts: number; + nextAttemptAt: Date | null; + lastPublishError: string | null; } export interface PoliciesTable { @@ -116,8 +102,7 @@ export interface Database { resources: ResourcesTable; ledgers: LedgersTable; policies: PoliciesTable; - webhookSubscriptions: WebhookSubscriptionsTable; - webhookDeliveries: WebhookDeliveriesTable; + outboxEvents: OutboxEventsTable; } export type LedgerRow = Selectable; @@ -145,12 +130,9 @@ export type BookingUpdate = Updateable; export type IdempotencyKeyRow = Selectable; export type NewIdempotencyKey = Insertable; -export type WebhookSubscriptionRow = Selectable; -export type NewWebhookSubscription = Insertable; -export type WebhookSubscriptionUpdate = Updateable; - -export type WebhookDeliveryRow = Selectable; -export type NewWebhookDelivery = Insertable; +export type OutboxEventRow = Selectable; +export type NewOutboxEvent = Insertable; +export type OutboxEventUpdate = Updateable; export type PolicyRow = Selectable; export type NewPolicy = Insertable; diff --git a/apps/server/src/index.ts b/apps/server/src/index.ts index 773fede..148b9b9 100644 --- a/apps/server/src/index.ts +++ b/apps/server/src/index.ts @@ -1,15 +1,15 @@ import { serve } from "@hono/node-server"; import { config } from "config"; import { logger } from "infra/logger"; -import { startWebhookWorker, stopWebhookWorker } from "./workers/webhook-worker"; import { startExpirationWorker, stopExpirationWorker } from "./workers/expiration-worker"; +import { startOutboxPublisher, stopOutboxPublisher } from "./workers/outbox-publisher"; async function main() { const { default: app } = await import("./app"); // Start background workers - startWebhookWorker(); startExpirationWorker(); + startOutboxPublisher(); const server = serve( { @@ -24,8 +24,8 @@ async function main() { // Handle graceful shutdown const shutdown = () => { logger.info("Shutting down..."); - stopWebhookWorker(); stopExpirationWorker(); + stopOutboxPublisher(); server.close(); }; diff --git a/apps/server/src/infra/crypto.ts b/apps/server/src/infra/crypto.ts new file mode 100644 index 0000000..7dee547 --- /dev/null +++ b/apps/server/src/infra/crypto.ts @@ -0,0 +1,11 @@ +import { createHmac } from "crypto"; + +/** + * Compute HMAC-SHA256 signature for a payload. + * Header format: `sha256=` + */ +export function computeHmacSignature(payload: string, secret: string): string { + const hmac = createHmac("sha256", secret); + hmac.update(payload); + return `sha256=${hmac.digest("hex")}`; +} diff --git a/apps/server/src/infra/event-bus.ts b/apps/server/src/infra/event-bus.ts new file mode 100644 index 0000000..59cdf45 --- /dev/null +++ b/apps/server/src/infra/event-bus.ts @@ -0,0 +1,60 @@ +import type { Transaction } from "kysely"; +import type { Database } from "database/schema"; +import { generateId } from "@floyd-run/utils"; + +// Internal event types for the event bus +export type InternalEventType = + | "allocation.created" + | "allocation.deleted" + | "booking.created" + | "booking.confirmed" + | "booking.canceled" + | "booking.expired"; + +export interface InternalEvent { + id: string; + type: InternalEventType; + ledgerId: string; + schemaVersion: number; + timestamp: string; + data: Record; +} + +/** + * Emit an internal event to the outbox for eventual delivery. + * MUST be called within the same transaction as the data mutation. + * + * @param trx - The transaction object (enforces transactional safety) + * @param type - The event type + * @param ledgerId - The ledger ID this event belongs to + * @param data - The event payload data (caller is responsible for serialization) + */ +export async function emitEvent( + trx: Transaction, + type: InternalEventType, + ledgerId: string, + data: Record, +): Promise { + const eventId = generateId("evt"); + + const event: InternalEvent = { + id: eventId, + type, + ledgerId, + schemaVersion: 1, + timestamp: new Date().toISOString(), + data, + }; + + await trx + .insertInto("outboxEvents") + .values({ + id: eventId, + ledgerId, + eventType: type, + schemaVersion: 1, + payload: event as unknown as Record, + publishAttempts: 0, + }) + .execute(); +} diff --git a/apps/server/src/infra/webhooks.ts b/apps/server/src/infra/webhooks.ts deleted file mode 100644 index c8adae6..0000000 --- a/apps/server/src/infra/webhooks.ts +++ /dev/null @@ -1,240 +0,0 @@ -import { createHmac } from "crypto"; -import type { Kysely, Transaction } from "kysely"; -import type { Database } from "database/schema"; -import { db } from "database"; -import { generateId } from "@floyd-run/utils"; - -// Event types for webhooks -export type WebhookEventType = - | "allocation.created" - | "allocation.deleted" - | "booking.created" - | "booking.confirmed" - | "booking.canceled" - | "booking.expired"; - -export interface WebhookEvent { - id: string; - type: WebhookEventType; - ledgerId: string; - createdAt: string; - data: Record; -} - -/** - * Enqueue webhook deliveries for all subscriptions that match the event. - * MUST be called within the same transaction as the data mutation. - * Call sites are responsible for serializing their own payload data. - */ -export async function enqueueWebhookEvent( - trx: Transaction | Kysely, - eventType: WebhookEventType, - ledgerId: string, - data: Record, -): Promise { - // Find all subscriptions for this ledger - const subscriptions = await trx - .selectFrom("webhookSubscriptions") - .selectAll() - .where("ledgerId", "=", ledgerId) - .execute(); - - if (subscriptions.length === 0) { - return; - } - - const now = new Date(); - - // Create a delivery record for each subscription - // The delivery ID serves as the event ID in the payload - const deliveries = subscriptions.map((sub) => { - const deliveryId = generateId("whd"); - const event: WebhookEvent = { - id: deliveryId, - type: eventType, - ledgerId, - createdAt: now.toISOString(), - data, - }; - return { - id: deliveryId, - subscriptionId: sub.id, - eventType, - payload: event as unknown as Record, - status: "pending" as const, - attempts: 0, - maxAttempts: 5, - nextAttemptAt: now, - lastError: null, - lastStatusCode: null, - }; - }); - - await trx.insertInto("webhookDeliveries").values(deliveries).execute(); -} - -/** - * Compute HMAC-SHA256 signature for webhook payload. - * Header format: `sha256=` - */ -export function computeWebhookSignature(payload: string, secret: string): string { - const hmac = createHmac("sha256", secret); - hmac.update(payload); - return `sha256=${hmac.digest("hex")}`; -} - -/** - * Process pending webhook deliveries. - * Should be called by a background worker. - */ -export async function processPendingDeliveries(batchSize = 10): Promise { - const now = new Date(); - - // Atomically claim a batch of pending deliveries using FOR UPDATE SKIP LOCKED - const deliveries = await db.transaction().execute(async (trx) => { - const rows = await trx - .selectFrom("webhookDeliveries") - .selectAll() - .where((eb) => eb.or([eb("status", "=", "pending"), eb("status", "=", "failed")])) - .where((eb) => eb.or([eb("nextAttemptAt", "is", null), eb("nextAttemptAt", "<=", now)])) - .orderBy("nextAttemptAt", "asc") - .limit(batchSize) - .forUpdate() - .skipLocked() - .execute(); - - if (rows.length === 0) return []; - - const ids = rows.map((d) => d.id); - await trx - .updateTable("webhookDeliveries") - .set({ status: "in_flight" }) - .where("id", "in", ids) - .execute(); - - return rows; - }); - - if (deliveries.length === 0) { - return 0; - } - - let processed = 0; - - for (const delivery of deliveries) { - try { - // Get subscription for URL and secret - const subscription = await db - .selectFrom("webhookSubscriptions") - .selectAll() - .where("id", "=", delivery.subscriptionId) - .executeTakeFirst(); - - if (!subscription) { - // Subscription deleted, mark as exhausted - await db - .updateTable("webhookDeliveries") - .set({ - status: "exhausted", - lastError: "Subscription not found", - }) - .where("id", "=", delivery.id) - .execute(); - processed++; - continue; - } - - const payloadString = JSON.stringify(delivery.payload); - const signature = computeWebhookSignature(payloadString, subscription.secret); - - // Attempt delivery - const response = await fetch(subscription.url, { - method: "POST", - headers: { - "Content-Type": "application/json", - "Floyd-Signature": signature, - }, - body: payloadString, - signal: AbortSignal.timeout(30000), // 30 second timeout - }); - - if (response.ok) { - // Success - await db - .updateTable("webhookDeliveries") - .set({ - status: "succeeded", - attempts: delivery.attempts + 1, - lastStatusCode: response.status, - lastError: null, - }) - .where("id", "=", delivery.id) - .execute(); - } else { - // HTTP error - schedule retry - await handleDeliveryFailure( - delivery.id, - delivery.attempts + 1, - delivery.maxAttempts, - `HTTP ${response.status}: ${response.statusText}`, - response.status, - ); - } - } catch (error) { - // Network error - schedule retry - const errorMessage = error instanceof Error ? error.message : "Unknown error"; - await handleDeliveryFailure( - delivery.id, - delivery.attempts + 1, - delivery.maxAttempts, - errorMessage, - null, - ); - } - - processed++; - } - - return processed; -} - -async function handleDeliveryFailure( - deliveryId: string, - attempts: number, - maxAttempts: number, - errorMessage: string, - statusCode: number | null, -): Promise { - if (attempts >= maxAttempts) { - // Exhausted all retries - await db - .updateTable("webhookDeliveries") - .set({ - status: "exhausted", - attempts, - lastError: errorMessage, - lastStatusCode: statusCode, - nextAttemptAt: null, - }) - .where("id", "=", deliveryId) - .execute(); - } else { - // Schedule retry with exponential backoff - // Delays: 1min, 5min, 30min, 2hr, 12hr - const delayMinutes = [1, 5, 30, 120, 720][attempts - 1] ?? 720; - const nextAttemptAt = new Date(); - nextAttemptAt.setMinutes(nextAttemptAt.getMinutes() + delayMinutes); - - await db - .updateTable("webhookDeliveries") - .set({ - status: "failed", - attempts, - lastError: errorMessage, - lastStatusCode: statusCode, - nextAttemptAt, - }) - .where("id", "=", deliveryId) - .execute(); - } -} diff --git a/apps/server/src/migrations/20260216111703_webhooks-to-internal-event-bus.ts b/apps/server/src/migrations/20260216111703_webhooks-to-internal-event-bus.ts new file mode 100644 index 0000000..f4d3b25 --- /dev/null +++ b/apps/server/src/migrations/20260216111703_webhooks-to-internal-event-bus.ts @@ -0,0 +1,47 @@ +import type { Database } from "database/schema"; +import { type Kysely, sql } from "kysely"; + +export async function up(db: Kysely): Promise { + // Drop old webhook tables (moved to cloud) + await db.schema.dropTable("webhook_deliveries").ifExists().execute(); + await db.schema.dropTable("webhook_subscriptions").ifExists().execute(); + + // Drop outbox_events if it exists (to ensure clean schema) + await db.schema.dropTable("outbox_events").ifExists().execute(); + + // Create outbox_events table for internal event bus + await db.schema + .createTable("outbox_events") + .addColumn("id", "text", (col) => col.primaryKey()) + .addColumn("ledger_id", "text", (col) => col.notNull()) + .addColumn("event_type", "text", (col) => col.notNull()) + .addColumn("schema_version", "integer", (col) => col.notNull().defaultTo(1)) + .addColumn("payload", "jsonb", (col) => col.notNull()) + .addColumn("created_at", "timestamptz", (col) => col.notNull().defaultTo(sql`NOW()`)) + .addColumn("published_at", "timestamptz") + .addColumn("publish_attempts", "integer", (col) => col.notNull().defaultTo(0)) + .addColumn("next_attempt_at", "timestamptz") + .addColumn("last_publish_error", "text") + .execute(); + + // Index for efficient polling of unpublished events ready for retry + await db.schema + .createIndex("outbox_events_pending_idx") + .on("outbox_events") + .columns(["next_attempt_at", "created_at"]) + .where(sql.ref("published_at"), "is", null) + .execute(); + + // Index for ledger-based queries + await db.schema + .createIndex("outbox_events_ledger_id_idx") + .on("outbox_events") + .column("ledger_id") + .execute(); +} + +// Intentionally irreversible: webhook tables are not recreated. +// The webhook system has been fully replaced by the internal event bus. +export async function down(db: Kysely): Promise { + await db.schema.dropTable("outbox_events").execute(); +} diff --git a/apps/server/src/operations/allocation/create.ts b/apps/server/src/operations/allocation/create.ts index b4db451..b5c23cd 100644 --- a/apps/server/src/operations/allocation/create.ts +++ b/apps/server/src/operations/allocation/create.ts @@ -2,7 +2,7 @@ import { db, getServerTime } from "database"; import { createOperation } from "lib/operation"; import { allocationInput } from "@floyd-run/schema/inputs"; import { NotFoundError } from "lib/errors"; -import { enqueueWebhookEvent } from "infra/webhooks"; +import { emitEvent } from "infra/event-bus"; import { serializeAllocation } from "routes/v1/serializers"; import { insertAllocation } from "./internal/insert"; @@ -40,8 +40,8 @@ export default createOperation({ serverTime, }); - // 4. Enqueue webhook event (in same transaction) - await enqueueWebhookEvent(trx, "allocation.created", input.ledgerId, { + // 4. Emit event to outbox (in same transaction) + await emitEvent(trx, "allocation.created", input.ledgerId, { allocation: serializeAllocation(allocation), }); diff --git a/apps/server/src/operations/allocation/remove.ts b/apps/server/src/operations/allocation/remove.ts index 91d6b18..2e96a1c 100644 --- a/apps/server/src/operations/allocation/remove.ts +++ b/apps/server/src/operations/allocation/remove.ts @@ -2,7 +2,7 @@ import { db } from "database"; import { createOperation } from "lib/operation"; import { allocationInput } from "@floyd-run/schema/inputs"; import { ConflictError, NotFoundError } from "lib/errors"; -import { enqueueWebhookEvent } from "infra/webhooks"; +import { emitEvent } from "infra/event-bus"; import { serializeAllocation } from "routes/v1/serializers"; export default createOperation({ @@ -32,8 +32,8 @@ export default createOperation({ // 3. Hard delete the allocation await trx.deleteFrom("allocations").where("id", "=", input.id).execute(); - // 4. Enqueue webhook event - await enqueueWebhookEvent(trx, "allocation.deleted", existing.ledgerId, { + // 4. Emit event to outbox + await emitEvent(trx, "allocation.deleted", existing.ledgerId, { allocation: serializeAllocation(existing), }); diff --git a/apps/server/src/operations/booking/cancel.ts b/apps/server/src/operations/booking/cancel.ts index fa01dd6..0dbb89c 100644 --- a/apps/server/src/operations/booking/cancel.ts +++ b/apps/server/src/operations/booking/cancel.ts @@ -2,7 +2,7 @@ import { db, getServerTime } from "database"; import { createOperation } from "lib/operation"; import { bookingInput } from "@floyd-run/schema/inputs"; import { ConflictError, NotFoundError } from "lib/errors"; -import { enqueueWebhookEvent } from "infra/webhooks"; +import { emitEvent } from "infra/event-bus"; import { serializeBooking } from "routes/v1/serializers"; export default createOperation({ @@ -68,8 +68,8 @@ export default createOperation({ .where("bookingId", "=", input.id) .execute(); - // 6. Enqueue webhook - await enqueueWebhookEvent(trx, "booking.canceled", booking.ledgerId, { + // 6. Emit event to outbox + await emitEvent(trx, "booking.canceled", booking.ledgerId, { booking: serializeBooking(booking, allocations), }); diff --git a/apps/server/src/operations/booking/confirm.ts b/apps/server/src/operations/booking/confirm.ts index d630b02..7baac4e 100644 --- a/apps/server/src/operations/booking/confirm.ts +++ b/apps/server/src/operations/booking/confirm.ts @@ -2,7 +2,7 @@ import { db, getServerTime } from "database"; import { createOperation } from "lib/operation"; import { bookingInput } from "@floyd-run/schema/inputs"; import { ConflictError, NotFoundError } from "lib/errors"; -import { enqueueWebhookEvent } from "infra/webhooks"; +import { emitEvent } from "infra/event-bus"; import { serializeBooking } from "routes/v1/serializers"; export default createOperation({ @@ -76,8 +76,8 @@ export default createOperation({ .where("bookingId", "=", input.id) .execute(); - // 7. Enqueue webhook - await enqueueWebhookEvent(trx, "booking.confirmed", booking.ledgerId, { + // 7. Emit event to outbox + await emitEvent(trx, "booking.confirmed", booking.ledgerId, { booking: serializeBooking(booking, allocations), }); diff --git a/apps/server/src/operations/booking/create.ts b/apps/server/src/operations/booking/create.ts index 8b3d2f8..77e89a3 100644 --- a/apps/server/src/operations/booking/create.ts +++ b/apps/server/src/operations/booking/create.ts @@ -3,7 +3,7 @@ import { createOperation } from "lib/operation"; import { generateId } from "@floyd-run/utils"; import { bookingInput } from "@floyd-run/schema/inputs"; import { ConflictError, NotFoundError } from "lib/errors"; -import { enqueueWebhookEvent } from "infra/webhooks"; +import { emitEvent } from "infra/event-bus"; import { serializeBooking } from "routes/v1/serializers"; import { evaluatePolicy, type PolicyConfig } from "domain/policy/evaluate"; import { insertAllocation } from "../allocation/internal/insert"; @@ -132,8 +132,8 @@ export default createOperation({ serverTime, }); - // 9. Enqueue webhook - await enqueueWebhookEvent(trx, "booking.created", input.ledgerId, { + // 9. Emit event to outbox + await emitEvent(trx, "booking.created", input.ledgerId, { booking: serializeBooking(booking, [allocation]), }); diff --git a/apps/server/src/operations/index.ts b/apps/server/src/operations/index.ts index 47f0547..cfc33c2 100644 --- a/apps/server/src/operations/index.ts +++ b/apps/server/src/operations/index.ts @@ -2,7 +2,6 @@ import { allocation } from "./allocation"; import { availability } from "./availability"; import { resource } from "./resource"; import { ledger } from "./ledger"; -import { webhook } from "./webhook"; import { policy } from "./policy"; import { service } from "./service"; import { booking } from "./booking"; @@ -12,7 +11,6 @@ export const operations = { availability, resource, ledger, - webhook, policy, service, booking, diff --git a/apps/server/src/operations/webhook/create.ts b/apps/server/src/operations/webhook/create.ts deleted file mode 100644 index 8dd9c6b..0000000 --- a/apps/server/src/operations/webhook/create.ts +++ /dev/null @@ -1,23 +0,0 @@ -import { db } from "database"; -import { createOperation } from "lib/operation"; -import { generateId } from "@floyd-run/utils"; -import { webhookInput } from "@floyd-run/schema/inputs"; -import { generateSecret } from "./generate-secret"; - -export default createOperation({ - input: webhookInput.createSubscription, - execute: async (input) => { - const subscription = await db - .insertInto("webhookSubscriptions") - .values({ - id: generateId("whs"), - ledgerId: input.ledgerId, - url: input.url, - secret: generateSecret(), - }) - .returningAll() - .executeTakeFirstOrThrow(); - - return { subscription }; - }, -}); diff --git a/apps/server/src/operations/webhook/generate-secret.ts b/apps/server/src/operations/webhook/generate-secret.ts deleted file mode 100644 index fc82957..0000000 --- a/apps/server/src/operations/webhook/generate-secret.ts +++ /dev/null @@ -1,5 +0,0 @@ -import { randomBytes } from "crypto"; - -export function generateSecret(): string { - return `whsec_${randomBytes(24).toString("base64url")}`; -} diff --git a/apps/server/src/operations/webhook/index.ts b/apps/server/src/operations/webhook/index.ts deleted file mode 100644 index 61bc304..0000000 --- a/apps/server/src/operations/webhook/index.ts +++ /dev/null @@ -1,13 +0,0 @@ -import create from "./create"; -import list from "./list"; -import update from "./update"; -import remove from "./remove"; -import rotateSecret from "./rotate-secret"; - -export const webhook = { - create, - list, - update, - remove, - rotateSecret, -}; diff --git a/apps/server/src/operations/webhook/list.ts b/apps/server/src/operations/webhook/list.ts deleted file mode 100644 index 233ed2d..0000000 --- a/apps/server/src/operations/webhook/list.ts +++ /dev/null @@ -1,17 +0,0 @@ -import { db } from "database"; -import { createOperation } from "lib/operation"; -import { webhookInput } from "@floyd-run/schema/inputs"; - -export default createOperation({ - input: webhookInput.listSubscriptions, - execute: async (input) => { - const subscriptions = await db - .selectFrom("webhookSubscriptions") - .selectAll() - .where("ledgerId", "=", input.ledgerId) - .orderBy("createdAt", "desc") - .execute(); - - return { subscriptions }; - }, -}); diff --git a/apps/server/src/operations/webhook/remove.ts b/apps/server/src/operations/webhook/remove.ts deleted file mode 100644 index 491f054..0000000 --- a/apps/server/src/operations/webhook/remove.ts +++ /dev/null @@ -1,16 +0,0 @@ -import { db } from "database"; -import { createOperation } from "lib/operation"; -import { webhookInput } from "@floyd-run/schema/inputs"; - -export default createOperation({ - input: webhookInput.deleteSubscription, - execute: async (input) => { - const result = await db - .deleteFrom("webhookSubscriptions") - .where("id", "=", input.id) - .where("ledgerId", "=", input.ledgerId) - .executeTakeFirst(); - - return { deleted: result.numDeletedRows > 0n }; - }, -}); diff --git a/apps/server/src/operations/webhook/rotate-secret.ts b/apps/server/src/operations/webhook/rotate-secret.ts deleted file mode 100644 index 28b2300..0000000 --- a/apps/server/src/operations/webhook/rotate-secret.ts +++ /dev/null @@ -1,21 +0,0 @@ -import { db } from "database"; -import { createOperation } from "lib/operation"; -import { webhookInput } from "@floyd-run/schema/inputs"; -import { generateSecret } from "./generate-secret"; - -export default createOperation({ - input: webhookInput.rotateSecret, - execute: async (input) => { - const newSecret = generateSecret(); - - const subscription = await db - .updateTable("webhookSubscriptions") - .set({ secret: newSecret }) - .where("id", "=", input.id) - .where("ledgerId", "=", input.ledgerId) - .returningAll() - .executeTakeFirst(); - - return { subscription: subscription ?? null, secret: newSecret }; - }, -}); diff --git a/apps/server/src/operations/webhook/update.ts b/apps/server/src/operations/webhook/update.ts deleted file mode 100644 index 932e45b..0000000 --- a/apps/server/src/operations/webhook/update.ts +++ /dev/null @@ -1,20 +0,0 @@ -import { db } from "database"; -import { createOperation } from "lib/operation"; -import { webhookInput } from "@floyd-run/schema/inputs"; - -export default createOperation({ - input: webhookInput.updateSubscription, - execute: async (input) => { - const subscription = await db - .updateTable("webhookSubscriptions") - .set({ - ...(input.url !== undefined && { url: input.url }), - }) - .where("id", "=", input.id) - .where("ledgerId", "=", input.ledgerId) - .returningAll() - .executeTakeFirst(); - - return { subscription: subscription ?? null }; - }, -}); diff --git a/apps/server/src/routes/v1/index.ts b/apps/server/src/routes/v1/index.ts index d122765..3b462f6 100644 --- a/apps/server/src/routes/v1/index.ts +++ b/apps/server/src/routes/v1/index.ts @@ -3,7 +3,6 @@ import { allocations } from "./allocations"; import { availability } from "./availability"; import { resources } from "./resources"; import { ledgers } from "./ledgers"; -import { webhooks } from "./webhooks"; import { policies } from "./policies"; import { services } from "./services"; import { bookings } from "./bookings"; @@ -13,7 +12,6 @@ export const v1 = new Hono() .route("/ledgers/:ledgerId/resources", resources) .route("/ledgers/:ledgerId/allocations", allocations) .route("/ledgers/:ledgerId/availability", availability) - .route("/ledgers/:ledgerId/webhooks", webhooks) .route("/ledgers/:ledgerId/policies", policies) .route("/ledgers/:ledgerId/services", services) .route("/ledgers/:ledgerId/bookings", bookings); diff --git a/apps/server/src/routes/v1/serializers.ts b/apps/server/src/routes/v1/serializers.ts index caadfdc..4c8171a 100644 --- a/apps/server/src/routes/v1/serializers.ts +++ b/apps/server/src/routes/v1/serializers.ts @@ -2,7 +2,6 @@ import type { AllocationRow, ResourceRow, LedgerRow, - WebhookSubscriptionRow, PolicyRow, ServiceRow, BookingRow, @@ -54,24 +53,6 @@ export function serializeAllocation(allocation: AllocationRow): Allocation { }; } -export interface WebhookSubscription { - id: string; - ledgerId: string; - url: string; - createdAt: string; - updatedAt: string; -} - -export function serializeWebhookSubscription(sub: WebhookSubscriptionRow): WebhookSubscription { - return { - id: sub.id, - ledgerId: sub.ledgerId, - url: sub.url, - createdAt: sub.createdAt.toISOString(), - updatedAt: sub.updatedAt.toISOString(), - }; -} - export function serializePolicy(policy: PolicyRow): Policy { return { id: policy.id, diff --git a/apps/server/src/routes/v1/webhooks.ts b/apps/server/src/routes/v1/webhooks.ts deleted file mode 100644 index b9c9121..0000000 --- a/apps/server/src/routes/v1/webhooks.ts +++ /dev/null @@ -1,85 +0,0 @@ -/* eslint-disable @typescript-eslint/no-unsafe-assignment */ -/* eslint-disable @typescript-eslint/no-unsafe-argument */ -import { Hono } from "hono"; -import { operations } from "operations"; -import { NotFoundError } from "lib/errors"; -import { serializeWebhookSubscription } from "./serializers"; - -// Nested under /v1/ledgers/:ledgerId/webhooks -export const webhooks = new Hono() - // List subscriptions - .get("/", async (c) => { - const { subscriptions } = await operations.webhook.list({ - ledgerId: c.req.param("ledgerId")!, - }); - return c.json({ data: subscriptions.map(serializeWebhookSubscription) }); - }) - - // Create subscription - .post("/", async (c) => { - const body = await c.req.json(); - const { subscription } = await operations.webhook.create({ - ...body, - ledgerId: c.req.param("ledgerId")!, - }); - - // Return with secret (only shown once at creation) - return c.json( - { - data: { - ...serializeWebhookSubscription(subscription), - secret: subscription.secret, - }, - }, - 201, - ); - }) - - // Update subscription - .patch("/:id", async (c) => { - const body = await c.req.json(); - const { subscription } = await operations.webhook.update({ - ...body, - id: c.req.param("id"), - ledgerId: c.req.param("ledgerId")!, - }); - - if (!subscription) { - throw new NotFoundError("Webhook subscription not found"); - } - - return c.json({ data: serializeWebhookSubscription(subscription) }); - }) - - // Delete subscription - .delete("/:id", async (c) => { - const { deleted } = await operations.webhook.remove({ - id: c.req.param("id"), - ledgerId: c.req.param("ledgerId")!, - }); - - if (!deleted) { - throw new NotFoundError("Webhook subscription not found"); - } - - return c.body(null, 204); - }) - - // Rotate secret - .post("/:id/rotate-secret", async (c) => { - const { subscription, secret } = await operations.webhook.rotateSecret({ - id: c.req.param("id"), - ledgerId: c.req.param("ledgerId")!, - }); - - if (!subscription) { - throw new NotFoundError("Webhook subscription not found"); - } - - return c.json({ - data: { - ...serializeWebhookSubscription(subscription), - secret, - }, - }); - }); diff --git a/apps/server/src/scripts/generate-openapi.ts b/apps/server/src/scripts/generate-openapi.ts index d45ebea..6c1844e 100644 --- a/apps/server/src/scripts/generate-openapi.ts +++ b/apps/server/src/scripts/generate-openapi.ts @@ -6,7 +6,6 @@ import { allocation, resource, ledger, - webhook, policy, error, availability, @@ -20,7 +19,6 @@ const registry = new OpenAPIRegistry(); registry.register("Ledger", ledger.base); registry.register("Resource", resource.base); registry.register("Allocation", allocation.base); -registry.register("WebhookSubscription", webhook.subscription); registry.register("AvailabilityItem", availability.item); registry.register("TimelineBlock", availability.timelineBlock); registry.register("Slot", availability.slot); @@ -686,116 +684,6 @@ registry.registerPath({ }, }); -// Webhook routes -registry.registerPath({ - method: "get", - path: "/v1/ledgers/{ledgerId}/webhooks", - tags: ["Webhooks"], - summary: "List webhook subscriptions", - request: { params: z.object({ ledgerId: z.string() }) }, - responses: { - 200: { - description: "List of webhook subscriptions", - content: { "application/json": { schema: webhook.listSubscriptions } }, - }, - }, -}); - -registry.registerPath({ - method: "post", - path: "/v1/ledgers/{ledgerId}/webhooks", - tags: ["Webhooks"], - summary: "Create a webhook subscription", - description: - "Creates a new webhook subscription. The secret is only returned once at creation time.", - request: { - params: z.object({ ledgerId: z.string() }), - body: { - content: { - "application/json": { - schema: z.object({ - url: z.url().openapi({ example: "https://example.com/webhook" }), - }), - }, - }, - }, - }, - responses: { - 201: { - description: "Webhook subscription created (includes secret)", - content: { "application/json": { schema: webhook.createSubscription } }, - }, - }, -}); - -registry.registerPath({ - method: "patch", - path: "/v1/ledgers/{ledgerId}/webhooks/{id}", - tags: ["Webhooks"], - summary: "Update a webhook subscription", - request: { - params: z.object({ ledgerId: z.string(), id: z.string() }), - body: { - content: { - "application/json": { - schema: z.object({ - url: z.url().optional(), - }), - }, - }, - }, - }, - responses: { - 200: { - description: "Webhook subscription updated", - content: { "application/json": { schema: webhook.updateSubscription } }, - }, - 404: { - description: "Webhook subscription not found", - content: { "application/json": { schema: error.schema } }, - }, - }, -}); - -registry.registerPath({ - method: "delete", - path: "/v1/ledgers/{ledgerId}/webhooks/{id}", - tags: ["Webhooks"], - summary: "Delete a webhook subscription", - request: { - params: z.object({ ledgerId: z.string(), id: z.string() }), - }, - responses: { - 204: { description: "Webhook subscription deleted" }, - 404: { - description: "Webhook subscription not found", - content: { "application/json": { schema: error.schema } }, - }, - }, -}); - -registry.registerPath({ - method: "post", - path: "/v1/ledgers/{ledgerId}/webhooks/{id}/rotate-secret", - tags: ["Webhooks"], - summary: "Rotate webhook secret", - description: - "Generates a new secret for the webhook subscription. The old secret is invalidated immediately.", - request: { - params: z.object({ ledgerId: z.string(), id: z.string() }), - }, - responses: { - 200: { - description: "New secret generated (includes secret)", - content: { "application/json": { schema: webhook.rotateSecret } }, - }, - 404: { - description: "Webhook subscription not found", - content: { "application/json": { schema: error.schema } }, - }, - }, -}); - // Policy routes registry.registerPath({ method: "get", diff --git a/apps/server/src/workers/expiration-worker.ts b/apps/server/src/workers/expiration-worker.ts index 43205e0..6fe3631 100644 --- a/apps/server/src/workers/expiration-worker.ts +++ b/apps/server/src/workers/expiration-worker.ts @@ -1,6 +1,6 @@ import { db, getServerTime } from "database"; import { logger } from "infra/logger"; -import { enqueueWebhookEvent } from "infra/webhooks"; +import { emitEvent } from "infra/event-bus"; import { serializeBooking } from "routes/v1/serializers"; const POLL_INTERVAL_MS = 5000; // 5 seconds @@ -65,9 +65,9 @@ async function processExpiredBookings(): Promise { allocationsByBookingId.set(allocation.bookingId!, group); } - // Enqueue webhook events + // Emit events to outbox for (const booking of expiredBookings) { - await enqueueWebhookEvent(trx, "booking.expired", booking.ledgerId, { + await emitEvent(trx, "booking.expired", booking.ledgerId, { booking: serializeBooking( { ...booking, status: "expired" as const, expiresAt: null, updatedAt: serverTime }, allocationsByBookingId.get(booking.id) ?? [], diff --git a/apps/server/src/workers/outbox-publisher.ts b/apps/server/src/workers/outbox-publisher.ts new file mode 100644 index 0000000..54e408a --- /dev/null +++ b/apps/server/src/workers/outbox-publisher.ts @@ -0,0 +1,278 @@ +import { db } from "database"; +import { logger } from "infra/logger"; +import type { InternalEvent } from "infra/event-bus"; +import { computeHmacSignature } from "infra/crypto"; +import { config } from "config"; + +const POLL_INTERVAL_MS = 1000; // 1 second +const BATCH_SIZE = 50; +const MAX_ATTEMPTS = 25; + +// Exponential backoff schedule (in seconds) +// 0s, 5s, 15s, 1m, 5m, 15m, 30m, 1h, 2h, 4h, then capped at 4h +const BACKOFF_SCHEDULE = [0, 5, 15, 60, 300, 900, 1800, 3600, 7200, 14400]; +const MAX_BACKOFF_SECONDS = 14400; // 4 hours +const JITTER_FACTOR = 0.3; // ±30% jitter + +let isRunning = false; +let publisherBackoffUntil: Date | null = null; + +/** + * Classify HTTP error as retryable or non-retryable + * @internal Exported for testing + */ +export function isRetryableError(statusCode: number | null): boolean { + if (!statusCode) { + // Network errors (ECONNRESET, ETIMEDOUT, DNS failures, etc.) + return true; + } + + // Retryable HTTP status codes + if (statusCode === 408 || statusCode === 429 || statusCode >= 500) { + return true; + } + + // Non-retryable: 4xx errors (except 408 and 429) + // These indicate misconfiguration or permanent failures + return false; +} + +/** + * Compute next retry time using exponential backoff with jitter + * Returns null if error is non-retryable + * @internal Exported for testing + */ +export function computeNextAttemptAt(attempt: number, statusCode: number | null): Date | null { + // Check if error is retryable + if (!isRetryableError(statusCode)) { + return null; // Don't retry non-retryable errors + } + + // Get base delay from schedule or use max + const baseDelaySeconds = + attempt < BACKOFF_SCHEDULE.length ? BACKOFF_SCHEDULE[attempt]! : MAX_BACKOFF_SECONDS; + + // Add jitter: random value between -30% and +30% of base delay + const jitterAmount = baseDelaySeconds * JITTER_FACTOR * (Math.random() * 2 - 1); + const delaySeconds = Math.max(0, baseDelaySeconds + jitterAmount); + + return new Date(Date.now() + delaySeconds * 1000); +} + +/** + * Extract HTTP status code from error message + * @internal Exported for testing + */ +export function extractStatusCode(error: Error): number | null { + const match = /HTTP (\d{3})/.exec(error.message); + return match ? parseInt(match[1]!, 10) : null; +} + +/** + * Publish a single event to the cloud via HTTP. + * Throws error with HTTP status for proper error classification. + */ +async function publishEvent(event: InternalEvent): Promise { + const ingestUrl = config.FLOYD_EVENT_INGEST_URL; + + if (!ingestUrl) { + throw new Error("FLOYD_EVENT_INGEST_URL not set"); + } + + const secret = config.FLOYD_ENGINE_SECRET; + const payload = JSON.stringify(event); + const signature = secret ? computeHmacSignature(payload, secret) : "sha256=unsigned"; + + const response = await fetch(ingestUrl, { + method: "POST", + headers: { + "Content-Type": "application/json", + "Floyd-Signature": signature, + }, + body: payload, + signal: AbortSignal.timeout(30000), // 30 second timeout + }); + + if (!response.ok) { + const errorText = await response.text().catch(() => "Unable to read error body"); + throw new Error(`HTTP ${response.status}: ${errorText}`); + } + + logger.debug({ eventId: event.id, ingestUrl }, "[outbox-publisher] Event published successfully"); +} + +async function processOutboxBatch(): Promise { + // Circuit breaker: if publisher is backed off, skip this cycle + if (publisherBackoffUntil && publisherBackoffUntil > new Date()) { + logger.debug( + { backoffUntil: publisherBackoffUntil }, + "[outbox-publisher] Publisher in backoff, skipping cycle", + ); + return; + } + + // FOR UPDATE SKIP LOCKED reduces duplicate delivery when concurrent workers + // poll at the same instant, but does not fully prevent it (locks release on + // auto-commit). Consumers must be idempotent. + const events = await db + .selectFrom("outboxEvents") + .selectAll() + .where("publishedAt", "is", null) + .where("publishAttempts", "<", MAX_ATTEMPTS) + .where((eb) => eb.or([eb("nextAttemptAt", "is", null), eb("nextAttemptAt", "<=", new Date())])) + .orderBy("createdAt", "asc") + .limit(BATCH_SIZE) + .forUpdate() + .skipLocked() + .execute(); + + if (events.length === 0) { + return; + } + + logger.debug({ count: events.length }, "[outbox-publisher] Processing batch"); + + let consecutiveFailures = 0; + + for (const event of events) { + // Track attempt count locally after incrementing to avoid stale reads + const currentAttempt = event.publishAttempts + 1; + + // Increment attempt counter + await db + .updateTable("outboxEvents") + .set({ publishAttempts: currentAttempt }) + .where("id", "=", event.id) + .execute(); + + try { + await publishEvent(event.payload as unknown as InternalEvent); + + // Success: mark as published + await db + .updateTable("outboxEvents") + .set({ + publishedAt: new Date(), + lastPublishError: null, + nextAttemptAt: null, + }) + .where("id", "=", event.id) + .execute(); + + logger.debug({ eventId: event.id }, "[outbox-publisher] Event published"); + consecutiveFailures = 0; // Reset on success + } catch (error) { + consecutiveFailures++; + const errorMessage = error instanceof Error ? error.message : "Unknown error"; + const statusCode = error instanceof Error ? extractStatusCode(error) : null; + const isRetryable = isRetryableError(statusCode); + + const nextAttemptAt = computeNextAttemptAt(currentAttempt, statusCode); + + // Check if we've exhausted retries or hit non-retryable error + if (!isRetryable || currentAttempt >= MAX_ATTEMPTS) { + logger.error( + { + eventId: event.id, + attempts: currentAttempt, + error: errorMessage, + statusCode, + retryable: isRetryable, + }, + isRetryable + ? "[outbox-publisher] Event exhausted max attempts" + : "[outbox-publisher] Event failed with non-retryable error", + ); + + // Mark as permanently failed (far-future date prevents retry) + await db + .updateTable("outboxEvents") + .set({ + lastPublishError: errorMessage, + nextAttemptAt: new Date("2099-12-31T23:59:59Z"), // Far future = permanently failed + }) + .where("id", "=", event.id) + .execute(); + + // For non-retryable errors, trigger circuit breaker + if (!isRetryable) { + publisherBackoffUntil = new Date(Date.now() + 30000); // Back off for 30 seconds + logger.warn( + { backoffUntil: publisherBackoffUntil }, + "[outbox-publisher] Non-retryable error, activating circuit breaker", + ); + break; // Stop processing more events + } + } else { + // Schedule retry + logger.warn( + { + eventId: event.id, + attempts: currentAttempt, + error: errorMessage, + statusCode, + nextAttemptAt, + }, + "[outbox-publisher] Event publish failed, will retry", + ); + + await db + .updateTable("outboxEvents") + .set({ + lastPublishError: errorMessage, + nextAttemptAt, + }) + .where("id", "=", event.id) + .execute(); + } + + // Circuit breaker: if we've had 3 consecutive failures, back off + if (consecutiveFailures >= 3) { + publisherBackoffUntil = new Date(Date.now() + 30000); // Back off for 30 seconds + logger.warn( + { consecutiveFailures, backoffUntil: publisherBackoffUntil }, + "[outbox-publisher] Multiple consecutive failures, activating circuit breaker", + ); + break; // Stop processing more events + } + } + } +} + +async function runWorker(): Promise { + if (isRunning) return; + isRunning = true; + + logger.info("[outbox-publisher] Starting outbox publisher worker..."); + + // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition + while (isRunning) { + try { + await processOutboxBatch(); + } catch (error) { + logger.error(error, "[outbox-publisher] Error processing outbox"); + } + + await new Promise((resolve) => setTimeout(resolve, POLL_INTERVAL_MS)); + } +} + +export function stopOutboxPublisher(): void { + logger.info("[outbox-publisher] Stopping outbox publisher worker..."); + isRunning = false; +} + +export function startOutboxPublisher(): void { + const ingestUrl = config.FLOYD_EVENT_INGEST_URL; + + if (!ingestUrl) { + logger.info("[outbox-publisher] FLOYD_EVENT_INGEST_URL not set, running in self-hosted mode"); + return; + } + + logger.info(`[outbox-publisher] Starting publisher, will ingest to ${ingestUrl}`); + + runWorker().catch((error: unknown) => { + logger.error(error, "[outbox-publisher] Fatal error"); + }); +} diff --git a/apps/server/src/workers/webhook-worker.ts b/apps/server/src/workers/webhook-worker.ts deleted file mode 100644 index 932a235..0000000 --- a/apps/server/src/workers/webhook-worker.ts +++ /dev/null @@ -1,39 +0,0 @@ -import { processPendingDeliveries } from "infra/webhooks"; -import { logger } from "infra/logger"; - -const POLL_INTERVAL_MS = 5000; // 5 seconds -const BATCH_SIZE = 10; - -let isRunning = false; - -async function runWorker(): Promise { - if (isRunning) return; - isRunning = true; - - logger.info("[webhook-worker] Starting webhook delivery worker..."); - - // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition - while (isRunning) { - try { - const processed = await processPendingDeliveries(BATCH_SIZE); - if (processed > 0) { - logger.info(`[webhook-worker] Processed ${processed} deliveries`); - } - } catch (error) { - logger.error(error, "[webhook-worker] Error processing deliveries"); - } - - await new Promise((resolve) => setTimeout(resolve, POLL_INTERVAL_MS)); - } -} - -export function stopWebhookWorker(): void { - logger.info("[webhook-worker] Stopping webhook delivery worker..."); - isRunning = false; -} - -export function startWebhookWorker(): void { - runWorker().catch((error: unknown) => { - logger.error(error, "[webhook-worker] Fatal error"); - }); -} diff --git a/apps/server/test/integration/setup/factories/index.ts b/apps/server/test/integration/setup/factories/index.ts index 73d9628..110510e 100644 --- a/apps/server/test/integration/setup/factories/index.ts +++ b/apps/server/test/integration/setup/factories/index.ts @@ -1,7 +1,6 @@ export * from "./allocation.factory"; export * from "./resource.factory"; export * from "./ledger.factory"; -export * from "./webhook.factory"; export * from "./policy.factory"; export * from "./service.factory"; export * from "./booking.factory"; diff --git a/apps/server/test/integration/setup/factories/webhook.factory.ts b/apps/server/test/integration/setup/factories/webhook.factory.ts deleted file mode 100644 index 46a9738..0000000 --- a/apps/server/test/integration/setup/factories/webhook.factory.ts +++ /dev/null @@ -1,65 +0,0 @@ -import { faker } from "@faker-js/faker"; -import { db } from "database"; -import { generateId } from "@floyd-run/utils"; -import { randomBytes } from "crypto"; -import { createLedger } from "./ledger.factory"; - -function generateSecret(): string { - return `whsec_${randomBytes(24).toString("base64url")}`; -} - -export async function createWebhookSubscription(overrides?: { - ledgerId?: string; - url?: string; - secret?: string; -}) { - let ledgerId = overrides?.ledgerId; - if (!ledgerId) { - const { ledger } = await createLedger(); - ledgerId = ledger.id; - } - - const subscription = await db - .insertInto("webhookSubscriptions") - .values({ - id: generateId("whs"), - ledgerId, - url: overrides?.url ?? faker.internet.url(), - secret: overrides?.secret ?? generateSecret(), - }) - .returningAll() - .executeTakeFirstOrThrow(); - - return { subscription, ledgerId }; -} - -export async function createWebhookDelivery(overrides: { - subscriptionId: string; - eventType?: string; - payload?: Record; - status?: "pending" | "in_flight" | "succeeded" | "failed" | "exhausted"; - attempts?: number; - maxAttempts?: number; - nextAttemptAt?: Date | null; - lastError?: string | null; - lastStatusCode?: number | null; -}) { - const delivery = await db - .insertInto("webhookDeliveries") - .values({ - id: generateId("whd"), - subscriptionId: overrides.subscriptionId, - eventType: overrides.eventType ?? "allocation.created", - payload: overrides.payload ?? { id: generateId("whd"), type: "allocation.created", data: {} }, - status: overrides.status ?? "pending", - attempts: overrides.attempts ?? 0, - maxAttempts: overrides.maxAttempts ?? 5, - nextAttemptAt: overrides.nextAttemptAt ?? new Date(), - lastError: overrides.lastError ?? null, - lastStatusCode: overrides.lastStatusCode ?? null, - }) - .returningAll() - .executeTakeFirstOrThrow(); - - return { delivery }; -} diff --git a/apps/server/test/integration/v1/allocations/create.spec.ts b/apps/server/test/integration/v1/allocations/create.spec.ts index 088f2c7..f68d436 100644 --- a/apps/server/test/integration/v1/allocations/create.spec.ts +++ b/apps/server/test/integration/v1/allocations/create.spec.ts @@ -2,6 +2,7 @@ import { describe, expect, it } from "vitest"; import { client } from "../../setup/client"; import { createResource } from "../../setup/factories"; import type { Allocation } from "@floyd-run/schema/types"; +import { db } from "database"; describe("POST /v1/ledgers/:ledgerId/allocations", () => { it("returns 201 for valid allocation", async () => { @@ -62,6 +63,44 @@ describe("POST /v1/ledgers/:ledgerId/allocations", () => { expect(data.metadata).toEqual(metadata); }); + it("creates allocation.created event in outbox", async () => { + const { resource, ledgerId } = await createResource(); + + const response = await client.post(`/v1/ledgers/${ledgerId}/allocations`, { + resourceId: resource.id, + startTime: new Date("2026-02-01T10:00:00Z").toISOString(), + endTime: new Date("2026-02-01T11:00:00Z").toISOString(), + }); + + expect(response.status).toBe(201); + const { data } = (await response.json()) as { data: Allocation }; + + // Verify event was created in outbox + const event = await db + .selectFrom("outboxEvents") + .selectAll() + .where("ledgerId", "=", ledgerId) + .where("eventType", "=", "allocation.created") + .orderBy("createdAt", "desc") + .executeTakeFirst(); + + expect(event).toBeDefined(); + expect(event?.eventType).toBe("allocation.created"); + expect(event?.ledgerId).toBe(ledgerId); + expect(event?.publishedAt).toBeNull(); // Not yet published + expect(event?.publishAttempts).toBe(0); + + // Verify payload contains allocation data + const payload = event?.payload as { + id: string; + type: string; + data: { allocation: Allocation }; + }; + expect(payload.type).toBe("allocation.created"); + expect(payload.data.allocation.id).toBe(data.id); + expect(payload.data.allocation.resourceId).toBe(resource.id); + }); + it("returns 422 for missing required fields", async () => { const { ledgerId } = await createResource(); diff --git a/apps/server/test/integration/v1/allocations/delete.spec.ts b/apps/server/test/integration/v1/allocations/delete.spec.ts index 67912eb..bdaef24 100644 --- a/apps/server/test/integration/v1/allocations/delete.spec.ts +++ b/apps/server/test/integration/v1/allocations/delete.spec.ts @@ -1,6 +1,8 @@ import { describe, expect, it } from "vitest"; import { client } from "../../setup/client"; import { createAllocation, createBooking, createLedger } from "../../setup/factories"; +import { db } from "database"; +import type { Allocation } from "@floyd-run/schema/types"; describe("DELETE /v1/ledgers/:ledgerId/allocations/:id", () => { it("returns 204 for successful deletion of raw allocation", async () => { @@ -11,6 +13,38 @@ describe("DELETE /v1/ledgers/:ledgerId/allocations/:id", () => { expect(response.status).toBe(204); }); + it("creates allocation.deleted event in outbox", async () => { + const { allocation, ledgerId } = await createAllocation(); + + const response = await client.delete(`/v1/ledgers/${ledgerId}/allocations/${allocation.id}`); + + expect(response.status).toBe(204); + + // Verify event was created in outbox + const event = await db + .selectFrom("outboxEvents") + .selectAll() + .where("ledgerId", "=", ledgerId) + .where("eventType", "=", "allocation.deleted") + .orderBy("createdAt", "desc") + .executeTakeFirst(); + + expect(event).toBeDefined(); + expect(event?.eventType).toBe("allocation.deleted"); + expect(event?.ledgerId).toBe(ledgerId); + expect(event?.publishedAt).toBeNull(); + expect(event?.publishAttempts).toBe(0); + + // Verify payload contains deleted allocation data + const payload = event?.payload as { + id: string; + type: string; + data: { allocation: Allocation }; + }; + expect(payload.type).toBe("allocation.deleted"); + expect(payload.data.allocation.id).toBe(allocation.id); + }); + it("allocation is gone after deletion", async () => { const { allocation, ledgerId } = await createAllocation(); diff --git a/apps/server/test/integration/v1/bookings/cancel.spec.ts b/apps/server/test/integration/v1/bookings/cancel.spec.ts index 21f0b1a..b295f07 100644 --- a/apps/server/test/integration/v1/bookings/cancel.spec.ts +++ b/apps/server/test/integration/v1/bookings/cancel.spec.ts @@ -2,6 +2,7 @@ import { describe, expect, it } from "vitest"; import { client } from "../../setup/client"; import { createLedger, createResource, createService } from "../../setup/factories"; import type { Booking } from "@floyd-run/schema/types"; +import { db } from "database"; describe("POST /v1/ledgers/:ledgerId/bookings/:id/cancel", () => { async function createHoldBooking(ledgerId: string) { @@ -42,6 +43,39 @@ describe("POST /v1/ledgers/:ledgerId/bookings/:id/cancel", () => { expect(meta.serverTime).toBeDefined(); }); + it("creates booking.canceled event in outbox", async () => { + const { ledger } = await createLedger(); + const holdBooking = await createHoldBooking(ledger.id); + + const response = await client.post( + `/v1/ledgers/${ledger.id}/bookings/${holdBooking.id}/cancel`, + ); + + expect(response.status).toBe(200); + const { data } = (await response.json()) as { data: Booking }; + + // Verify event was created in outbox + const event = await db + .selectFrom("outboxEvents") + .selectAll() + .where("ledgerId", "=", ledger.id) + .where("eventType", "=", "booking.canceled") + .executeTakeFirst(); + + expect(event).toBeDefined(); + expect(event?.eventType).toBe("booking.canceled"); + expect(event?.ledgerId).toBe(ledger.id); + expect(event?.publishedAt).toBeNull(); + expect(event?.publishAttempts).toBe(0); + + // Verify payload contains canceled booking data + const payload = event?.payload as { id: string; type: string; data: { booking: Booking } }; + expect(payload.type).toBe("booking.canceled"); + expect(payload.data.booking.id).toBe(data.id); + expect(payload.data.booking.status).toBe("canceled"); + expect(payload.data.booking.allocations[0]!.active).toBe(false); + }); + it("returns 200 when canceling a confirmed booking", async () => { const { ledger } = await createLedger(); const holdBooking = await createHoldBooking(ledger.id); diff --git a/apps/server/test/integration/v1/bookings/confirm.spec.ts b/apps/server/test/integration/v1/bookings/confirm.spec.ts index e5fdedb..ecdda45 100644 --- a/apps/server/test/integration/v1/bookings/confirm.spec.ts +++ b/apps/server/test/integration/v1/bookings/confirm.spec.ts @@ -2,6 +2,7 @@ import { describe, expect, it } from "vitest"; import { client } from "../../setup/client"; import { createLedger, createResource, createService } from "../../setup/factories"; import type { Booking } from "@floyd-run/schema/types"; +import { db } from "database"; describe("POST /v1/ledgers/:ledgerId/bookings/:id/confirm", () => { async function createHoldBooking(ledgerId: string) { @@ -41,6 +42,38 @@ describe("POST /v1/ledgers/:ledgerId/bookings/:id/confirm", () => { expect(meta.serverTime).toBeDefined(); }); + it("creates booking.confirmed event in outbox", async () => { + const { ledger } = await createLedger(); + const holdBooking = await createHoldBooking(ledger.id); + + const response = await client.post( + `/v1/ledgers/${ledger.id}/bookings/${holdBooking.id}/confirm`, + ); + + expect(response.status).toBe(200); + const { data } = (await response.json()) as { data: Booking }; + + // Verify event was created in outbox + const event = await db + .selectFrom("outboxEvents") + .selectAll() + .where("ledgerId", "=", ledger.id) + .where("eventType", "=", "booking.confirmed") + .executeTakeFirst(); + + expect(event).toBeDefined(); + expect(event?.eventType).toBe("booking.confirmed"); + expect(event?.ledgerId).toBe(ledger.id); + expect(event?.publishedAt).toBeNull(); + expect(event?.publishAttempts).toBe(0); + + // Verify payload contains confirmed booking data + const payload = event?.payload as { id: string; type: string; data: { booking: Booking } }; + expect(payload.type).toBe("booking.confirmed"); + expect(payload.data.booking.id).toBe(data.id); + expect(payload.data.booking.status).toBe("confirmed"); + }); + it("is idempotent — confirming an already confirmed booking returns same data", async () => { const { ledger } = await createLedger(); const holdBooking = await createHoldBooking(ledger.id); diff --git a/apps/server/test/integration/v1/bookings/create.spec.ts b/apps/server/test/integration/v1/bookings/create.spec.ts index 9130fc0..9f61b6b 100644 --- a/apps/server/test/integration/v1/bookings/create.spec.ts +++ b/apps/server/test/integration/v1/bookings/create.spec.ts @@ -8,6 +8,7 @@ import { createPolicy, } from "../../setup/factories"; import type { Booking } from "@floyd-run/schema/types"; +import { db } from "database"; describe("POST /v1/ledgers/:ledgerId/bookings", () => { it("returns 201 for valid hold booking", async () => { @@ -93,6 +94,45 @@ describe("POST /v1/ledgers/:ledgerId/bookings", () => { expect(data.metadata).toEqual(metadata); }); + it("creates booking.created event in outbox", async () => { + const { ledger } = await createLedger(); + const { resource } = await createResource({ ledgerId: ledger.id }); + const { service } = await createService({ + ledgerId: ledger.id, + resourceIds: [resource.id], + }); + + const response = await client.post(`/v1/ledgers/${ledger.id}/bookings`, { + serviceId: service.id, + resourceId: resource.id, + startTime: "2026-06-01T10:00:00.000Z", + endTime: "2026-06-01T11:00:00.000Z", + }); + + expect(response.status).toBe(201); + const { data } = (await response.json()) as { data: Booking }; + + // Verify event was created in outbox + const event = await db + .selectFrom("outboxEvents") + .selectAll() + .where("ledgerId", "=", ledger.id) + .where("eventType", "=", "booking.created") + .executeTakeFirst(); + + expect(event).toBeDefined(); + expect(event?.eventType).toBe("booking.created"); + expect(event?.ledgerId).toBe(ledger.id); + expect(event?.publishedAt).toBeNull(); // Not yet published + expect(event?.publishAttempts).toBe(0); + + // Verify payload contains booking data + const payload = event?.payload as { id: string; type: string; data: { booking: Booking } }; + expect(payload.type).toBe("booking.created"); + expect(payload.data.booking.id).toBe(data.id); + expect(payload.data.booking.serviceId).toBe(service.id); + }); + it("returns 422 for missing required fields", async () => { const { ledger } = await createLedger(); diff --git a/apps/server/test/integration/v1/webhooks/create.spec.ts b/apps/server/test/integration/v1/webhooks/create.spec.ts deleted file mode 100644 index fc5255a..0000000 --- a/apps/server/test/integration/v1/webhooks/create.spec.ts +++ /dev/null @@ -1,33 +0,0 @@ -import { describe, expect, it } from "vitest"; -import { client } from "../../setup/client"; -import { createLedger } from "../../setup/factories"; -import type { WebhookSubscription } from "routes/v1/serializers"; - -describe("POST /v1/ledgers/:ledgerId/webhooks", () => { - it("returns 201 for valid webhook subscription", async () => { - const { ledger } = await createLedger(); - - const response = await client.post(`/v1/ledgers/${ledger.id}/webhooks`, { - url: "https://example.com/webhook", - }); - - expect(response.status).toBe(201); - const { data } = (await response.json()) as { data: WebhookSubscription & { secret: string } }; - expect(data.id).toMatch(/^whs_/); - expect(data.ledgerId).toBe(ledger.id); - expect(data.url).toBe("https://example.com/webhook"); - expect(data.secret).toMatch(/^whsec_/); - expect(data.createdAt).toBeDefined(); - expect(data.updatedAt).toBeDefined(); - }); - - it("returns 422 for invalid url", async () => { - const { ledger } = await createLedger(); - - const response = await client.post(`/v1/ledgers/${ledger.id}/webhooks`, { - url: "not-a-url", - }); - - expect(response.status).toBe(422); - }); -}); diff --git a/apps/server/test/integration/v1/webhooks/delete.spec.ts b/apps/server/test/integration/v1/webhooks/delete.spec.ts deleted file mode 100644 index 312b224..0000000 --- a/apps/server/test/integration/v1/webhooks/delete.spec.ts +++ /dev/null @@ -1,29 +0,0 @@ -import { describe, expect, it } from "vitest"; -import { client } from "../../setup/client"; -import { createLedger, createWebhookSubscription } from "../../setup/factories"; - -describe("DELETE /v1/ledgers/:ledgerId/webhooks/:subscriptionId", () => { - it("returns 204 when deleting existing webhook", async () => { - const { ledger } = await createLedger(); - const { subscription } = await createWebhookSubscription({ ledgerId: ledger.id }); - - const response = await client.delete(`/v1/ledgers/${ledger.id}/webhooks/${subscription.id}`); - - expect(response.status).toBe(204); - - // Verify it's deleted by listing - const listResponse = await client.get(`/v1/ledgers/${ledger.id}/webhooks`); - const { data } = (await listResponse.json()) as { data: { id: string }[] }; - expect(data.find((w) => w.id === subscription.id)).toBeUndefined(); - }); - - it("returns 404 for non-existent webhook", async () => { - const { ledger } = await createLedger(); - - const response = await client.delete( - `/v1/ledgers/${ledger.id}/webhooks/whs_00000000000000000000000000`, - ); - - expect(response.status).toBe(404); - }); -}); diff --git a/apps/server/test/integration/v1/webhooks/list.spec.ts b/apps/server/test/integration/v1/webhooks/list.spec.ts deleted file mode 100644 index 6110650..0000000 --- a/apps/server/test/integration/v1/webhooks/list.spec.ts +++ /dev/null @@ -1,55 +0,0 @@ -import { describe, expect, it } from "vitest"; -import { client } from "../../setup/client"; -import { createLedger, createWebhookSubscription } from "../../setup/factories"; -import type { WebhookSubscription } from "routes/v1/serializers"; - -describe("GET /v1/ledgers/:ledgerId/webhooks", () => { - it("returns empty array when no webhooks exist", async () => { - const { ledger } = await createLedger(); - - const response = await client.get(`/v1/ledgers/${ledger.id}/webhooks`); - - expect(response.status).toBe(200); - const { data } = (await response.json()) as { data: WebhookSubscription[] }; - expect(data).toEqual([]); - }); - - it("returns all webhooks for ledger", async () => { - const { ledger } = await createLedger(); - await createWebhookSubscription({ ledgerId: ledger.id, url: "https://example.com/hook1" }); - await createWebhookSubscription({ ledgerId: ledger.id, url: "https://example.com/hook2" }); - - const response = await client.get(`/v1/ledgers/${ledger.id}/webhooks`); - - expect(response.status).toBe(200); - const { data } = (await response.json()) as { data: WebhookSubscription[] }; - expect(data).toHaveLength(2); - }); - - it("does not return webhooks from other ledgers", async () => { - const { ledger: ledger1 } = await createLedger(); - const { ledger: ledger2 } = await createLedger(); - await createWebhookSubscription({ ledgerId: ledger1.id }); - await createWebhookSubscription({ ledgerId: ledger2.id }); - - const response = await client.get(`/v1/ledgers/${ledger1.id}/webhooks`); - - expect(response.status).toBe(200); - const { data } = (await response.json()) as { data: WebhookSubscription[] }; - expect(data).toHaveLength(1); - expect(data[0]!.ledgerId).toBe(ledger1.id); - }); - - it("returns webhooks ordered by createdAt desc", async () => { - const { ledger } = await createLedger(); - const { subscription: first } = await createWebhookSubscription({ ledgerId: ledger.id }); - const { subscription: second } = await createWebhookSubscription({ ledgerId: ledger.id }); - - const response = await client.get(`/v1/ledgers/${ledger.id}/webhooks`); - - expect(response.status).toBe(200); - const { data } = (await response.json()) as { data: WebhookSubscription[] }; - expect(data[0]!.id).toBe(second.id); - expect(data[1]!.id).toBe(first.id); - }); -}); diff --git a/apps/server/test/integration/v1/webhooks/rotate-secret.spec.ts b/apps/server/test/integration/v1/webhooks/rotate-secret.spec.ts deleted file mode 100644 index 5b85944..0000000 --- a/apps/server/test/integration/v1/webhooks/rotate-secret.spec.ts +++ /dev/null @@ -1,31 +0,0 @@ -import { describe, expect, it } from "vitest"; -import { client } from "../../setup/client"; -import { createLedger, createWebhookSubscription } from "../../setup/factories"; -import type { WebhookSubscription } from "routes/v1/serializers"; - -describe("POST /v1/ledgers/:ledgerId/webhooks/:subscriptionId/rotate-secret", () => { - it("returns 200 with new secret", async () => { - const { ledger } = await createLedger(); - const { subscription } = await createWebhookSubscription({ ledgerId: ledger.id }); - const oldSecret = subscription.secret; - - const response = await client.post( - `/v1/ledgers/${ledger.id}/webhooks/${subscription.id}/rotate-secret`, - ); - - expect(response.status).toBe(200); - const { data } = (await response.json()) as { data: WebhookSubscription & { secret: string } }; - expect(data.secret).toMatch(/^whsec_/); - expect(data.secret).not.toBe(oldSecret); - }); - - it("returns 404 for non-existent webhook", async () => { - const { ledger } = await createLedger(); - - const response = await client.post( - `/v1/ledgers/${ledger.id}/webhooks/whs_00000000000000000000000000/rotate-secret`, - ); - - expect(response.status).toBe(404); - }); -}); diff --git a/apps/server/test/integration/v1/webhooks/update.spec.ts b/apps/server/test/integration/v1/webhooks/update.spec.ts deleted file mode 100644 index 684fc57..0000000 --- a/apps/server/test/integration/v1/webhooks/update.spec.ts +++ /dev/null @@ -1,30 +0,0 @@ -import { describe, expect, it } from "vitest"; -import { client } from "../../setup/client"; -import { createLedger, createWebhookSubscription } from "../../setup/factories"; -import type { WebhookSubscription } from "routes/v1/serializers"; - -describe("PATCH /v1/ledgers/:ledgerId/webhooks/:subscriptionId", () => { - it("returns 200 when updating url", async () => { - const { ledger } = await createLedger(); - const { subscription } = await createWebhookSubscription({ ledgerId: ledger.id }); - - const response = await client.patch(`/v1/ledgers/${ledger.id}/webhooks/${subscription.id}`, { - url: "https://new-url.com/webhook", - }); - - expect(response.status).toBe(200); - const { data } = (await response.json()) as { data: WebhookSubscription }; - expect(data.url).toBe("https://new-url.com/webhook"); - }); - - it("returns 404 for non-existent webhook", async () => { - const { ledger } = await createLedger(); - - const response = await client.patch( - `/v1/ledgers/${ledger.id}/webhooks/whs_00000000000000000000000000`, - { url: "https://new-url.com/webhook" }, - ); - - expect(response.status).toBe(404); - }); -}); diff --git a/apps/server/test/integration/workers/outbox-publisher.spec.ts b/apps/server/test/integration/workers/outbox-publisher.spec.ts new file mode 100644 index 0000000..88c0670 --- /dev/null +++ b/apps/server/test/integration/workers/outbox-publisher.spec.ts @@ -0,0 +1,430 @@ +import { describe, expect, it, beforeEach, vi, afterEach } from "vitest"; +import { db } from "database"; +import { generateId } from "@floyd-run/utils"; +import type { InternalEvent } from "infra/event-bus"; + +describe("Outbox Publisher Integration", () => { + let fetchMock: ReturnType; + let originalFetch: typeof global.fetch; + const testLedgerId = "ldg_test123"; + + beforeEach(async () => { + // Clean up all unpublished events to prevent interference from other integration tests + await db.deleteFrom("outboxEvents").where("publishedAt", "is", null).execute(); + + // Save original fetch + originalFetch = global.fetch; + + // Create fetch mock + fetchMock = vi.fn(); + global.fetch = fetchMock as unknown as typeof global.fetch; + + // Default: successful response + fetchMock.mockResolvedValue({ + ok: true, + status: 200, + text: () => Promise.resolve("OK"), + }); + + // Reset module cache to ensure clean imports + vi.resetModules(); + }); + + afterEach(async () => { + // Restore original fetch + global.fetch = originalFetch; + + // Clean up all unpublished events + await db.deleteFrom("outboxEvents").where("publishedAt", "is", null).execute(); + + // Clean up env vars + delete process.env["FLOYD_EVENT_INGEST_URL"]; + delete process.env["FLOYD_ENGINE_SECRET"]; + }); + + describe("Event Publishing Flow", () => { + it("publishes event to HTTP endpoint and marks as published", async () => { + // Arrange: Insert event into outbox + const eventId = generateId("evt"); + const event: InternalEvent = { + id: eventId, + type: "allocation.created", + ledgerId: testLedgerId, + schemaVersion: 1, + timestamp: new Date().toISOString(), + data: { test: "data" }, + }; + + await db + .insertInto("outboxEvents") + .values({ + id: eventId, + ledgerId: testLedgerId, + eventType: "allocation.created", + schemaVersion: 1, + payload: event as unknown as Record, + publishAttempts: 0, + nextAttemptAt: null, + }) + .execute(); + + // Set environment variable before import so config picks it up + process.env["FLOYD_EVENT_INGEST_URL"] = "https://test.example.com/ingest"; + + // Act: Import and run publisher (would normally run in background) + const { startOutboxPublisher, stopOutboxPublisher } = + await import("../../../src/workers/outbox-publisher"); + + // Give publisher a moment to process + startOutboxPublisher(); + await new Promise((resolve) => setTimeout(resolve, 1500)); // Wait for poll + publish + stopOutboxPublisher(); + + // Assert: Event was published via HTTP + expect(fetchMock).toHaveBeenCalledWith( + "https://test.example.com/ingest", + expect.objectContaining({ + method: "POST", + // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment + headers: expect.objectContaining({ + "Content-Type": "application/json", + }), + }), + ); + + // Assert: Event marked as published in database + const publishedEvent = await db + .selectFrom("outboxEvents") + .selectAll() + .where("id", "=", eventId) + .executeTakeFirst(); + + expect(publishedEvent?.publishedAt).not.toBeNull(); + expect(publishedEvent?.lastPublishError).toBeNull(); + expect(publishedEvent?.nextAttemptAt).toBeNull(); + }); + + it("sets nextAttemptAt on retryable failure", async () => { + // Arrange: Insert event and mock 500 error + const eventId = generateId("evt"); + const event: InternalEvent = { + id: eventId, + type: "booking.created", + ledgerId: testLedgerId, + schemaVersion: 1, + timestamp: new Date().toISOString(), + data: { test: "data" }, + }; + + await db + .insertInto("outboxEvents") + .values({ + id: eventId, + ledgerId: testLedgerId, + eventType: "booking.created", + schemaVersion: 1, + payload: event as unknown as Record, + publishAttempts: 0, + nextAttemptAt: null, + }) + .execute(); + + // Mock 500 error (retryable) + fetchMock.mockResolvedValue({ + ok: false, + status: 500, + text: () => Promise.resolve("Internal Server Error"), + }); + + process.env["FLOYD_EVENT_INGEST_URL"] = "https://test.example.com/ingest"; + + // Act: Run publisher + const { startOutboxPublisher, stopOutboxPublisher } = + await import("../../../src/workers/outbox-publisher"); + + startOutboxPublisher(); + await new Promise((resolve) => setTimeout(resolve, 1500)); + stopOutboxPublisher(); + + // Assert: Event NOT published, but has nextAttemptAt scheduled + const failedEvent = await db + .selectFrom("outboxEvents") + .selectAll() + .where("id", "=", eventId) + .executeTakeFirst(); + + expect(failedEvent?.publishedAt).toBeNull(); + expect(failedEvent?.publishAttempts).toBeGreaterThan(0); + expect(failedEvent?.nextAttemptAt).not.toBeNull(); + expect(failedEvent?.lastPublishError).toContain("500"); + }); + + it("does not schedule retry for non-retryable errors", async () => { + // Arrange: Insert event and mock 401 error + const eventId = generateId("evt"); + const event: InternalEvent = { + id: eventId, + type: "booking.created", + ledgerId: testLedgerId, + schemaVersion: 1, + timestamp: new Date().toISOString(), + data: { test: "data" }, + }; + + await db + .insertInto("outboxEvents") + .values({ + id: eventId, + ledgerId: testLedgerId, + eventType: "booking.created", + schemaVersion: 1, + payload: event as unknown as Record, + publishAttempts: 0, + nextAttemptAt: null, + }) + .execute(); + + // Mock 401 error (non-retryable) + fetchMock.mockResolvedValue({ + ok: false, + status: 401, + text: () => Promise.resolve("Unauthorized"), + }); + + process.env["FLOYD_EVENT_INGEST_URL"] = "https://test.example.com/ingest"; + + // Act: Run publisher + const { startOutboxPublisher, stopOutboxPublisher } = + await import("../../../src/workers/outbox-publisher"); + + startOutboxPublisher(); + await new Promise((resolve) => setTimeout(resolve, 1500)); + stopOutboxPublisher(); + + // Assert: Event NOT published, nextAttemptAt is null (won't retry) + const blockedEvent = await db + .selectFrom("outboxEvents") + .selectAll() + .where("id", "=", eventId) + .executeTakeFirst(); + + expect(blockedEvent?.publishedAt).toBeNull(); + expect(blockedEvent?.publishAttempts).toBeGreaterThan(0); + // Non-retryable errors get far-future date to prevent retry + expect(blockedEvent?.nextAttemptAt).toEqual(new Date("2099-12-31T23:59:59.000Z")); + expect(blockedEvent?.lastPublishError).toContain("401"); + }); + + it("includes Floyd-Signature header when secret is set", async () => { + // Arrange + const eventId = generateId("evt"); + const event: InternalEvent = { + id: eventId, + type: "allocation.created", + ledgerId: testLedgerId, + schemaVersion: 1, + timestamp: new Date().toISOString(), + data: { test: "data" }, + }; + + await db + .insertInto("outboxEvents") + .values({ + id: eventId, + ledgerId: testLedgerId, + eventType: "allocation.created", + schemaVersion: 1, + payload: event as unknown as Record, + publishAttempts: 0, + nextAttemptAt: null, + }) + .execute(); + + process.env["FLOYD_EVENT_INGEST_URL"] = "https://test.example.com/ingest"; + process.env["FLOYD_ENGINE_SECRET"] = "test-secret-key"; + + // Act + const { startOutboxPublisher, stopOutboxPublisher } = + await import("../../../src/workers/outbox-publisher"); + + startOutboxPublisher(); + await new Promise((resolve) => setTimeout(resolve, 1500)); + stopOutboxPublisher(); + + // Assert: Floyd-Signature header is present and not "unsigned" + expect(fetchMock).toHaveBeenCalledWith( + "https://test.example.com/ingest", + expect.objectContaining({ + // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment + headers: expect.objectContaining({ + // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment + "Floyd-Signature": expect.stringMatching(/^sha256=/), + }), + }), + ); + + const call = fetchMock.mock.calls[0] as [string, { headers: Record }]; + expect(call[1].headers["Floyd-Signature"]).not.toBe("sha256=unsigned"); + + // Cleanup + delete process.env["FLOYD_ENGINE_SECRET"]; + }); + + it("respects nextAttemptAt and does not publish before scheduled time", async () => { + // Arrange: Insert event with future nextAttemptAt + const eventId = generateId("evt"); + const futureTime = new Date(Date.now() + 60000); // 1 minute in future + + await db + .insertInto("outboxEvents") + .values({ + id: eventId, + ledgerId: testLedgerId, + eventType: "booking.created", + schemaVersion: 1, + payload: { + id: eventId, + type: "booking.created", + ledgerId: testLedgerId, + schemaVersion: 1, + timestamp: new Date().toISOString(), + data: {}, + } as unknown as Record, + publishAttempts: 1, + nextAttemptAt: futureTime, + }) + .execute(); + + process.env["FLOYD_EVENT_INGEST_URL"] = "https://test.example.com/ingest"; + + // Act: Run publisher + const { startOutboxPublisher, stopOutboxPublisher } = + await import("../../../src/workers/outbox-publisher"); + + startOutboxPublisher(); + await new Promise((resolve) => setTimeout(resolve, 1500)); + stopOutboxPublisher(); + + // Assert: Event was NOT published (nextAttemptAt is in future) + expect(fetchMock).not.toHaveBeenCalled(); + + // Event still in outbox unpublished + const event = await db + .selectFrom("outboxEvents") + .selectAll() + .where("id", "=", eventId) + .executeTakeFirst(); + + expect(event?.publishedAt).toBeNull(); + expect(event?.publishAttempts).toBe(1); // Unchanged + }); + + it("stops retrying after MAX_ATTEMPTS", async () => { + // Arrange: Insert event with MAX_ATTEMPTS - 1 + const eventId = generateId("evt"); + + await db + .insertInto("outboxEvents") + .values({ + id: eventId, + ledgerId: testLedgerId, + eventType: "booking.created", + schemaVersion: 1, + payload: { + id: eventId, + type: "booking.created", + ledgerId: testLedgerId, + schemaVersion: 1, + timestamp: new Date().toISOString(), + data: {}, + } as unknown as Record, + publishAttempts: 24, // One below MAX_ATTEMPTS (25) + nextAttemptAt: null, + }) + .execute(); + + // Mock 500 error + fetchMock.mockResolvedValue({ + ok: false, + status: 500, + text: () => Promise.resolve("Server Error"), + }); + + process.env["FLOYD_EVENT_INGEST_URL"] = "https://test.example.com/ingest"; + + // Act: Run publisher - give it more time to process + const { startOutboxPublisher, stopOutboxPublisher } = + await import("../../../src/workers/outbox-publisher"); + + startOutboxPublisher(); + await new Promise((resolve) => setTimeout(resolve, 2000)); // Extra time + stopOutboxPublisher(); + + // Assert: Event exhausted retries (may still be at 24 or reached 25) + const exhaustedEvent = await db + .selectFrom("outboxEvents") + .selectAll() + .where("id", "=", eventId) + .executeTakeFirst(); + + expect(exhaustedEvent?.publishedAt).toBeNull(); + // The worker may not increment beyond 24 if it determines it's already at max + expect(exhaustedEvent?.publishAttempts).toBeGreaterThanOrEqual(24); + // Exhausted events get far-future date to prevent retry + expect(exhaustedEvent?.nextAttemptAt).toEqual(new Date("2099-12-31T23:59:59.000Z")); + expect(exhaustedEvent?.lastPublishError).toContain("500"); + }); + }); + + describe("Circuit Breaker", () => { + it("activates circuit breaker after consecutive failures", async () => { + // Arrange: Insert 5 events + await Promise.all( + Array.from({ length: 5 }, async (_, i) => { + const eventId = generateId("evt"); + await db + .insertInto("outboxEvents") + .values({ + id: eventId, + ledgerId: testLedgerId, + eventType: "booking.created", + schemaVersion: 1, + payload: { + id: eventId, + type: "booking.created", + ledgerId: testLedgerId, + schemaVersion: 1, + timestamp: new Date().toISOString(), + data: { index: i }, + } as unknown as Record, + publishAttempts: 0, + nextAttemptAt: null, + }) + .execute(); + return eventId; + }), + ); + + // Mock 500 error for all requests + fetchMock.mockResolvedValue({ + ok: false, + status: 500, + text: () => Promise.resolve("Server Error"), + }); + + process.env["FLOYD_EVENT_INGEST_URL"] = "https://test.example.com/ingest"; + + // Act + const { startOutboxPublisher, stopOutboxPublisher } = + await import("../../../src/workers/outbox-publisher"); + + startOutboxPublisher(); + await new Promise((resolve) => setTimeout(resolve, 1500)); + stopOutboxPublisher(); + + // Assert: Circuit breaker should activate after 3 consecutive failures + // So at most 3 requests should be made, not all 5 + expect(fetchMock.mock.calls.length).toBeLessThanOrEqual(3); + }); + }); +}); diff --git a/apps/server/test/unit/vitest.config.ts b/apps/server/test/unit/vitest.config.ts index 17dab78..75641d5 100644 --- a/apps/server/test/unit/vitest.config.ts +++ b/apps/server/test/unit/vitest.config.ts @@ -11,6 +11,13 @@ export default defineConfig({ resolve: { alias: { domain: path.resolve(__dirname, "../../src/domain"), + database: path.resolve(__dirname, "../../src/database"), + infra: path.resolve(__dirname, "../../src/infra"), + workers: path.resolve(__dirname, "../../src/workers"), + lib: path.resolve(__dirname, "../../src/lib"), + config: path.resolve(__dirname, "../../src/config"), + operations: path.resolve(__dirname, "../../src/operations"), + routes: path.resolve(__dirname, "../../src/routes"), }, }, }); diff --git a/apps/server/test/unit/workers/outbox-publisher.spec.ts b/apps/server/test/unit/workers/outbox-publisher.spec.ts new file mode 100644 index 0000000..6989289 --- /dev/null +++ b/apps/server/test/unit/workers/outbox-publisher.spec.ts @@ -0,0 +1,310 @@ +import { describe, expect, it, beforeEach, vi } from "vitest"; + +// Mock database and other dependencies before importing the worker +vi.mock("database", () => ({ + db: {}, +})); + +vi.mock("infra/logger", () => ({ + logger: { + warn: vi.fn(), + error: vi.fn(), + debug: vi.fn(), + info: vi.fn(), + }, +})); + +vi.mock("infra/crypto", () => ({ + computeHmacSignature: vi.fn(() => "mocked-signature"), +})); + +vi.mock("config", () => ({ + config: {}, +})); + +import { + isRetryableError, + computeNextAttemptAt, + extractStatusCode, +} from "workers/outbox-publisher"; + +// ─── Constants ───────────────────────────────────────────────────────────────── + +const SECOND = 1000; + +// ─── Tests ───────────────────────────────────────────────────────────────────── + +describe("outbox-publisher retry logic", () => { + describe("extractStatusCode", () => { + it("extracts status code from error message", () => { + const error = new Error("HTTP 404: Not Found"); + expect(extractStatusCode(error)).toBe(404); + }); + + it("extracts status code from error with complex message", () => { + const error = new Error('HTTP 500: {"error":"Internal Server Error"}'); + expect(extractStatusCode(error)).toBe(500); + }); + + it("returns null for network errors without status", () => { + const error = new Error("ECONNRESET"); + expect(extractStatusCode(error)).toBeNull(); + }); + + it("returns null for DNS errors", () => { + const error = new Error("getaddrinfo ENOTFOUND"); + expect(extractStatusCode(error)).toBeNull(); + }); + + it("handles malformed error messages", () => { + const error = new Error("Something went wrong"); + expect(extractStatusCode(error)).toBeNull(); + }); + }); + + describe("isRetryableError", () => { + describe("retryable errors", () => { + it("treats network errors (no status code) as retryable", () => { + expect(isRetryableError(null)).toBe(true); + }); + + it("treats 408 Request Timeout as retryable", () => { + expect(isRetryableError(408)).toBe(true); + }); + + it("treats 429 Too Many Requests as retryable", () => { + expect(isRetryableError(429)).toBe(true); + }); + + it("treats 500 Internal Server Error as retryable", () => { + expect(isRetryableError(500)).toBe(true); + }); + + it("treats 502 Bad Gateway as retryable", () => { + expect(isRetryableError(502)).toBe(true); + }); + + it("treats 503 Service Unavailable as retryable", () => { + expect(isRetryableError(503)).toBe(true); + }); + + it("treats 504 Gateway Timeout as retryable", () => { + expect(isRetryableError(504)).toBe(true); + }); + }); + + describe("non-retryable errors", () => { + it("treats 400 Bad Request as non-retryable", () => { + expect(isRetryableError(400)).toBe(false); + }); + + it("treats 401 Unauthorized as non-retryable", () => { + expect(isRetryableError(401)).toBe(false); + }); + + it("treats 403 Forbidden as non-retryable", () => { + expect(isRetryableError(403)).toBe(false); + }); + + it("treats 404 Not Found as non-retryable", () => { + expect(isRetryableError(404)).toBe(false); + }); + + it("treats 422 Unprocessable Entity as non-retryable", () => { + expect(isRetryableError(422)).toBe(false); + }); + + it("treats 405 Method Not Allowed as non-retryable", () => { + expect(isRetryableError(405)).toBe(false); + }); + }); + }); + + describe("computeNextAttemptAt", () => { + beforeEach(() => { + // Fix the current time for deterministic tests + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-02-16T12:00:00Z")); + }); + + describe("non-retryable errors", () => { + it("returns null for 401 Unauthorized", () => { + const result = computeNextAttemptAt(1, 401); + expect(result).toBeNull(); + }); + + it("returns null for 403 Forbidden", () => { + const result = computeNextAttemptAt(1, 403); + expect(result).toBeNull(); + }); + + it("returns null for 400 Bad Request", () => { + const result = computeNextAttemptAt(1, 400); + expect(result).toBeNull(); + }); + + it("returns null for 404 Not Found", () => { + const result = computeNextAttemptAt(1, 404); + expect(result).toBeNull(); + }); + }); + + describe("retryable errors - backoff schedule", () => { + it("attempt 0: ~0 seconds (immediate)", () => { + const result = computeNextAttemptAt(0, 500); + expect(result).not.toBeNull(); + if (result) { + const delayMs = result.getTime() - Date.now(); + // 0s base + jitter (±30% of 0 = 0), so should be immediate + expect(delayMs).toBeLessThan(100); // Allow small variance + } + }); + + it("attempt 1: ~5 seconds ± jitter", () => { + const result = computeNextAttemptAt(1, 500); + expect(result).not.toBeNull(); + if (result) { + const delayMs = result.getTime() - Date.now(); + // 5s base ± 30% = 3.5s to 6.5s + expect(delayMs).toBeGreaterThanOrEqual(3.5 * SECOND); + expect(delayMs).toBeLessThanOrEqual(6.5 * SECOND); + } + }); + + it("attempt 2: ~15 seconds ± jitter", () => { + const result = computeNextAttemptAt(2, 500); + expect(result).not.toBeNull(); + if (result) { + const delayMs = result.getTime() - Date.now(); + // 15s base ± 30% = 10.5s to 19.5s + expect(delayMs).toBeGreaterThanOrEqual(10.5 * SECOND); + expect(delayMs).toBeLessThanOrEqual(19.5 * SECOND); + } + }); + + it("attempt 3: ~1 minute ± jitter", () => { + const result = computeNextAttemptAt(3, 500); + expect(result).not.toBeNull(); + if (result) { + const delayMs = result.getTime() - Date.now(); + // 60s base ± 30% = 42s to 78s + expect(delayMs).toBeGreaterThanOrEqual(42 * SECOND); + expect(delayMs).toBeLessThanOrEqual(78 * SECOND); + } + }); + + it("attempt 4: ~5 minutes ± jitter", () => { + const result = computeNextAttemptAt(4, 500); + expect(result).not.toBeNull(); + if (result) { + const delayMs = result.getTime() - Date.now(); + // 300s base ± 30% = 210s to 390s + expect(delayMs).toBeGreaterThanOrEqual(210 * SECOND); + expect(delayMs).toBeLessThanOrEqual(390 * SECOND); + } + }); + + it("attempt 10: capped at 4 hours ± jitter", () => { + const result = computeNextAttemptAt(10, 500); + expect(result).not.toBeNull(); + if (result) { + const delayMs = result.getTime() - Date.now(); + // 14400s (4h) base ± 30% = 10080s to 18720s + expect(delayMs).toBeGreaterThanOrEqual(10080 * SECOND); + expect(delayMs).toBeLessThanOrEqual(18720 * SECOND); + } + }); + + it("attempt 25: still capped at 4 hours ± jitter", () => { + const result = computeNextAttemptAt(25, 500); + expect(result).not.toBeNull(); + if (result) { + const delayMs = result.getTime() - Date.now(); + // Should still be capped at 4h ± 30% + expect(delayMs).toBeGreaterThanOrEqual(10080 * SECOND); + expect(delayMs).toBeLessThanOrEqual(18720 * SECOND); + } + }); + }); + + describe("retryable error types", () => { + it("computes backoff for network errors (null status)", () => { + const result = computeNextAttemptAt(1, null); + expect(result).not.toBeNull(); + if (result) { + const delayMs = result.getTime() - Date.now(); + // Should follow attempt 1 schedule: ~5s ± 30% + expect(delayMs).toBeGreaterThanOrEqual(3.5 * SECOND); + expect(delayMs).toBeLessThanOrEqual(6.5 * SECOND); + } + }); + + it("computes backoff for 429 Too Many Requests", () => { + const result = computeNextAttemptAt(2, 429); + expect(result).not.toBeNull(); + if (result) { + const delayMs = result.getTime() - Date.now(); + // Should follow attempt 2 schedule: ~15s ± 30% + expect(delayMs).toBeGreaterThanOrEqual(10.5 * SECOND); + expect(delayMs).toBeLessThanOrEqual(19.5 * SECOND); + } + }); + + it("computes backoff for 503 Service Unavailable", () => { + const result = computeNextAttemptAt(3, 503); + expect(result).not.toBeNull(); + if (result) { + const delayMs = result.getTime() - Date.now(); + // Should follow attempt 3 schedule: ~60s ± 30% + expect(delayMs).toBeGreaterThanOrEqual(42 * SECOND); + expect(delayMs).toBeLessThanOrEqual(78 * SECOND); + } + }); + }); + + describe("jitter randomization", () => { + it("produces different delays for same attempt (jitter)", () => { + const results: number[] = []; + for (let i = 0; i < 10; i++) { + const result = computeNextAttemptAt(4, 500); + if (result) { + results.push(result.getTime() - Date.now()); + } + } + + // All results should be within the valid range + results.forEach((delay) => { + expect(delay).toBeGreaterThanOrEqual(210 * SECOND); // 5min - 30% + expect(delay).toBeLessThanOrEqual(390 * SECOND); // 5min + 30% + }); + + // Results should not all be identical (jitter adds randomness) + const uniqueValues = new Set(results); + expect(uniqueValues.size).toBeGreaterThan(1); + }); + }); + + describe("edge cases", () => { + it("handles attempt 0 (should use first schedule entry)", () => { + const result = computeNextAttemptAt(0, 500); + expect(result).not.toBeNull(); + if (result) { + const delayMs = result.getTime() - Date.now(); + // Should use schedule[0] = 0s + expect(delayMs).toBeLessThan(100); + } + }); + + it("never returns negative delay", () => { + // Even with jitter, delay should never be negative + for (let attempt = 0; attempt < 30; attempt++) { + const result = computeNextAttemptAt(attempt, 500); + if (result) { + const delayMs = result.getTime() - Date.now(); + expect(delayMs).toBeGreaterThanOrEqual(0); + } + } + }); + }); + }); +}); diff --git a/docs/bookings.md b/docs/bookings.md index 3e61100..2f0a7e3 100644 --- a/docs/bookings.md +++ b/docs/bookings.md @@ -113,7 +113,7 @@ Hold bookings expire automatically when `expiresAt` elapses. The expiration work 1. Finds hold bookings where `expiresAt <= now()` 2. Sets `status = expired` 3. Deactivates allocations (`active = false`) -4. Enqueues a `booking.expired` webhook event +4. Emits a `booking.expired` event After expiration, the time slot is available for new bookings. diff --git a/docs/events.md b/docs/events.md new file mode 100644 index 0000000..fc40039 --- /dev/null +++ b/docs/events.md @@ -0,0 +1,212 @@ +# Events + +Floyd Engine emits events when state changes occur (bookings created, allocations deleted, etc.). Events are stored durably using the **Transactional Outbox Pattern** and can be consumed via multiple transports. + +## Event types + +### Booking events + +| Event | Description | +| ------------------- | ---------------------------- | +| `booking.created` | A new booking was created | +| `booking.confirmed` | A hold booking was confirmed | +| `booking.canceled` | A booking was canceled | +| `booking.expired` | A hold booking expired | + +### Allocation events + +| Event | Description | +| -------------------- | ---------------------------- | +| `allocation.created` | A raw allocation was created | +| `allocation.deleted` | A raw allocation was deleted | + +## Event guarantees + +- **Transactional safety** - Events are written in the same database transaction as the state change. If the transaction rolls back, no event is emitted. +- **At-least-once delivery** - Events are retried automatically on failure +- **Ordering** - Events are emitted in `created_at` order within a ledger, but may be delivered out of order when retries occur. Use `event.id` (ULID) or `event.timestamp` to reconstruct order on the consumer side. +- **Durability** - Events survive crashes and restarts + +## Consuming events + +### Option 1: Poll the outbox table + +Query `outbox_events` directly from the database: + +```sql +SELECT * FROM outbox_events +WHERE published_at IS NULL +ORDER BY created_at ASC +LIMIT 100; +``` + +After processing, mark events as published: + +```sql +UPDATE outbox_events +SET published_at = NOW() +WHERE id = 'evt_...'; +``` + +### Option 2: HTTP push to external service + +Set `FLOYD_EVENT_INGEST_URL` to push events to an HTTP endpoint: + +```bash +FLOYD_EVENT_INGEST_URL=https://your-service.com/events +FLOYD_ENGINE_SECRET=your-secret-key +``` + +Events will be POSTed with an HMAC signature in the `Floyd-Signature` header. + +### Option 3: Build a custom worker + +Read from the outbox table and process events in a background worker: + +```typescript +import { db } from "./database"; + +async function processEvents() { + const events = await db + .selectFrom("outboxEvents") + .selectAll() + .where("publishedAt", "is", null) + .orderBy("createdAt", "asc") + .limit(100) + .execute(); + + for (const event of events) { + try { + await handleEvent(event); + + await db + .updateTable("outboxEvents") + .set({ publishedAt: new Date() }) + .where("id", "=", event.id) + .execute(); + } catch (error) { + console.error(`Failed to process event ${event.id}:`, error); + } + } +} + +setInterval(processEvents, 5000); // Poll every 5 seconds +``` + +## Event payload structure + +All events follow this schema: + +```typescript +interface Event { + id: string; // Unique event ID (evt_...) + type: string; // Event type (booking.created, etc.) + ledgerId: string; // Which ledger + timestamp: string; // ISO 8601 timestamp + schemaVersion: number; // Payload schema version (currently 1) + data: { + booking?: Booking; // Present for booking events + allocation?: Allocation; // Present for allocation events + }; +} +``` + +### Example: booking.created + +```json +{ + "id": "evt_01abc123...", + "type": "booking.created", + "ledgerId": "ldg_01xyz789...", + "timestamp": "2026-01-15T10:00:00Z", + + "schemaVersion": 1, + "data": { + "booking": { + "id": "bkg_01def456...", + "serviceId": "svc_01ghi789...", + "status": "hold", + "expiresAt": "2026-01-15T10:15:00Z", + "allocations": [ + { + "id": "alc_01jkl012...", + "resourceId": "rsc_01mno345...", + "startTime": "2026-01-15T14:00:00Z", + "endTime": "2026-01-15T15:00:00Z" + } + ], + "createdAt": "2026-01-15T10:00:00Z" + } + } +} +``` + +### Example: allocation.deleted + +```json +{ + "id": "evt_01abc123...", + "type": "allocation.deleted", + "ledgerId": "ldg_01xyz789...", + "timestamp": "2026-01-15T10:00:00Z", + + "schemaVersion": 1, + "data": { + "allocation": { + "id": "alc_01def456...", + "resourceId": "rsc_01ghi789...", + "startTime": "2026-01-15T14:00:00Z", + "endTime": "2026-01-15T15:00:00Z", + "active": false + } + } +} +``` + +## Configuration + +Configure event publishing via environment variables: + +```bash +# HTTP push to external endpoint (default if FLOYD_EVENT_INGEST_URL is set) +FLOYD_EVENT_INGEST_URL=https://your-service.com/events +FLOYD_ENGINE_SECRET=your-secret-key # Optional, for HMAC signing + +# If not set, events accumulate in outbox_events without being published +# (useful for self-hosted deployments that poll the table directly) +``` + +## Verifying HTTP signatures + +When `FLOYD_ENGINE_SECRET` is set, events are signed with HMAC-SHA256: + +```typescript +import { createHmac } from "crypto"; + +function verifySignature(payload: string, signature: string, secret: string): boolean { + const expected = "sha256=" + createHmac("sha256", secret).update(payload).digest("hex"); + return signature === expected; +} + +// In your HTTP handler +app.post("/events", (req, res) => { + const signature = req.headers["floyd-signature"]; + const payload = JSON.stringify(req.body); + + if (!verifySignature(payload, signature, process.env.FLOYD_ENGINE_SECRET)) { + return res.status(401).json({ error: "Invalid signature" }); + } + + const event = req.body; + console.log(`Received ${event.type}`); + + res.status(200).send("OK"); +}); +``` + +## Best practices + +1. **Idempotency** - Use `event.id` to deduplicate events. The engine may emit the same event multiple times. +2. **Error handling** - If processing fails, leave `published_at` as NULL so the event can be retried. +3. **Ordering** - Events are ordered by `created_at` within a ledger, but may be processed out of order across ledgers. +4. **Monitoring** - Track the `outbox_events` table size. If it grows unbounded, events aren't being processed fast enough. diff --git a/docs/introduction.md b/docs/introduction.md index e45472c..3d7f0a9 100644 --- a/docs/introduction.md +++ b/docs/introduction.md @@ -44,6 +44,6 @@ See the [Quickstart](./quickstart) for full setup instructions. - [Allocations](./allocations) - Raw time blocks - [Policies](./policies) - Scheduling rules - [Availability](./availability) - Slots, windows, and free/busy timelines -- [Webhooks](./webhooks) - Real-time notifications +- [Events](./events) - Event system and consumption - [Idempotency](./idempotency) - Safe retries - [Errors](./errors) - Error handling diff --git a/docs/quickstart.md b/docs/quickstart.md index ab0029f..24c1b43 100644 --- a/docs/quickstart.md +++ b/docs/quickstart.md @@ -254,5 +254,5 @@ curl "$FLOYD_BASE_URL/v1/ledgers/$LEDGER_ID/bookings" - [Availability](./availability.md) - Slots, windows, and free/busy timelines - [Policies](./policies.md) - Scheduling rules - [Idempotency](./idempotency.md) - Safe retries -- [Webhooks](./webhooks.md) - Real-time notifications +- [Events](./events.md) - Event system and consumption - [Errors](./errors.md) - Error handling diff --git a/docs/webhooks.md b/docs/webhooks.md deleted file mode 100644 index 1dc4926..0000000 --- a/docs/webhooks.md +++ /dev/null @@ -1,138 +0,0 @@ -# Webhooks - -Floyd sends webhook notifications when booking and allocation events occur. Subscribe to receive real-time updates. - -## Events - -### Booking events - -| Event | Description | -| ------------------- | ---------------------------- | -| `booking.created` | A new booking was created | -| `booking.confirmed` | A hold booking was confirmed | -| `booking.canceled` | A booking was canceled | -| `booking.expired` | A hold booking expired | - -### Allocation events - -| Event | Description | -| -------------------- | ---------------------------- | -| `allocation.created` | A raw allocation was created | -| `allocation.deleted` | A raw allocation was deleted | - -## Payload format - -### Booking event payload - -```json -{ - "id": "whd_01abc123...", - "type": "booking.created", - "ledgerId": "ldg_01xyz789...", - "createdAt": "2026-01-15T10:00:00Z", - "data": { - "booking": { - "id": "bkg_01def456...", - "ledgerId": "ldg_01xyz789...", - "serviceId": "svc_01ghi789...", - "status": "hold", - "expiresAt": "2026-01-15T10:15:00Z", - "allocations": [ - { - "id": "alc_01jkl012...", - "resourceId": "rsc_01mno345...", - "startTime": "2026-01-15T14:00:00Z", - "endTime": "2026-01-15T15:00:00Z", - "active": true - } - ], - "metadata": null, - "createdAt": "2026-01-15T10:00:00Z", - "updatedAt": "2026-01-15T10:00:00Z" - } - } -} -``` - -### Allocation event payload - -```json -{ - "id": "whd_01abc123...", - "type": "allocation.created", - "ledgerId": "ldg_01xyz789...", - "createdAt": "2026-01-15T10:00:00Z", - "data": { - "allocation": { - "id": "alc_01def456...", - "ledgerId": "ldg_01xyz789...", - "resourceId": "rsc_01ghi789...", - "bookingId": null, - "active": true, - "startTime": "2026-01-15T14:00:00Z", - "endTime": "2026-01-15T15:00:00Z", - "buffer": { "beforeMs": 0, "afterMs": 0 }, - "expiresAt": null, - "metadata": null, - "createdAt": "2026-01-15T10:00:00Z", - "updatedAt": "2026-01-15T10:00:00Z" - } - } -} -``` - -## Headers - -| Header | Description | -| ----------------- | -------------------------------------- | -| `Floyd-Signature` | HMAC-SHA256 signature for verification | -| `Content-Type` | `application/json` | - -## Verifying signatures - -Verify the `Floyd-Signature` header to ensure the request is from Floyd: - -```javascript -import { createHmac } from "crypto"; - -function verifySignature(payload, signature, secret) { - const expected = "sha256=" + createHmac("sha256", secret).update(payload).digest("hex"); - return signature === expected; -} - -// In your handler -app.post("/webhook", (req, res) => { - const signature = req.headers["floyd-signature"]; - const payload = JSON.stringify(req.body); - - if (!verifySignature(payload, signature, process.env.WEBHOOK_SECRET)) { - return res.status(401).send("Invalid signature"); - } - - const event = req.body; - console.log(`Received ${event.type}`); - - res.status(200).send("OK"); -}); -``` - -## Retry behavior - -Failed deliveries are retried with exponential backoff: - -| Attempt | Delay | -| ------- | ---------- | -| 1 | 1 minute | -| 2 | 5 minutes | -| 3 | 30 minutes | -| 4 | 2 hours | -| 5 | 12 hours | - -After 5 failed attempts, the delivery is marked as exhausted. - -## Best practices - -1. **Respond quickly** - Return a 2xx status within 30 seconds -2. **Process asynchronously** - Queue the event and process later -3. **Handle duplicates** - Use the event `id` for idempotency -4. **Verify signatures** - Always validate the `Floyd-Signature` header diff --git a/packages/schema/constants/index.ts b/packages/schema/constants/index.ts index 288b436..385399a 100644 --- a/packages/schema/constants/index.ts +++ b/packages/schema/constants/index.ts @@ -3,14 +3,6 @@ export const IdempotencyStatus = { COMPLETED: "completed", } as const; -export const WebhookDeliveryStatus = { - PENDING: "pending", - IN_FLIGHT: "in_flight", - SUCCEEDED: "succeeded", - FAILED: "failed", - EXHAUSTED: "exhausted", -} as const; - export const BookingStatus = { HOLD: "hold", CONFIRMED: "confirmed", diff --git a/packages/schema/inputs/index.ts b/packages/schema/inputs/index.ts index 72e8568..186fe35 100644 --- a/packages/schema/inputs/index.ts +++ b/packages/schema/inputs/index.ts @@ -2,7 +2,6 @@ export * as allocationInput from "./allocation"; export * as availabilityInput from "./availability"; export * as resourceInput from "./resource"; export * as ledgerInput from "./ledger"; -export * as webhookInput from "./webhook"; export * as policyInput from "./policy"; export * as serviceInput from "./service"; export * as bookingInput from "./booking"; diff --git a/packages/schema/inputs/webhook.ts b/packages/schema/inputs/webhook.ts deleted file mode 100644 index 2ecfc4e..0000000 --- a/packages/schema/inputs/webhook.ts +++ /dev/null @@ -1,33 +0,0 @@ -import z from "zod"; -import { isValidId } from "@floyd-run/utils"; - -export const createSubscription = z.object({ - ledgerId: z.string().refine((id) => isValidId(id, "ldg"), { message: "Invalid ledger ID" }), - url: z.string().url(), -}); - -export const listSubscriptions = z.object({ - ledgerId: z.string().refine((id) => isValidId(id, "ldg"), { message: "Invalid ledger ID" }), -}); - -export const updateSubscription = z.object({ - id: z - .string() - .refine((id) => isValidId(id, "whs"), { message: "Invalid webhook subscription ID" }), - ledgerId: z.string().refine((id) => isValidId(id, "ldg"), { message: "Invalid ledger ID" }), - url: z.string().url().optional(), -}); - -export const deleteSubscription = z.object({ - id: z - .string() - .refine((id) => isValidId(id, "whs"), { message: "Invalid webhook subscription ID" }), - ledgerId: z.string().refine((id) => isValidId(id, "ldg"), { message: "Invalid ledger ID" }), -}); - -export const rotateSecret = z.object({ - id: z - .string() - .refine((id) => isValidId(id, "whs"), { message: "Invalid webhook subscription ID" }), - ledgerId: z.string().refine((id) => isValidId(id, "ldg"), { message: "Invalid ledger ID" }), -}); diff --git a/packages/schema/outputs/index.ts b/packages/schema/outputs/index.ts index d13ddbb..c8cab49 100644 --- a/packages/schema/outputs/index.ts +++ b/packages/schema/outputs/index.ts @@ -2,7 +2,6 @@ export * as allocation from "./allocation"; export * as availability from "./availability"; export * as resource from "./resource"; export * as ledger from "./ledger"; -export * as webhook from "./webhook"; export * as error from "./error"; export * as policy from "./policy"; export * as service from "./service"; diff --git a/packages/schema/outputs/webhook.ts b/packages/schema/outputs/webhook.ts deleted file mode 100644 index 9330aaf..0000000 --- a/packages/schema/outputs/webhook.ts +++ /dev/null @@ -1,29 +0,0 @@ -import { z } from "./zod"; - -export const subscription = z.object({ - id: z.string(), - ledgerId: z.string(), - url: z.string(), - createdAt: z.string(), - updatedAt: z.string(), -}); - -export const subscriptionWithSecret = subscription.extend({ - secret: z.string(), -}); - -export const createSubscription = z.object({ - data: subscriptionWithSecret, -}); - -export const listSubscriptions = z.object({ - data: z.array(subscription), -}); - -export const updateSubscription = z.object({ - data: subscription, -}); - -export const rotateSecret = z.object({ - data: subscriptionWithSecret, -}); diff --git a/packages/schema/types/index.ts b/packages/schema/types/index.ts index 6bd9696..1150286 100644 --- a/packages/schema/types/index.ts +++ b/packages/schema/types/index.ts @@ -1,16 +1,10 @@ import type { z } from "zod"; import type * as outputs from "../outputs"; -import { - BookingStatus, - IdempotencyStatus, - WebhookDeliveryStatus, - ScheduleDefault, -} from "../constants"; +import { BookingStatus, IdempotencyStatus, ScheduleDefault } from "../constants"; import { ConstantType } from "./utils"; export type BookingStatus = ConstantType; export type IdempotencyStatus = ConstantType; -export type WebhookDeliveryStatus = ConstantType; export type ScheduleDefault = ConstantType; export type Allocation = z.infer; diff --git a/packages/utils/id.ts b/packages/utils/id.ts index 8f1dc47..d016fb5 100644 --- a/packages/utils/id.ts +++ b/packages/utils/id.ts @@ -1,13 +1,13 @@ import { ulid } from "ulid"; -export type IdPrefix = "ldg" | "rsc" | "alc" | "whs" | "whd" | "pol" | "svc" | "bkg"; +export type IdPrefix = "ldg" | "rsc" | "alc" | "pol" | "svc" | "bkg" | "evt"; export function generateId(prefix: IdPrefix): string { return `${prefix}_${ulid().toLowerCase()}`; } export function parseId(id: string): { prefix: IdPrefix; ulid: string } | null { - const match = id.match(/^(ldg|rsc|alc|whs|whd|pol|svc|bkg)_([a-z0-9]{26})$/); + const match = id.match(/^(ldg|rsc|alc|pol|svc|bkg|evt)_([a-z0-9]{26})$/); if (!match) return null; return { prefix: match[1] as IdPrefix, ulid: match[2]! }; }