feat: replace ledger webhooks with internal event bus

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);

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<Date>;
 }
 
-export interface WebhookSubscriptionsTable {
+export interface OutboxEventsTable {
   id: string;
   ledgerId: string;
-  url: string;
-  secret: string;
-  createdAt: Generated<Date>;
-  updatedAt: Generated<Date>;
-}
-
-export interface WebhookDeliveriesTable {
-  id: string;
-  subscriptionId: string;
   eventType: string;
+  schemaVersion: number;
   payload: Record<string, unknown>;
-  status: WebhookDeliveryStatus;
-  attempts: number;
-  maxAttempts: number;
-  nextAttemptAt: Date | null;
-  lastError: string | null;
-  lastStatusCode: number | null;
   createdAt: Generated<Date>;
+  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<LedgersTable>;
@@ -145,12 +130,9 @@ export type BookingUpdate = Updateable<BookingsTable>;
 export type IdempotencyKeyRow = Selectable<IdempotencyKeysTable>;
 export type NewIdempotencyKey = Insertable<IdempotencyKeysTable>;
 
-export type WebhookSubscriptionRow = Selectable<WebhookSubscriptionsTable>;
-export type NewWebhookSubscription = Insertable<WebhookSubscriptionsTable>;
-export type WebhookSubscriptionUpdate = Updateable<WebhookSubscriptionsTable>;
-
-export type WebhookDeliveryRow = Selectable<WebhookDeliveriesTable>;
-export type NewWebhookDelivery = Insertable<WebhookDeliveriesTable>;
+export type OutboxEventRow = Selectable<OutboxEventsTable>;
+export type NewOutboxEvent = Insertable<OutboxEventsTable>;
+export type OutboxEventUpdate = Updateable<OutboxEventsTable>;
 
 export type PolicyRow = Selectable<PoliciesTable>;
 export type NewPolicy = Insertable<PoliciesTable>;

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();
   };
 

apps/server/src/infra/crypto.ts

@@ -0,0 +1,11 @@
+import { createHmac } from "crypto";
+
+/**
+ * Compute HMAC-SHA256 signature for a payload.
+ * Header format: `sha256=<hex>`
+ */
+export function computeHmacSignature(payload: string, secret: string): string {
+  const hmac = createHmac("sha256", secret);
+  hmac.update(payload);
+  return `sha256=${hmac.digest("hex")}`;
+}

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<string, unknown>;
+}
+
+/**
+ * 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<Database>,
+  type: InternalEventType,
+  ledgerId: string,
+  data: Record<string, unknown>,
+): Promise<void> {
+  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<string, unknown>,
+      publishAttempts: 0,
+    })
+    .execute();
+}