From 388d34e86dcafd14a4074591d0fc87ff3183df4b Mon Sep 17 00:00:00 2001 From: Benoit TRAVERS Date: Thu, 3 Sep 2026 19:37:48 +0200 Subject: [PATCH] =?UTF-8?q?feat(testing):=20`InMemoryAmqpBroker`=20?= =?UTF-8?q?=E2=80=94=20run=20a=20contract=20without=20Docker?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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` in core — the eight members the two facades actually use, with a compile-time assertion that `AmqpClient` still satisfies it. Both `create()`s take `transport?` beside `urls`, and exactly one is required: preferring one silently would let a test that passes a transport AND inherits a `urls` default reach a real broker while believing it had not. The fake is deliberately not kinder than a broker — an unroutable publish is dropped and confirmed, an unbound DLX loses the message. Topology refusals, reconnection and flow control are not modelled and stay the integration suite's job. `packages/core` stops declaring `@amqp-contract/testing`: that package now depends on core, so the edge back would be a cycle turbo refuses. Its specs reach the fixtures through a tsconfig path, a vitest alias, an explicit globalSetup path, a turbo edge and a knip ignore — five places, each commented, and the whole integration suite still passes. Closes #541. Claude-Session: https://claude.ai/code/session_01GGixjxi5AQ2cNK62bBymfF --- .changeset/in-memory-broker.md | 47 +++ docs/.vitepress/config.ts | 1 + docs/how-to/test-with-rabbitmq.md | 4 +- docs/how-to/test-without-a-broker.md | 100 +++++ knip.json | 4 + packages/client/src/client.ts | 30 +- packages/core/package.json | 1 - packages/core/src/index.ts | 1 + packages/core/src/transport.ts | 113 ++++++ packages/core/tsconfig.json | 11 +- packages/core/vitest.config.ts | 8 + packages/testing/package.json | 10 +- packages/testing/src/in-memory.ts | 551 +++++++++++++++++++++++++++ packages/testing/src/index.ts | 1 + packages/worker/src/retry.ts | 6 +- packages/worker/src/worker.ts | 28 +- pnpm-lock.yaml | 12 +- tests/src/in-memory.spec.ts | 169 ++++++++ turbo.json | 6 + vitest.shared.ts | 21 +- 20 files changed, 1104 insertions(+), 20 deletions(-) create mode 100644 .changeset/in-memory-broker.md create mode 100644 docs/how-to/test-without-a-broker.md create mode 100644 packages/core/src/transport.ts create mode 100644 packages/testing/src/in-memory.ts create mode 100644 tests/src/in-memory.spec.ts diff --git a/.changeset/in-memory-broker.md b/.changeset/in-memory-broker.md new file mode 100644 index 00000000..7f8a1c09 --- /dev/null +++ b/.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. diff --git a/docs/.vitepress/config.ts b/docs/.vitepress/config.ts index dd06d5b0..975fb2de 100644 --- a/docs/.vitepress/config.ts +++ b/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" }, diff --git a/docs/how-to/test-with-rabbitmq.md b/docs/how-to/test-with-rabbitmq.md index 933dfef7..9e3af487 100644 --- a/docs/how-to/test-with-rabbitmq.md +++ b/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 diff --git a/docs/how-to/test-without-a-broker.md b/docs/how-to/test-without-a-broker.md new file mode 100644 index 00000000..0008e267 --- /dev/null +++ b/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). diff --git a/knip.json b/knip.json index a8311fa4..768c6dc7 100644 --- a/knip.json +++ b/knip.json @@ -5,6 +5,10 @@ "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"] diff --git a/packages/client/src/client.ts b/packages/client/src/client.ts index 2191b0fe..651e2170 100644 --- a/packages/client/src/client.ts +++ b/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 = { 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 { 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 { static create({ contract, urls, + transport, connectionOptions, defaultPublishOptions, logger, @@ -231,8 +252,9 @@ export class TypedAmqpClient { return OkAsync(undefined).flatMap(() => { const client = new TypedAmqpClient( contract, - new AmqpClient(contract, { + resolveTransport(contract, { urls, + transport, connectionOptions, connectTimeoutMs, publishTimeoutMs, diff --git a/packages/core/package.json b/packages/core/package.json index 20573c49..33c0a8bb 100644 --- a/packages/core/package.json +++ b/packages/core/package.json @@ -82,7 +82,6 @@ "amqplib": "catalog:" }, "devDependencies": { - "@amqp-contract/testing": "workspace:*", "@arethetypeswrong/cli": "catalog:", "@btravstack/tsconfig": "catalog:", "@btravstack/typedoc": "catalog:", diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 7e57a2e9..5c5b1d8a 100644 --- a/packages/core/src/index.ts +++ b/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, diff --git a/packages/core/src/transport.ts b/packages/core/src/transport.ts new file mode 100644 index 00000000..20212707 --- /dev/null +++ b/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; + publish( + target: { exchange: string; routingKey: string }, + content: Buffer | unknown, + options?: AmqpPublishOptions, + ): AsyncResult; + consume( + queue: string, + callback: ConsumeCallback, + options?: AmqpConsumeOptions, + ): AsyncResult; + cancel(consumerTag: string): AsyncResult; + ack(msg: ConsumeMessageLike, options?: AckOptions): void; + nack(msg: ConsumeMessageLike, options?: NackOptions): void; + close(): AsyncResult; + /** + * 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[0]; +type AckOptions = NonNullable[1]>; +type NackOptions = NonNullable[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 & { + 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 }); +}; diff --git a/packages/core/tsconfig.json b/packages/core/tsconfig.json index ae65d214..b44d03c0 100644 --- a/packages/core/tsconfig.json +++ b/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/**/*"] } diff --git a/packages/core/vitest.config.ts b/packages/core/vitest.config.ts index 75bcdd59..37c7134d 100644 --- a/packages/core/vitest.config.ts +++ b/packages/core/vitest.config.ts @@ -2,10 +2,18 @@ import { defineConfig } from "vitest/config"; import { sharedVitestConfig } from "../../vitest.shared.js"; +// The alias half of the arrangement `tsconfig.json` documents: with no +// devDependency to resolve through, the specs' `@amqp-contract/testing/*` +// imports are pointed at the source. +const testingSource = (entry: string) => + new URL(`../testing/src/${entry}.ts`, import.meta.url).pathname; + export default defineConfig( sharedVitestConfig({ thresholds: { statements: 20, branches: 10, functions: 12, lines: 20 }, integration: true, setupFile: "./src/vitest.setup.ts", + alias: { "@amqp-contract/testing/extension": testingSource("extension") }, + globalSetup: testingSource("global-setup"), }), ); diff --git a/packages/testing/package.json b/packages/testing/package.json index 2e8fefa3..bac45130 100644 --- a/packages/testing/package.json +++ b/packages/testing/package.json @@ -43,15 +43,21 @@ "types": "./dist/extension.d.mts", "import": "./dist/extension.mjs" }, + "./in-memory": { + "types": "./dist/in-memory.d.mts", + "import": "./dist/in-memory.mjs" + }, "./package.json": "./package.json" }, "scripts": { - "build": "tsdown src/index.ts src/global-setup.ts src/extension.ts --format esm --dts --clean", + "build": "tsdown src/index.ts src/global-setup.ts src/extension.ts src/in-memory.ts --format esm --dts --clean", "build:docs": "typedoc", "typecheck": "tsc --noEmit", "check:package": "publint --strict && attw --pack . --profile esm-only" }, "dependencies": { + "@amqp-contract/contract": "workspace:*", + "@amqp-contract/core": "workspace:*", "amqplib": "catalog:", "testcontainers": "catalog:" }, @@ -65,9 +71,11 @@ "typedoc": "catalog:", "typedoc-plugin-markdown": "catalog:", "typescript": "catalog:", + "unthrown": "catalog:", "vitest": "catalog:" }, "peerDependencies": { + "unthrown": "^5.0.0", "vitest": "^4" }, "engines": { diff --git a/packages/testing/src/in-memory.ts b/packages/testing/src/in-memory.ts new file mode 100644 index 00000000..a124440c --- /dev/null +++ b/packages/testing/src/in-memory.ts @@ -0,0 +1,551 @@ +import { + type ContractDefinition, + deriveTtlBackoffInfrastructure, + type QueueDefinition, +} from "@amqp-contract/contract"; +import type { + AmqpConsumeOptions, + AmqpPublishOptions, + AmqpTransport, + ConnectionError, + ConsumeCallback, +} from "@amqp-contract/core"; +import type { ConsumeMessage } from "amqplib"; +import { type AsyncResult, OkAsync } from "unthrown"; + +/** + * The `replyTo` value RabbitMQ treats as direct reply-to. A publisher naming + * it is asking for a per-channel pseudo-queue rather than a declared one. + */ +const DIRECT_REPLY_TO = "amq.rabbitmq.reply-to"; + +type Binding = { + readonly queue: string; + readonly routingKey: string; + readonly arguments?: Record | undefined; +}; + +type Exchange = { + readonly name: string; + readonly type: "topic" | "direct" | "fanout" | "headers"; + readonly bindings: Binding[]; +}; + +type Queue = { + readonly name: string; + readonly arguments: Record; + /** Ready messages, oldest first. */ + readonly ready: Delivery[]; + /** Consumers, in the order they subscribed — deliveries round-robin. */ + readonly consumers: Consumer[]; +}; + +type Consumer = { + readonly tag: string; + readonly callback: ConsumeCallback; + readonly noAck: boolean; + /** The transport that opened it, so a reply can find its way home. */ + readonly owner: InMemoryTransport; +}; + +type Delivery = { + readonly content: Buffer; + readonly properties: ConsumeMessage["properties"]; + readonly routingKey: string; + readonly exchange: string; + redelivered: boolean; + deliveryCount: number; +}; + +/** Deliveries are handed out asynchronously, as a broker's would be. */ +const soon = (run: () => void): void => { + queueMicrotask(run); +}; + +/** + * Topic matching: `*` is exactly one word, `#` is zero or more. Built as a + * regular expression over the dot-separated key, which is what RabbitMQ's + * trie computes the same answer for. + */ +const topicMatches = (pattern: string, routingKey: string): boolean => { + const source = pattern + .split(".") + .map((word) => (word === "*" ? "[^.]+" : word === "#" ? ".*" : escapeWord(word))) + .join("\\.") + // `#` spans separators, so a `#` segment must be able to eat the dot + // beside it — `a.#` matches `a`, not just `a.something`. + .replace(/\\\.\.\*/g, "(?:\\..*)?") + .replace(/^\.\*\\\./, "(?:.*\\.)?"); + return new RegExp(`^${source}$`).test(routingKey); +}; + +const escapeWord = (word: string): string => word.replace(/[.*+?^${}()|[\]\\]/g, "\\$&"); + +/** + * Headers matching, per AMQP's `x-match`: `all` requires every declared + * header, `any` requires one. Keys beginning `x-` are the matcher's own + * configuration and never participate. + */ +const headersMatch = ( + binding: Record, + message: Record, +): boolean => { + const mode = binding["x-match"] === "any" ? "any" : "all"; + const required = Object.entries(binding).filter(([key]) => !key.startsWith("x-")); + if (required.length === 0) return mode === "all"; + const matched = required.filter(([key, value]) => message[key] === value); + return mode === "any" ? matched.length > 0 : matched.length === required.length; +}; + +/** + * An in-memory AMQP broker: enough of one to run a contract end to end + * without Docker. + * + * It is shared state plus a routing table, and every transport it hands out + * talks to the same state — so a client and a worker built from one broker + * see each other exactly as they would across a real connection. What runs + * for real is everything above the wire: serialization, both validation + * passes, interceptors and middleware, RPC correlation, retry routing and + * dead-lettering. + * + * What it is NOT is a RabbitMQ. It models the routing and settlement rules + * the contract pipeline depends on; it models no cluster, no flow control, + * no channel errors, and no persistence. Topology *refusals* — the 406 a + * real broker answers when a queue is redeclared with different arguments — + * are a broker behaviour and stay the integration suite's job. + * + * @example + * ```ts + * const broker = new InMemoryAmqpBroker(); + * const worker = await TypedAmqpWorker.create({ + * contract, + * handlers, + * transport: broker.createTransport(contract), + * }); + * const client = await TypedAmqpClient.create({ + * contract, + * transport: broker.createTransport(contract), + * }); + * ``` + */ +export class InMemoryAmqpBroker { + private readonly exchanges = new Map(); + private readonly queues = new Map(); + /** Direct reply-to pseudo-queues, one per transport that consumes one. */ + private readonly replyQueues = new Map(); + private nextTag = 0; + + /** + * Register a contract's topology and hand back a transport bound to it. + * + * Declaring is idempotent and additive, exactly as `assertExchange` / + * `assertQueue` are: two transports built from the same contract converge + * on one set of exchanges, queues and bindings rather than two. + */ + createTransport(contract: ContractDefinition): AmqpTransport { + this.declare(contract); + return new InMemoryTransport(this); + } + + /** Every queue currently declared, for a spec that wants to look inside. */ + queueNames(): readonly string[] { + return [...this.queues.keys()].sort(); + } + + /** The messages sitting unconsumed on a queue — a DLQ, most usefully. */ + peek(queue: string): readonly ConsumeMessage[] { + return (this.queues.get(queue)?.ready ?? []).map((delivery, index) => + this.toConsumeMessage(delivery, index + 1), + ); + } + + /** Register exchanges, queues, TTL-backoff wait queues and bindings. */ + private declare(contract: ContractDefinition): void { + for (const exchange of Object.values(contract.exchanges ?? {})) { + // The default exchange is implicit and never declared, here or on a + // real broker; `route` handles it directly. + if (exchange.name === "") continue; + if (!this.exchanges.has(exchange.name)) { + this.exchanges.set(exchange.name, { + name: exchange.name, + type: exchange.type, + bindings: [], + }); + } + } + + for (const queue of Object.values(contract.queues ?? {})) { + this.declareQueue(queue.name, queueArgumentsOf(queue)); + for (const wait of deriveTtlBackoffInfrastructure(queue)?.waitQueues ?? []) { + this.declareQueue(wait.name, { + "x-message-ttl": wait.messageTtlMs, + "x-dead-letter-exchange": "", + "x-dead-letter-routing-key": queue.name, + }); + } + } + + for (const binding of Object.values(contract.bindings ?? {})) { + // Exchange-to-exchange bindings are declarable but carry no traffic + // this fake routes: every publish here names a leaf exchange or the + // default one. A spec that needs the hop is a broker-behaviour test. + if (binding.type !== "queue") continue; + const exchange = this.exchanges.get(binding.exchange.name); + if (!exchange) continue; + const routingKey = binding.routingKey ?? ""; + const already = exchange.bindings.some( + (b) => b.queue === binding.queue.name && b.routingKey === routingKey, + ); + if (!already) { + exchange.bindings.push({ + queue: binding.queue.name, + routingKey, + arguments: binding.arguments, + }); + } + } + } + + private declareQueue(name: string, args: Record): void { + if (!this.queues.has(name)) { + this.queues.set(name, { name, arguments: args, ready: [], consumers: [] }); + } + } + + /** + * Route one publish to every matching queue. + * + * An unroutable publish is **dropped and confirmed**, which is AMQP's own + * behaviour without a mandatory flag or an alternate exchange — and the + * hazard the contract's define-time guard exists to catch, so the fake must + * not be kinder than the broker. + */ + publish( + from: InMemoryTransport, + exchange: string, + routingKey: string, + content: Buffer, + options: AmqpPublishOptions | undefined, + ): void { + const properties = { + ...options, + headers: { ...(options?.headers as Record | undefined) }, + } as ConsumeMessage["properties"]; + + // Direct reply-to: the publisher's `replyTo` names a pseudo-queue rather + // than a declared one, and a real broker rewrites it per channel. Doing + // the same here is what lets a reply find the transport that asked. + if (properties.replyTo === DIRECT_REPLY_TO) { + properties.replyTo = from.replyQueueName; + this.replyQueues.set(from.replyQueueName, from); + } + + for (const queue of this.match(exchange, routingKey, properties, content)) { + this.enqueue(queue, { + content, + properties, + routingKey, + exchange, + redelivered: false, + deliveryCount: 0, + }); + } + } + + /** Which queues a message reaches, by exchange type. */ + private match( + exchange: string, + routingKey: string, + properties: ConsumeMessage["properties"], + content: Buffer = Buffer.alloc(0), + ): Queue[] { + // The default exchange routes by queue name, and is what every RPC + // request and every retry republish travels through. + if (exchange === "") { + const direct = this.queues.get(routingKey); + if (direct) return [direct]; + // A direct reply-to pseudo-queue is not a queue: it has no backlog and + // no settlement, and a reply arriving after its consumer is gone is + // dropped rather than parked — which is the late-reply case the client + // records a metric for. + this.replyQueues.get(routingKey)?.deliverReply(content, properties, routingKey); + return []; + } + + const declared = this.exchanges.get(exchange); + if (!declared) return []; + + const headers = (properties.headers ?? {}) as Record; + const names = declared.bindings + .filter((binding) => { + switch (declared.type) { + case "fanout": + return true; + case "direct": + return binding.routingKey === routingKey; + case "topic": + return topicMatches(binding.routingKey, routingKey); + case "headers": + return headersMatch(binding.arguments ?? {}, headers); + } + }) + .map((binding) => binding.queue); + + // A queue bound twice to one exchange receives one copy, as on a broker. + return [...new Set(names)].flatMap((name) => { + const queue = this.queues.get(name); + return queue ? [queue] : []; + }); + } + + /** + * Put a message on a queue, honouring its TTL, then try to deliver. + * + * The effective TTL is the smaller of the per-message `expiration` and the + * queue's `x-message-ttl` — which is what makes TTL-backoff retry work: + * the republish carries `expiration`, the wait queue carries the ceiling, + * and whichever is shorter decides when it dead-letters back. + */ + private enqueue(queue: Queue, delivery: Delivery): void { + const ttl = effectiveTtl(queue, delivery); + if (ttl !== undefined) { + const timer = setTimeout(() => { + const index = queue.ready.indexOf(delivery); + if (index === -1) return; + queue.ready.splice(index, 1); + this.deadLetter(queue, delivery); + }, ttl); + // A pending TTL must never hold a test process open. + timer.unref?.(); + } + queue.ready.push(delivery); + soon(() => { + this.drain(queue); + }); + } + + /** Hand ready messages to consumers, round-robin. */ + private drain(queue: Queue): void { + while (queue.ready.length > 0 && queue.consumers.length > 0) { + const delivery = queue.ready.shift(); + if (!delivery) return; + const consumer = queue.consumers[this.nextTag++ % queue.consumers.length]; + if (!consumer) return; + const message = this.toConsumeMessage(delivery, this.nextTag); + consumer.owner.track(message, queue, delivery, consumer.noAck); + void consumer.callback(message); + } + } + + /** + * Route a message to the queue's dead-letter exchange, or drop it. + * + * A DLX naming an exchange nothing is bound to loses the message, exactly + * as a real broker does — the hazard `dlx-routability` exists to prove, and + * one this fake must reproduce rather than paper over. + */ + private deadLetter(queue: Queue, delivery: Delivery): void { + const exchange = queue.arguments["x-dead-letter-exchange"]; + if (typeof exchange !== "string") return; + const routingKey = + typeof queue.arguments["x-dead-letter-routing-key"] === "string" + ? (queue.arguments["x-dead-letter-routing-key"] as string) + : delivery.routingKey; + for (const target of this.match(exchange, routingKey, delivery.properties, delivery.content)) { + this.enqueue(target, { ...delivery, redelivered: false }); + } + } + + /** Settle a delivery: ack drops it, nack requeues or dead-letters. */ + settle(queue: Queue, delivery: Delivery, requeue: boolean | undefined): void { + if (requeue !== true) { + this.deadLetter(queue, delivery); + return; + } + // Quorum semantics: a redelivery is marked, and its delivery count is + // what the worker's immediate-requeue retry budget counts. + delivery.redelivered = true; + delivery.deliveryCount += 1; + delivery.properties.headers = { + ...delivery.properties.headers, + "x-delivery-count": delivery.deliveryCount, + }; + queue.ready.unshift(delivery); + soon(() => { + this.drain(queue); + }); + } + + register(queue: string, consumer: Consumer): string | undefined { + const declared = this.queues.get(queue); + if (!declared) return undefined; + declared.consumers.push(consumer); + soon(() => { + this.drain(declared); + }); + return consumer.tag; + } + + unregister(tag: string): void { + for (const queue of this.queues.values()) { + const index = queue.consumers.findIndex((consumer) => consumer.tag === tag); + if (index !== -1) queue.consumers.splice(index, 1); + } + } + + mintTag(): string { + this.nextTag += 1; + return `in-memory-ctag-${this.nextTag}`; + } + + private toConsumeMessage(delivery: Delivery, deliveryTag: number): ConsumeMessage { + return { + content: delivery.content, + fields: { + consumerTag: "", + deliveryTag, + redelivered: delivery.redelivered, + exchange: delivery.exchange, + routingKey: delivery.routingKey, + }, + properties: delivery.properties, + } as ConsumeMessage; + } +} + +/** The queue arguments `setupAmqpTopology` would have asserted. */ +const queueArgumentsOf = (queue: QueueDefinition): Record => ({ + ...queue.arguments, + "x-queue-type": queue.type, + ...(queue.deadLetter && { + "x-dead-letter-exchange": queue.deadLetter.exchange.name, + ...(queue.deadLetter.routingKey !== undefined && { + "x-dead-letter-routing-key": queue.deadLetter.routingKey, + }), + }), +}); + +/** The smaller of the per-message `expiration` and the queue's ceiling. */ +const effectiveTtl = (queue: Queue, delivery: Delivery): number | undefined => { + const perMessage = Number(delivery.properties.expiration); + const perQueue = Number(queue.arguments["x-message-ttl"]); + const candidates = [perMessage, perQueue].filter((value) => Number.isFinite(value) && value >= 0); + return candidates.length === 0 ? undefined : Math.min(...candidates); +}; + +/** + * One "connection" onto an {@link InMemoryAmqpBroker}, satisfying + * `AmqpTransport`. + * + * 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, and what + * lets `close()` retire one side without disturbing the other. + */ +class InMemoryTransport implements AmqpTransport { + /** The pseudo-queue a direct reply-to publish is rewritten to. */ + readonly replyQueueName = `amq.rabbitmq.reply-to.${Math.random().toString(36).slice(2)}`; + private readonly outstanding = new Map(); + private replyConsumer: ConsumeCallback | undefined; + private closed = false; + + constructor(private readonly broker: InMemoryAmqpBroker) {} + + /** + * Always zero: nothing here reconnects, so no delivery is ever stale. The + * epoch guard in `ack`/`nack` therefore never fires, which is correct — a + * reconnect is exactly the broker behaviour this fake does not model, and + * pretending otherwise would make the guard untestable in both directions. + */ + readonly currentChannelEpoch = 0; + + waitForConnect(): AsyncResult { + return OkAsync(); + } + + publish( + target: { exchange: string; routingKey: string }, + content: Buffer | unknown, + options?: AmqpPublishOptions, + ): AsyncResult { + // Byte-for-byte what `AmqpClient.encodeContent` does, so a compressed + // payload survives the round trip unchanged and the decompression path + // runs for real. + const encoded = Buffer.isBuffer(content) ? content : Buffer.from(JSON.stringify(content)); + this.broker.publish(this, target.exchange, target.routingKey, encoded, options); + return OkAsync(); + } + + consume( + queue: string, + callback: ConsumeCallback, + options?: AmqpConsumeOptions, + ): AsyncResult { + if (queue === DIRECT_REPLY_TO) { + this.replyConsumer = callback; + return OkAsync(this.broker.mintTag()); + } + const tag = this.broker.mintTag(); + const registered = this.broker.register(queue, { + tag, + callback, + noAck: options?.noAck === true, + owner: this, + }); + // Consuming a queue the contract never declared is a broker error on the + // real thing; here it is a tag that receives nothing, which is the + // closest honest analogue without inventing a channel-error path. + return OkAsync(registered ?? tag); + } + + cancel(consumerTag: string): AsyncResult { + this.broker.unregister(consumerTag); + return OkAsync(); + } + + ack(msg: Parameters[0]) { + this.outstanding.delete(msg.fields.deliveryTag); + } + + nack(msg: Parameters[0], options?: { requeue?: boolean | undefined }) { + const held = this.outstanding.get(msg.fields.deliveryTag); + if (!held) return; + this.outstanding.delete(msg.fields.deliveryTag); + this.broker.settle(held.queue, held.delivery, options?.requeue); + } + + close(): AsyncResult { + this.closed = true; + this.replyConsumer = undefined; + return OkAsync(); + } + + /** Remember an unsettled delivery so `nack` can put it back. */ + track(message: ConsumeMessage, queue: Queue, delivery: Delivery, noAck: boolean): void { + if (noAck) return; + this.outstanding.set(message.fields.deliveryTag, { queue, delivery }); + } + + /** A direct reply-to message coming home to this transport's consumer. */ + deliverReply( + content: Buffer, + properties: ConsumeMessage["properties"], + routingKey: string, + ): void { + const consumer = this.replyConsumer; + if (this.closed || !consumer) return; + const message = { + content, + fields: { + consumerTag: "", + deliveryTag: 0, + redelivered: false, + exchange: "", + routingKey, + }, + properties, + } as ConsumeMessage; + soon(() => { + void consumer(message); + }); + } +} diff --git a/packages/testing/src/index.ts b/packages/testing/src/index.ts index a3a0a004..f1f28569 100644 --- a/packages/testing/src/index.ts +++ b/packages/testing/src/index.ts @@ -9,3 +9,4 @@ export { default as globalSetup } from "./global-setup.js"; export { it } from "./extension.js"; +export { InMemoryAmqpBroker } from "./in-memory.js"; diff --git a/packages/worker/src/retry.ts b/packages/worker/src/retry.ts index 106a70d5..20769682 100644 --- a/packages/worker/src/retry.ts +++ b/packages/worker/src/retry.ts @@ -6,18 +6,18 @@ import { ttlBackoffWaitQueueName, } from "@amqp-contract/contract"; import { _internal_queueHasDeadLetterExchange } from "@amqp-contract/contract/internal"; -import type { AmqpClient, Logger } from "@amqp-contract/core"; +import type { AmqpTransport, Logger } from "@amqp-contract/core"; import type { ConsumeMessage } from "amqplib"; import { OkAsync, type AsyncResult } from "unthrown"; import { NonRetryableError } from "./errors.js"; type RetryContext = { - amqpClient: AmqpClient; + amqpClient: AmqpTransport; logger?: Logger | undefined; /** * Channel epoch captured when the message was delivered - * ({@link AmqpClient.currentChannelEpoch}). Stamped onto every ack/nack so + * ({@link AmqpTransport.currentChannelEpoch}). Stamped onto every ack/nack so * a settle that lands after a reconnect is skipped instead of targeting a * foreign delivery tag on the new channel. */ diff --git a/packages/worker/src/worker.ts b/packages/worker/src/worker.ts index 1c5ebbfc..a576becb 100644 --- a/packages/worker/src/worker.ts +++ b/packages/worker/src/worker.ts @@ -8,7 +8,7 @@ import { } from "@amqp-contract/contract"; import { _internal_queueHasDeadLetterExchange } from "@amqp-contract/contract/internal"; import { - AmqpClient, + type AmqpTransport, type AmqpConsumeOptions, type ConnectionError, type Logger, @@ -21,6 +21,7 @@ import { endSpanSuccess, isRpcError, recordConsumeMetric, + resolveTransport, safeJsonParse, startConsumeSpan, technicalDefect, @@ -259,8 +260,23 @@ export type CreateWorkerOptions< * substitutes the message payload, re-validated before the handler runs. */ middleware?: WorkerMiddleware | readonly AnyWorkerMiddleware[] | undefined; - /** AMQP broker URL(s). Multiple URLs provide failover support */ - 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 consume through instead of dialling a broker — an + * {@link AmqpTransport}, which the real `AmqpClient` satisfies. + * + * `@amqp-contract/testing`'s `InMemoryAmqpBroker` hands one back, so a spec + * can exercise the whole worker pipeline — validation, middleware, retry + * routing, dead-lettering — with no Docker. Supplying one makes `urls`, + * `connectionOptions` and `connectTimeoutMs` meaningless; passing both is + * refused rather than silently preferring one. + */ + transport?: AmqpTransport | undefined; /** Optional connection configuration (heartbeat, reconnect settings, etc.) */ connectionOptions?: AmqpConnectionManagerOptions | undefined; /** Optional logger for logging message consumption and errors */ @@ -365,7 +381,7 @@ export class TypedAmqpWorker { private constructor( private readonly contract: TContract, - private readonly amqpClient: AmqpClient, + private readonly amqpClient: AmqpTransport, handlers: WorkerInferHandlers, private readonly defaultConsumerOptions: ConsumerOptions, private readonly logger?: Logger, @@ -479,6 +495,7 @@ export class TypedAmqpWorker { createContext, middleware, urls, + transport, connectionOptions, defaultConsumerOptions, logger, @@ -538,8 +555,9 @@ export class TypedAmqpWorker { return OkAsync(undefined).flatMap(() => { const worker = new TypedAmqpWorker( contract, - new AmqpClient(contract, { + resolveTransport(contract, { urls, + transport, connectionOptions, connectTimeoutMs, publishTimeoutMs, diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 18a1f963..4340afa3 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -549,9 +549,6 @@ importers: specifier: 'catalog:' version: 2.0.1 devDependencies: - '@amqp-contract/testing': - specifier: workspace:* - version: link:../testing '@arethetypeswrong/cli': specifier: 'catalog:' version: 0.18.5 @@ -597,6 +594,12 @@ importers: packages/testing: dependencies: + '@amqp-contract/contract': + specifier: workspace:* + version: link:../contract + '@amqp-contract/core': + specifier: workspace:* + version: link:../core amqplib: specifier: 'catalog:' version: 2.0.1 @@ -631,6 +634,9 @@ importers: typescript: specifier: 'catalog:' version: 6.0.3 + unthrown: + specifier: 'catalog:' + version: 5.7.0 vitest: specifier: 'catalog:' version: 4.1.10(@opentelemetry/api@1.9.1)(@types/node@26.2.0)(@vitest/coverage-v8@4.1.10)(jiti@2.7.0)(tsx@4.23.12)(yaml@2.9.0) diff --git a/tests/src/in-memory.spec.ts b/tests/src/in-memory.spec.ts new file mode 100644 index 00000000..9199532b --- /dev/null +++ b/tests/src/in-memory.spec.ts @@ -0,0 +1,169 @@ +import { TypedAmqpClient } from "@amqp-contract/client"; +import { + defineContract, + defineEventConsumer, + defineEventPublisher, + defineExchange, + defineMessage, + defineQueue, + defineQueueBinding, + defineRpc, +} from "@amqp-contract/contract"; +import { InMemoryAmqpBroker } from "@amqp-contract/testing/in-memory"; +import { NonRetryableError, TypedAmqpWorker } from "@amqp-contract/worker"; +import { ErrAsync, OkAsync } from "unthrown"; +import { describe, expect, it, vi } from "vitest"; +import { z } from "zod"; + +/** + * The whole point of these: they exercise the contract pipeline — both + * validation passes, serialization, RPC correlation, dead-lettering — in the + * **unit** project, where nothing may need a broker. The integration suite + * keeps the broker behaviours a fake cannot honestly claim. + */ +const orderPlaced = defineMessage(z.object({ orderId: z.string(), total: z.number() })); + +const events = defineExchange("events", { type: "topic", durable: false }); +const parked = defineExchange("parked", { durable: false }); +const orders = defineQueue("orders", { + type: "classic", + durable: false, + deadLetter: { exchange: parked }, +}); +const dlq = defineQueue("orders-dlq", { type: "classic", durable: false }); + +const placeOrder = defineEventPublisher(events, orderPlaced, { routingKey: "order.placed" }); + +const pubSubContract = defineContract({ + publishers: { placeOrder }, + consumers: { onOrder: defineEventConsumer(placeOrder, orders) }, + queues: { dlq }, + bindings: { dlqBinding: defineQueueBinding(dlq, parked, { routingKey: "#" }) }, +}); + +describe("the in-memory broker carries a contract end to end", () => { + it("delivers a published event to the consumer that is bound for it", async () => { + // GIVEN a worker and a client on one broker, with no Docker anywhere + const broker = new InMemoryAmqpBroker(); + const received: unknown[] = []; + const worker = await TypedAmqpWorker.create({ + contract: pubSubContract, + handlers: { + onOrder: (_helpers, { payload }) => { + received.push(payload); + return OkAsync(); + }, + }, + transport: broker.createTransport(pubSubContract), + }).getOrThrow(); + const client = await TypedAmqpClient.create({ + contract: pubSubContract, + transport: broker.createTransport(pubSubContract), + }).getOrThrow(); + + // WHEN an event is published + await client.publish("placeOrder", { orderId: "o-1", total: 42 }).getOrThrow(); + await vi.waitUntil(() => received.length > 0); + + // THEN it arrives validated and parsed, the routing key having matched + expect(received).toEqual([{ orderId: "o-1", total: 42 }]); + await worker.close().get(); + await client.close().get(); + }); + + it("dead-letters a handler failure to the queue's dead-letter exchange", async () => { + // GIVEN a handler that refuses every message + const broker = new InMemoryAmqpBroker(); + const worker = await TypedAmqpWorker.create({ + contract: pubSubContract, + handlers: { onOrder: () => ErrAsync(new NonRetryableError("nope")) }, + transport: broker.createTransport(pubSubContract), + }).getOrThrow(); + const client = await TypedAmqpClient.create({ + contract: pubSubContract, + transport: broker.createTransport(pubSubContract), + }).getOrThrow(); + + // WHEN an event is published + await client.publish("placeOrder", { orderId: "o-2", total: 1 }).getOrThrow(); + await vi.waitUntil(() => broker.peek("orders-dlq").length > 0); + + // THEN the message is parked on the DLQ, body intact + expect( + broker.peek("orders-dlq").map((message) => JSON.parse(message.content.toString())), + ).toEqual([{ orderId: "o-2", total: 1 }]); + await worker.close({ drainTimeoutMs: 200 }).get(); + await client.close().get(); + }); +}); + +const rpcQueue = defineQueue("calculate", { + type: "classic", + durable: false, + deadLetter: { exchange: parked }, +}); +const calculate = defineRpc(rpcQueue, { + request: defineMessage(z.object({ a: z.number(), b: z.number() })), + response: defineMessage(z.object({ sum: z.number() })), + errors: { OVERFLOW: { data: z.object({ limit: z.number() }) } }, +}); + +const rpcContract = defineContract({ + rpcs: { calculate }, + queues: { dlq }, + bindings: { dlqBinding: defineQueueBinding(dlq, parked, { routingKey: "#" }) }, +}); + +describe("the in-memory broker carries an RPC", () => { + it("routes a reply back through direct reply-to", async () => { + // GIVEN an RPC worker and a client, both on the in-memory broker + const broker = new InMemoryAmqpBroker(); + const worker = await TypedAmqpWorker.create({ + contract: rpcContract, + handlers: { calculate: (_helpers, { payload }) => OkAsync({ sum: payload.a + payload.b }) }, + transport: broker.createTransport(rpcContract), + }).getOrThrow(); + const client = await TypedAmqpClient.create({ + contract: rpcContract, + transport: broker.createTransport(rpcContract), + }).getOrThrow(); + + // WHEN the client calls + // THEN the reply comes home, correlated, through the reply pseudo-queue + const outcome = await client.call("calculate", { a: 2, b: 3 }, { timeoutMs: 2_000 }); + + expect(outcome.isOk() ? outcome.value : outcome.isDefect() ? outcome.cause : outcome).toEqual({ + sum: 5, + }); + await worker.close({ drainTimeoutMs: 200 }).get(); + await client.close().get(); + }); + + it("carries a declared RPC error back as a typed failure", async () => { + // GIVEN a handler that answers a declared error code + const broker = new InMemoryAmqpBroker(); + const worker = await TypedAmqpWorker.create({ + contract: rpcContract, + handlers: { + calculate: (helpers) => ErrAsync(helpers.errors.OVERFLOW({ limit: 100 })), + }, + transport: broker.createTransport(rpcContract), + }).getOrThrow(); + const client = await TypedAmqpClient.create({ + contract: rpcContract, + transport: broker.createTransport(rpcContract), + }).getOrThrow(); + + // WHEN the client calls + const outcome = await client.call("calculate", { a: 1, b: 1 }, { timeoutMs: 2_000 }); + + // THEN the failure arrives as the declared code, not as a timeout + // `code` is the discriminator only on the declared-error arm; the others + // are the timeout and validation failures the same channel can carry. + expect(outcome.isErr() && "code" in outcome.error ? outcome.error.code : outcome).toBe( + "OVERFLOW", + ); + await worker.close({ drainTimeoutMs: 200 }).get(); + await client.close().get(); + }); +}); diff --git a/turbo.json b/turbo.json index 48922a00..14c03ead 100644 --- a/turbo.json +++ b/turbo.json @@ -47,6 +47,12 @@ "check:package": { "dependsOn": ["build"], "outputs": [] + }, + "@amqp-contract/core#typecheck": { + "dependsOn": ["^build", "@amqp-contract/testing#build"] + }, + "@amqp-contract/core#test:integration": { + "dependsOn": ["^build", "@amqp-contract/testing#build"] } } } diff --git a/vitest.shared.ts b/vitest.shared.ts index e525dfa3..4bb9dddd 100644 --- a/vitest.shared.ts +++ b/vitest.shared.ts @@ -35,6 +35,19 @@ export type SharedConfigOptions = { integration?: boolean; /** Setup file applied to every project, relative to the package root. */ setupFile?: string; + /** + * Module aliases, for a workspace that must reach a sibling it cannot + * declare a dependency on. `packages/core` is the one: it consumes + * `@amqp-contract/testing`'s fixtures, and that package depends on core, so + * the edge back would be a cycle. + */ + alias?: Record | undefined; + /** + * Where the integration project's global setup lives. `resolve.alias` does + * not reach this — vitest resolves it as a path of its own — so a workspace + * that cannot declare the dependency has to name the file. + */ + globalSetup?: string | undefined; }; /** Type tests are typechecked, never executed — see the `include` note below. */ @@ -45,10 +58,14 @@ export function sharedVitestConfig({ typecheck = false, integration = false, setupFile, + alias, + globalSetup = "@amqp-contract/testing/global-setup", }: SharedConfigOptions = {}) { const setupFiles = setupFile ? { setupFiles: [setupFile] } : {}; + const resolve = alias ? { resolve: { alias } } : {}; return { + ...resolve, test: { environment: "node", reporters: ["default"], @@ -70,6 +87,7 @@ export function sharedVitestConfig({ ? { projects: [ { + ...resolve, test: { // Runs in the main gate. No broker: nothing here may need one. name: "unit", @@ -81,11 +99,12 @@ export function sharedVitestConfig({ }, }, { + ...resolve, test: { name: "integration", environment: "node", ...setupFiles, - globalSetup: "@amqp-contract/testing/global-setup", + globalSetup, include: ["src/**/__tests__/*.spec.ts"], testTimeout: 10_000, hookTimeout: 10_000,