feat(testing): InMemoryAmqpBroker — run a contract without Docker

.changeset/in-memory-broker.md

@@ -0,0 +1,47 @@
+---
+"@amqp-contract/testing": minor
+"@amqp-contract/client": minor
+"@amqp-contract/worker": minor
+"@amqp-contract/core": minor
+---
+
+`InMemoryAmqpBroker`: run a contract end to end with no Docker.
+
+```ts
+import { InMemoryAmqpBroker } from "@amqp-contract/testing";
+
+const broker = new InMemoryAmqpBroker();
+const worker = await TypedAmqpWorker.create({
+  contract,
+  handlers,
+  transport: broker.createTransport(contract),
+}).getOrThrow();
+const client = await TypedAmqpClient.create({
+  contract,
+  transport: broker.createTransport(contract),
+}).getOrThrow();
+```
+
+Testing a contract and its handlers meant a container, and a container is a
+30-second tax on a question the broker was never going to answer differently.
+What runs for real here is everything above the wire: routing, both validation
+passes, middleware and interceptors, RPC correlation over direct reply-to,
+retry routing, TTL dead-lettering.
+
+**The seam is `AmqpTransport`**, new in `@amqp-contract/core`: the eight
+members `TypedAmqpClient` and `TypedAmqpWorker` actually use, out of
+`AmqpClient`'s full surface. A compile-time assertion keeps `AmqpClient`
+satisfying it, so a signature change there is a type error rather than a
+substitute that silently stops matching. Both facades now take
+`transport?: AmqpTransport` beside `urls`, and **exactly one is required** —
+passing both is refused rather than silently preferring one, because a test
+that supplies a transport and inherits a `urls` default would otherwise reach
+a real broker while believing it had not.
+
+`urls` becomes optional on both option types. Existing code is unaffected.
+
+The fake is deliberately not kinder than a broker: an unroutable publish is
+dropped and confirmed, and a dead-letter exchange with nothing bound loses the
+message. Topology refusals, reconnection, flow control and persistence are not
+modelled and stay the integration suite's job — the boundary is written down
+in the new `how-to/test-without-a-broker` page.

docs/.vitepress/config.ts

@@ -88,6 +88,7 @@ const GUIDE_SIDEBAR = [
       { text: "Instrument with OpenTelemetry", link: "/how-to/instrument-with-opentelemetry" },
       { text: "Generate AsyncAPI", link: "/how-to/generate-asyncapi" },
       { text: "Test with RabbitMQ", link: "/how-to/test-with-rabbitmq" },
+      { text: "Test without a broker", link: "/how-to/test-without-a-broker" },
       { text: "Tune performance", link: "/how-to/tune-performance" },
       { text: "Upgrade", link: "/how-to/upgrade" },
       { text: "Troubleshoot", link: "/how-to/troubleshoot" },

docs/how-to/test-with-rabbitmq.md

@@ -7,7 +7,9 @@ description: Run integration tests against a real broker with the Vitest extensi
 
 `@amqp-contract/testing` runs your tests against a real RabbitMQ in a container, one isolated virtual host per test. Testing messaging against a mock mostly tests the mock; this tests routing, bindings, acknowledgement and dead-lettering as the broker actually implements them.
 
-Requires Docker and Vitest 4+.
+Requires Docker and Vitest 4+. For the tests that are about your contract and
+your handlers rather than the broker, there is a container-free path:
+[test without a broker](/how-to/test-without-a-broker).
 
 ## Set it up
 

docs/how-to/test-without-a-broker.md

@@ -0,0 +1,100 @@
+---
+title: Test without a broker - amqp-contract
+description: Run the contract pipeline against an in-memory broker — no Docker, in the unit suite.
+---
+
+# Test without a broker
+
+`InMemoryAmqpBroker` runs a contract end to end with no container: publish,
+routing, both validation passes, middleware and interceptors, RPC correlation,
+retry routing and dead-lettering. It is for the tests that are about **your
+contract and your handlers**, where a container is a 30-second tax on a
+question the broker was never going to answer differently.
+
+It does not replace [testing against RabbitMQ](/how-to/test-with-rabbitmq) —
+see [what it does not model](#what-it-does-not-model) below.
+
+## Use it
+
+```ts
+import { InMemoryAmqpBroker } from "@amqp-contract/testing";
+import { TypedAmqpClient } from "@amqp-contract/client";
+import { TypedAmqpWorker } from "@amqp-contract/worker";
+
+const broker = new InMemoryAmqpBroker();
+
+const worker = await TypedAmqpWorker.create({
+  contract,
+  handlers,
+  transport: broker.createTransport(contract),
+}).getOrThrow();
+
+const client = await TypedAmqpClient.create({
+  contract,
+  transport: broker.createTransport(contract),
+}).getOrThrow();
+
+await client.publish("placeOrder", { orderId: "o-1", total: 42 }).getOrThrow();
+```
+
+`transport` replaces `urls`, and exactly one of the two is required — passing
+both is refused rather than silently preferring one, so a test that supplies a
+transport can never quietly reach a real broker instead.
+
+A transport per facade, as a real deployment has a connection per facade: that
+is what gives direct reply-to somewhere to route back to.
+
+## Look inside the broker
+
+Two methods exist for assertions a consumer cannot make:
+
+```ts
+broker.queueNames(); // every queue the contract declared, wait queues included
+broker.peek("orders-dlq"); // the messages parked on a queue, unconsumed
+```
+
+`peek` is the one worth reaching for: a dead-letter assertion otherwise needs a
+consumer on the DLQ, which changes what you are testing.
+
+## What it models
+
+- **Routing** — topic (`*` one word, `#` zero or more), direct, fanout, headers
+  (`x-match` `all`/`any`), and the default exchange, where the routing key is a
+  queue name. That last one carries every RPC request and every retry
+  republish.
+- **Direct reply-to** — `amq.rabbitmq.reply-to` is rewritten to a per-transport
+  pseudo-queue and routed back to the transport that published, which is what
+  RabbitMQ does per channel.
+- **Settlement** — `ack` drops the delivery; `nack(requeue: false)`
+  dead-letters through the queue's `x-dead-letter-exchange` or drops;
+  `nack(requeue: true)` redelivers with `redelivered: true` and an incremented
+  `x-delivery-count`, which is the header a quorum queue's retry budget counts.
+- **TTL** — per-message `expiration` and queue-level `x-message-ttl`, whichever
+  is shorter, dead-lettering on expiry. That is what makes TTL-backoff retry
+  run for real: the republish carries the `expiration`, the wait queue carries
+  the ceiling.
+- **Serialization** — a Buffer passes through byte for byte and anything else
+  is `JSON.stringify`d, exactly as `AmqpClient` encodes, so a compressed
+  payload survives the round trip and the decompression path runs.
+- **Asynchronous delivery**, so nothing is delivered re-entrantly inside the
+  publish that caused it.
+
+## What it does not model
+
+An unroutable publish is **dropped and confirmed**, as AMQP does without a
+mandatory flag — the fake is not kinder than the broker.
+
+Beyond that it is not a RabbitMQ, and these stay the integration suite's job:
+
+- **Topology refusals.** A real broker answers `406 PRECONDITION_FAILED` when a
+  queue is redeclared with different arguments; here declaring is idempotent
+  and additive.
+- **Reconnection.** `currentChannelEpoch` is always `0`, so the stale-delivery
+  guard never fires. A reconnect is precisely the behaviour this does not have.
+- **Flow control, prefetch limits, persistence and clustering.** `prefetch` is
+  accepted and ignored.
+- **Exchange-to-exchange bindings** are declarable but carry no traffic.
+
+The rule of thumb: if the assertion is about **your** contract, handlers or
+middleware, use this; if it is about what the **broker** does, use
+[a real one](/how-to/test-with-rabbitmq).

knip.json

@@ -1,10 +1,14 @@
 {
   "$schema": "https://unpkg.com/knip@6/schema.json",
   "ignoreExportsUsedInFile": true,
   "workspaces": {
     "packages/*": {
       "project": ["src/**/*.ts"]
     },
+    "packages/core": {
+      "project": ["src/**/*.ts"],
+      "ignoreDependencies": ["@amqp-contract/testing"]
+    },
     "examples/basic-order-processing-contract": {
       "entry": ["src/index.ts", "scripts/*.ts"],
       "project": ["src/**/*.ts", "scripts/**/*.ts"]

packages/client/src/client.ts

@@ -9,7 +9,8 @@ import {
   type RpcErrorMap,
 } from "@amqp-contract/contract";
 import {
-  AmqpClient,
+  type AmqpClient,
+  type AmqpTransport,
   type ConnectionError,
   type AmqpPublishOptions,
   type Logger,
@@ -23,6 +24,7 @@ import {
   endSpanSuccess,
   recordLateRpcReply,
   recordPublishMetric,
+  resolveTransport,
   safeJsonParse,
   startPublishSpan,
   technicalDefect,
@@ -110,7 +112,25 @@ export type PublishOptions = AmqpPublishOptions & {
  */
 export type CreateClientOptions<TContract extends ContractDefinition> = {
   contract: TContract;
-  urls: ConnectionUrl[];
+  /**
+   * AMQP broker URL(s). Multiple URLs provide failover support.
+   *
+   * Exactly one connection source is required: `urls` **or** {@link transport}.
+   */
+  urls?: ConnectionUrl[] | undefined;
+  /**
+   * A transport to use instead of dialling a broker — an
+   * {@link AmqpTransport}, which the real {@link AmqpClient} satisfies.
+   *
+   * The reason this exists is testing: `@amqp-contract/testing`'s
+   * `InMemoryAmqpBroker` hands back a transport that runs the whole contract
+   * pipeline — serialization, both validation passes, interceptors, RPC
+   * correlation, retry routing — with no Docker. Supplying one makes `urls`,
+   * `connectionOptions` and `connectTimeoutMs` meaningless, since nothing is
+   * being dialled, and passing both is refused rather than silently
+   * preferring one.
+   */
+  transport?: AmqpTransport | undefined;
   connectionOptions?: AmqpConnectionManagerOptions | undefined;
   logger?: Logger | undefined;
   /**
@@ -195,7 +215,7 @@ export class TypedAmqpClient<TContract extends ContractDefinition> {
 
   private constructor(
     private readonly contract: TContract,
-    private readonly amqpClient: AmqpClient,
+    private readonly amqpClient: AmqpTransport,
     private readonly defaultPublishOptions: PublishOptions,
     private readonly logger?: Logger,
     private readonly telemetry: TelemetryProvider = defaultTelemetryProvider,
@@ -216,6 +236,7 @@ export class TypedAmqpClient<TContract extends ContractDefinition> {
   static create<TContract extends ContractDefinition>({
     contract,
     urls,
+    transport,
     connectionOptions,
     defaultPublishOptions,
     logger,
@@ -231,8 +252,9 @@ export class TypedAmqpClient<TContract extends ContractDefinition> {
     return OkAsync(undefined).flatMap(() => {
       const client = new TypedAmqpClient(
         contract,
-        new AmqpClient(contract, {
+        resolveTransport(contract, {
           urls,
+          transport,
           connectionOptions,
           connectTimeoutMs,
           publishTimeoutMs,

packages/core/package.json

@@ -82,7 +82,6 @@
     "amqplib": "catalog:"
   },
   "devDependencies": {
-    "@amqp-contract/testing": "workspace:*",
     "@arethetypeswrong/cli": "catalog:",
     "@btravstack/tsconfig": "catalog:",
     "@btravstack/typedoc": "catalog:",

packages/core/src/index.ts

@@ -9,6 +9,7 @@ export {
   DEFAULT_PUBLISH_TIMEOUT_MS,
 } from "./amqp-client.js";
 export { type ConnectionLease } from "./connection-manager.js";
+export { type AmqpTransport, resolveTransport, type TransportSource } from "./transport.js";
 export { technicalDefect } from "./defect.js";
 export {
   ConnectionError,

packages/core/src/transport.ts

@@ -0,0 +1,113 @@
+import type { ContractDefinition } from "@amqp-contract/contract";
+import type { AsyncResult } from "unthrown";
+
+import {
+  AmqpClient,
+  type AmqpClientOptions,
+  type AmqpConsumeOptions,
+  type AmqpPublishOptions,
+  type ConsumeCallback,
+} from "./amqp-client.js";
+import type { ConnectionError } from "./errors.js";
+import { TechnicalError } from "./errors.js";
+
+/**
+ * The transport surface `TypedAmqpClient` and `TypedAmqpWorker` actually use.
+ *
+ * {@link AmqpClient} is the one that speaks to a broker, and this is the
+ * subset of it the typed facades depend on — eight members out of its full
+ * surface. Naming that subset is what lets a test substitute an in-memory
+ * broker for a real one without either facade knowing.
+ *
+ * It is deliberately **structural and small**. `sendToQueue`, `addSetup`,
+ * `on` and `getConnection` are absent because no facade calls them; an
+ * implementation is free to have them, and a future facade that reaches for
+ * one has to widen this type first, which is the point.
+ *
+ * The compile-time assertion below is what keeps it honest: `AmqpClient` must
+ * satisfy it, so a signature change there is a type error here rather than a
+ * substitution that silently stops matching.
+ */
+export type AmqpTransport = {
+  /**
+   * Resolve once the transport is ready to carry messages. The one member
+   * with a modeled error: a broker that cannot be dialled is an anticipated
+   * outcome, where every other failure here is infrastructure and rides the
+   * defect channel.
+   */
+  waitForConnect(): AsyncResult<void, ConnectionError>;
+  publish(
+    target: { exchange: string; routingKey: string },
+    content: Buffer | unknown,
+    options?: AmqpPublishOptions,
+  ): AsyncResult<void, never>;
+  consume(
+    queue: string,
+    callback: ConsumeCallback,
+    options?: AmqpConsumeOptions,
+  ): AsyncResult<string, never>;
+  cancel(consumerTag: string): AsyncResult<void, never>;
+  ack(msg: ConsumeMessageLike, options?: AckOptions): void;
+  nack(msg: ConsumeMessageLike, options?: NackOptions): void;
+  close(): AsyncResult<void, never>;
+  /**
+   * Bumped whenever the underlying channel is re-established. A delivery
+   * carries the epoch it arrived on, and `ack`/`nack` drop a delivery whose
+   * epoch has moved — the broker has already requeued it, so acknowledging
+   * would settle a different message.
+   */
+  readonly currentChannelEpoch: number;
+};
+
+/** What `ack` / `nack` accept — amqplib's `ConsumeMessage`, structurally. */
+type ConsumeMessageLike = Parameters<AmqpClient["ack"]>[0];
+type AckOptions = NonNullable<Parameters<AmqpClient["ack"]>[1]>;
+type NackOptions = NonNullable<Parameters<AmqpClient["nack"]>[1]>;
+
+// The seam is only worth having if the real client still fits through it.
+// A signature change in `AmqpClient` fails here rather than silently
+// diverging from every substitute.
+type AssertAmqpClientIsATransport = AmqpClient extends AmqpTransport ? true : never;
+const _assertAmqpClientIsATransport: AssertAmqpClientIsATransport = true;
+void _assertAmqpClientIsATransport;
+
+/** What a facade was handed as its connection source. */
+export type TransportSource = Omit<AmqpClientOptions, "urls"> & {
+  urls?: AmqpClientOptions["urls"] | undefined;
+  transport?: AmqpTransport | undefined;
+};
+
+/**
+ * Pick the transport a facade will run on: the supplied one, or a real
+ * {@link AmqpClient} dialled from `urls`.
+ *
+ * **Exactly one source**, and neither degenerate case is allowed to pass
+ * quietly. With neither there is nothing to dial and every later call would
+ * fail one at a time; with both, preferring one silently would mean a test
+ * that passes a transport AND inherits a `urls` default connects to a real
+ * broker while believing it did not — the failure this option exists to
+ * prevent, arriving in disguise.
+ *
+ * @throws TechnicalError when the sources are not exactly one. Both facades
+ * call this inside their `OkAsync(...).flatMap(...)` safety net, so it
+ * surfaces as a Defect from `create()` rather than as a raw throw.
+ */
+export const resolveTransport = (
+  contract: ContractDefinition,
+  { urls, transport, ...options }: TransportSource,
+): AmqpTransport => {
+  if (transport !== undefined && urls !== undefined) {
+    // oxlint-disable-next-line unthrown/no-throw -- a synchronous misuse at the create() boundary; both facades adopt it as a Defect through their OkAsync safety net (documented @throws)
+    throw new TechnicalError(
+      "Both `urls` and `transport` were supplied. Pass exactly one connection source: `urls` to dial a broker, or `transport` to run on a supplied one.",
+    );
+  }
+  if (transport !== undefined) return transport;
+  if (urls === undefined) {
+    // oxlint-disable-next-line unthrown/no-throw -- same boundary, same adoption
+    throw new TechnicalError(
+      "No connection source. Pass `urls` to dial a broker, or `transport` to run on a supplied one.",
+    );
+  }
+  return new AmqpClient(contract, { ...options, urls });
+};

packages/core/tsconfig.json

@@ -3,7 +3,16 @@
   "compilerOptions": {
     "outDir": "./dist",
     "rootDir": "./src",
-    "types": ["node"]
+    "types": ["node"],
+    // `@amqp-contract/testing` is NOT a devDependency here, and cannot be:
+    // testing depends on this package for `AmqpTransport`, so declaring the
+    // edge back would be a package cycle turbo refuses. The integration specs
+    // still need its `it` extension, so the type checker is pointed at the
+    // built declarations and vitest at the source (see `vitest.config.ts`).
+    "paths": {
+      "@amqp-contract/testing/extension": ["../testing/dist/extension.d.mts"],
+      "@amqp-contract/testing/global-setup": ["../testing/dist/global-setup.d.mts"]
+    }
   },
   "include": ["src/**/*"]
 }