@amqp-contract/worker

@amqp-contract/worker

Classes

MessageValidationError

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

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/index.d.mts:43

Parameters

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

Returns

MessageValidationError

Overrides

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

Properties

Property	Modifier	Type	Inherited from	Defined in
`_tag`	`readonly`	`"@amqp-contract/MessageValidationError"`	`MessageValidationError_base._tag`	node_modules/.pnpm/unthrown@4.1.0/node_modules/unthrown/dist/index.d.mts:1456
`cause?`	`public`	`unknown`	`MessageValidationError_base.cause`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es2022.error.d.ts:24
`issues`	`readonly`	`unknown`	`MessageValidationError_base.issues`	packages/core/dist/index.d.mts:41
`message`	`public`	`string`	`MessageValidationError_base.message`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1075
`name`	`public`	`string`	`MessageValidationError_base.name`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1074
`source`	`readonly`	`string`	`MessageValidationError_base.source`	packages/core/dist/index.d.mts:40
`stack?`	`public`	`string`	`MessageValidationError_base.stack`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1076

NonRetryableError

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

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:42

Parameters

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

Returns

NonRetryableError

Overrides

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

Properties

Property	Modifier	Type	Inherited from	Defined in
`_tag`	`readonly`	`"@amqp-contract/NonRetryableError"`	`TaggedError("@amqp-contract/NonRetryableError", { name: "NonRetryableError", })._tag`	node_modules/.pnpm/unthrown@4.1.0/node_modules/unthrown/dist/index.d.mts:1456
`cause?`	`public`	`unknown`	`MessageValidationError`.`cause`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es2022.error.d.ts:24
`message`	`public`	`string`	`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`	`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`	`TaggedError("@amqp-contract/NonRetryableError", { name: "NonRetryableError", }).stack`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1076

RetryableError

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

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 matchTags) for exhaustive dispatch; 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:22

Parameters

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

Returns

RetryableError

Overrides

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

Properties

Property	Modifier	Type	Inherited from	Defined in
`_tag`	`readonly`	`"@amqp-contract/RetryableError"`	`TaggedError("@amqp-contract/RetryableError", { name: "RetryableError", })._tag`	node_modules/.pnpm/unthrown@4.1.0/node_modules/unthrown/dist/index.d.mts:1456
`cause?`	`public`	`unknown`	`MessageValidationError`.`cause`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es2022.error.d.ts:24
`message`	`public`	`string`	`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`	`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`	`TaggedError("@amqp-contract/RetryableError", { name: "RetryableError", }).stack`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1076

RpcError

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

A typed, contract-declared RPC error — the business-failure channel of an RPC, as opposed to the transport failures modeled by TechnicalError.

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 matchTags; 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/index.d.mts:78

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	Overrides	Inherited from	Defined in
`_tag`	`readonly`	`"@amqp-contract/RpcError"`	-	`RpcError_base._tag`	node_modules/.pnpm/unthrown@4.1.0/node_modules/unthrown/dist/index.d.mts:1456
`cause?`	`public`	`unknown`	-	`RpcError_base.cause`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es2022.error.d.ts:24
`code`	`readonly`	`TCode`	`RpcError_base.code`	-	packages/core/dist/index.d.mts:76
`data`	`readonly`	`TData`	`RpcError_base.data`	-	packages/core/dist/index.d.mts:77
`message`	`public`	`string`	-	`RpcError_base.message`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1075
`name`	`public`	`string`	-	`RpcError_base.name`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1074
`stack?`	`public`	`string`	-	`RpcError_base.stack`	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1076

TypedAmqpWorker

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

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 result = await TypedAmqpWorker.create({
  contract,
  handlers: {
    processOrder: ({ payload }) => {
      console.log('Processing order', payload.orderId);
      return OkAsync(undefined);
    },
  },
  urls: ['amqp://localhost'],
});

const worker = result.getOrThrow();

// Close when done
await worker.close();

Type Parameters

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

Methods

close()

close(): AsyncResult<void, TechnicalError>;

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

Close the AMQP channel and connection.

This gracefully closes the connection to the AMQP broker, stopping all message consumption and cleaning up resources.

Returns

AsyncResult<void, TechnicalError>

Example

typescript

const closeResult = await worker.close();
if (closeResult.isOk()) {
  console.log('Worker closed successfully');
}

create()

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

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

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>, TechnicalError>

A AsyncResult that resolves to the worker or a TechnicalError.

Example

typescript

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

Type Aliases

ConsumerOptions

type ConsumerOptions = AmqpClientConsumerOptions;

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

CreateWorkerOptions

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

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

Options for creating a type-safe AMQP worker.

Example

typescript

const options: CreateWorkerOptions<typeof contract> = {
  contract: myContract,
  handlers: {
    // Simple handler
    processOrder: ({ payload }) => {
      console.log('Processing order:', payload.orderId);
      return OkAsync(undefined);
    },
    // Handler with prefetch configuration
    processPayment: [
      ({ 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.

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`	-

Properties

Property	Type	Description	Defined in
`connectionOptions?`	`AmqpConnectionManagerOptions`	Optional connection configuration (heartbeat, reconnect settings, etc.)	packages/worker/src/worker.ts:203
`connectTimeoutMs?`	`number` \| `null`	Maximum time in ms to wait for the AMQP connection to become ready before `create()` resolves to an `Err(TechnicalError)`. 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:223
`contract`	`TContract`	The AMQP contract definition specifying consumers and their message schemas	packages/worker/src/worker.ts:156
`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:185
`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:216
`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 the middleware-produced context as a third argument (an empty object when no `middleware` is configured). Use `defineHandler` / `defineHandlers` to create handlers with full type inference.	packages/worker/src/worker.ts:173
`logger?`	`Logger`	Optional logger for logging message consumption and errors	packages/worker/src/worker.ts:205
`middleware?`	`WorkerMiddleware`<`TCreated`, `TContext`>	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); compose multiple middleware with `composeMiddleware(outermost, ..., innermost)` — the chain's final context type is what handlers receive in `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:199
`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:211
`urls`	`ConnectionUrl`[]	AMQP broker URL(s). Multiple URLs provide failover support	packages/worker/src/worker.ts:201

EmptyContext

type EmptyContext = Record<never, never>;

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

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:58

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 matchTags.

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.

WorkerConsumedMessage

type WorkerConsumedMessage<TPayload, THeaders> = object;

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

A consumed message containing parsed payload and headers.

This type represents the first argument passed to consumer handlers. It contains the validated payload and (if defined in the message schema) the validated headers.

Example

typescript

const handler = defineHandler(contract, 'processOrder', (message, rawMessage) => {
  console.log(message.payload.orderId);  // Typed payload
  console.log(message.headers?.priority); // Typed headers (if defined)
  console.log(rawMessage.fields.deliveryTag); // Raw AMQP message
  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:237
`payload`	`TPayload`	The validated message payload	packages/worker/src/types.ts:235

WorkerCreateContextInfo

type WorkerCreateContextInfo = object;

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

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:79
`isRpc`	`boolean`	True when the handler is an RPC server.	packages/worker/src/worker.ts:81
`message`	`object`	The validated message (payload and headers already schema-checked).	packages/worker/src/worker.ts:83
`message.headers`	`unknown`	-	packages/worker/src/worker.ts:83
`message.payload`	`unknown`	-	packages/worker/src/worker.ts:83
`rawMessage`	`ConsumeMessage`	The raw amqplib message.	packages/worker/src/worker.ts:85

WorkerHandlerHelpers

type WorkerHandlerHelpers<TContext, TErrors> = object;

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

The helpers object every handler receives as its third argument: the middleware-produced context and (for RPC handlers with a declared errors map) the typed error constructors.

Type Parameters

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

Properties

Property	Modifier	Type	Description	Defined in
`context`	`readonly`	`TContext`	Context produced by `createContext` and the middleware chain.	packages/worker/src/types.ts:167
`errors`	`readonly`	`TErrors`	Typed constructors for the contract-declared errors (empty for consumers).	packages/worker/src/types.ts:169

WorkerInferConsumedMessage

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

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

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> = (message, rawMessage, helpers) => AsyncResult<void, HandlerError>;

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

Handler signature for a regular consumer (event/command). Returns AsyncResult<void, HandlerError> — there is no response message.

Type Parameters

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

Parameters

Parameter	Type
`message`	`WorkerInferConsumedMessage`<`TContract`, `TName`>
`rawMessage`	`ConsumeMessage`
`helpers`	`WorkerHandlerHelpers`<`TContext`>

Returns

AsyncResult<void, HandlerError>

WorkerInferConsumerHandlerEntry

type WorkerInferConsumerHandlerEntry<TContract, TName, TContext> = 
  | WorkerInferConsumerHandler<TContract, TName, TContext>
  | readonly [WorkerInferConsumerHandler<TContract, TName, TContext>, ConsumerOptions];

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

Handler entry for a regular consumer — function or [handler, options].

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:96

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:356

All handlers for a contract: one entry per consumers key plus one entry per rpcs key. The two name spaces are disjoint so the resulting object type is unambiguous.

TContext is the context produced by the worker's middleware chain; the third handler argument is typed with it.

Type Parameters

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

Example

typescript

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

WorkerInferRpcConsumedMessage

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

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

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, QueueEntry, infer TErrors> ? TErrors extends RpcErrorMap ? { [K in keyof TErrors & string]: (data: InferSchemaInput<TErrors[K]["payload"]>, message?: string) => RpcError<K, InferSchemaInput<TErrors[K]["payload"]>> } : EmptyContext : EmptyContext;

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

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, QueueEntry, infer TErrors> ? TErrors extends RpcErrorMap ? { [K in keyof TErrors & string]: RpcError<K, InferSchemaInput<TErrors[K]["payload"]>> }[keyof TErrors & string] : never : never;

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

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> = (message, rawMessage, helpers) => AsyncResult<WorkerInferRpcResponse<TContract, TName>, 
  | HandlerError
  | WorkerInferRpcErrors<TContract, TName>>;

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

Handler signature for an RPC. Returns AsyncResult<TResponse, HandlerError | RpcError> where TResponse is the inferred response payload and the RpcError members come from the RPC's declared errors map (absent when none are declared). The worker validates the response against the RPC's response schema and publishes it back to msg.properties.replyTo with the same correlationId; a declared RpcError is validated, published as an error reply, and the request is acked (business errors are not retried).

Type Parameters

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

Parameters

Parameter	Type
`message`	`WorkerInferRpcConsumedMessage`<`TContract`, `TName`>
`rawMessage`	`ConsumeMessage`
`helpers`	`WorkerHandlerHelpers`<`TContext`, `WorkerInferRpcErrorConstructors`<`TContract`, `TName`>>

Returns

AsyncResult<WorkerInferRpcResponse<TContract, TName>, | HandlerError | WorkerInferRpcErrors<TContract, TName>>

WorkerInferRpcHandlerEntry

type WorkerInferRpcHandlerEntry<TContract, TName, TContext> = 
  | WorkerInferRpcHandler<TContract, TName, TContext>
  | readonly [WorkerInferRpcHandler<TContract, TName, TContext>, ConsumerOptions];

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

Handler entry for an RPC — function or [handler, options].

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:118

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:104

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 ? InferSchemaOutput<TResponse["payload"]> : never : never;

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

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

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:88

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 third argument — or short-circuit by returning without calling next:

Err(retryable(...)) / Err(nonRetryable(...)) 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 { defineMiddleware, nonRetryable } from '@amqp-contract/worker';
import { ErrAsync } from 'unthrown';

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

WorkerMiddlewareArgs

type WorkerMiddlewareArgs<TContextIn> = object;

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

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:29
`handlerName`	`string`	The `consumers` / `rpcs` key being dispatched.	packages/worker/src/middleware.ts:25
`isRpc`	`boolean`	True when the handler is an RPC server (its result is published as a reply).	packages/worker/src/middleware.ts:27
`message`	`object`	The validated message (payload and headers already schema-checked).	packages/worker/src/middleware.ts:21
`message.headers`	`unknown`	-	packages/worker/src/middleware.ts:21
`message.payload`	`unknown`	-	packages/worker/src/middleware.ts:21
`rawMessage`	`ConsumeMessage`	The raw amqplib message (delivery tag, AMQP properties, raw headers, …).	packages/worker/src/middleware.ts:23

WorkerMiddlewareNext

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

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

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.

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>

Functions

composeMiddleware()

Call Signature

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

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

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:143

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:148

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:158

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:170

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:184

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:200

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:218

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

defineHandler()

Call Signature

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

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

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 }]

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 { defineHandler, RetryableError, NonRetryableError } from '@amqp-contract/worker';
import { fromPromise } from 'unthrown';

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

RPC handler

typescript

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

Call Signature

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

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

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 { defineHandler, RetryableError, NonRetryableError } from '@amqp-contract/worker';
import { fromPromise } from 'unthrown';

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

RPC handler

typescript

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

Call Signature

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

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

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 { defineHandler, RetryableError, NonRetryableError } from '@amqp-contract/worker';
import { fromPromise } from 'unthrown';

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

RPC handler

typescript

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

Call Signature

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

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

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 { defineHandler, RetryableError, NonRetryableError } from '@amqp-contract/worker';
import { fromPromise } from 'unthrown';

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

RPC handler

typescript

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

defineHandlers()

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

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

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 { defineHandlers, RetryableError } from '@amqp-contract/worker';
import { fromPromise, OkAsync } from 'unthrown';

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

defineMiddleware()

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

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

Identity helper that pins a middleware's context types without a variable annotation — defineMiddleware<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>

isHandlerError()

function isHandlerError(error): error is HandlerError;

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

Type guard to check if an error is any HandlerError (RetryableError or NonRetryableError).

Parameters

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

Returns

error is HandlerError

True if the error is a HandlerError

Example

typescript

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

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

isNonRetryableError()

function isNonRetryableError(error): error is NonRetryableError;

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

Type guard to check if an error is a NonRetryableError.

Use this to check error types in catch blocks or error handlers.

Parameters

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

Returns

error is NonRetryableError

True if the error is a NonRetryableError

Example

typescript

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

try {
  await processMessage();
} catch (error) {
  if (isNonRetryableError(error)) {
    console.log('Will not retry:', error.message);
  }
}

isRetryableError()

function isRetryableError(error): error is RetryableError;

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

Type guard to check if an error is a RetryableError.

Use this to check error types in catch blocks or error handlers.

Parameters

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

Returns

error is RetryableError

True if the error is a RetryableError

Example

typescript

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

try {
  await processMessage();
} catch (error) {
  if (isRetryableError(error)) {
    console.log('Will retry:', error.message);
  } else {
    console.log('Permanent failure:', error);
  }
}

isRpcError()

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

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

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 matchTags on the _tag.

Parameters

Parameter	Type
`error`	`unknown`

Returns

error is RpcError<string, unknown>

nonRetryable()

function nonRetryable(message, cause?): NonRetryableError;

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

Create a NonRetryableError with less verbosity.

This is a shorthand factory function for creating NonRetryableError instances. Use it for cleaner error creation in handlers.

Parameters

Parameter	Type	Description
`message`	`string`	Error message describing the failure
`cause?`	`unknown`	Optional underlying error that caused this failure

Returns

NonRetryableError

A new NonRetryableError instance

Example

typescript

import { nonRetryable } from '@amqp-contract/worker';
import { ErrAsync, OkAsync } from 'unthrown';

const handler = ({ payload }) => {
  if (!isValidPayload(payload)) {
    return ErrAsync(nonRetryable('Invalid payload format'));
  }
  return OkAsync(undefined);
};

// Equivalent to:
// return ErrAsync(new NonRetryableError('Invalid payload format'));

qualifyNonRetryable()

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

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

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 = ({ payload }) =>
  fromPromise(chargeCard(payload), qualifyNonRetryable('Card permanently declined'))
    .map(() => undefined);

qualifyRetryable()

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

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

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 = ({ payload }) =>
  fromPromise(processPayment(payload), qualifyRetryable('Payment service unavailable'))
    .map(() => undefined);

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

retryable()

function retryable(message, cause?): RetryableError;

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

Create a RetryableError with less verbosity.

This is a shorthand factory function for creating RetryableError instances. Use it for cleaner error creation in handlers.

Parameters

Parameter	Type	Description
`message`	`string`	Error message describing the failure
`cause?`	`unknown`	Optional underlying error that caused this failure

Returns

RetryableError

A new RetryableError instance

Example

typescript

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

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

// Equivalent to:
// fromPromise(processPayment(payload), (e) => new RetryableError('...', e))

rpcError()

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

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

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 = ({ payload }) => {
  if (!orders.has(payload.orderId)) {
    return ErrAsync(rpcError('ORDER_NOT_FOUND', { orderId: payload.orderId }));
  }
  // ...
};

@amqp-contract/worker ​

Classes ​

MessageValidationError ​

Param ​

Param ​

Extends ​

Constructors ​

Constructor ​

Parameters ​

Returns ​

Overrides ​

Properties ​

NonRetryableError ​

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 ​

TypedAmqpWorker ​

Example ​

Type Parameters ​

Methods ​

close() ​

Returns ​

Example ​

create() ​

Type Parameters ​

Parameters ​

Returns ​

Example ​

Type Aliases ​

ConsumerOptions ​

CreateWorkerOptions ​

Example ​

Type Parameters ​

Properties ​

EmptyContext ​

HandlerError ​

WorkerConsumedMessage ​

Example ​

Type Parameters ​

Properties ​

WorkerCreateContextInfo ​

Properties ​

WorkerHandlerHelpers ​

Type Parameters ​

Properties ​

WorkerInferConsumedMessage ​

Type Parameters ​

WorkerInferConsumerHandler ​

Type Parameters ​

Parameters ​

Returns ​

WorkerInferConsumerHandlerEntry ​

Type Parameters ​

WorkerInferConsumerHeaders ​

Type Parameters ​

WorkerInferHandlers ​

Type Parameters ​

Example ​

WorkerInferRpcConsumedMessage ​

@amqp-contract/worker

Classes

MessageValidationError

Param

Param

Extends

Constructors

Constructor

Parameters

Returns

Overrides

Properties

NonRetryableError

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

TypedAmqpWorker

Example

Type Parameters

Methods

close()

Returns

Example

create()

Type Parameters

Parameters

Returns

Example

Type Aliases

ConsumerOptions

CreateWorkerOptions

Example

Type Parameters

Properties

EmptyContext

HandlerError

WorkerConsumedMessage

Example

Type Parameters

Properties

WorkerCreateContextInfo

Properties

WorkerHandlerHelpers

Type Parameters

Properties

WorkerInferConsumedMessage

Type Parameters

WorkerInferConsumerHandler

Type Parameters

Parameters

Returns

WorkerInferConsumerHandlerEntry

Type Parameters

WorkerInferConsumerHeaders

Type Parameters

WorkerInferHandlers

Type Parameters

Example

WorkerInferRpcConsumedMessage