@cosyte/pathways

attempt

readonly attempt: number

Which attempt this was, from 1.

Inherited from

channel

readonly channel: string

The channel's name.

Inherited from

deliveryId

readonly deliveryId: number

Inherited from

destination

readonly destination: number

The destination's place in the channel's destinations: list, from 0.

Inherited from

messageId

readonly messageId: number

Inherited from

resend?

readonly optional resend?: object

Set when this attempt was a possible duplicate: an earlier one was cut off after the send, so the receiver may have the message already. previousAttempts counts the attempts before it. Absent otherwise.

previousAttempts

readonly previousAttempts: number

Inherited from

type

readonly type: "delivered"

Overrides

DeliveryEventBase

What every event about one delivery carries: the message, the destination and the delivery.

Example

const key = (event: DeliveryEventBase): string => `${event.channel}/${event.destination}/${event.deliveryId}`;

Extends

LifecycleEventBase

Properties

at

readonly at: string

When it happened, as an ISO 8601 time in UTC, such as 2026-11-07T12:00:00.000Z.

Inherited from

attempt

readonly attempt: number

Which attempt this was, from 1.

channel

readonly channel: string

The channel's name.

Inherited from

deliveryId

readonly deliveryId: number

destination

readonly destination: number

The destination's place in the channel's destinations: list, from 0.

messageId

readonly messageId: number

resend?

readonly optional resend?: object

previousAttempts

readonly previousAttempts: number

type

readonly type: "received" | "receive-failed" | "transformed" | "filtered" | "transform-failed" | "delivered" | "delivery-retry" | "delivery-redirected" | "delivery-rejected" | "delivery-exhausted" | "source-down" | "source-up" | "audit"

Inherited from

DeliveryExhaustedEvent

A delivery ran out of retries: it had its retry.limit of attempts, or its next attempt would come later than retry.maxAge after it was queued. It is marked failed, its body kept for an operator to resend, and the destination moves on to the next. Only a destination whose retry settings set a limit or an age sends this.

Example

if (event.type === "delivery-exhausted") page(`${event.channel}: gave up after ${event.attempt}`);

Extends

DeliveryEventBase

Properties

at

readonly at: string

When it happened, as an ISO 8601 time in UTC, such as 2026-11-07T12:00:00.000Z.

Inherited from

attempt

readonly attempt: number

Which attempt this was, from 1.

Inherited from

channel

readonly channel: string

The channel's name.

Inherited from

code

readonly code: string | undefined

The last error's code, such as ECONNREFUSED or HTTP_503.

deliveryId

readonly deliveryId: number

Inherited from

destination

readonly destination: number

The destination's place in the channel's destinations: list, from 0.

Inherited from

messageId

readonly messageId: number

Inherited from

name

readonly name: string

The last error's name, such as MllpDeliveryError.

reason

readonly reason: "limit" | "maxAge"

limit when it had its attempts, maxAge when it was too old to wait for another.

resend?

readonly optional resend?: object

previousAttempts

readonly previousAttempts: number

Inherited from

type

readonly type: "delivery-exhausted"

Overrides

DeliveryRedirectedEvent

An HTTP destination answered with a redirect (a 3xx). Redirects are never followed, since following one could send a message to a host nobody configured: the delivery is held and tried again after the longest wait, five minutes, and its later deliveries wait behind it, until the destination's URL is fixed. It names the Location's host only, never its path or query.

Example

if (event.type === "delivery-redirected") page(`${event.channel} moved to ${event.locationHost}`);

Extends

DeliveryEventBase

Properties

at

readonly at: string

When it happened, as an ISO 8601 time in UTC, such as 2026-11-07T12:00:00.000Z.

Inherited from

attempt

readonly attempt: number

Which attempt this was, from 1.

Inherited from

channel

readonly channel: string

The channel's name.

Inherited from

deliveryId

readonly deliveryId: number

Inherited from

destination

readonly destination: number

The destination's place in the channel's destinations: list, from 0.

Inherited from

locationHost

readonly locationHost: string | undefined

The host the Location names, with its port when it has one, or undefined when it names none.

messageId

readonly messageId: number

Inherited from

resend?

readonly optional resend?: object

previousAttempts

readonly previousAttempts: number

Inherited from

retryAt

readonly retryAt: string

When the next attempt is due, as an ISO 8601 time in UTC.

retryInMs

readonly retryInMs: number

How long until the next attempt, in milliseconds.

status

readonly status: number

The 3xx status, such as 301 or 308.

type

readonly type: "delivery-redirected"

Overrides

DeliveryRejectedEvent

A destination rejected an output for good (an MLLP AE, an HTTP 4xx): it is marked failed and the destination moves on to the next.

Example

if (event.type === "delivery-rejected") page(`${event.channel}: rejected (${event.reason})`);

Extends

DeliveryEventBase

Properties

at

readonly at: string

When it happened, as an ISO 8601 time in UTC, such as 2026-11-07T12:00:00.000Z.

Inherited from

attempt

readonly attempt: number

Which attempt this was, from 1.

Inherited from

channel

readonly channel: string

The channel's name.

Inherited from

deliveryId

readonly deliveryId: number

Inherited from

destination

readonly destination: number

The destination's place in the channel's destinations: list, from 0.

Inherited from

messageId

readonly messageId: number

Inherited from

reason

readonly reason: string

The destination's short, stable reason, such as AE or HTTP 400.

resend?

readonly optional resend?: object

previousAttempts

readonly previousAttempts: number

Inherited from

type

readonly type: "delivery-rejected"

Overrides

DeliveryRetryEvent

A delivery failed in a way worth trying again (a refused connection, a timeout, an MLLP AR, an HTTP 408, 425, 429 or 5xx). The destination retries it after retryInMs, and its later deliveries wait behind it. retryInMs is the backoff, or the wait a 429's or a 503's Retry-After asked for when that is longer, up to five minutes.

Example

if (event.type === "delivery-retry" && event.attempt === 5) page(`${event.channel} is retrying`);

Extends

DeliveryEventBase

Properties

at

readonly at: string

When it happened, as an ISO 8601 time in UTC, such as 2026-11-07T12:00:00.000Z.

Inherited from

attempt

readonly attempt: number

Which attempt this was, from 1.

Inherited from

channel

readonly channel: string

The channel's name.

Inherited from

code

readonly code: string | undefined

The error's code, such as ECONNREFUSED, ACK_TIMEOUT or HTTP_503.

deliveryId

readonly deliveryId: number

Inherited from

destination

readonly destination: number

The destination's place in the channel's destinations: list, from 0.

Inherited from

messageId

readonly messageId: number

Inherited from

name

readonly name: string

The error's name, such as MllpDeliveryError.

resend?

readonly optional resend?: object

previousAttempts

readonly previousAttempts: number

Inherited from

retryAt

readonly retryAt: string

When the next attempt is due, as an ISO 8601 time in UTC.

retryInMs

readonly retryInMs: number

How long until the next attempt, in milliseconds.

type

readonly type: "delivery-retry"

Overrides

SecretProvider.resolve

DestinationConnector

Sends messages out. Throw to fail an attempt: by default the engine retries it with growing waits and holds this destination's later messages behind it. Throw a DeliveryRejected when sending the same message again cannot succeed (the receiver rejected its content), and the engine marks it failed and moves on. A custom destination's module builds one with a DestinationFactory.

Example

const destination: DestinationConnector = {
  async deliver(message) {
    await upload(`${message.messageId}-${message.seq}`, message.body);
  },
};

Methods

deliver()

deliver(message): Promise<void>

Parameters

message

OutgoingMessage

Returns

Promise<void>

stop()?

optional stop(): Promise<void>

Releases what the destination keeps open between deliveries, such as connections.

Returns

Promise<void>

ExpireResult

What retention deleted, or would delete, from one channel.

Example

const { messages, deliveries }: ExpireResult = await store.expire("adt", cutoff, true);

Properties

deliveries

readonly deliveries: number

messages

readonly messages: number

ExtensionContext

What every extension factory is given beside its options.

Example

const token = context.env["INBOX_TOKEN"];
const folder = path.resolve(context.dir, "./in");

Extended by

ChannelContext
HookContext

Properties

dir

readonly dir: string

The folder of the YAML file that names the extension: the channel's folder, or the project root for pathways.yaml. Resolve relative paths in options against it.

env

readonly env: Readonly<Record<string, string | undefined>>

Where to read credentials and other settings from: process.env when the engine runs.

logger

readonly logger: ExtensionLogger

ExtensionLogger

The engine's logger, handed to an extension. Lines land in the same JSON log as the engine's, tagged with the channel and which extension wrote them.

Example

context.logger.warn({ event: "inbox-read-failed", code: "EACCES" });

Methods

error()

error(fields): void

Parameters

fields

LogFields

Returns

void

info()

info(fields): void

Parameters

fields

LogFields

Returns

void

warn()

warn(fields): void

Parameters

fields

LogFields

Returns

void

FakeSecret

One version of a secret a fakeSecretProvider holds: its value, and optionally its version's name (by default its place in the list, from "1"), when it expires and the identity that read it.

Example

const secret: FakeSecret = { value: "synthetic-token", version: "v2" };

Properties

expiresAt?

readonly optional expiresAt?: Date

identity?

readonly optional identity?: string

The principal it answers as having read it, as ResolvedSecret.identity.

value

readonly value: string

version?

readonly optional version?: string

FakeSecretProvider

A SecretProvider held in memory, for tests: no network, no cloud account. It answers as the contract asks of every provider, so a channel that reads { secret: { provider, name } } can be tested against it in place of a cloud store.

Example

const vault = fakeSecretProvider({ "lab/api": '{"token":"synthetic-token"}' });
await vault.resolve({ name: "lab/api", field: "token" }, AbortSignal.timeout(1_000));
vault.calls; // [{ name: "lab/api", field: "token" }]

Extends

SecretProvider

Properties

calls

readonly calls: readonly SecretRef[]

Every request it was given, in order.

stopped

readonly stopped: number

How many times stop was called.

Methods

delete()

delete(name): void

Removes name, so a later request for it is not found.

Parameters

name

string

Returns

void

fail()

fail(name, error): void

Makes every later request for name reject with error, as a store that fails would.

Parameters

name

string

error

Error

Returns

void

resolve()

resolve(ref, signal): Promise<ResolvedSecret>

Resolves one secret.

Parameters

ref

SecretRef

signal

AbortSignal

Returns

Promise<ResolvedSecret>

Inherited from

set()

set(name, entry): void

Sets or replaces name; a single value or secret adds it as a new current version.

Parameters

name

string

entry

FakeSecretEntry

Returns

void

stop()?

optional stop(): void | Promise<void>

Releases what the provider holds open. Called once start has resolved every secret.

Returns

void | Promise<void>

Inherited from

SecretProvider.stop

FileSourceFacts

What a file source knows about a message: the file's name as it was dropped, its size in bytes, and when it was stored.

Example

const name = (source: FileSourceFacts): string => source.name;

Properties

kind

readonly kind: "file"

name

readonly name: string

The file's name as it was dropped in the folder. It can carry a patient's name: never log it.

receivedAt

readonly receivedAt: string

When the message was stored, as an ISO 8601 time.

size

readonly size: number

The body's size in bytes.

FilteredEvent

A message's transform returned no output, so nothing is delivered.

Example

if (event.type === "filtered") count("filtered", 1);

Extends

LifecycleEventBase

Properties

at

readonly at: string

When it happened, as an ISO 8601 time in UTC, such as 2026-11-07T12:00:00.000Z.

Inherited from

channel

readonly channel: string

The channel's name.

Inherited from

messageId

readonly messageId: number

type

readonly type: "filtered"

Overrides

ExtensionContext.logger

HookContext

What a hook factory is given beside its options: the context, and the channel for a hook a channel's YAML names (undefined for an engine-wide one in pathways.yaml).

Example

const scope = context.channel ?? "engine";

Extends

ExtensionContext

Properties

channel

readonly channel: string | undefined

dir

readonly dir: string

The folder of the YAML file that names the extension: the channel's folder, or the project root for pathways.yaml. Resolve relative paths in options against it.

Inherited from

ExtensionContext.dir

env

readonly env: Readonly<Record<string, string | undefined>>

Where to read credentials and other settings from: process.env when the engine runs.

Inherited from

ExtensionContext.env

logger

readonly logger: ExtensionLogger

Inherited from

HttpSourceFacts

What an HTTP source knows about a message: the request's method, its path as the source matched it, the sender's address, and the headers the channel's contextHeaders names, by lower-case name. Authorization, Proxy-Authorization and Cookie are never among them.

Example

const requestId = (source: HttpSourceFacts): string | undefined => source.headers["x-request-id"];

Properties

headers

readonly headers: Readonly<Record<string, string>>

The listed headers the request carried, by lower-case name; several values joined by , .

kind

readonly kind: "http"

method

readonly method: string

POST or PUT.

path

readonly path: string

The path the request was sent to, as the source matched it, without a query.

receivedAt

readonly receivedAt: string

When the message was stored, as an ISO 8601 time.

remoteAddress

readonly remoteAddress: string | undefined

The sender's IP address, when the connection still had one.

KeyProvider

Where master keys live. Pathways encrypts each stored value with a fresh data key and asks the provider to wrap that key under its current master key; to read the value back, it asks the provider to unwrap it. A master key never has to leave the provider, so a cloud key service (AWS KMS, Azure Key Vault, Google Cloud KMS) can implement this interface by calling its own encrypt and decrypt. The built-in provider holds keys read from files or environment variables.

Rotation is the provider's to offer: it keeps unwrapping data keys under its older master keys while it wraps new ones under the current key.

Pathways overwrites a data key with zeros once it is done with it: wrapKey must not keep the key it is given, and unwrapKey must return a buffer of its own.

Example

const kms: KeyProvider = {
  currentKeyId: "kms-2026-11",
  wrapKey: async (dataKey) => ({ keyId: "kms-2026-11", wrapped: await myKms.encrypt(dataKey) }),
  unwrapKey: async (_keyId, wrapped) => myKms.decrypt(wrapped),
};

Properties

currentKeyId

readonly currentKeyId: string

The id of the master key that wraps new data keys.

keyIds?

readonly optional keyIds?: readonly string[]

Every master key id the provider holds, the current one among them: a metadata search derives its index tags under each, so rows tagged under an older key are still found. Optional; without it, a search runs under the current key alone.

Methods

deriveIndexKey()?

optional deriveIndexKey(keyId, info): Promise<Uint8Array<ArrayBufferLike>>

Derives a 32-byte key for one searchable index from the master key keyId, with HKDF-SHA-256 and info naming the index. Optional: a provider whose master keys never leave a key service cannot derive one, and a channel with an index: block then refuses to start on it. The returned buffer is the caller's, which overwrites it with zeros once it is done.

Parameters

keyId

string

info

Uint8Array

Returns

Promise<Uint8Array<ArrayBufferLike>>

unwrapKey()

unwrapKey(keyId, wrapped): Promise<Uint8Array<ArrayBufferLike>>

Unwraps a data key that the master key keyId wrapped, returning its 32 bytes. Rejects with an EncryptionError: ERR_PATHWAYS_UNKNOWN_KEY when it holds no key keyId, ERR_PATHWAYS_INTEGRITY when the wrapped key fails its check.

Parameters

keyId

string

wrapped

Uint8Array

Returns

Promise<Uint8Array<ArrayBufferLike>>

wrapKey()

wrapKey(dataKey): Promise<WrappedKey>

Wraps a 32-byte data key under the current master key.

Parameters

dataKey

Uint8Array

Returns

Promise<WrappedKey>

LeaseClaimant

Who asks for a lease: the node, an id each engine process draws at start, and its host name.

Example

const claimant: LeaseClaimant = { node: "3f9a1c2b", host: "engine-a" };

Properties

host

readonly host: string

node

readonly node: string

LeaseHolder

A lease's last holder, by the database's clock: who took it, when, under which epoch, and how long it has left. expiresInMs of 0 or less means it has lapsed or was released, and the next acquisition takes it.

Example

const held = (holder: LeaseHolder): boolean => holder.expiresInMs > 0;

Properties

epoch

readonly epoch: number

expiresInMs

readonly expiresInMs: number

host

readonly host: string

node

readonly node: string

since

readonly since: string

When it was taken, as an ISO 8601 time in UTC.

LifecycleEventBase

What every lifecycle event about a channel carries: its type, when it happened and the channel it happened on. An AuditEvent can concern no one channel, so it stands on its own. Events carry ids, counts, error names, codes and times, never message content and never an error's own text, so a hook can send them anywhere.

Example

const line = (event: LifecycleEventBase): string => `${event.at} ${event.channel} ${event.type}`;

Extended by

DeliveryEventBase
FilteredEvent
ReceivedEvent
ReceiveFailedEvent
SourceDownEvent
SourceUpEvent
TransformedEvent
TransformFailedEvent

Properties

at

readonly at: string

When it happened, as an ISO 8601 time in UTC, such as 2026-11-07T12:00:00.000Z.

channel

readonly channel: string

The channel's name.

type

readonly type: "received" | "receive-failed" | "transformed" | "filtered" | "transform-failed" | "delivered" | "delivery-retry" | "delivery-redirected" | "delivery-rejected" | "delivery-exhausted" | "source-down" | "source-up" | "audit"

LifecycleHook

A lifecycle hook: told of each LifecycleEvent, one at a time and in order. Paging, webhooks and forwarding to a SIEM are hooks.

A hook cannot hold the engine up. Each call has a time limit (the hook entry's timeout, 5 seconds unless set): past it, signal aborts, the call is abandoned and logged, and the hook gets the next event. A call that throws is logged by error name and code, and the next event goes ahead. Events queue for a busy hook; past 1,000 waiting, new ones are dropped and logged. On stop, the engine gives a hook up to its time limit to finish what is queued, then calls stop.

Hooks run in the engine's own thread: an endless synchronous loop stops everything, so a hook awaits its I/O and does little work itself.

Example

// A webhook, much as the built-in `webhook:` hook does it.
import type { HookFactory } from "@cosyte/pathways";

export default ((options, context) => ({
  async onEvent(event, signal) {
    const response = await fetch(options.url, {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
        Authorization: `Bearer ${context.env["OPS_WEBHOOK_TOKEN"] ?? ""}`,
      },
      body: JSON.stringify(event),
      signal,
    });
    if (!response.ok) throw Object.assign(new Error("webhook refused"), { code: `HTTP_${response.status}` });
  },
})) satisfies HookFactory<{ url: string }>;

Methods

onEvent()

onEvent(event, signal): void | Promise<void>

Handles one event. Resolve when done; reject to have the failure logged. signal aborts when the call runs past its time limit.

Parameters

event

LifecycleEvent

signal

AbortSignal

Returns

void | Promise<void>

stop()?

optional stop(): void | Promise<void>

Releases what the hook keeps open, such as connections. Called once, after the last event.

Returns

void | Promise<void>

LogFields

One structured log line: an event name, then fields that are ids, counts, codes and error names. Never message content, never an error's own text, never a credential.

Example

const fields: LogFields = { event: "inbox-read-failed", code: "EACCES" };

Indexable

[field: string]: LogValue

Properties

event

readonly event: string

MessageQuery

Which messages MessageStore.listMessages returns: newest first, every filter optional.

Example

const query: MessageQuery = { channel: "adt", state: "failed", since: "2026-11-01T00:00:00.000Z", limit: 50 };

Properties

before?

readonly optional before?: number

Only ids below this one: the next of the page before.

channel?

readonly optional channel?: string

ids?

readonly optional ids?: readonly number[]

Only these messages, by id.

limit

readonly limit: number

since?

readonly optional since?: string

Received at or after this time (ISO 8601, UTC).

state?

readonly optional state?: "received" | "filtered" | "delivered" | "failed" | "cleared" | "dry-run" | "queued" | "retrying"

until?

readonly optional until?: string

Received before this time (ISO 8601, UTC).

MessageSink

What a source hands each incoming message to, and tells when its connection drops and comes back.

Example

const id = await sink.receive(body); // durable once this resolves
await acknowledge(id);

socket.on("close", () => {
  const wait = sink.sourceDown?.("ECONNRESET") ?? 1_000;
  setTimeout(reconnect, wait);
});
socket.on("ready", () => sink.sourceUp?.());

Methods

receive()

receive(body, options?): Promise<number>

Stores the message and resolves with its id once it is durable. Acknowledge the sender (move the file, send the ACK) only after this resolves: that ordering is what keeps a crash from losing a message. If it rejects, do not acknowledge; the sender should send again.

options is optional, so a source that passes only body keeps working unchanged.

Parameters

body

Uint8Array

options?

ReceiveOptions

Returns

Promise<number>

sourceDown()?

optional sourceDown(code): number

Says the source lost what it reads from (a broker, a remote server, a socket it dialled) and cannot take messages in until it is back. Until sourceUp, the health endpoint shows the channel's source down and answers 503, readiness fails, and the hooks are sent a source-down event with name: "SourceDown". code is a short code such as ECONNRESET (letters, digits and _.:-, at most 64 characters), never an error's text: anything else is kept as UNSPECIFIED.

Returns how long to wait before trying to connect again, in milliseconds: 1 second, doubling with each report in a row, to 5 minutes. The engine's own sink always has this method; it is optional so a sink a test builds by hand need not.

Parameters

code

string

Returns

number

sourceUp()?

optional sourceUp(): void

Says a source that reported sourceDown is connected again: health shows it up, and the hooks are sent a source-up event. Nothing happens when it was not down.

Returns

void

MessageStore

What the engine and the operator commands need from storage. Store implements it on SQLite, PostgreSQL, MySQL or MariaDB, and SQL Server; a storage adapter for anything else implements the same methods, and its module's StoreFactory returns it. A layer around a store (encryption at rest) wraps them: every method that takes or returns a body or an error's text is here, so a wrapping layer sees them all.

Every method is the engine's only path to that data, so an adapter keeps these promises: receive is durable before it resolves; completeTransform writes all its deliveries and the message's new status, or none of them; nextDelivery returns the lowest pending id for that destination, so deliveries stay in order; a new delivery is due at once (nextAttemptAt 0); markSending is durable before it resolves, nextDelivery returns the marker it wrote, and markDelivered, scheduleRetry and failDelivery clear it in the same write as their outcome. For the operator commands: requeue and clear change nothing unless the message still stands where the caller found it, and write their audit entry in the same transaction as their change; expire deletes only done messages, with their deliveries, and never an audit entry. A message whose content was dropped (markDeliveredAndDropContent) is never replayed, resent or expired.

The audit log has two methods, append and read, and no way to change or remove an entry.

The lease methods are optional. A store with acquireLease, renewLease, releaseLease and leaseHolder lets pathways start refuse a second engine and run a standby; each judges expiry by one clock, the database's, in the same statement as its change. Given an epoch, the fenced writes (completeTransform, failTransform, markSending, markDelivered, scheduleRetry and failDelivery) change their row only while the engine lease still carries that epoch, in the same statement, and reject with a LeaseLostError when they change nothing. A store without the lease methods runs as before, with a warning at start, and cannot run a standby.

Example

class PoolStore implements MessageStore {
  async receive(channel: string, body: Uint8Array, options?: StoreReceiveOptions) {
    const id = await this.pool.insertMessage(channel, body, options?.failure);
    return { id, duplicateOf: undefined };
  }
  // ...and the other methods, each against the same database.
}

Methods

acquireLease()?

optional acquireLease(name, claimant, ttlMs): Promise<LeaseClaim>

Takes the lease name for ttlMs when it is free or has lapsed by the database's clock, raising its epoch by one, in one statement; else answers with its holder. Optional.

Parameters

name

string

claimant

LeaseClaimant

ttlMs

number

Returns

Promise<LeaseClaim>

appendAudit()

appendAudit(actor, entries): Promise<AuditEntry[]>

Appends entries to the audit log, stamped with the time and actor.

Parameters

actor

string

entries

readonly AuditInput[]

Returns

Promise<AuditEntry[]>

auditEntries()

auditEntries(query): Promise<Page<AuditEntry>>

A page of the audit log, newest first.

Parameters

query

AuditQuery

Returns

Promise<Page<AuditEntry>>

channels()

channels(): Promise<string[]>

Every channel with a stored message, by name.

Returns

Promise<string[]>

channelStates()?

optional channelStates(): Promise<Record<string, ChannelRunState>>

Every channel an operator paused or stopped, by name; a channel missing from it runs. Optional, with setChannelState: a store without them keeps a channel's state in the engine only, so a restart forgets it.

Returns

Promise<Record<string, ChannelRunState>>

clear()

clear(messageId, actor): Promise<ClearResult>

Marks a failed message, and its deliveries that failed for good, cleared. Refuses when there is nothing failed. The audit entry is written in the same transaction.

Parameters

messageId

number

actor

string

Returns

Promise<ClearResult>

close()

close(): Promise<void>

Returns

Promise<void>

completeTransform()

completeTransform(messageId, outputs, destinations, epoch?): Promise<void>

Fenced by epoch when it is given: see the lease methods above.

Parameters

messageId

number

outputs

readonly Uint8Array<ArrayBufferLike>[]

destinations

number

epoch?

number

Returns

Promise<void>

deliveries()

deliveries(messageId): Promise<StoredDelivery[]>

A message's deliveries, every generation, oldest first, without bodies.

Parameters

messageId

number

Returns

Promise<StoredDelivery[]>

deliveryCounts()

deliveryCounts(channel): Promise<DeliveryCounts>

How many deliveries of a channel stand at each status; dry-run only when some ran dry.

Parameters

channel

string

Returns

Promise<DeliveryCounts>

expire()

expire(channel, cutoff, dryRun): Promise<ExpireResult>

Deletes a channel's messages that are done (delivered, dry-run, filtered or cleared) and whose last activity is before cutoff (ISO 8601), with their deliveries, leaving those whose content was dropped. With dryRun, counts them only.

Parameters

channel

string

cutoff

string

dryRun

boolean

Returns

Promise<ExpireResult>

failDelivery()

failDelivery(deliveryId, error, epoch?): Promise<void>

Fenced by epoch when it is given.

Parameters

deliveryId

number

error

string

epoch?

number

Returns

Promise<void>

failTransform()

failTransform(messageId, error, epoch?): Promise<void>

Fenced by epoch when it is given.

Parameters

messageId

number

error

string

epoch?

number

Returns

Promise<void>

leaseHolder()?

optional leaseHolder(name): Promise<LeaseHolder | undefined>

The lease's last holder, or undefined when it was never taken. Optional, with acquireLease.

Parameters

name

string

Returns

Promise<LeaseHolder | undefined>

listMessages()

listMessages(query): Promise<Page<MessageSummary>>

A page of messages, newest first, without bodies or errors.

Parameters

query

MessageQuery

Returns

Promise<Page<MessageSummary>>

listRetrying()

listRetrying(channel, limit): Promise<RetryingDelivery[]>

A channel's retrying deliveries, oldest first, at most limit.

Parameters

channel

string

limit

number

Returns

Promise<RetryingDelivery[]>

markDelivered()

markDelivered(deliveryId, epoch?): Promise<void>

Fenced by epoch when it is given.

Parameters

deliveryId

number

epoch?

number

Returns

Promise<void>

markDeliveredAndDropContent()?

optional markDeliveredAndDropContent(deliveryId, epoch?): Promise<boolean>

markDelivered, and when that was the last delivery of the message's current generation still to be delivered, also drops the bodies of the message and of all its deliveries in the same transaction, keeping its metadata: ids, times, statuses, sizes, the content hash and the possible-duplicate flag. Its ctx.meta and that metadata's index tags go with the bodies, since they hold identifiers. Keeps every body while any delivery of the current generation has failed or waits to be tried again. Resolves to whether it dropped them. Fenced by epoch when it is given. Optional: a channel with storage.content: dropAfterDelivery needs it, and the engine refuses to start such a channel on a store without it.

Parameters

deliveryId

number

epoch?

number

Returns

Promise<boolean>

markDryRun()?

optional markDryRun(deliveryId, epoch?): Promise<void>

Marks a delivery dry-run: its output stays recorded, it was never sent, and its destination moves on to the next. Clears the marker. Fenced by epoch when it is given. Optional: an engine with a destination that runs dry refuses to start on a store without it.

Parameters

deliveryId

number

epoch?

number

Returns

Promise<void>

markSending()

markSending(deliveryId, start, epoch?): Promise<void>

Writes the marker before a delivery is handed to its destination, in one write: the attempt and its start, and the attempt found cut off when start.interruptedAttempt is set. A delivery nextDelivery returns still marked was cut off, and its next send is flagged. Fenced by epoch when it is given.

Parameters

deliveryId

number

start

SendStart

epoch?

number

Returns

Promise<void>

message()

message(id): Promise<StoredMessage | undefined>

One message, without its body, or undefined if there is none. Its error is raw: see there.

Parameters

id

number

Returns

Promise<StoredMessage | undefined>

messageBody()

messageBody(id): Promise<Uint8Array<ArrayBufferLike> | undefined>

One message's body as it arrived, or undefined if there is none or it was dropped.

Parameters

id

number

Returns

Promise<Uint8Array<ArrayBufferLike> | undefined>

messageCounts()

messageCounts(channel): Promise<Record<MessageStatus, number>>

How many messages of a channel stand at each status. The health endpoint reports these.

Parameters

channel

string

Returns

Promise<Record<MessageStatus, number>>

metaCandidates()?

optional metaCandidates(query): Promise<MetaCandidate[]>

Up to query.limit messages that carry the tags query names, newest first, each with its metadata as stored, for a metadata search to confirm. Optional: a store without it cannot be searched by metadata.

Parameters

query

MetaCandidateQuery

Returns

Promise<MetaCandidate[]>

nextDelivery()

nextDelivery(channel, destination, options?): Promise<PendingDelivery | undefined>

Parameters

channel

string

destination

number

options?

NextDeliveryOptions

Returns

Promise<PendingDelivery | undefined>

nextToTransform()

nextToTransform(channel): Promise<PendingMessage | undefined>

Parameters

channel

string

Returns

Promise<PendingMessage | undefined>

outputs()

outputs(messageId): Promise<Uint8Array<ArrayBufferLike>[]>

The outputs of a message's current generation, in order; none when it has none or they were dropped.

Parameters

messageId

number

Returns

Promise<Uint8Array<ArrayBufferLike>[]>

receive()

receive(channel, body, options?): Promise<ReceivedMessage>

Stores an incoming message, durably, before its source acknowledges it: as received, or as failed when options.failure is set. Flags it as a possible duplicate when an earlier message of the channel has the same options.duplicateKey. Given options.resumed, when a message of the channel was already stored with the same options.idempotencyKey, stores nothing and returns that message's id with alreadyStored: true. A store that ignores the key still works, and a source that stops between storing and acknowledging then stores that message twice.

Parameters

channel

string

body

Uint8Array

options?

StoreReceiveOptions

Returns

Promise<ReceivedMessage>

releaseLease()?

optional releaseLease(name, epoch?): Promise<boolean>

Frees the lease name, so the next acquisition takes it at once: only while it carries epoch when that is given, whoever holds it when not (pathways lease release --force). Returns whether it freed it. Optional, with acquireLease.

Parameters

name

string

epoch?

number

Returns

Promise<boolean>

renewLease()?

optional renewLease(name, epoch, ttlMs): Promise<boolean>

Extends the lease name to ttlMs from now, by the database's clock, while it still carries epoch and has not lapsed. Returns whether it did. Optional, with acquireLease.

Parameters

name

string

epoch

number

ttlMs

number

Returns

Promise<boolean>

requeue()

requeue(messageId, request, actor): Promise<RequeueResult>

Records a replay or a resend: the message's new status, a new generation, and fresh deliveries at the back of each destination's queue. Refuses, changing nothing, when the message has moved since the caller read it or has deliveries still pending. The audit entry is written in the same transaction, whichever way it goes.

Parameters

messageId

number

request

RequeueRequest

actor

string

Returns

Promise<RequeueResult>

retryingDeliveries()

retryingDeliveries(channel): Promise<number>

How many of a channel's pending deliveries have failed at least once and wait for a retry.

Parameters

channel

string

Returns

Promise<number>

scheduleRetry()

scheduleRetry(deliveryId, attempts, nextAttemptAt, error, epoch?): Promise<void>

Fenced by epoch when it is given.

Parameters

deliveryId

number

attempts

number

nextAttemptAt

number

error

string

epoch?

number

Returns

Promise<void>

setChannelState()?

optional setChannelState(channel, state): Promise<void>

Sets a channel's state; running removes its row. Optional, with channelStates.

Parameters

channel

string

state

ChannelRunState

Returns

Promise<void>

MessageSummary

What storage holds about one message, without its body or its error: safe to list.

Example

const { items } = await store.listMessages({ state: "failed", limit: 20 });
const ids = items.map((message: MessageSummary) => message.id);

Extended by

StoredMessage

Properties

channel

readonly channel: string

contentDroppedAt?

readonly optional contentDroppedAt?: string

When the message's body and its deliveries' bodies were dropped, once its last delivery was delivered (storage.content: dropAfterDelivery). Undefined while they are kept. Such a message cannot be replayed or resent, and its content cannot be shown.

contentHash?

readonly optional contentHash?: string

The hash of the body as received, when one was stored with it: see StoreReceiveOptions.

deliveries

readonly deliveries: DeliveryTally

The deliveries of the current generation, by status.

duplicateOf

readonly duplicateOf: number | null

The latest earlier message with the same sender and control ID, when there is one.

generation

readonly generation: number

0 until the message is first replayed or resent; one more each time.

id

readonly id: number

receivedAt

readonly receivedAt: string

size

readonly size: number | null

The body's size in bytes, or null for a message stored before sizes were kept.

state

readonly state: "received" | "filtered" | "delivered" | "failed" | "cleared" | "dry-run" | "queued" | "retrying"

The status and the current deliveries, as an operator sees them.

status

readonly status: MessageStatus

updatedAt

readonly updatedAt: string

When the message last changed: transformed, failed, replayed or cleared.

MetaCandidate

A message whose index tags match a search, and its metadata as stored, which the search decrypts to confirm the match: tags are truncated, so a row can carry a tag and hold another value.

Example

const candidate: MetaCandidate = { id: 12, meta: storedJson };

Properties

id

readonly id: number

meta

readonly meta: string

MetaCandidateQuery

Which messages MessageStore.metaCandidates returns: those of channel that carry, for every condition, one of its tags, newest first, with the other filters of a listing.

Example

const query: MetaCandidateQuery = { channel: "adt", conditions: [condition], limit: 100 };

Properties

before?

readonly optional before?: number

Only ids below this one.

channel

readonly channel: string

conditions

readonly conditions: readonly MetaCondition[]

limit

readonly limit: number

since?

readonly optional since?: string

Received at or after this time (ISO 8601, UTC).

state?

readonly optional state?: "received" | "filtered" | "delivered" | "failed" | "cleared" | "dry-run" | "queued" | "retrying"

until?

readonly optional until?: string

Received before this time (ISO 8601, UTC).

MetaCondition

What a metadata search looks up for one condition: the index, and its tag under each master key id the provider holds. A row matches the condition when it carries any of them.

Example

const condition: MetaCondition = { index: "mrn", tags: [{ keyId: "prod-2026", tag: "a3f0" }] };

Properties

index

readonly index: string

MetaStore

What pathways rekey needs from storage to rewrite metadata: every message's metadata and tags, and a way to replace them. Store implements it.

Example

for (const row of await store.storedMeta(0, 100)) await store.replaceMeta(row, next(row));

Methods

replaceMeta()

replaceMeta(current, next): Promise<boolean>

Replaces a message's metadata and all its tags with next, in one transaction, only while the metadata is still what current read. Returns whether it did.

Parameters

current

StoredMeta

meta

string

Returns

Promise<boolean>

storedMeta()

storedMeta(after, limit): Promise<StoredMeta[]>

Up to limit messages with metadata and an id above after, oldest first, as stored.

Parameters

after

number

limit

number

Returns

Promise<StoredMeta[]>

MetaTag

One index tag, as storage keeps it: the index (the metadata key it covers), the master key id it was derived under, and the tag, the first bits bits of an HMAC-SHA-256 in lowercase hex.

Example

const tag: MetaTag = { index: "mrn", keyId: "prod-2026", tag: "a3f0" };

Properties

index

readonly index: string

keyId

readonly keyId: string

tag

readonly tag: string

MllpSourceFacts

What an MLLP source knows about a message: the sender's address and, over mutual TLS, the common name in the subject of the certificate the sender presented.

Example

const peer = (source: MllpSourceFacts): string => source.tlsPeerSubject ?? "unknown";

Properties

kind

readonly kind: "mllp"

receivedAt

readonly receivedAt: string

When the message was stored, as an ISO 8601 time.

remoteAddress

readonly remoteAddress: string | undefined

The sender's IP address, when the connection had one.

tlsPeerSubject

readonly tlsPeerSubject: string | undefined

The common name (CN) in the subject of the sender's TLS certificate, when it presented one.

ModuleSourceFacts

What a custom source knows about a message: whatever it passed as source to sink.receive.

Example

const queue = (source: ModuleSourceFacts): JsonValue | undefined => source.facts["queue"];

Properties

facts

readonly facts: Readonly<Record<string, JsonValue>>

What the source passed, or {} when it passed nothing.

kind

readonly kind: "module"

receivedAt

readonly receivedAt: string

When the message was stored, as an ISO 8601 time.

NextDeliveryOptions

What MessageStore.nextDelivery looks for. Without dueBy it returns the oldest pending delivery, due or not, which keeps a destination in order.

Example

const due = await store.nextDelivery("adt", 0, { dueBy: Date.now() });

Properties

dueBy?

readonly optional dueBy?: number

Only a delivery due by this time, in epoch milliseconds: the oldest whose nextAttemptAt is at or before it, passing over any still waiting for a retry. The engine asks this for a destination with retry.rotate: true. A store that ignores it keeps that destination in order.

OutgoingMessage

One output of one message, on its way to one destination.

Example

const name = (message: OutgoingMessage): string => `${message.messageId}-${message.seq}.hl7`;

Properties

body

readonly body: Uint8Array

messageId

readonly messageId: number

The id of the stored message this came from.

resend?

readonly optional resend?: Resend

Set when this is a possible duplicate: an earlier attempt was cut off after the send, so the receiver may have it already. A destination may tell the receiver so (an HTTP destination's resendHeader); it sends the message either way. Undefined on every other send.

seq

readonly seq: number

Which output of that message's transform this is, from 0.

signal?

readonly optional signal?: AbortSignal

Aborted when an operator halts the channel (pathways channel halt) while this send is in flight. A destination that honours it gives up the send at once; the engine stops waiting either way, records no outcome, and sends the message again after the channel resumes, flagged as a possible duplicate when the send was marked.

Page

One page of results, newest first. next is the before that fetches the page after it.

Example

const first: Page<MessageSummary> = await store.listMessages({ limit: 50 });
const second = await store.listMessages({ limit: 50, before: first.next });

Type Parameters

T

T

Properties

items

readonly items: readonly T[]

readonly next: number | undefined

PendingDelivery

The oldest delivery still pending for one destination of one channel.

Example

const due = (delivery: PendingDelivery): boolean => delivery.nextAttemptAt <= Date.now();

Properties

attempts

readonly attempts: number

body

readonly body: Uint8Array

id

readonly id: number

interruptedAttempt?

readonly optional interruptedAttempt?: number

The latest attempt found cut off, kept after it: see StoredDelivery.interruptedAttempt.

messageId

readonly messageId: number

nextAttemptAt

readonly nextAttemptAt: number

Epoch milliseconds before which this delivery must not be tried again.

sending?

readonly optional sending?: SendingMarker

The marker of an attempt whose outcome was never recorded: set means it was cut off after it was handed to the destination, which may have it. Undefined when there is none, and from a store that keeps no marker.

seq

readonly seq: number

PendingMessage

A stored message waiting for its transform.

Example

const next: PendingMessage | undefined = await store.nextToTransform("adt");

Properties

body

readonly body: Uint8Array

id

readonly id: number

ReceivedEvent

A message was stored, before its source acknowledged it.

Example

if (event.type === "received" && event.duplicateOf !== undefined) flagDuplicate(event.messageId);

Extends

LifecycleEventBase

Properties

at

readonly at: string

When it happened, as an ISO 8601 time in UTC, such as 2026-11-07T12:00:00.000Z.

Inherited from

bytes

readonly bytes: number

The message's size in bytes.

channel

readonly channel: string

The channel's name.

Inherited from

duplicateOf

readonly duplicateOf: number | undefined

The earlier message this one may repeat (same sender and control ID), when there is one.

failed

readonly failed: boolean

True when the source stored it as failed, such as an MLLP message with no usable header.

messageId

readonly messageId: number

type

readonly type: "received"

Overrides

ReceivedMessage

An incoming message, stored.

Example

const { id, duplicateOf } = await store.receive("adt", body);

Properties

alreadyStored?

readonly optional alreadyStored?: boolean

True when a message of the channel was already stored with the same idempotency key: id is that message's, and nothing new was stored.

duplicateOf

readonly duplicateOf: number | undefined

The latest earlier message of the channel with the same duplicate key, when there is one.

id

readonly id: number

ReceiveFailedEvent

A message could not be stored, so its source did not acknowledge it and its sender should send it again. Storage being unreachable shows up here first.

Example

if (event.type === "receive-failed") page(`storage refused a message on ${event.channel}`);

Extends

LifecycleEventBase

Properties

at

readonly at: string

When it happened, as an ISO 8601 time in UTC, such as 2026-11-07T12:00:00.000Z.

Inherited from

channel

readonly channel: string

The channel's name.

Inherited from

code

readonly code: string | undefined

The error's code, when it has one, such as SQLITE_FULL.

name

readonly name: string

The error's name, such as SqliteError.

type

readonly type: "receive-failed"

Overrides

SendingMarker.attempt

ReceiveOptions

What a source can tell the engine about a message it hands over. Every field is optional.

Example

await sink.receive(body, { duplicateKey: `${sender}|${controlId}` });
await sink.receive(body, { failure: "inbound: NO_MSH_SEGMENT" });
await sink.receive(body, { idempotencyKey: `queue|${deliveryTag}` });
await sink.receive(body, { idempotencyKey: `queue|${deliveryTag}`, resumed: true });
await sink.receive(body, { source: { queue: "inbound", deliveryTag } });

Properties

duplicateKey?

readonly optional duplicateKey?: string

What identifies the message to its sender, such as the sender and its MSH-10 control ID. When an earlier message of the channel carried the same key, this one is stored and processed as usual, flagged as a possible duplicate: never dropped. The engine keeps a hash of the key, never the key itself.

failure?

readonly optional failure?: string

Stores the message as failed, with this reason, instead of handing it to the transform: for a message the source received but could not accept, such as one with no usable header. A short code, never message content.

idempotencyKey?

readonly optional idempotencyKey?: string

What names this one message to the source, the same every time the source hands it in. It is stored with the message, so that when the source hands the message in again with resumed, after a stop between storing it and acknowledging it, it is stored once. The engine keeps a hash of the key, never the key itself. A custom store may ignore it, and then stores such a message twice.

resumed?

readonly optional resumed?: boolean

True when the source hands in again, under the same idempotencyKey, a message that an earlier run may have stored already: one it took in and never acknowledged. When a message of the channel was stored with that key, nothing new is stored and that message's id comes back. Only these messages are looked up, so a message handed in for the first time costs no lookup.

source?

readonly optional source?: Readonly<Record<string, JsonValue>>

What the source knows about the message, which the channel's transform reads as ctx.source.facts: a map of plain JSON values (strings, finite numbers, booleans, null, lists and maps). Anything else is refused with the code INVALID_SOURCE_FACTS before the message is stored. The engine keeps a copy taken at the call, so the source may reuse or change its object afterwards. It is never logged, sent to a hook or stored: the engine holds it in memory until the message is transformed, so a message taken in before the engine restarts is transformed without it.

RequeueRequest

What a replay or a resend asks storage to record. Storage checks the message still stands where the caller found it, and changes nothing otherwise.

Example

const request: RequeueRequest = { mode: "resend", expected: "transformed", outputs, destinations: 2 };

Properties

destinations

readonly destinations: number

How many destinations the channel has now.

expected

readonly expected: MessageStatus

The status the caller read before transforming.

failure?

readonly optional failure?: string

What the replayed transform failed with, as stored; the message is marked failed.

mode

readonly mode: "replay" | "resend"

replay ran the transform again; resend queues the stored outputs again.

outputs

readonly outputs: readonly Uint8Array<ArrayBufferLike>[]

The outputs to queue, in order, for each of destinations. None, without a failure, means the transform filtered the message out.

Resend

Set on a message that is being sent again after an attempt was cut off: the engine stopped after handing it to the destination and before it recorded the outcome, so the receiver may have it already. The message is the same; only the flag tells it apart.

Example

const header = (resend: Resend | undefined): string | undefined =>
  resend === undefined ? undefined : String(resend.previousAttempts);

Properties

previousAttempts

readonly previousAttempts: number

How many attempts were made before this one, the one cut off among them.

ResolvedSecret

What a provider answers: the value, the version it is, and when it stops being good, if it does.

Example

const secret: ResolvedSecret = { value: "synthetic-token", version: "3", expiresAt: new Date("2026-12-01T00:00:00Z") };

Properties

expiresAt?

readonly optional expiresAt?: Date

When the value stops being good, for a lease or a short-lived credential.

identity?

readonly optional identity?: string

Who read it, when the provider can tell: the cloud principal its call ran as, such as a role's ARN, a managed identity's id or a service account's email. Never secret. pathways secrets check prints it, when it is printable ASCII with no space, 1 to 256 characters.

value

readonly value: string

The secret itself. Pathways never logs it, and never puts it in an error.

version

readonly version: string

The version resolved, as the provider names it; never secret.

RetryingDelivery

A pending delivery that has failed at least once and waits to be tried again.

Example

const late = (d: RetryingDelivery): boolean => d.nextAttemptAt > Date.now() + 60_000;

Properties

attempts

readonly attempts: number

destination

readonly destination: number

id

readonly id: number

messageId

readonly messageId: number

nextAttemptAt

readonly nextAttemptAt: number

Epoch milliseconds of the next try.

SecretProvider

Where secrets come from: an environment variable, a file, or a secret store such as a cloud provider's secrets manager. pathways.yaml names each provider under secrets: by a module whose default export builds it, and a { secret: { provider, name, field, version } } reference anywhere a secret is read asks it for one.

resolve is called once per secret when the engine starts, with a signal that aborts when the provider's time limit passes or the start is cancelled. It rejects when the secret cannot be had, preferably with a SecretError. Pathways reports a rejection by the reference's file and key, the secret's name and the error's name and code, never the error's text, so an error may say anything; a value must still never be put in one.

Example

const store: SecretProvider = {
  async resolve(ref, signal) {
    const found = await myStore.get(ref.name, { version: ref.version, signal });
    return { value: found.text, version: found.version };
  },
};

Extended by

FakeSecretProvider

Methods

resolve()

resolve(ref, signal): Promise<ResolvedSecret>

Resolves one secret.

Parameters

ref

SecretRef

signal

AbortSignal

Returns

Promise<ResolvedSecret>

stop()?

optional stop(): void | Promise<void>

Releases what the provider holds open. Called once start has resolved every secret.

Returns

void | Promise<void>

SecretRef

One secret a provider is asked for: its name in the provider, and optionally one field of it and a version. A provider that has no fields or versions refuses a request that names one.

Example

const ref: SecretRef = { name: "lab/api", field: "token" };

Properties

field?

readonly optional field?: string

One field of a structured secret, or of a value that is a JSON object.

name

readonly name: string

The secret's name in the provider: a variable, a path, a key. Never secret itself.

version?

readonly optional version?: string

A version the provider keeps. Undefined asks for the current one.

SendingMarker

The marker written on a delivery before it is handed to its destination: which attempt it is and when it started. The outcome write clears it, so a delivery found still marked was cut off between the send and the record of it, and the receiver may have it.

Example

await store.markSending(delivery.id, { attempt: delivery.attempts + 1, at: Date.now() });

Extended by

SendStart

Properties

at

readonly at: number

When it started, in epoch milliseconds.

attempt

readonly attempt: number

The attempt's number, from 1.

SendStart

What MessageStore.markSending writes: the marker, and, when the delivery was found marked, the number of the attempt that was cut off.

Example

const start: SendStart = { attempt: 2, at: Date.now(), interruptedAttempt: 1 };

Extends

SendingMarker

Properties

at

readonly at: number

When it started, in epoch milliseconds.

Inherited from

SendingMarker.at

attempt

readonly attempt: number

The attempt's number, from 1.

Inherited from

interruptedAttempt?

readonly optional interruptedAttempt?: number

The attempt found cut off, kept on the delivery so every later send of it is flagged.

SourceConnector

Brings messages in. A source that polls (a folder, an inbox) implements poll, which the engine calls on each pass its schedule makes due (the channel's poll:, or the engine-wide pollInterval). A source that listens (MLLP, HTTP) implements start and stop. A custom source's module builds one with a SourceFactory.

Example

const source: SourceConnector = {
  async poll(sink, signal) {
    for (const body of await takeWaiting({ signal })) await sink.receive(body);
  },
};

Methods

poll()?

optional poll(sink, signal?): Promise<void>

One pass: takes in what is waiting, each message through sink. The engine never starts a pass while the last one's call has not returned. At the pass's deadline (poll: { deadline }, 5 minutes by default) the engine stops waiting: signal aborts, which a source may pass to its client to give up a hung listing or query, and sink.receive refuses every later message with the code POLL_DEADLINE, so the source keeps it for the next pass.

Parameters

sink

MessageSink

signal?

AbortSignal

Returns

Promise<void>

start()?

optional start(sink): Promise<void>

Parameters

sink

MessageSink

Returns

Promise<void>

stop()?

optional stop(): Promise<void>

Returns

Promise<void>

SourceDownEvent

A source could not start listening (its port is taken, say), or a running source said it lost its connection (MessageSink.sourceDown; then name is SourceDown). A listener is tried again after retryInMs; a source that lost its connection is told to wait that long. Each is reported again on each failure in a row, counting up attempt.

Example

if (event.type === "source-down" && event.attempt === 1) page(`${event.channel} cannot listen`);

Extends

LifecycleEventBase

Properties

at

readonly at: string

When it happened, as an ISO 8601 time in UTC, such as 2026-11-07T12:00:00.000Z.

Inherited from

attempt

readonly attempt: number

How many tries have failed in a row, from 1.

channel

readonly channel: string

The channel's name.

Inherited from

code

readonly code: string | undefined

The error's code, such as EADDRINUSE, or the code the source passed, such as ECONNRESET.

name

readonly name: string

The error's name, such as Error, or SourceDown when the source said it lost its connection.

retryAt

readonly retryAt: string

When the next try is due, as an ISO 8601 time in UTC.

retryInMs

readonly retryInMs: number

type

readonly type: "source-down"

Overrides

SourceUpEvent

A source that was down is listening, or connected, again. A source that starts on the first try reports nothing.

Example

if (event.type === "source-up") resolveIncident(event.channel, "source");

Extends

LifecycleEventBase

Properties

at

readonly at: string

When it happened, as an ISO 8601 time in UTC, such as 2026-11-07T12:00:00.000Z.

Inherited from

channel

readonly channel: string

The channel's name.

Inherited from

type

readonly type: "source-up"

Overrides

MessageSummary.channel

StoredContent

One row's content, exactly as storage holds it: a message's body and error, or a delivery's body and last error. With encryption on, these are the encrypted values.

Example

const row: StoredContent = { table: "messages", id: 1, body, error: null };

Properties

body

readonly body: Uint8Array

error

readonly error: string | null

id

readonly id: number

table

readonly table: ContentTable

StoredDelivery

One delivery of a message, without its body.

Example

const waiting = (await store.deliveries(12)).filter((d: StoredDelivery) => d.status === "pending");

Properties

attempts

readonly attempts: number

deliveredAt

readonly deliveredAt: string | null

destination

readonly destination: number

error

readonly error: string | null

The last failure, as stored: a stage and a code, such as delivery: ECONNREFUSED.

generation

readonly generation: number

id

readonly id: number

interruptedAttempt?

readonly optional interruptedAttempt?: number

Set when an attempt of this delivery was cut off, its outcome never recorded, so the receiver may have it more than once: the number of the latest such attempt. Every later send of it is flagged as a possible duplicate.

nextAttemptAt

readonly nextAttemptAt: number

Epoch milliseconds before which it is not tried again; 0 when due at once.

seq

readonly seq: number

Which output of the transform this is, from 0.

status

readonly status: DeliveryStatus

StoredMessage

What storage holds about one message, without its body.

Example

const message: StoredMessage | undefined = await store.message(12);

Extends

MessageSummary

Properties

channel

readonly channel: string

Inherited from

contentDroppedAt?

readonly optional contentDroppedAt?: string

Inherited from

MessageSummary.contentDroppedAt

contentHash?

readonly optional contentHash?: string

The hash of the body as received, when one was stored with it: see StoreReceiveOptions.

Inherited from

MessageSummary.contentHash

deliveries

readonly deliveries: DeliveryTally

The deliveries of the current generation, by status.

Inherited from

MessageSummary.deliveries

duplicateOf

readonly duplicateOf: number | null

The latest earlier message with the same sender and control ID, when there is one.

Inherited from

MessageSummary.duplicateOf

error

readonly error: string | null

Why the message failed, as stored. The part after the second ": " is the error's own text, which can quote the message: never print it.

generation

readonly generation: number

0 until the message is first replayed or resent; one more each time.

Inherited from

MessageSummary.generation

id

readonly id: number

Inherited from

MessageSummary.id

receivedAt

readonly receivedAt: string

Inherited from

MessageSummary.receivedAt

size

readonly size: number | null

The body's size in bytes, or null for a message stored before sizes were kept.

Inherited from

MessageSummary.size

state

readonly state: "received" | "filtered" | "delivered" | "failed" | "cleared" | "dry-run" | "queued" | "retrying"

The status and the current deliveries, as an operator sees them.

Inherited from

MessageSummary.state

status

readonly status: MessageStatus

Inherited from

MessageSummary.status

updatedAt

readonly updatedAt: string

When the message last changed: transformed, failed, replayed or cleared.

Inherited from

MessageSummary.updatedAt

StoredMeta

One message's metadata and its index tags, as stored: what pathways rekey reads to re-encrypt the metadata and derive its tags again under the current key.

Example

const row: StoredMeta = { id: 12, channel: "adt", meta: storedJson, tags: [] };

Properties

channel

readonly channel: string

id

readonly id: number

meta

readonly meta: string

StoreReceiveOptions

What the engine tells storage about an incoming message. Every field is optional.

Example

await store.receive("adt", body, { duplicateKey: hashOfSenderAndControlId });

Properties

contentHash?

readonly optional contentHash?: string

A hash of the body as it arrived (the engine passes its SHA-256 in hex, for a channel that drops content after delivery), kept with the message's metadata after its body is dropped.

duplicateKey?

readonly optional duplicateKey?: string

An opaque key for who sent the message and its control ID (the engine passes a hash). When an earlier message of the channel has the same key, this one is stored flagged as a possible duplicate of it.

failure?

readonly optional failure?: string

Stores the message as failed, with this reason, so it is kept but never transformed.

idempotencyKey?

readonly optional idempotencyKey?: string

An opaque key that names this one message to its source (the engine passes a hash), stored with the message. With resumed, when a message of the channel was already stored with the same key, nothing is stored and that message's id comes back with alreadyStored: a source that stored a message and then stopped before it could acknowledge it hands it in again on restart, and it is stored once.

meta?

readonly optional meta?: Readonly<Record<string, string | number | boolean>>

The message's metadata. A store wrapper (encryption at rest) turns it into storedMeta and metaTags before it reaches the store underneath; a store given it bare keeps it as JSON.

metaTags?

readonly optional metaTags?: readonly MetaTag[]

The metadata's index tags, stored with the message in the same transaction. Set by a wrapper.

resumed?

readonly optional resumed?: boolean

True when the message may have been stored already under idempotencyKey, by an earlier run: only then is the key looked up.

size?

readonly optional size?: number

The body's size in bytes as it arrived; defaults to body.byteLength. A storage layer that changes the body on its way in, such as encryption, passes the original size along.

storedMeta?

readonly optional storedMeta?: string

The metadata as the store keeps it: JSON, encrypted under encryption. Set by a wrapper.

TransformConfig

The configuration a transform reads as ctx.config: the config: block of pathways.yaml, with the channel's own config: block over it key by key, and each { env: NAME } read from the engine's environment when it starts. It is empty here: name your keys and their types by declaration merging, once per project, and ctx.config is typed by them.

Example

declare module "@cosyte/pathways" {
  interface TransformConfig {
    readonly facility: string;
    readonly labCodes: readonly string[];
  }
}

Properties

facility?

readonly optional facility?: string

TransformContext

The second argument a transform receives: a frozen plain object, built for each message.

Nothing in it is logged, sent to a hook or shown on the health endpoint. A transform reads it; to change what is sent, it changes what it returns.

Example

import type { Transform } from "@cosyte/pathways";

export default ((msg, ctx) => {
  if (ctx?.source?.kind === "mllp" && ctx.source.tlsPeerSubject !== "lab.example.test") return null;
  return msg;
}) satisfies Transform;

Properties

attempt

readonly attempt: number

Which run of the transform this is for the message: 1 the first time, and one more for each replay or resend before this one.

config

readonly config: Readonly<TransformConfig>

The channel's configuration, resolved when the engine (or the command) started.

messageId

readonly messageId: number

The stored message's id. 0 under pathways test, which stores nothing.

meta

readonly meta: Readonly<Record<string, string | number | boolean>>

A per-message map of flat values. Always empty in this release: nothing sets it yet.

source

readonly source: SourceFacts | undefined

What the source knew about the message when it took it in. Undefined when that is not known: under pathways test, on a replay, and for a message taken in before the engine last started, since the facts are kept in memory until the message is transformed and not stored; and for the oldest of a channel's messages past the 4,096 whose facts the engine holds while they wait.

TransformedEvent

A message's transform returned one or more outputs, each now queued for every destination.

Example

if (event.type === "transformed") count("outputs", event.outputs);

Extends

LifecycleEventBase

Properties

at

readonly at: string

When it happened, as an ISO 8601 time in UTC, such as 2026-11-07T12:00:00.000Z.

Inherited from

channel

readonly channel: string

The channel's name.

Inherited from

messageId

readonly messageId: number

outputs

readonly outputs: number

How many outputs the transform returned.

type

readonly type: "transformed"

Overrides

TransformFailedEvent

A message's transform threw, timed out or ran out of memory, or its input or output could not be read. The message is marked failed and the channel moves on.

Example

if (event.type === "transform-failed") page(`${event.channel}: ${event.name} at ${event.where ?? "?"}`);

Extends

LifecycleEventBase

Properties

at

readonly at: string

When it happened, as an ISO 8601 time in UTC, such as 2026-11-07T12:00:00.000Z.

Inherited from

channel

readonly channel: string

The channel's name.

Inherited from

code

readonly code: string | undefined

The error's code, when it has one, such as a parser's NO_MSH_SEGMENT.

kind

readonly kind: TransformFailureKind

The stage that failed.

messageId

readonly messageId: number

name

readonly name: string

The error's name, such as TypeError.

type

readonly type: "transform-failed"

Overrides