@amqp-contract/worker

---

@amqp-contract/worker

Classes

ConnectionError

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

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

Parameters

Parameter	Type
message	string
cause?	unknown

Returns

ConnectionError

Overrides

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

Properties

Property	Modifier	Type	Inherited from	Defined in
_tag	readonly	"@amqp-contract/ConnectionError"	ConnectionError_base._tag	node_modules/.pnpm/unthrown@5.9.0/node_modules/unthrown/dist/index.d.mts:2188
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	ConnectionError_base.message	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1075
name	public	string	ConnectionError_base.name	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1074
stack?	public	string	ConnectionError_base.stack	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1076

---

MessageValidationError

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

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

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@5.9.0/node_modules/unthrown/dist/index.d.mts:2188
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:71
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:70
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:39

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

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@5.9.0/node_modules/unthrown/dist/index.d.mts:2188
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:19

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

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@5.9.0/node_modules/unthrown/dist/index.d.mts:2188
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:118

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

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@5.9.0/node_modules/unthrown/dist/index.d.mts:2188
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:122
data	readonly	TData	RpcError_base.data	-	packages/core/dist/index.d.mts:123
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

---

TechnicalError

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

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

Parameters

Parameter	Type
message	string
cause?	unknown

Returns

TechnicalError

Overrides

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

Properties

Property	Modifier	Type	Inherited from	Defined in
_tag	readonly	"@amqp-contract/TechnicalError"	TechnicalError_base._tag	node_modules/.pnpm/unthrown@5.9.0/node_modules/unthrown/dist/index.d.mts:2188
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	TechnicalError_base.message	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1075
name	public	string	TechnicalError_base.name	node_modules/.pnpm/typescript@6.0.3/node_modules/typescript/lib/lib.es5.d.ts:1074
stack?	public	string	TechnicalError_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:347

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

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

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

await worker.close().get();

create()

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

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

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

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.

---

ConsumerOptions

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

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

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> = object;

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

Options for creating a type-safe AMQP worker.

Example

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.

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:265
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:285
contract	TContract	The AMQP contract definition specifying consumers and their message schemas	packages/worker/src/worker.ts:210
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:242
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:278
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:230
logger?	Logger	Optional logger for logging message consumption and errors	packages/worker/src/worker.ts:267
maxDecompressedBytes?	number	Cap on the decompressed size (bytes) of a single inbound message. 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 DEFAULT_MAX_DECOMPRESSED_BYTES (64 MiB).	packages/worker/src/worker.ts:302
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:261
publishTimeoutMs?	number | null	Maximum time in ms a worker-side publish (retry republish, RPC reply) may sit buffered waiting for the broker before its promise settles with a timeout failure (surfaced as a Defect). 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:295
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:273
urls	ConnectionUrl[]	AMQP broker URL(s). Multiple URLs provide failover support	packages/worker/src/worker.ts:263

---

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

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

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

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

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

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

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

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

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

---

TelemetryProvider

type TelemetryProvider = object;

Defined in: packages/core/dist/telemetry-BtfQAlOJ.d.mts:59

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-BtfQAlOJ.d.mts:74
getConsumeLatencyHistogram	() => Histogram | undefined	Get a histogram for consume/process latency. Returns undefined if OpenTelemetry is not available.	packages/core/dist/telemetry-BtfQAlOJ.d.mts:84
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-BtfQAlOJ.d.mts:90
getPublishCounter	() => Counter | undefined	Get a counter for messages published. Returns undefined if OpenTelemetry is not available.	packages/core/dist/telemetry-BtfQAlOJ.d.mts:69
getPublishLatencyHistogram	() => Histogram | undefined	Get a histogram for publish latency. Returns undefined if OpenTelemetry is not available.	packages/core/dist/telemetry-BtfQAlOJ.d.mts:79
getTracer	() => Tracer | undefined	Get a tracer instance for creating spans. Returns undefined if OpenTelemetry is not available.	packages/core/dist/telemetry-BtfQAlOJ.d.mts:64

---

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

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

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

---

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

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

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
helpers	WorkerHandlerHelpers<TContext, EmptyContext, WorkerInferConsumedMessage<TContract, TName>>
message	WorkerInferConsumedMessage<TContract, TName>

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

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

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

const handlers: WorkerInferHandlers<typeof contract> = {
  processOrder: ({ input: { payload } }) =>
    fromPromise(
      processPayment(payload),
      (error) => new RetryableError('Payment failed', error),
    ).map(() => undefined),
  calculate: ({ input: { 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: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> = (helpers, message) => AsyncResult<WorkerInferRpcResponse<TContract, TName>, 
  | HandlerError
  | WorkerInferRpcErrors<TContract, TName>>;

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

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
helpers	WorkerHandlerHelpers<TContext, WorkerInferRpcErrorConstructors<TContract, TName>, WorkerInferRpcConsumedMessage<TContract, TName>>
message	WorkerInferRpcConsumedMessage<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:365

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

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

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

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

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
TB extends Record<string, unknown> | EmptyContext

Parameters

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

Returns

WorkerMiddleware<TIn, TB>

Example

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

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

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

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

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

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

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

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

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

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

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

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

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:

1. Simple handler: just the function
2. 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

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

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.

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:

1. Simple handler: just the function
2. 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>.
options	ConsumerOptions	Optional consumer options (prefetch)

Returns

WorkerInferConsumerHandlerEntry<TContract, TName, TContext>

A type-safe handler that can be used with TypedAmqpWorker

Examples

Consumer handler

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

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.

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:

1. Simple handler: just the function
2. 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	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

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

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.

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:

1. Simple handler: just the function
2. 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	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

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

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.

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.

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

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

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

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

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

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

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

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

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

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

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

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

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

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 }));
  }
  // ...
};