GitHub - btravstack/amqp-contract: Type-safe contracts for AMQP/RabbitMQ messaging with TypeScript

// contract.ts
import {
  defineContract,
  defineEventConsumer,
  defineEventPublisher,
  defineExchange,
  defineMessage,
  defineQueue,
  defineQueueBinding,
} from "@amqp-contract/contract";
import { z } from "zod";

// 1. Define resources with Dead Letter Exchange and retry configuration
const ordersExchange = defineExchange("orders");
const ordersDlx = defineExchange("orders-dlx");
const orderProcessingQueue = defineQueue("order-processing", {
  deadLetter: { exchange: ordersDlx, routingKey: "order.failed" },
  retry: { mode: "ttl-backoff", maxRetries: 3, initialDelayMs: 1000 }, // Retry configured at queue level
});
// A dead-letter exchange with nothing bound to it drops what it receives, so the
// DLQ and its binding are what make `deadLetter` actually keep anything. The
// binding key must match the dead-letter routing key set above.
const orderDlq = defineQueue("order-processing-dlq");

// 2. Define message with schema validation
const orderMessage = defineMessage(
  z.object({
    orderId: z.string(),
    amount: z.number(),
  }),
);

// 3. Event pattern: publisher broadcasts, consumers subscribe
const orderCreatedEvent = defineEventPublisher(ordersExchange, orderMessage, {
  routingKey: "order.created",
});

// 4. Define contract - only publishers and consumers needed
//    Exchanges, queues, and bindings are automatically extracted
export const contract = defineContract({
  publishers: {
    orderCreated: orderCreatedEvent,
  },
  consumers: {
    processOrder: defineEventConsumer(orderCreatedEvent, orderProcessingQueue),
  },
  // The DLQ is declared but never consumed — standalone topology
  queues: { orderDlq },
  bindings: {
    orderDlq: defineQueueBinding(orderDlq, ordersDlx, { routingKey: "order.failed" }),
  },
});

import { TypedAmqpClient } from "@amqp-contract/client";
import { TypedAmqpWorker } from "@amqp-contract/worker";
import { OkAsync } from "unthrown";

import { contract } from "./contract.js";

// 5. Type-safe consuming with automatic retry (configured at queue level)
const worker = await TypedAmqpWorker.create({
  contract,
  handlers: {
    processOrder: ({ payload }) => {
      console.log(payload.orderId); // ✅ TypeScript knows!
      return OkAsync();
    },
  },
  urls: ["amqp://localhost"],
}).get();

// 6. Type-safe publishing with validation
const client = await TypedAmqpClient.create({
  contract,
  urls: ["amqp://localhost"],
}).get();

// publish() returns an AsyncResult instead of throwing — awaiting it yields a
// Result. create()/close() have an empty error channel (E = never), so .get()
// is correct there; publish() still has a modeled error, so extract it with
// .getOrThrow() (or handle it with .match()). See the error model guide.
await client
  .publish("orderCreated", {
    orderId: "ORD-123", // ✅ TypeScript knows!
    amount: 99.99,
  })
  .getOrThrow();

// 7. Clean up
await client.close().get();
await worker.close().get();

pnpm add @amqp-contract/contract @amqp-contract/client @amqp-contract/worker unthrown zod

docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:4-management