@amqp-contract/worker

@amqp-contract/worker

Classes

ConnectionError

Defined in: packages/core/dist/telemetry-U0DumV05.d.mts:63

The broker could not be reached: refused, unresolvable, unauthorized, or still not ready when connectTimeoutMs elapsed.

Modeled, not a defect — unlike TechnicalError. An unreachable broker is the anticipated failure of dialing one: it is what a wrong URL, a rotated credential or a cluster that has not come up yet look like, every one of them an operator's business rather than a bug in the caller. So TypedAmqpWorker.create and TypedAmqpClient.create report it on the E channel, where a start-up path can triage it by tag and turn it into an exit code, a retry, or a health probe — and the defect channel keeps its meaning: the failures nobody anticipated.

Carries a _tag of "@amqp-contract/ConnectionError"; the human-facing Error.name is kept bare ("ConnectionError"). The underlying amqplib rejection is on cause.

Extends

ConnectionError_base<{ cause?: unknown; }>

Constructors

Constructor

new ConnectionError(message, cause?): ConnectionError;

Defined in: packages/core/dist/telemetry-U0DumV05.d.mts:68

Parameters

Parameter	Type
`message`	`string`
`cause?`	`unknown`

Returns

ConnectionError

Overrides

ConnectionError_base<{ cause?: unknown; }>.constructor

Properties

Property	Modifier	Type	Default value	Description	Inherited from	Defined in
`_tag`	`readonly`	`"@amqp-contract/ConnectionError"`	`undefined`	-	`ConnectionError_base._tag`	node_modules/.pnpm/unthrown@5.12.2/node_modules/unthrown/dist/index.d.mts:2363
`cause?`	`public`	`unknown`	`undefined`	-	`MessageValidationError`.`cause`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es2022.error.d.ts:24
`message`	`public`	`string`	`undefined`	-	`ConnectionError_base.message`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1075
`name`	`public`	`string`	`undefined`	-	`ConnectionError_base.name`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1074
`stack?`	`public`	`string`	`undefined`	-	`ConnectionError_base.stack`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1076
`tag`	`static readonly`	`"@amqp-contract/ConnectionError"`	`"@amqp-contract/ConnectionError"`	The `_tag`, for `P.tag(ConnectionError.tag)` without a raw string.	-	packages/core/dist/telemetry-U0DumV05.d.mts:67

MessageValidationError

Defined in: packages/core/dist/telemetry-U0DumV05.d.mts:126

Error thrown when message validation fails (payload or headers).

Used by both the client (publish-time payload validation) and the worker (consume-time payload and headers validation). Carries a _tag of "@amqp-contract/MessageValidationError" (namespaced to avoid collisions); the Error.name is kept bare ("MessageValidationError").

Param

source

The name of the publisher or consumer that triggered the validation

Param

issues

The validation issues from the Standard Schema validation

Extends

MessageValidationError_base<{ issues: unknown; source: string; }>

Constructors

Constructor

new MessageValidationError(source, issues): MessageValidationError;

Defined in: packages/core/dist/telemetry-U0DumV05.d.mts:132

Parameters

Parameter	Type
`source`	`string`
`issues`	`unknown`

Returns

MessageValidationError

Overrides

MessageValidationError_base<{ source: string; issues: unknown; }>.constructor

Properties

Property	Modifier	Type	Default value	Description	Inherited from	Defined in
`_tag`	`readonly`	`"@amqp-contract/MessageValidationError"`	`undefined`	-	`MessageValidationError_base._tag`	node_modules/.pnpm/unthrown@5.12.2/node_modules/unthrown/dist/index.d.mts:2363
`cause?`	`public`	`unknown`	`undefined`	-	`MessageValidationError_base.cause`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es2022.error.d.ts:24
`issues`	`readonly`	`unknown`	`undefined`	-	`MessageValidationError_base.issues`	packages/core/dist/telemetry-U0DumV05.d.mts:128
`message`	`public`	`string`	`undefined`	-	`MessageValidationError_base.message`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1075
`name`	`public`	`string`	`undefined`	-	`MessageValidationError_base.name`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1074
`source`	`readonly`	`string`	`undefined`	-	`MessageValidationError_base.source`	packages/core/dist/telemetry-U0DumV05.d.mts:127
`stack?`	`public`	`string`	`undefined`	-	`MessageValidationError_base.stack`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1076
`tag`	`static readonly`	`"@amqp-contract/MessageValidationError"`	`"@amqp-contract/MessageValidationError"`	The `_tag`, for `P.tag(MessageValidationError.tag)` without a raw string.	-	packages/core/dist/telemetry-U0DumV05.d.mts:131

NonRetryableError

Defined in: packages/worker/src/errors.ts:44

Non-retryable errors - permanent failures that should not be retried Examples: invalid data, business rule violations, permanent external failures

Use this error type when retrying would not help - the message will be immediately sent to the dead letter queue (DLQ) if configured. Carries a namespaced _tag of "@amqp-contract/NonRetryableError"; the Error.name is kept bare ("NonRetryableError").

Extends

TaggedErrorInstance<"@amqp-contract/NonRetryableError", { cause?: unknown; }>

Constructors

Constructor

new NonRetryableError(message, cause?): NonRetryableError;

Defined in: packages/worker/src/errors.ts:52

Parameters

Parameter	Type
`message`	`string`
`cause?`	`unknown`

Returns

NonRetryableError

Overrides

TaggedError("@amqp-contract/NonRetryableError", { name: "NonRetryableError", })<{ cause?: unknown; }>.constructor

Properties

Property	Modifier	Type	Default value	Description	Inherited from	Defined in
`_tag`	`readonly`	`"@amqp-contract/NonRetryableError"`	`undefined`	-	`TaggedError("@amqp-contract/NonRetryableError", { name: "NonRetryableError", })._tag`	node_modules/.pnpm/unthrown@5.12.2/node_modules/unthrown/dist/index.d.mts:2363
`cause?`	`public`	`unknown`	`undefined`	-	`MessageValidationError`.`cause`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es2022.error.d.ts:24
`message`	`public`	`string`	`undefined`	-	`TaggedError("@amqp-contract/NonRetryableError", { name: "NonRetryableError", }).message`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1075
`name`	`public`	`string`	`undefined`	-	`TaggedError("@amqp-contract/NonRetryableError", { name: "NonRetryableError", }).name`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1074
`stack?`	`public`	`string`	`undefined`	-	`TaggedError("@amqp-contract/NonRetryableError", { name: "NonRetryableError", }).stack`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1076
`tag`	`static readonly`	`"@amqp-contract/NonRetryableError"`	`"@amqp-contract/NonRetryableError"`	The `_tag`, for `P.tag(NonRetryableError.tag)` without a raw string.	-	packages/worker/src/errors.ts:50

PublishError

Defined in: packages/core/dist/telemetry-U0DumV05.d.mts:100

The broker side of a publish failed: timed out, nacked, or the channel closed under it. (A full write buffer is NOT one: on the confirm channel the wrapper only reports it after the broker confirmed the message.)

Modeled, not a defect — a broker that is down, overloaded or refusing a message is an operational condition a publisher is expected to handle (buffer, retry, shed load, surface a 503), not a bug. Returned on the E channel of AmqpClient.publish, TypedAmqpClient.publish and TypedAmqpClient.call; switch on PublishError.reason. A failure core cannot classify (an unencodable payload, an unknown rejection) stays a Defect with a TechnicalError cause.

Carries a _tag of "@amqp-contract/PublishError"; the Error.name is kept bare ("PublishError"). The underlying rejection, if any, is on cause.

Extends

PublishError_base<{ cause?: unknown; reason: PublishFailureReason; target: string; }>

Constructors

Constructor

new PublishError(props): PublishError;

Defined in: packages/core/dist/telemetry-U0DumV05.d.mts:108

Parameters

Parameter	Type
`props`	{ `cause?`: `unknown`; `reason`: `PublishFailureReason`; `target`: `string`; }
`props.cause?`	`unknown`
`props.reason`	`PublishFailureReason`
`props.target`	`string`

Returns

PublishError

Overrides

PublishError_base<{ reason: PublishFailureReason; /** Where the message was going, e.g. `exchange "orders" (routing key "order.created")`. */ target: string; cause?: unknown; }>.constructor

Properties

Property	Modifier	Type	Default value	Description	Inherited from	Defined in
`_tag`	`readonly`	`"@amqp-contract/PublishError"`	`undefined`	-	`PublishError_base._tag`	node_modules/.pnpm/unthrown@5.12.2/node_modules/unthrown/dist/index.d.mts:2363
`cause?`	`public`	`unknown`	`undefined`	-	`MessageValidationError`.`cause`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es2022.error.d.ts:24
`message`	`public`	`string`	`undefined`	-	`PublishError_base.message`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1075
`name`	`public`	`string`	`undefined`	-	`PublishError_base.name`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1074
`reason`	`readonly`	`PublishFailureReason`	`undefined`	-	`PublishError_base.reason`	packages/core/dist/telemetry-U0DumV05.d.mts:101
`stack?`	`public`	`string`	`undefined`	-	`PublishError_base.stack`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1076
`target`	`readonly`	`string`	`undefined`	Where the message was going, e.g. `exchange "orders" (routing key "order.created")`.	`PublishError_base.target`	packages/core/dist/telemetry-U0DumV05.d.mts:103
`tag`	`static readonly`	`"@amqp-contract/PublishError"`	`"@amqp-contract/PublishError"`	The `_tag`, for `P.tag(PublishError.tag)` without a raw string.	-	packages/core/dist/telemetry-U0DumV05.d.mts:107

RetryableError

Defined in: packages/worker/src/errors.ts:20

Retryable errors - transient failures that may succeed on retry Examples: network timeouts, rate limiting, temporary service unavailability

Use this error type when the operation might succeed if retried. The worker will apply exponential backoff and retry the message.

Built on unthrown's TaggedError, so it carries a namespaced _tag of "@amqp-contract/RetryableError" (to avoid colliding with other libraries' tags in a shared error matcher) for exhaustive dispatch via result.match({ ok, defect, errCases: (matcher) => matcher.with(P.tag("@amqp-contract/RetryableError"), …) }); the Error.name is kept bare ("RetryableError").

Extends

TaggedErrorInstance<"@amqp-contract/RetryableError", { cause?: unknown; }>

Constructors

Constructor

new RetryableError(message, cause?): RetryableError;

Defined in: packages/worker/src/errors.ts:28

Parameters

Parameter	Type
`message`	`string`
`cause?`	`unknown`

Returns

RetryableError

Overrides

TaggedError("@amqp-contract/RetryableError", { name: "RetryableError", })<{ cause?: unknown; }>.constructor

Properties

Property	Modifier	Type	Default value	Description	Inherited from	Defined in
`_tag`	`readonly`	`"@amqp-contract/RetryableError"`	`undefined`	-	`TaggedError("@amqp-contract/RetryableError", { name: "RetryableError", })._tag`	node_modules/.pnpm/unthrown@5.12.2/node_modules/unthrown/dist/index.d.mts:2363
`cause?`	`public`	`unknown`	`undefined`	-	`MessageValidationError`.`cause`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es2022.error.d.ts:24
`message`	`public`	`string`	`undefined`	-	`TaggedError("@amqp-contract/RetryableError", { name: "RetryableError", }).message`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1075
`name`	`public`	`string`	`undefined`	-	`TaggedError("@amqp-contract/RetryableError", { name: "RetryableError", }).name`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1074
`stack?`	`public`	`string`	`undefined`	-	`TaggedError("@amqp-contract/RetryableError", { name: "RetryableError", }).stack`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1076
`tag`	`static readonly`	`"@amqp-contract/RetryableError"`	`"@amqp-contract/RetryableError"`	The `_tag`, for `P.tag(RetryableError.tag)` without a raw string.	-	packages/worker/src/errors.ts:26

RpcError

Defined in: packages/core/dist/telemetry-U0DumV05.d.mts:177

A typed, contract-declared RPC error — the business-failure channel of an RPC, as opposed to transport failures (which surface as a Defect with a TechnicalError cause).

Declared per-RPC via defineRpc(queue, { request, response, errors }), where each error code maps to a message definition validating the error's data payload. A worker handler surfaces one by returning Err(rpcError(code, data)); the worker validates data against the declared schema, publishes an error reply, and acks the request (business errors are not retried). The caller's client.call(...) resolves to Err(RpcError<code, data>) with data re-validated on arrival.

Carries a _tag of "@amqp-contract/RpcError" for exhaustive dispatch via the error matcher (result.match({ ok, defect, errCases: (matcher) => matcher.with(P.tag("@amqp-contract/RpcError"), …) })); the Error.name is kept bare ("RpcError"). Discriminate between codes on the code property.

Extends

RpcError_base<{ code: string; data: unknown; }>

Type Parameters

Type Parameter	Default type
`TCode` extends `string`	`string`
`TData`	`unknown`

Constructors

Constructor

new RpcError<TCode, TData>(
   code, 
   data, 
   message?
): RpcError<TCode, TData>;

Defined in: packages/core/dist/telemetry-U0DumV05.d.mts:185

Parameters

Parameter	Type
`code`	`TCode`
`data`	`TData`
`message?`	`string`

Returns

RpcError<TCode, TData>

Overrides

RpcError_base<{ code: string; data: unknown; }>.constructor

Properties

Property	Modifier	Type	Default value	Description	Overrides	Inherited from	Defined in
`_tag`	`readonly`	`"@amqp-contract/RpcError"`	`undefined`	-	-	`RpcError_base._tag`	node_modules/.pnpm/unthrown@5.12.2/node_modules/unthrown/dist/index.d.mts:2363
`cause?`	`public`	`unknown`	`undefined`	-	-	`RpcError_base.cause`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es2022.error.d.ts:24
`code`	`readonly`	`TCode`	`undefined`	-	`RpcError_base.code`	-	packages/core/dist/telemetry-U0DumV05.d.mts:183
`data`	`readonly`	`TData`	`undefined`	-	`RpcError_base.data`	-	packages/core/dist/telemetry-U0DumV05.d.mts:184
`message`	`public`	`string`	`undefined`	-	-	`RpcError_base.message`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1075
`name`	`public`	`string`	`undefined`	-	-	`RpcError_base.name`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1074
`stack?`	`public`	`string`	`undefined`	-	-	`RpcError_base.stack`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1076
`tag`	`static readonly`	`"@amqp-contract/RpcError"`	`"@amqp-contract/RpcError"`	The `_tag`, for `P.tag(RpcError.tag)` without a raw string.	-	-	packages/core/dist/telemetry-U0DumV05.d.mts:182

TechnicalError

Defined in: packages/core/dist/telemetry-U0DumV05.d.mts:38

Error for technical/runtime failures that cannot be prevented by TypeScript.

This includes channel issues, compression/parse faults, and other unexpected runtime errors. Shared across core, worker, and client packages. Dialing the broker is NOT one of them any more — that failure is anticipated, and it is modeled as ConnectionError.

These failures are unexpected, so @amqp-contract surfaces them through unthrown's defect channel, not the modeled E channel: a TechnicalError instance is carried as the cause of a Defect (so its message/cause survive for logging), and is handled in the defect arm of result.match({ ok, errCases, defect }) — or via recoverDefect / tapDefect — never matched in errCases. It is deliberately absent from every operation's E (only anticipated domain failures live there).

Built on unthrown's TaggedError, so it carries a _tag of "@amqp-contract/TechnicalError" (namespaced to avoid colliding with other libraries' tags); the human-facing Error.name is kept bare ("TechnicalError"). Remains a real Error.

Extends

TechnicalError_base<{ cause?: unknown; }>

Constructors

Constructor

new TechnicalError(message, cause?): TechnicalError;

Defined in: packages/core/dist/telemetry-U0DumV05.d.mts:43

Parameters

Parameter	Type
`message`	`string`
`cause?`	`unknown`

Returns

TechnicalError

Overrides

TechnicalError_base<{ cause?: unknown; }>.constructor

Properties

Property	Modifier	Type	Default value	Description	Inherited from	Defined in
`_tag`	`readonly`	`"@amqp-contract/TechnicalError"`	`undefined`	-	`TechnicalError_base._tag`	node_modules/.pnpm/unthrown@5.12.2/node_modules/unthrown/dist/index.d.mts:2363
`cause?`	`public`	`unknown`	`undefined`	-	`MessageValidationError`.`cause`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es2022.error.d.ts:24
`message`	`public`	`string`	`undefined`	-	`TechnicalError_base.message`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1075
`name`	`public`	`string`	`undefined`	-	`TechnicalError_base.name`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1074
`stack?`	`public`	`string`	`undefined`	-	`TechnicalError_base.stack`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1076
`tag`	`static readonly`	`"@amqp-contract/TechnicalError"`	`"@amqp-contract/TechnicalError"`	The `_tag`, for `P.tag(TechnicalError.tag)` without a raw string.	-	packages/core/dist/telemetry-U0DumV05.d.mts:42

TypedAmqpWorker

Defined in: packages/worker/src/worker.ts:384

Type-safe AMQP worker for consuming messages from RabbitMQ.

This class provides automatic message validation, connection management, and error handling for consuming messages based on a contract definition.

Example

typescript

import { TypedAmqpWorker } from '@amqp-contract/worker';
import { defineQueue, defineMessage, defineContract, defineConsumer } from '@amqp-contract/contract';
import { OkAsync } from 'unthrown';
import { z } from 'zod';

const orderQueue = defineQueue('order-processing');
const orderMessage = defineMessage(z.object({
  orderId: z.string(),
  amount: z.number()
}));

const contract = defineContract({
  consumers: {
    processOrder: defineConsumer(orderQueue, orderMessage)
  }
});

const worker = await TypedAmqpWorker.create({
  contract,
  handlers: {
    processOrder: ({ input: { payload } }) => {
      console.log('Processing order', payload.orderId);
      return OkAsync(undefined);
    },
  },
  urls: ['amqp://localhost'],
}).getOrThrow();

// Close when done (drains in-flight handlers first)
await worker.close().get();

Type Parameters

Type Parameter	Description
`TContract` extends `ContractDefinition`	The contract definition type

Methods

close()

close(options?): AsyncResult<void, never>;

Defined in: packages/worker/src/worker.ts:653

Close the AMQP channel and connection.

Graceful shutdown in three steps: cancel every consumer (no new deliveries), drain in-flight handlers so their acks/nacks land on the still-open channel, then close the channel and release the connection.

The drain waits up to drainTimeoutMs (default DEFAULT_DRAIN_TIMEOUT_MS) for in-flight handlers. On timeout the teardown proceeds anyway — the un-acked deliveries are redelivered by the broker, preserving at-least-once semantics — and a warning is logged. Pass null to wait indefinitely.

Parameters

Parameter	Type
`options?`	{ `drainTimeoutMs?`: `number` \| `null`; }
`options.drainTimeoutMs?`	`number` \| `null`

Returns

AsyncResult<void, never>

Example

typescript

await worker.close().get();

isConnected()

isConnected(): boolean;

Defined in: packages/worker/src/worker.ts:631

Whether the broker connection is currently up — for a readiness/health probe. false while reconnecting (consumers resume on reconnect).

Returns

boolean

create()

static create<TContract, TCreated, TContext>(__namedParameters): AsyncResult<TypedAmqpWorker<TContract>, ConnectionError>;

Defined in: packages/worker/src/worker.ts:512

Create a type-safe AMQP worker from a contract.

Connection management (including automatic reconnection) is handled internally by amqp-connection-manager via the AmqpClient. The worker will set up consumers for all contract-defined handlers asynchronously in the background once the underlying connection and channels are ready.

Connections are automatically shared across clients and workers with the same URLs and connection options, following RabbitMQ best practices.

Type Parameters

Type Parameter	Default type
`TContract` extends `ContractDefinition`	-
`TCreated` extends `Record`<`string`, `unknown`> \| `EmptyContext`	`EmptyContext`
`TContext` extends `Record`<`string`, `unknown`> \| `EmptyContext`	`TCreated`

Parameters

Parameter	Type
`__namedParameters`	`CreateWorkerOptions`<`TContract`, `TCreated`, `TContext`>

Returns

AsyncResult<TypedAmqpWorker<TContract>, ConnectionError>

An AsyncResult that resolves to the worker. An unreachable broker is a modeled Err({@link ConnectionError}) — the anticipated failure of dialing one, and the case a start-up path wants to branch on. Everything else that can go wrong here (a bad option, a topology assert the broker refuses, a bug in a provider) stays on the Defect channel with a TechnicalError cause.

Example

typescript

const result = await TypedAmqpWorker.create({
  contract: myContract,
  handlers: {
    processOrder: ({ input: { payload } }) => OkAsync(undefined),
  },
  urls: ['amqp://localhost'],
});

Type Aliases

AnyWorkerMiddleware

type AnyWorkerMiddleware = WorkerMiddleware<Record<string, unknown>, Record<string, unknown>>;

Defined in: packages/worker/src/middleware.ts:108

Widened middleware shape used internally by the dispatcher, where context types have been erased. Public call sites keep the typed WorkerMiddleware form.

ConnectionSource

type ConnectionSource = object;

Defined in: packages/core/dist/index.d.mts:47

Where a client's connection comes from — exactly one of (checked at construction; passing both or neither is a defect):

urls (+ optional connectionOptions): a connection dialled and pooled by amqp-contract, released when the last client using it closes. Multiple URLs provide failover.
connection: an AmqpConnectionManager the caller created and owns. The client only opens a channel on it; close() never closes it. Pass the same one to a client and a worker to share a TCP connection explicitly.

A flat type rather than a union so Omit<…>-style wrappers keep working.

Properties

Property	Type	Defined in
`connection?`	`AmqpConnectionManager`	packages/core/dist/index.d.mts:50
`connectionOptions?`	`AmqpConnectionManagerOptions`	packages/core/dist/index.d.mts:49
`urls?`	`ConnectionUrl`[]	packages/core/dist/index.d.mts:48

ConsumerHandler

type ConsumerHandler<TPayload, THeaders, TContext> = (helpers, message) => AsyncResult<void, HandlerError>;

Defined in: packages/worker/src/types.ts:315

A consumer handler for messages whose validated payload is TPayload (and headers THeaders): the helpers record first, the message second, an AsyncResult<void, HandlerError> back. This short alias — over the RESOLVED payload — is what a mistake in a handler reports (ConsumerHandler<{ to: string; }, …>), never the whole contract type.

Type Parameters

Type Parameter	Default type
`TPayload`	-
`THeaders`	`undefined`
`TContext` extends `Record`<`string`, `unknown`> \| `EmptyContext`	`EmptyContext`

Parameters

Parameter	Type
`helpers`	`WorkerHandlerHelpers`<`TContext`, `EmptyContext`, `WorkerConsumedMessage`<`TPayload`, `THeaders`>>
`message`	`WorkerConsumedMessage`<`TPayload`, `THeaders`>

Returns

AsyncResult<void, HandlerError>

ConsumerHandlerEntry

type ConsumerHandlerEntry<TPayload, THeaders, TContext> = 
  | ConsumerHandler<TPayload, THeaders, TContext>
  | readonly [ConsumerHandler<TPayload, THeaders, TContext>, ConsumerOptions];

Defined in: packages/worker/src/types.ts:325

A ConsumerHandler, or a [handler, consumerOptions] tuple.

Type Parameters

Type Parameter	Default type
`TPayload`	-
`THeaders`	`undefined`
`TContext` extends `Record`<`string`, `unknown`> \| `EmptyContext`	`EmptyContext`

ConsumerOptions

type ConsumerOptions = Pick<AmqpConsumeOptions, "prefetch" | "priority" | "arguments" | "consumerTag" | "exclusive">;

Defined in: packages/worker/src/worker.ts:156

Consumer options a worker handler may configure — a curated subset of the AMQP consume options. noAck and noLocal are deliberately excluded: noAck: true would silently break the worker's ack-exactly-once and retry/DLQ invariants (deliveries would be considered settled on send), and noLocal is not supported by RabbitMQ.

CreateWorkerOptions

type CreateWorkerOptions<TContract, TCreated, TContext> = ConnectionSource & object;

Defined in: packages/worker/src/worker.ts:214

Options for creating a type-safe AMQP worker.

Type Parameters

Type Parameter	Default type	Description
`TContract` extends `ContractDefinition`	-	The contract definition type
`TCreated` extends `Record`<`string`, `unknown`> \| `EmptyContext`	`EmptyContext`	-
`TContext` extends `TCreated`	`TCreated`	-

Type Declaration

Name	Type	Description	Defined in
`connectTimeoutMs?`	`number` \| `null`	Maximum time in ms to wait for the AMQP connection to become ready before `create()` answers `Err(ConnectionError)`. Defaults to 30s (the AmqpClient's `DEFAULT_CONNECT_TIMEOUT_MS`). Pass `null` to disable the timeout and let amqp-connection-manager retry indefinitely.	packages/worker/src/worker.ts:303
`contract`	`TContract`	The AMQP contract definition specifying consumers and their message schemas	packages/worker/src/worker.ts:220
`createContext()?`	(`info`) => `TCreated` \| `Promise`<`TCreated`>	Build the per-message dependency context — the seed of the middleware chain (and, without middleware, the context handlers receive directly in `helpers.context`). Invoked once per message after validation, so it can produce request-scoped values (correlation-id loggers, per-message transactions); close over singletons for per-worker dependencies. A rejection/throw routes the message to the DLQ as a NonRetryableError. demesne's `Layer.forkScope` is the recommended implementation for DI-managed graphs.	packages/worker/src/worker.ts:252
`defaultConsumerOptions?`	`ConsumerOptions`	Optional default consumer options applied to all consumer handlers. Handler-specific options provided in tuple form override these defaults.	packages/worker/src/worker.ts:296
`handlers`	`WorkerInferHandlers`<`TContract`, `TContext`>	Handlers for each `consumers` and `rpcs` entry in the contract. - Regular consumers return `AsyncResult<void, HandlerError>`. - RPC handlers return `AsyncResult<TResponse, HandlerError>` where `TResponse` is inferred from the RPC's response message schema. When the RPC declares an `errors` map, the error channel additionally accepts the declared `RpcError<code, data>` members (otherwise it stays plain `HandlerError`). Handlers receive one record FIRST — `{ input, context, errors, raw, retryable, nonRetryable }`, where `input` is the validated message — and that message again as the second parameter. `context` is an empty object when no `createContext` and no `middleware` are configured. Use `declareHandler` / `declareHandlers` to create handlers with full type inference.	packages/worker/src/worker.ts:240
`logger?`	`Logger`	Optional logger for logging message consumption and errors	packages/worker/src/worker.ts:285
`maxDecompressedBytes?`	`number`	Deprecated Renamed maxMessageBytes (it caps plain bodies too).	packages/worker/src/worker.ts:324
`maxMessageBytes?`	`number`	Cap on the size (bytes) of a single inbound message — the plain body, or the decompressed one (guards against a decompression bomb: a few-KB payload that expands to gigabytes before schema validation runs). Over-cap messages follow the poison-message DLQ path. Defaults to core's `DEFAULT_MAX_MESSAGE_BYTES` (16 MiB).	packages/worker/src/worker.ts:322
`middleware?`	\| `WorkerMiddleware`<`TCreated`, `TContext`> \| readonly `AnyWorkerMiddleware`[]	Optional middleware wrapping every handler invocation (consumers and RPCs), applied after message validation. The chain is seeded with the `createContext` result (an empty object when none is configured). Accepts either a single middleware or an array (first entry = outermost, mirroring the client's interceptor arrays). The array form composes at runtime exactly like `composeMiddleware(...)`, but cannot thread the stepwise context types across entries — when middleware accumulate typed context for the handlers, pre-compose with `composeMiddleware(outermost, ..., innermost)` so the chain's final context type is inferred into `helpers.context`. A middleware can short-circuit by returning without calling `next`: handler-style errors route through retry/DLQ (or a typed RPC error reply), and an `Ok(value)` skips the handler entirely. `next({ payload })` substitutes the message payload, re-validated before the handler runs.	packages/worker/src/worker.ts:271
`publishTimeoutMs?`	`number` \| `null`	Maximum time in ms a worker-side publish (retry republish, RPC reply) may sit buffered waiting for the broker before it fails with a `PublishError` (reason `"timeout"`): a failed retry republish requeues the original, a failed RPC reply dead-letters the request. Maps to amqp-connection-manager's channel-level `publishTimeout`. Defaults to 30s (the AmqpClient's `DEFAULT_PUBLISH_TIMEOUT_MS`). Pass `null` to disable, restoring unbounded buffering — a publish issued during an outage then never settles.	packages/worker/src/worker.ts:314
`rpc?`	`object`	RPC server options.	packages/worker/src/worker.ts:326
`rpc.allowReplyTo()?`	(`replyTo`) => `boolean`	Which `replyTo` addresses the worker may publish a reply to. By default only RabbitMQ direct reply-to (`amq.rabbitmq.reply-to`, delivered as `amq.rabbitmq.reply-to.<token>`), which is what `client.call()` uses. A request whose `replyTo` is refused is dead-lettered with the reason logged — never replied to — so a forged request cannot make the worker publish into an arbitrary queue. Pass a predicate to allow other reply queues.	packages/worker/src/worker.ts:337
`telemetry?`	`TelemetryProvider`	Optional telemetry provider for tracing and metrics. If not provided, uses the default provider which attempts to load OpenTelemetry. OpenTelemetry instrumentation is automatically enabled if @opentelemetry/api is installed.	packages/worker/src/worker.ts:291
`topology?`	`TopologyMode`	What the worker does with its topology on every (re)connect. The worker only touches what its consumers need: the queues it consumes (with their retry wait queues and the bindings into them, and the exchanges those bind to) plus their dead-letter exchanges and DLQs — never publisher-only exchanges or unrelated queues. - `"assert"` (default) — declare them. - `"passive"` — only check they exist; `create()` fails if one is missing. - `"none"` — touch nothing (topology is provisioned elsewhere).	packages/worker/src/worker.ts:283

Example

typescript

const options: CreateWorkerOptions<typeof contract> = {
  contract: myContract,
  handlers: {
    // Simple handler
    processOrder: ({ input: { payload } }) => {
      console.log('Processing order:', payload.orderId);
      return OkAsync(undefined);
    },
    // Handler with prefetch configuration
    processPayment: [
      ({ input: { payload } }) => {
        console.log('Processing payment:', payload.paymentId);
        return OkAsync(undefined);
      },
      { prefetch: 10 }
    ]
  },
  urls: ['amqp://localhost'],
  defaultConsumerOptions: {
    prefetch: 5,
  },
  connectionOptions: {
    heartbeatIntervalInSeconds: 30
  },
  logger: myLogger
};

Note: Retry configuration is defined at the queue level in the contract, not at the handler level. See QueueDefinition.retry for configuration options.

EmptyContext

type EmptyContext = Record<never, never>;

Defined in: packages/worker/src/middleware.ts:13

The middleware context handlers see when no middleware injects anything.

Record<never, never> rather than {} so an empty context is a real "no properties" type instead of the anything-goes empty-object type.

HandlerError

type HandlerError = 
  | RetryableError
  | NonRetryableError;

Defined in: packages/worker/src/errors.ts:70

Any handler-signalled error — the union a handler may put in the Err channel of its AsyncResult. Discriminate on _tag ("@amqp-contract/RetryableError" / "@amqp-contract/NonRetryableError"), e.g. with the error matcher (matcher.with(P.tag("@amqp-contract/RetryableError"), …)).

Previously an abstract base class; now a tagged union, because unthrown's TaggedError mints a distinct base class per tag. Use isHandlerError for runtime narrowing instead of instanceof HandlerError.

Logger

type Logger = object;

Defined in: packages/core/dist/telemetry-U0DumV05.d.mts:249

Logger interface for amqp-contract packages.

Provides a simple logging abstraction that can be implemented by users to integrate with their preferred logging framework.

Example

typescript

// Simple console logger implementation
const logger: Logger = {
  debug: (message, context) => console.debug(message, context),
  info: (message, context) => console.info(message, context),
  warn: (message, context) => console.warn(message, context),
  error: (message, context) => console.error(message, context),
};

Methods

debug()

debug(message, context?): void;

Defined in: packages/core/dist/telemetry-U0DumV05.d.mts:255

Log debug level messages

Parameters

Parameter	Type	Description
`message`	`string`	The log message
`context?`	`LoggerContext`	Optional context to include with the log

Returns

void

error()

error(message, context?): void;

Defined in: packages/core/dist/telemetry-U0DumV05.d.mts:273

Log error level messages

Parameters

Parameter	Type	Description
`message`	`string`	The log message
`context?`	`LoggerContext`	Optional context to include with the log

Returns

void

info()

info(message, context?): void;

Defined in: packages/core/dist/telemetry-U0DumV05.d.mts:261

Log info level messages

Parameters

Parameter	Type	Description
`message`	`string`	The log message
`context?`	`LoggerContext`	Optional context to include with the log

Returns

void

warn()

warn(message, context?): void;

Defined in: packages/core/dist/telemetry-U0DumV05.d.mts:267

Log warning level messages

Parameters

Parameter	Type	Description
`message`	`string`	The log message
`context?`	`LoggerContext`	Optional context to include with the log

Returns

void

LoggerContext

type LoggerContext = Record<string, unknown> & object;

Defined in: packages/core/dist/telemetry-U0DumV05.d.mts:229

Context object for logger methods.

This type includes reserved keys that provide consistent naming for common logging context properties.

Type Declaration

Name	Type	Defined in
`error?`	`unknown`	packages/core/dist/telemetry-U0DumV05.d.mts:230

PublishFailureReason

type PublishFailureReason = "timeout" | "nacked" | "channel-closed";

Defined in: packages/core/dist/telemetry-U0DumV05.d.mts:82

Why the broker side of a publish failed — each one something core can actually observe on amqp-connection-manager's confirm channel:

"timeout" — no confirm arrived within publishTimeoutMs. Ambiguous: the message may have been written and be confirmed late, so republishing it can produce a duplicate. The worker's retry path accepts that (at-least-once): it requeues the original, and if the late copy did land in the wait queue, the handler runs twice.
"nacked" — the broker refused the message (basic.nack).
"channel-closed" — the channel closed before the message was confirmed.

RpcHandler

type RpcHandler<TRequest, TResponse, TErrors, THeaders, TContext> = (helpers, message) => AsyncResult<TResponse, HandlerError | TErrors>;

Defined in: packages/worker/src/types.ts:339

An RPC handler for requests whose validated payload is TRequest, answering TResponse — or one of the declared errors TErrors (an RpcError union), whose constructors it receives as helpers.errors. Like ConsumerHandler, a short alias over resolved types.

Type Parameters

Type Parameter	Default type
`TRequest`	-
`TResponse`	-
`TErrors` extends `RpcError`	`never`
`THeaders`	`undefined`
`TContext` extends `Record`<`string`, `unknown`> \| `EmptyContext`	`EmptyContext`

Parameters

Parameter	Type
`helpers`	`WorkerHandlerHelpers`<`TContext`, `RpcErrorConstructors`<`TErrors`>, `WorkerConsumedMessage`<`TRequest`, `THeaders`>>
`message`	`WorkerConsumedMessage`<`TRequest`, `THeaders`>

Returns

AsyncResult<TResponse, HandlerError | TErrors>

RpcHandlerEntry

type RpcHandlerEntry<TRequest, TResponse, TErrors, THeaders, TContext> = 
  | RpcHandler<TRequest, TResponse, TErrors, THeaders, TContext>
  | readonly [RpcHandler<TRequest, TResponse, TErrors, THeaders, TContext>, ConsumerOptions];

Defined in: packages/worker/src/types.ts:355

A RpcHandler, or a [handler, consumerOptions] tuple.

Type Parameters

Type Parameter	Default type
`TRequest`	-
`TResponse`	-
`TErrors` extends `RpcError`	`never`
`THeaders`	`undefined`
`TContext` extends `Record`<`string`, `unknown`> \| `EmptyContext`	`EmptyContext`

TelemetryProvider

type TelemetryProvider = object;

Defined in: packages/core/dist/telemetry-U0DumV05.d.mts:417

Telemetry provider for AMQP operations. Uses lazy loading to gracefully handle cases where OpenTelemetry is not installed.

Properties

Property	Type	Description	Defined in
`getConsumeCounter`	() => `Counter` \| `undefined`	Get a counter for messages consumed. Returns undefined if OpenTelemetry is not available.	packages/core/dist/telemetry-U0DumV05.d.mts:432
`getConsumeLatencyHistogram`	() => `Histogram` \| `undefined`	Get a histogram for consume/process latency. Returns undefined if OpenTelemetry is not available.	packages/core/dist/telemetry-U0DumV05.d.mts:442
`getLateRpcReplyCounter`	() => `Counter` \| `undefined`	Get a counter for RPC replies that arrive after the caller has gone away (timeout, cancellation, or unknown correlationId). Returns undefined if OpenTelemetry is not available.	packages/core/dist/telemetry-U0DumV05.d.mts:448
`getPublishCounter`	() => `Counter` \| `undefined`	Get a counter for messages published. Returns undefined if OpenTelemetry is not available.	packages/core/dist/telemetry-U0DumV05.d.mts:427
`getPublishLatencyHistogram`	() => `Histogram` \| `undefined`	Get a histogram for publish latency. Returns undefined if OpenTelemetry is not available.	packages/core/dist/telemetry-U0DumV05.d.mts:437
`getRpcCallLatencyHistogram?`	() => `Histogram` \| `undefined`	Get a histogram for RPC round-trip duration (request publish → validated reply), kept apart from the publish histogram so a slow handler does not read as a slow broker. Optional so an existing custom provider keeps compiling; omitted means "not recorded".	packages/core/dist/telemetry-U0DumV05.d.mts:455
`getTracer`	() => `Tracer` \| `undefined`	Get a tracer instance for creating spans. Returns undefined if OpenTelemetry is not available.	packages/core/dist/telemetry-U0DumV05.d.mts:422

TopologyMode

type TopologyMode = "assert" | "passive" | "none";

Defined in: packages/core/dist/telemetry-U0DumV05.d.mts:287

What topology setup does with the contract's resources on (re)connect:

"assert" (default) — declare them (assertExchange / assertQueue, plus bindings): the broker creates what is missing and refuses a mismatch.
"passive" — only check they exist (checkExchange / checkQueue) and declare nothing, for an application whose credentials may not configure the broker; bindings cannot be checked passively and are skipped. A missing resource fails create().
"none" — touch nothing: topology is owned elsewhere (IaC, definitions import, another service).

WorkerConsumedMessage

type WorkerConsumedMessage<TPayload, THeaders> = object;

Defined in: packages/worker/src/types.ts:259

A consumed message containing parsed payload and headers.

What a handler receives as input on its record, and again as the second positional argument. It contains the validated payload and (if defined in the message schema) the validated headers.

Example

typescript

const handler = declareHandler(contract, 'processOrder', ({ raw, input }) => {
  console.log(input.payload.orderId);  // Typed payload
  console.log(input.headers?.priority); // Typed headers (if defined)
  console.log(raw.fields.deliveryTag); // Raw AMQP delivery
  return OkAsync(undefined);
});

Type Parameters

Type Parameter	Default type	Description
`TPayload`	-	The inferred payload type from the message schema
`THeaders`	`undefined`	The inferred headers type from the message schema (undefined if not defined)

Properties

Property	Type	Description	Defined in
`headers`	`THeaders` extends `undefined` ? `undefined` : `THeaders`	The validated message headers (present only when headers schema is defined)	packages/worker/src/types.ts:263
`payload`	`TPayload`	The validated message payload	packages/worker/src/types.ts:261

WorkerCreateContextInfo

type WorkerCreateContextInfo = object;

Defined in: packages/worker/src/worker.ts:123

Per-message information handed to the createContext factory — enough to derive request-scoped dependencies (correlation-id loggers, per-message transactions) without closing over the dispatch loop.

Properties

Property	Type	Description	Defined in
`handlerName`	`string`	The `consumers` / `rpcs` key being dispatched.	packages/worker/src/worker.ts:125
`isRpc`	`boolean`	True when the handler is an RPC server.	packages/worker/src/worker.ts:127
`message`	`object`	The validated message (payload and headers already schema-checked).	packages/worker/src/worker.ts:129
`message.headers`	`unknown`	-	packages/worker/src/worker.ts:129
`message.payload`	`unknown`	-	packages/worker/src/worker.ts:129
`rawMessage`	`ConsumeMessage`	The raw amqplib message.	packages/worker/src/worker.ts:131

WorkerHandlerHelpers

type WorkerHandlerHelpers<TContext, TErrors, TMessage> = object;

Defined in: packages/worker/src/types.ts:159

The helpers record every handler receives as its FIRST argument — everything the delivery carries, the validated message included, with that message repeated as the second parameter.

That is oRPC's shape, and the one this family converged on: ProcedureHandlerOptions carries input and the handler still takes it positionally, so ({ errors, input }) => ... and ({ errors }, message) => ... are the same call. raw rides here rather than in a third parameter for the same reason — one record for everything the delivery carries.

Type Parameters

Type Parameter	Default type
`TContext` extends `Record`<`string`, `unknown`> \| `EmptyContext`	`EmptyContext`
`TErrors`	`EmptyContext`
`TMessage`	`unknown`

Properties

Property	Modifier	Type	Description	Defined in
`context`	`readonly`	`TContext`	Context produced by `createContext` and the middleware chain.	packages/worker/src/types.ts:174
`errors`	`readonly`	`TErrors`	Typed constructors for the contract-declared errors (empty for consumers).	packages/worker/src/types.ts:176
`input`	`readonly`	`TMessage`	The validated message — the SAME value the second parameter carries. It is on the record so a whole handler is one destructuring, which is oRPC's own shape and its own word for it: `ProcedureHandlerOptions` carries `input`, and the handler still takes it positionally. One name across the three transports is the point — a developer moving between them destructures `input` in each.	packages/worker/src/types.ts:172
`nonRetryable`	`readonly`	(`message`, `cause?`) => `NonRetryableError`	"This will never work" — straight to the dead-letter queue, no retry budget spent. The permanent twin of WorkerHandlerHelpers.retryable.	packages/worker/src/types.ts:189
`raw`	`readonly`	`ConsumeMessage`	The raw AMQP delivery — `fields`, `properties`, and the untouched `content`.	packages/worker/src/types.ts:178
`retryable`	`readonly`	(`message`, `cause?`) => `RetryableError`	"Infrastructure comes back" — the failure the retry schedule is for, handed over rather than imported and constructed. `ErrAsync(retryable(...))` is `ErrAsync(new RetryableError(...))` without the import.	packages/worker/src/types.ts:184

WorkerInferConsumedMessage

type WorkerInferConsumedMessage<TContract, TName> = WorkerConsumedMessage<WorkerInferConsumerPayload<TContract, TName>, WorkerInferConsumerHeaders<TContract, TName>>;

Defined in: packages/worker/src/types.ts:269

Infer the full consumed message type for a regular consumer.

Type Parameters

Type Parameter
`TContract` extends `ContractDefinition`
`TName` extends `InferConsumerNames`<`TContract`>

WorkerInferConsumerHandler

type WorkerInferConsumerHandler<TContract, TName, TContext> = [WorkerInferConsumerPayload<TContract, TName>, WorkerInferConsumerHeaders<TContract, TName>] extends [infer TPayload, infer THeaders] ? ConsumerHandler<TPayload, THeaders, TContext> : never;

Defined in: packages/worker/src/types.ts:380

Handler type for the consumer TName of TContract: a ConsumerHandler over that consumer's resolved payload and headers.

Type Parameters

Type Parameter	Default type
`TContract` extends `ContractDefinition`	-
`TName` extends `InferConsumerNames`<`TContract`>	-
`TContext` extends `Record`<`string`, `unknown`> \| `EmptyContext`	`EmptyContext`

WorkerInferConsumerHandlerEntry

type WorkerInferConsumerHandlerEntry<TContract, TName, TContext> = [WorkerInferConsumerPayload<TContract, TName>, WorkerInferConsumerHeaders<TContract, TName>] extends [infer TPayload, infer THeaders] ? ConsumerHandlerEntry<TPayload, THeaders, TContext> : never;

Defined in: packages/worker/src/types.ts:409

A WorkerInferConsumerHandler, or a [handler, consumerOptions] tuple.

Type Parameters

Type Parameter	Default type
`TContract` extends `ContractDefinition`	-
`TName` extends `InferConsumerNames`<`TContract`>	-
`TContext` extends `Record`<`string`, `unknown`> \| `EmptyContext`	`EmptyContext`

WorkerInferConsumerHeaders

type WorkerInferConsumerHeaders<TContract, TName> = ConsumerInferHeadersOutput<InferConsumer<TContract, TName>>;

Defined in: packages/worker/src/types.ts:86

Infer the headers type for a regular consumer. Returns undefined if no headers schema is defined.

Type Parameters

Type Parameter
`TContract` extends `ContractDefinition`
`TName` extends `InferConsumerNames`<`TContract`>

WorkerInferHandlers

type WorkerInferHandlers<TContract, TContext> = ([InferConsumerNames<TContract>] extends [never] ? object : { [K in InferConsumerNames<TContract>]: WorkerInferConsumerHandlerEntry<TContract, K, TContext> }) & ([InferRpcNames<TContract>] extends [never] ? object : { [K in InferRpcNames<TContract>]: WorkerInferRpcHandlerEntry<TContract, K, TContext> });

Defined in: packages/worker/src/types.ts:434

Type Parameters

Type Parameter	Default type
`TContract` extends `ContractDefinition`	-
`TContext` extends `Record`<`string`, `unknown`> \| `EmptyContext`	`EmptyContext`

WorkerInferRpcConsumedMessage

type WorkerInferRpcConsumedMessage<TContract, TName> = WorkerConsumedMessage<WorkerInferRpcRequest<TContract, TName>, WorkerInferRpcHeaders<TContract, TName>>;

Defined in: packages/worker/src/types.ts:281

Infer the consumed message type for an RPC handler — payload + headers from the request side of the RPC.

Type Parameters

Type Parameter
`TContract` extends `ContractDefinition`
`TName` extends `InferRpcNames`<`TContract`>

WorkerInferRpcErrorConstructors

type WorkerInferRpcErrorConstructors<TContract, TName> = InferRpc<TContract, TName> extends RpcDefinition<MessageDefinition, MessageDefinition, QueueDefinition, infer TErrors> ? TErrors extends RpcErrorMap ? { [K in keyof TErrors & string]: (data: InferSchemaInput<TErrors[K]["data"]>, message?: string) => RpcError<K, InferSchemaInput<TErrors[K]["data"]>> } : EmptyContext : EmptyContext;

Defined in: packages/worker/src/types.ts:127

Typed constructors for an RPC's declared errors, handed to the handler via its helpers argument: errors.ORDER_NOT_FOUND({ orderId }) builds the RpcError with per-code data inference and autocomplete — the constructor-bag form of the free rpcError(code, data) factory (org DNA, mirroring temporal-contract's helpers.errors).

Type Parameters

Type Parameter
`TContract` extends `ContractDefinition`
`TName` extends `InferRpcNames`<`TContract`>

WorkerInferRpcErrors

type WorkerInferRpcErrors<TContract, TName> = InferRpc<TContract, TName> extends RpcDefinition<MessageDefinition, MessageDefinition, QueueDefinition, infer TErrors> ? TErrors extends RpcErrorMap ? { [K in keyof TErrors & string]: RpcError<K, InferSchemaInput<TErrors[K]["data"]>> }[keyof TErrors & string] : never : never;

Defined in: packages/worker/src/types.ts:199

Infer the typed error union for an RPC handler — one RpcError<code, data> member per entry in the RPC's errors map, with data typed as the declared schema's input (the worker validates before replying). Resolves to never when the RPC declares no errors, leaving the handler's error channel as plain HandlerError.

Type Parameters

Type Parameter
`TContract` extends `ContractDefinition`
`TName` extends `InferRpcNames`<`TContract`>

WorkerInferRpcHandler

type WorkerInferRpcHandler<TContract, TName, TContext> = [WorkerInferRpcRequest<TContract, TName>, WorkerInferRpcResponse<TContract, TName>, WorkerInferRpcErrors<TContract, TName>, WorkerInferRpcHeaders<TContract, TName>] extends [infer TRequest, infer TResponse, infer TErrors, infer THeaders] ? RpcHandler<TRequest, TResponse, TErrors, THeaders, TContext> : never;

Defined in: packages/worker/src/types.ts:395

Handler type for the RPC TName of TContract: a RpcHandler over its resolved request, response, declared errors and headers.

Type Parameters

Type Parameter	Default type
`TContract` extends `ContractDefinition`	-
`TName` extends `InferRpcNames`<`TContract`>	-
`TContext` extends `Record`<`string`, `unknown`> \| `EmptyContext`	`EmptyContext`

WorkerInferRpcHandlerEntry

type WorkerInferRpcHandlerEntry<TContract, TName, TContext> = [WorkerInferRpcRequest<TContract, TName>, WorkerInferRpcResponse<TContract, TName>, WorkerInferRpcErrors<TContract, TName>, WorkerInferRpcHeaders<TContract, TName>] extends [infer TRequest, infer TResponse, infer TErrors, infer THeaders] ? RpcHandlerEntry<TRequest, TResponse, TErrors, THeaders, TContext> : never;

Defined in: packages/worker/src/types.ts:421

A WorkerInferRpcHandler, or a [handler, consumerOptions] tuple.

Type Parameters

Type Parameter	Default type
`TContract` extends `ContractDefinition`	-
`TName` extends `InferRpcNames`<`TContract`>	-
`TContext` extends `Record`<`string`, `unknown`> \| `EmptyContext`	`EmptyContext`

WorkerInferRpcHeaders

type WorkerInferRpcHeaders<TContract, TName> = InferRpc<TContract, TName> extends RpcDefinition<infer TRequest, MessageDefinition> ? TRequest extends MessageDefinition<infer _TPayload, infer THeaders> ? THeaders extends StandardSchemaV1<Record<string, unknown>> ? InferSchemaOutput<THeaders> : undefined : undefined : undefined;

Defined in: packages/worker/src/types.ts:108

Infer the request headers type for an RPC. Returns undefined unless the RPC's request MessageDefinition declares a headers schema.

Type Parameters

Type Parameter
`TContract` extends `ContractDefinition`
`TName` extends `InferRpcNames`<`TContract`>

WorkerInferRpcRequest

type WorkerInferRpcRequest<TContract, TName> = InferRpc<TContract, TName> extends RpcDefinition<infer TRequest, MessageDefinition> ? TRequest extends MessageDefinition ? InferSchemaOutput<TRequest["payload"]> : never : never;

Defined in: packages/worker/src/types.ts:94

Infer the request payload type for an RPC.

Type Parameters

Type Parameter
`TContract` extends `ContractDefinition`
`TName` extends `InferRpcNames`<`TContract`>

WorkerInferRpcResponse

type WorkerInferRpcResponse<TContract, TName> = InferRpc<TContract, TName> extends RpcDefinition<MessageDefinition, infer TResponse> ? TResponse extends MessageDefinition ? InferSchemaInput<TResponse["payload"]> : never : never;

Defined in: packages/worker/src/types.ts:225

Infer the response payload type for an RPC. The handler must return an AsyncResult<TResponse, HandlerError> matching this shape.

Typed as the schema's input — the handler supplies the pre-validation shape (defaults optional, transforms not yet applied); the worker validates against the response schema before publishing the reply. Same convention as RPC error data.

Type Parameters

Type Parameter
`TContract` extends `ContractDefinition`
`TName` extends `InferRpcNames`<`TContract`>

WorkerMiddleware

type WorkerMiddleware<TContextIn, TContextOut> = (args, next) => AsyncResult<unknown, HandlerError | RpcError>;

Defined in: packages/worker/src/middleware.ts:95

A worker middleware: wraps every handler invocation (consumers and RPCs) after message validation.

Middleware follow the guard-and-narrow pattern: check something, then call next({ context }) to run the rest of the chain with typed fields injected into the handler's helpers record — or short-circuit by returning without calling next:

Err(new RetryableError(...)) / Err(new NonRetryableError(...)) routes through the normal retry/DLQ pipeline, exactly as if the handler had returned it.
Err(rpcError(code, data)) (on an RPC with a declared errors map) publishes a typed error reply.
Ok(value) skips the handler entirely; for an RPC, value is validated against the response schema and published as the reply (cache pattern).

Type Parameters

Type Parameter	Default type	Description
`TContextIn` extends `Record`<`string`, `unknown`> \| `EmptyContext`	`EmptyContext`	context provided by outer middleware
`TContextOut` extends `TContextIn`	`TContextIn`	context this middleware passes downstream

Parameters

Parameter	Type
`args`	`WorkerMiddlewareArgs`<`TContextIn`>
`next`	`WorkerMiddlewareNext`<`TContextOut`>

Returns

AsyncResult<unknown, HandlerError | RpcError>

Example

typescript

import { declareMiddleware, NonRetryableError } from '@amqp-contract/worker';
import { ErrAsync } from 'unthrown';

const auth = declareMiddleware<EmptyContext, { tenantId: string }>((args, next) => {
  const tenantId = args.rawMessage.properties.headers?.['x-tenant-id'];
  if (typeof tenantId !== 'string') {
    return ErrAsync(new NonRetryableError('Missing x-tenant-id header'));
  }
  return next({ context: { tenantId } });
});

WorkerMiddlewareArgs

type WorkerMiddlewareArgs<TContextIn> = object;

Defined in: packages/worker/src/middleware.ts:20

Arguments passed to a worker middleware — everything the downstream handler will receive, plus dispatch metadata and the context accumulated by outer middleware.

Type Parameters

Type Parameter
`TContextIn` extends `Record`<`string`, `unknown`> \| `EmptyContext`

Properties

Property	Type	Description	Defined in
`context`	`TContextIn`	Context accumulated by outer middleware (empty for the outermost one).	packages/worker/src/middleware.ts:30
`handlerName`	`string`	The `consumers` / `rpcs` key being dispatched.	packages/worker/src/middleware.ts:26
`isRpc`	`boolean`	True when the handler is an RPC server (its result is published as a reply).	packages/worker/src/middleware.ts:28
`message`	`object`	The validated message (payload and headers already schema-checked).	packages/worker/src/middleware.ts:22
`message.headers`	`unknown`	-	packages/worker/src/middleware.ts:22
`message.payload`	`unknown`	-	packages/worker/src/middleware.ts:22
`rawMessage`	`ConsumeMessage`	The raw amqplib message (delivery tag, AMQP properties, raw headers, …).	packages/worker/src/middleware.ts:24

WorkerMiddlewareNext

type WorkerMiddlewareNext<TContextOut> = (opts?) => AsyncResult<unknown, HandlerError | RpcError>;

Defined in: packages/worker/src/middleware.ts:56

Continuation a middleware calls to run the rest of the chain (inner middleware, then the handler). opts.context is the full context the downstream sees — pass { ...args.context, ...injected } (or just the injected fields; the dispatcher merges over the current context either way).

opts.payload substitutes the message payload seen downstream. Inner middleware observe the substituted payload as-is; the dispatcher re-validates it against the consumer's payload schema before the handler runs — an invalid substitution fails terminally as a NonRetryableError (DLQ), so middleware cannot smuggle unvalidated data past the contract boundary.

What counts as a substitution is the KEY being present, not its value: next({ payload: undefined }) substitutes undefined and sends it through the payload schema like any other substitution — refused by a schema that demands a shape, accepted by one that admits undefined — where next({}) and next() leave the message alone.

The returned AsyncResult carries the handler outcome: undefined for a regular consumer, the (not yet validated) response for an RPC. A middleware can transform or inspect it before returning.

Type Parameters

Type Parameter
`TContextOut` extends `Record`<`string`, `unknown`> \| `EmptyContext`

Parameters

Parameter	Type
`opts?`	{ `context?`: `TContextOut`; `payload?`: `unknown`; }
`opts.context?`	`TContextOut`
`opts.payload?`	`unknown`

Returns

AsyncResult<unknown, HandlerError | RpcError>

Variables

DEFAULT_DRAIN_TIMEOUT_MS

const DEFAULT_DRAIN_TIMEOUT_MS: 30000 = 30_000;

Defined in: packages/worker/src/worker.ts:167

Default time close() waits for in-flight handlers before tearing the channel down anyway. Finite by default so a hung handler cannot wedge shutdown — the un-acked deliveries are redelivered by the broker (at-least-once semantics). Pass drainTimeoutMs: null to wait forever.

Functions

composeMiddleware()

Call Signature

function composeMiddleware<TIn, TA>(m1): WorkerMiddleware<TIn, TA>;

Defined in: packages/worker/src/middleware.ts:146

Compose middleware left-to-right: the first argument is the outermost (runs first, sees the emptiest context), the last is the innermost (its injected context is what handlers receive). Context types accumulate across the chain — each middleware's TContextIn must match what the previous one produced.

Typed overloads cover up to 8 middleware. For longer chains, nest: a composed chain is itself a WorkerMiddleware<EmptyContext, T> and can be the first argument of an outer composeMiddleware call — composeMiddleware(composeMiddleware(a, ..., h), i, j) — preserving context-type accumulation at any length.

Type Parameters

Type Parameter
`TIn` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TA` extends `Record`<`string`, `unknown`> \| `EmptyContext`

Parameters

Parameter	Type
`m1`	`WorkerMiddleware`<`TIn`, `TA`>

Returns

WorkerMiddleware<TIn, TA>

Example

typescript

const middleware = composeMiddleware(logging, auth, idempotency);
// handlers receive the context injected by all three

Call Signature

function composeMiddleware<TIn, TA, TB>(m1, m2): WorkerMiddleware<TIn, TB>;

Defined in: packages/worker/src/middleware.ts:150

Type Parameters

Type Parameter
`TIn` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TA` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TB` extends `Record`<`string`, `unknown`> \| `EmptyContext`

Parameters

Parameter	Type
`m1`	`WorkerMiddleware`<`TIn`, `TA`>
`m2`	`WorkerMiddleware`<`TA`, `TB`>

Returns

WorkerMiddleware<TIn, TB>

Example

typescript

const middleware = composeMiddleware(logging, auth, idempotency);
// handlers receive the context injected by all three

Call Signature

function composeMiddleware<TIn, TA, TB, TC>(
   m1, 
   m2, 
   m3
): WorkerMiddleware<TIn, TC>;

Defined in: packages/worker/src/middleware.ts:155

Type Parameters

Type Parameter
`TIn` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TA` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TB` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TC` extends `Record`<`string`, `unknown`> \| `EmptyContext`

Parameters

Parameter	Type
`m1`	`WorkerMiddleware`<`TIn`, `TA`>
`m2`	`WorkerMiddleware`<`TA`, `TB`>
`m3`	`WorkerMiddleware`<`TB`, `TC`>

Returns

WorkerMiddleware<TIn, TC>

Example

typescript

const middleware = composeMiddleware(logging, auth, idempotency);
// handlers receive the context injected by all three

Call Signature

function composeMiddleware<TIn, TA, TB, TC, TD>(
   m1, 
   m2, 
   m3, 
   m4
): WorkerMiddleware<TIn, TD>;

Defined in: packages/worker/src/middleware.ts:165

Type Parameters

Type Parameter
`TIn` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TA` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TB` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TC` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TD` extends `Record`<`string`, `unknown`> \| `EmptyContext`

Parameters

Parameter	Type
`m1`	`WorkerMiddleware`<`TIn`, `TA`>
`m2`	`WorkerMiddleware`<`TA`, `TB`>
`m3`	`WorkerMiddleware`<`TB`, `TC`>
`m4`	`WorkerMiddleware`<`TC`, `TD`>

Returns

WorkerMiddleware<TIn, TD>

Example

typescript

const middleware = composeMiddleware(logging, auth, idempotency);
// handlers receive the context injected by all three

Call Signature

function composeMiddleware<TIn, TA, TB, TC, TD, TE>(
   m1, 
   m2, 
   m3, 
   m4, 
   m5
): WorkerMiddleware<TIn, TE>;

Defined in: packages/worker/src/middleware.ts:177

Type Parameters

Type Parameter
`TIn` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TA` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TB` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TC` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TD` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TE` extends `Record`<`string`, `unknown`> \| `EmptyContext`

Parameters

Parameter	Type
`m1`	`WorkerMiddleware`<`TIn`, `TA`>
`m2`	`WorkerMiddleware`<`TA`, `TB`>
`m3`	`WorkerMiddleware`<`TB`, `TC`>
`m4`	`WorkerMiddleware`<`TC`, `TD`>
`m5`	`WorkerMiddleware`<`TD`, `TE`>

Returns

WorkerMiddleware<TIn, TE>

Example

typescript

const middleware = composeMiddleware(logging, auth, idempotency);
// handlers receive the context injected by all three

Call Signature

function composeMiddleware<TIn, TA, TB, TC, TD, TE, TF>(
   m1, 
   m2, 
   m3, 
   m4, 
   m5, 
   m6
): WorkerMiddleware<TIn, TF>;

Defined in: packages/worker/src/middleware.ts:191

Type Parameters

Type Parameter
`TIn` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TA` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TB` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TC` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TD` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TE` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TF` extends `Record`<`string`, `unknown`> \| `EmptyContext`

Parameters

Parameter	Type
`m1`	`WorkerMiddleware`<`TIn`, `TA`>
`m2`	`WorkerMiddleware`<`TA`, `TB`>
`m3`	`WorkerMiddleware`<`TB`, `TC`>
`m4`	`WorkerMiddleware`<`TC`, `TD`>
`m5`	`WorkerMiddleware`<`TD`, `TE`>
`m6`	`WorkerMiddleware`<`TE`, `TF`>

Returns

WorkerMiddleware<TIn, TF>

Example

typescript

const middleware = composeMiddleware(logging, auth, idempotency);
// handlers receive the context injected by all three

Call Signature

function composeMiddleware<TIn, TA, TB, TC, TD, TE, TF, TG>(
   m1, 
   m2, 
   m3, 
   m4, 
   m5, 
   m6, 
   m7
): WorkerMiddleware<TIn, TG>;

Defined in: packages/worker/src/middleware.ts:207

Type Parameters

Type Parameter
`TIn` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TA` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TB` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TC` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TD` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TE` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TF` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TG` extends `Record`<`string`, `unknown`> \| `EmptyContext`

Parameters

Parameter	Type
`m1`	`WorkerMiddleware`<`TIn`, `TA`>
`m2`	`WorkerMiddleware`<`TA`, `TB`>
`m3`	`WorkerMiddleware`<`TB`, `TC`>
`m4`	`WorkerMiddleware`<`TC`, `TD`>
`m5`	`WorkerMiddleware`<`TD`, `TE`>
`m6`	`WorkerMiddleware`<`TE`, `TF`>
`m7`	`WorkerMiddleware`<`TF`, `TG`>

Returns

WorkerMiddleware<TIn, TG>

Example

typescript

const middleware = composeMiddleware(logging, auth, idempotency);
// handlers receive the context injected by all three

Call Signature

function composeMiddleware<TIn, TA, TB, TC, TD, TE, TF, TG, TH>(
   m1, 
   m2, 
   m3, 
   m4, 
   m5, 
   m6, 
   m7, 
   m8
): WorkerMiddleware<TIn, TH>;

Defined in: packages/worker/src/middleware.ts:225

Type Parameters

Type Parameter
`TIn` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TA` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TB` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TC` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TD` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TE` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TF` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TG` extends `Record`<`string`, `unknown`> \| `EmptyContext`
`TH` extends `Record`<`string`, `unknown`> \| `EmptyContext`

Parameters

Parameter	Type
`m1`	`WorkerMiddleware`<`TIn`, `TA`>
`m2`	`WorkerMiddleware`<`TA`, `TB`>
`m3`	`WorkerMiddleware`<`TB`, `TC`>
`m4`	`WorkerMiddleware`<`TC`, `TD`>
`m5`	`WorkerMiddleware`<`TD`, `TE`>
`m6`	`WorkerMiddleware`<`TE`, `TF`>
`m7`	`WorkerMiddleware`<`TF`, `TG`>
`m8`	`WorkerMiddleware`<`TG`, `TH`>

Returns

WorkerMiddleware<TIn, TH>

Example

typescript

const middleware = composeMiddleware(logging, auth, idempotency);
// handlers receive the context injected by all three

declareHandler()

Call Signature

function declareHandler<TContract, TName, TContext>(
   contract, 
   name, 
   handler
): WorkerInferConsumerHandlerEntry<TContract, TName, TContext>;

Defined in: packages/worker/src/handlers.ts:212

Define a type-safe handler for a specific consumer or RPC in a contract.

Recommended: This function creates handlers that return AsyncResult<void, HandlerError> (consumers) or AsyncResult<TResponse, HandlerError> (RPCs), providing explicit error handling and better control over retry behavior.

Supports two patterns:

Simple handler: just the function
Handler with options: [handler, { prefetch: 10 }]

The leaf takes the helpers record first and the validated message second, with that message on the record as well — oRPC's shape — so ({ errors, input }) => ... and ({ errors }, message) => ... are the same call — oRPC offers both — and a handler that wants only its message is ({ input: { payload } }) => ..., with no placeholder to spell.

Type Parameters

Type Parameter	Default type	Description
`TContract` extends `ContractDefinition`	-	The contract definition type
`TName` extends `string` \| `number` \| `symbol`	-	The consumer or RPC name from the contract
`TContext` extends `Record`<`string`, `unknown`> \| `EmptyContext`	`EmptyContext`	-

Parameters

Parameter	Type	Description
`contract`	`TContract`	The contract definition containing the consumer or RPC
`name`	`TName`	The name of the consumer or RPC from the contract
`handler`	`WorkerInferConsumerHandler`<`TContract`, `TName`, `TContext`>	The handler function — for consumers, returns `AsyncResult<void, HandlerError>`; for RPCs, returns `AsyncResult<TResponse, HandlerError>`.

Returns

WorkerInferConsumerHandlerEntry<TContract, TName, TContext>

A type-safe handler that can be used with TypedAmqpWorker

Examples

Consumer handler

typescript

import { declareHandler, RetryableError, NonRetryableError } from '@amqp-contract/worker';
import { fromPromise } from 'unthrown';

const processOrderHandler = declareHandler(
  orderContract,
  'processOrder',
  ({ input: { payload } }) =>
    fromPromise(
      processPayment(payload),
      (error) => new RetryableError('Payment failed', error),
    ).map(() => undefined),
);

RPC handler

typescript

const calculateHandler = declareHandler(
  rpcContract,
  'calculate',
  ({ input: { payload } }) => OkAsync({ sum: payload.a + payload.b }),
);

Call Signature

function declareHandler<TContract, TName, TContext>(
   contract, 
   name, 
   handler, 
   options
): WorkerInferConsumerHandlerEntry<TContract, TName, TContext>;

Defined in: packages/worker/src/handlers.ts:221

Define a type-safe handler for a specific consumer or RPC in a contract.

Supports two patterns:

Simple handler: just the function
Handler with options: [handler, { prefetch: 10 }]

Type Parameters

Type Parameter	Default type	Description
`TContract` extends `ContractDefinition`	-	The contract definition type
`TName` extends `string` \| `number` \| `symbol`	-	The consumer or RPC name from the contract
`TContext` extends `Record`<`string`, `unknown`> \| `EmptyContext`	`EmptyContext`	-

Parameters

Parameter	Type	Description
`contract`	`TContract`	The contract definition containing the consumer or RPC
`name`	`TName`	The name of the consumer or RPC from the contract
`handler`	`WorkerInferConsumerHandler`<`TContract`, `TName`, `TContext`>	The handler function — for consumers, returns `AsyncResult<void, HandlerError>`; for RPCs, returns `AsyncResult<TResponse, HandlerError>`.
`options`	`ConsumerOptions`	Optional consumer options (prefetch)

Returns

WorkerInferConsumerHandlerEntry<TContract, TName, TContext>

A type-safe handler that can be used with TypedAmqpWorker

Examples

Consumer handler

typescript

import { declareHandler, RetryableError, NonRetryableError } from '@amqp-contract/worker';
import { fromPromise } from 'unthrown';

const processOrderHandler = declareHandler(
  orderContract,
  'processOrder',
  ({ input: { payload } }) =>
    fromPromise(
      processPayment(payload),
      (error) => new RetryableError('Payment failed', error),
    ).map(() => undefined),
);

RPC handler

typescript

const calculateHandler = declareHandler(
  rpcContract,
  'calculate',
  ({ input: { payload } }) => OkAsync({ sum: payload.a + payload.b }),
);

Call Signature

function declareHandler<TContract, TName, TContext>(
   contract, 
   name, 
   handler
): WorkerInferRpcHandlerEntry<TContract, TName, TContext>;

Defined in: packages/worker/src/handlers.ts:231

Define a type-safe handler for a specific consumer or RPC in a contract.

Supports two patterns:

Simple handler: just the function
Handler with options: [handler, { prefetch: 10 }]

Type Parameters

Type Parameter	Default type	Description
`TContract` extends `ContractDefinition`	-	The contract definition type
`TName` extends `string` \| `number` \| `symbol`	-	The consumer or RPC name from the contract
`TContext` extends `Record`<`string`, `unknown`> \| `EmptyContext`	`EmptyContext`	-

Parameters

Parameter	Type	Description
`contract`	`TContract`	The contract definition containing the consumer or RPC
`name`	`TName`	The name of the consumer or RPC from the contract
`handler`	`WorkerInferRpcHandler`<`TContract`, `TName`, `TContext`>	The handler function — for consumers, returns `AsyncResult<void, HandlerError>`; for RPCs, returns `AsyncResult<TResponse, HandlerError>`.

Returns

WorkerInferRpcHandlerEntry<TContract, TName, TContext>

A type-safe handler that can be used with TypedAmqpWorker

Examples

Consumer handler

typescript

import { declareHandler, RetryableError, NonRetryableError } from '@amqp-contract/worker';
import { fromPromise } from 'unthrown';

const processOrderHandler = declareHandler(
  orderContract,
  'processOrder',
  ({ input: { payload } }) =>
    fromPromise(
      processPayment(payload),
      (error) => new RetryableError('Payment failed', error),
    ).map(() => undefined),
);

RPC handler

typescript

const calculateHandler = declareHandler(
  rpcContract,
  'calculate',
  ({ input: { payload } }) => OkAsync({ sum: payload.a + payload.b }),
);

Call Signature

function declareHandler<TContract, TName, TContext>(
   contract, 
   name, 
   handler, 
   options
): WorkerInferRpcHandlerEntry<TContract, TName, TContext>;

Defined in: packages/worker/src/handlers.ts:240

Define a type-safe handler for a specific consumer or RPC in a contract.

Supports two patterns:

Simple handler: just the function
Handler with options: [handler, { prefetch: 10 }]

Type Parameters

Type Parameter	Default type	Description
`TContract` extends `ContractDefinition`	-	The contract definition type
`TName` extends `string` \| `number` \| `symbol`	-	The consumer or RPC name from the contract
`TContext` extends `Record`<`string`, `unknown`> \| `EmptyContext`	`EmptyContext`	-

Parameters

Parameter	Type	Description
`contract`	`TContract`	The contract definition containing the consumer or RPC
`name`	`TName`	The name of the consumer or RPC from the contract
`handler`	`WorkerInferRpcHandler`<`TContract`, `TName`, `TContext`>	The handler function — for consumers, returns `AsyncResult<void, HandlerError>`; for RPCs, returns `AsyncResult<TResponse, HandlerError>`.
`options`	`ConsumerOptions`	Optional consumer options (prefetch)

Returns

WorkerInferRpcHandlerEntry<TContract, TName, TContext>

A type-safe handler that can be used with TypedAmqpWorker

Examples

Consumer handler

typescript

import { declareHandler, RetryableError, NonRetryableError } from '@amqp-contract/worker';
import { fromPromise } from 'unthrown';

const processOrderHandler = declareHandler(
  orderContract,
  'processOrder',
  ({ input: { payload } }) =>
    fromPromise(
      processPayment(payload),
      (error) => new RetryableError('Payment failed', error),
    ).map(() => undefined),
);

RPC handler

typescript

const calculateHandler = declareHandler(
  rpcContract,
  'calculate',
  ({ input: { payload } }) => OkAsync({ sum: payload.a + payload.b }),
);

declareHandlers()

function declareHandlers<TContract, TContext>(contract, handlers): WorkerInferHandlers<TContract, TContext>;

Defined in: packages/worker/src/handlers.ts:293

Define multiple type-safe handlers for consumers and RPCs in a contract.

The handlers object must contain exactly one entry per consumers and rpcs key in the contract — see WorkerInferHandlers.

Type Parameters

Type Parameter	Default type	Description
`TContract` extends `ContractDefinition`	-	The contract definition type
`TContext` extends `Record`<`string`, `unknown`> \| `EmptyContext`	`EmptyContext`	-

Parameters

Parameter	Type	Description
`contract`	`TContract`	The contract definition containing the consumers and RPCs
`handlers`	`WorkerInferHandlers`<`TContract`, `TContext`>	An object with handler functions for each consumer and RPC

Returns

WorkerInferHandlers<TContract, TContext>

A type-safe handlers object that can be used with TypedAmqpWorker

Example

typescript

import { declareHandlers, RetryableError } from '@amqp-contract/worker';
import { fromPromise, OkAsync } from 'unthrown';

const handlers = declareHandlers(orderContract, {
  processOrder: ({ input: { payload } }) =>
    fromPromise(
      processPayment(payload),
      (error) => new RetryableError('Payment failed', error),
    ).map(() => undefined),
  calculate: ({ input: { payload } }) => OkAsync({ sum: payload.a + payload.b }),
});

declareMiddleware()

function declareMiddleware<TContextIn, TContextOut>(middleware): WorkerMiddleware<TContextIn, TContextOut>;

Defined in: packages/worker/src/middleware.ts:118

Identity helper that pins a middleware's context types without a variable annotation — declareMiddleware<In, Out>(fn) reads better than const mw: WorkerMiddleware<In, Out> = fn.

Type Parameters

Type Parameter	Default type
`TContextIn` extends `Record`<`string`, `unknown`> \| `EmptyContext`	`EmptyContext`
`TContextOut` extends `Record`<`string`, `unknown`> \| `EmptyContext`	`TContextIn`

Parameters

Parameter	Type
`middleware`	`WorkerMiddleware`<`TContextIn`, `TContextOut`>

Returns

WorkerMiddleware<TContextIn, TContextOut>

isConnectionError()

function isConnectionError(error): error is ConnectionError;

Defined in: packages/core/dist/telemetry-U0DumV05.d.mts:143

Type guard to check if an error is a ConnectionError — the modeled failure of dialing the broker.

Parameters

Parameter	Type
`error`	`unknown`

Returns

error is ConnectionError

isHandlerError()

function isHandlerError(error): error is HandlerError;

Defined in: packages/worker/src/errors.ts:102

Narrow an unknown value to HandlerError.

HandlerError is a union type with no runtime counterpart, so there is no instanceof HandlerError to reach for — this guard is the only one-step way to ask "is this either of the two handler errors?".

The per-class questions need no helper: RetryableError and NonRetryableError are exported classes, so error instanceof RetryableError says it directly.

Parameters

Parameter	Type	Description
`error`	`unknown`	The error to check

Returns

error is HandlerError

True if the error is a RetryableError or a NonRetryableError

Example

typescript

import { isHandlerError } from '@amqp-contract/worker';

function report(error: unknown) {
  if (isHandlerError(error)) {
    // error is RetryableError | NonRetryableError
    console.log('Handler error:', error.name, error.message);
  }
}

isMessageValidationError()

function isMessageValidationError(error): error is MessageValidationError;

Defined in: packages/core/dist/telemetry-U0DumV05.d.mts:147

Type guard to check if an error is a MessageValidationError.

Parameters

Parameter	Type
`error`	`unknown`

Returns

error is MessageValidationError

isRpcError()

function isRpcError(error): error is RpcError<string, unknown>;

Defined in: packages/core/dist/telemetry-U0DumV05.d.mts:194

Type guard to check if an error is an RpcError.

Narrowing to a specific code (and thus a typed data) is done on the code property after the guard, or via the error matcher on the _tag (matcher.with(P.tag("@amqp-contract/RpcError"), …)).

Parameters

Parameter	Type
`error`	`unknown`

Returns

error is RpcError<string, unknown>

isTechnicalError()

function isTechnicalError(error): error is TechnicalError;

Defined in: packages/core/dist/telemetry-U0DumV05.d.mts:138

Type guard to check if an error is a TechnicalError — the cause carried by every infrastructure Defect this library produces.

Parameters

Parameter	Type
`error`	`unknown`

Returns

error is TechnicalError

qualifyNonRetryable()

function qualifyNonRetryable(message): (cause) => NonRetryableError;

Defined in: packages/worker/src/errors.ts:150

Build a fromPromise qualifier that wraps the rejection cause in a NonRetryableError with the given message — the permanent-failure counterpart of qualifyRetryable.

Parameters

Parameter	Type
`message`	`string`

Returns

(cause) => NonRetryableError

Example

typescript

import { qualifyNonRetryable } from '@amqp-contract/worker';
import { fromPromise } from 'unthrown';

const handler = ({ input: { payload } }) =>
  fromPromise(chargeCard(payload), qualifyNonRetryable('Card permanently declined'))
    .map(() => undefined);

qualifyRetryable()

function qualifyRetryable(message): (cause) => RetryableError;

Defined in: packages/worker/src/errors.ts:131

Build a fromPromise qualifier that wraps the rejection cause in a RetryableError with the given message.

fromPromise requires a qualify mapper as its second argument; writing it by hand every time ((e) => retryable("...", e)) is the most re-introduced mistake in this codebase. The factory removes the boilerplate:

Parameters

Parameter	Type
`message`	`string`

Returns

(cause) => RetryableError

Example

typescript

import { qualifyRetryable } from '@amqp-contract/worker';
import { fromPromise } from 'unthrown';

const handler = ({ input: { payload } }) =>
  fromPromise(processPayment(payload), qualifyRetryable('Payment service unavailable'))
    .map(() => undefined);

// Equivalent to:
// fromPromise(processPayment(payload), (e) => retryable('Payment service unavailable', e))

rpcError()

function rpcError<TCode, TData>(
   code, 
   data, 
   message?
): RpcError<TCode, TData>;

Defined in: packages/core/dist/telemetry-U0DumV05.d.mts:220

Create an RpcError with less verbosity.

The code/data pair must match one of the entries declared in the RPC's errors map — the handler's return type enforces this at compile time, and the worker validates data against the declared schema at runtime before replying.

Type Parameters

Type Parameter
`TCode` extends `string`
`TData`

Parameters

Parameter	Type	Description
`code`	`TCode`	The error code, as declared in the RPC's `errors` map
`data`	`TData`	The error data, validated against the declared schema
`message?`	`string`	Optional human-readable message (defaults to a generic one)

Returns

RpcError<TCode, TData>

Example

typescript

import { rpcError } from '@amqp-contract/worker';
import { ErrAsync } from 'unthrown';

const handler = ({ input: { payload } }) => {
  if (!orders.has(payload.orderId)) {
    return ErrAsync(rpcError('ORDER_NOT_FOUND', { orderId: payload.orderId }));
  }
  // ...
};

@amqp-contract/worker ​

Classes ​

ConnectionError ​

Extends ​

Constructors ​

Constructor ​

Parameters ​

Returns ​

Overrides ​

Properties ​

MessageValidationError ​

Param ​

Param ​

Extends ​

Constructors ​

Constructor ​

Parameters ​

Returns ​

Overrides ​

Properties ​

NonRetryableError ​

Extends ​

Constructors ​

Constructor ​

Parameters ​

Returns ​

Overrides ​

Properties ​

PublishError ​

Extends ​

Constructors ​

Constructor ​

Parameters ​

Returns ​

Overrides ​

Properties ​

RetryableError ​

Extends ​

Constructors ​

Constructor ​

Parameters ​

Returns ​

Overrides ​

Properties ​

RpcError ​

Extends ​

Type Parameters ​

Constructors ​

Constructor ​

Parameters ​

Returns ​

Overrides ​

Properties ​

TechnicalError ​

Extends ​

Constructors ​

Constructor ​

Parameters ​

Returns ​

Overrides ​

Properties ​

TypedAmqpWorker ​

Example ​

Type Parameters ​

Methods ​

close() ​

Parameters ​

Returns ​

Example ​

isConnected() ​

Returns ​

create() ​

Type Parameters ​

Parameters ​

Returns ​

Example ​

Type Aliases ​

AnyWorkerMiddleware ​

ConnectionSource ​

Properties ​

@amqp-contract/worker

Classes

ConnectionError

Extends

Constructors

Constructor

Parameters

Returns

Overrides

Properties

MessageValidationError

Param

Param

Extends

Constructors

Constructor

Parameters

Returns

Overrides

Properties

NonRetryableError

Extends

Constructors

Constructor

Parameters

Returns

Overrides

Properties

PublishError

Extends

Constructors

Constructor

Parameters

Returns

Overrides

Properties

RetryableError

Extends

Constructors

Constructor

Parameters

Returns

Overrides

Properties

RpcError

Extends

Type Parameters

Constructors

Constructor

Parameters

Returns

Overrides

Properties

TechnicalError

Extends

Constructors

Constructor

Parameters

Returns

Overrides

Properties

TypedAmqpWorker

Example

Type Parameters

Methods

close()

Parameters

Returns

Example

isConnected()

Returns

create()

Type Parameters

Parameters

Returns

Example

Type Aliases

AnyWorkerMiddleware

ConnectionSource

Properties