Creating adapters

When the built-in adapters do not cover a use case, you can write your own. Adapters are plain TypeScript classes (or objects) that implement one or more of a small set of role interfaces. The role model is directional:

• Source streams data IN and starts the pipeline (.from()).
• Destination pushes the exchange OUT, per exchange (.to() / .tap()). send is strictly void; the body flows through unchanged.
• Enricher pulls a value IN, per exchange (.enrich(), also accepted by .to() and .tap()). fetch produces a value.
• Processor / Transformer reshape the exchange in the middle.

The operation keyword selects the role: .from() resolves subscribe, .to() and .tap() prefer send and fall back to fetch, .enrich() resolves fetch.

Source

A source produces data and starts the pipeline. Implement the Source interface:

import { type Source, type Subscription } from '@routecraft/routecraft'

class MyQueueSourceAdapter implements Source<Message> {
  readonly adapterId = 'acme.adapter.my-queue'

  async subscribe(sub: Subscription<Message>) {
    sub.ready()
    while (!sub.signal.aborted) {
      const message = await queue.receive()
      await sub.emit({ message })
    }
  }
}

Destination

A destination pushes the exchange out to an external system (a queue publish, a database insert, an SMTP send). Implement the Destination interface:

import { type Destination } from '@routecraft/routecraft'

class MyStorageDestinationAdapter implements Destination<Record<string, unknown>> {
  readonly adapterId = 'acme.adapter.my-storage'

  async send(exchange) {
    await storage.write(exchange.body)
  }
}

send is strictly void: the body always flows through a .to() step unchanged. A send that produces a receipt (a message id, an etag, a created-resource URL) surfaces it through the second argument, a SendContext with a setHeader sink; the .to() step merges the collected headers onto the continuing exchange. .tap() provides the same sink but discards the headers along with its snapshot.

import { type Destination, type SendContext } from '@routecraft/routecraft'

class MyStorageDestinationAdapter implements Destination<Record<string, unknown>> {
  readonly adapterId = 'acme.adapter.my-storage'

  async send(exchange, ctx?: SendContext) {
    const receipt = await storage.write(exchange.body)
    ctx?.setHeader('acme.storage.id', receipt.id)
  }
}

The ctx parameter is optional at the call site (adapters invoked directly in tests may omit it), so always guard with ctx?.setHeader(...). ctx.signal aborts when an enclosing .timeout() expires; forward it into cancellation-aware IO.

Enricher

An adapter whose purpose is to produce data implements Enricher instead (or additionally). fetch receives the exchange and returns a value:

import { type Enricher } from '@routecraft/routecraft'

class MyLookupEnricherAdapter implements Enricher<InputType, ExtraFields> {
  readonly adapterId = 'acme.adapter.my-lookup'

  async fetch(exchange) {
    return fetchExtra(exchange.body.id)
  }
}

// Bare .enrich(): the fetched value REPLACES the body
.enrich(myLookup({ apiKey: process.env.ENRICH_KEY }))

// Merge instead: pass an aggregator such as only()
.enrich(myLookup({ apiKey: process.env.ENRICH_KEY }), only((extra) => extra, 'extra'))

// .to() accepts a fetch-only enricher too: the result replaces the body
.to(myLookup({ apiKey: process.env.ENRICH_KEY }))

Do not return data from send to simulate an enricher: .to() ignores anything a send returns. If an adapter both pushes out and can report data back, give it both slots; .to() prefers send, .enrich() uses fetch.

Processor

A processor sits in the middle of a pipeline and modifies the exchange. Implement the Processor interface. Use this when you need header or context access alongside body reshaping -- for body-only changes, .transform() is the simpler choice:

import { type Processor } from '@routecraft/routecraft'

class MyTransformAdapter implements Processor<InputType, OutputType> {
  readonly adapterId = 'acme.adapter.my-transform'

  async process(exchange) {
    const tenantId = exchange.headers['x-tenant']
    return { ...exchange, body: { ...exchange.body, tenantId } }
  }
}

Factory function

Expose your adapter as a factory function so it reads naturally in the DSL. The recommended pattern is one factory per adapter -- one name, one import:

// adapters/my-storage.ts
export function myStorage(options?: MyStorageOptions) {
  return new MyStorageDestinationAdapter(options)
}

// Usage -- destination
.to(myStorage({ bucket: 'uploads' }))

// adapters/my-queue.ts
export function myQueue(options?: MyQueueOptions) {
  return new MyQueueSourceAdapter(options)
}

// Usage -- source
.from(myQueue({ queue: 'orders' }))

Keeping one factory per adapter makes imports predictable and avoids a proliferation of role-suffixed exports (myQueueSource, myQueueDestination, etc.). The adapter carries the role slots -- the factory just wires up the options, and the position in the route selects the role.

An adapter can carry multiple role slots on one honest combined type when it makes sense. The built-in file() is the canonical example: Source<string> & Destination<unknown> & Enricher<unknown, string>, where .from() reads, .to() writes, and .enrich() reads mid-route. A queue adapter may combine source and destination the same way:

class MyQueueAdapter implements Source<Message>, Destination<Message> {
  readonly adapterId = 'acme.adapter.my-queue'

  async subscribe(sub: Subscription<Message>) {
    sub.ready()
    while (!sub.signal.aborted) {
      const message = await queue.receive(this.options.queue)
      await sub.emit({ message })
    }
  }

  async send(exchange) {
    await queue.send(this.options.queue, exchange.body)
  }
}

export function myQueue(options: MyQueueOptions) {
  return new MyQueueAdapter(options)
}

// Same factory, different positions
.from(myQueue({ queue: 'orders' }))
.to(myQueue({ queue: 'results' }))

Option laws

When a factory returns different shapes for different call forms, a few rules keep the surface predictable:

• The operation keyword selects the role. Never make an option choose between roles that both exist on the returned adapter: .from() subscribes, .to() sends, .enrich() fetches. There is no mode: 'read' | 'write' option and no per-role type alias (MyThingReadAdapter); the combined type plus the keyword is the whole story.
• Overload by key presence, never by an option's value. http() splits on path (server) vs url (client); json() becomes a file adapter because path is present; mail() fetches because folder is present. Discriminate structurally -- arguments.length, typeof, 'key' in options -- so the returned type is knowable at compile time.
• Behavior variants are boolean flags, not an enum. A send that can also append or delete takes append?: boolean / delete?: boolean, defaulting to the primary behavior (overwrite). Validate mutually exclusive flags at construction and throw RC5003 with a suggestion, so misconfiguration fails at the call site rather than mid-route.
• Shape-changing flags demand a literal. When a flag changes the emitted type (like chunked: true switching a source from T[] to per-item T), type the overload against the literal true so a widened boolean is a compile error and dynamic switching is an explicit branch at the call site.
• A slot that cannot work in some configuration still fails loudly. The source role of a file-family adapter needs a static string path; a dynamic (function) path keeps the honest combined type but its subscribe throws a clear error lazily instead of surfacing an undefined-property TypeError.

Class names carry the role, {Concept}{Role}Adapter (MyQueueSourceAdapter, MyQueueDestinationAdapter, MyLookupEnricherAdapter), even for single-role adapters, so growing a role later stays additive.

File structure

A non-trivial adapter is a folder named for its concept, with one file per role it plays:

adapters/
  my-queue/
    index.ts          # public factory + exports -- the only file consumers import
    types.ts          # exported option and result types
    source.ts         # MyQueueSourceAdapter -- present because it can be a .from() source
    destination.ts    # MyQueueDestinationAdapter -- present because it can be a .to() destination
    enricher.ts       # MyQueueEnricherAdapter -- present because it can be an .enrich() pull-in
    shared.ts         # option parsing / helpers shared between the role files

The files present are the documentation: a folder with source.ts, destination.ts, and enricher.ts is visibly a three-role adapter, while one with only source.ts is source-only. Adding a role later means adding a file, not reshaping the existing ones.

A trivial single-role adapter with no shared helpers can stay a single file (adapters/my-queue.ts), the same shorthand the examples above use. Reach for the folder once the adapter grows a second role, shared helpers, or a types module.

Options naming

When an adapter plays two sides with different options for each, name the option types so the side is readable from the type alone. Interfaces use Source/Destination/Enricher; option types use Server/Client:

Both role types carry the base. A role that adds nothing of its own can alias it (type MyQueueClientOptions = MyQueueBaseOptions). If the roles share no fields at all, declare each independently and drop the base. A single-role adapter needs only MyQueueOptions, plus an optional MyQueueResult.

Making your adapter mockable

Tag every adapter instance your factory returns so consumers can mock it with mockAdapter(yourFactory, ...) instead of having to import the internal adapter class. Tagging is a one-line addition per return path:

import { tagAdapter, factoryArgs } from '@routecraft/routecraft'

export function myQueue(options: MyQueueOptions) {
  return tagAdapter(new MyQueueAdapter(options), myQueue, factoryArgs(options))
}

tagAdapter stamps the instance with two non-enumerable symbol properties: a reference to the factory function (so mockAdapter(myQueue, ...) can match instances back to their factory) and the args the user passed at the call site (so mock handlers can receive them via meta.args and discriminate same-factory call sites).

factoryArgs(...) builds the args tuple and trims trailing undefined so call.args.length reflects what the user actually typed. Use it rather than hand-building an array so your adapter behaves consistently with the framework's built-in adapters.

For a multi-role factory, tag at every return path:

export function myQueue(options: MyQueueOptions) {
  if ('consumerGroup' in options) {
    return tagAdapter(new MyQueueSourceAdapter(options), myQueue, factoryArgs(options))
  }
  return tagAdapter(new MyQueueDestinationAdapter(options), myQueue, factoryArgs(options))
}

Consumers can then write a single mock that covers both roles:

const queueMock = mockAdapter(myQueue, {
  source: [{ id: 1 }, { id: 2 }],
  send: async (exchange, { args }) => {
    // args[0] is whatever the user passed to myQueue() at this call site,
    // so you can assert on it or branch behaviour per call site.
  },
})

Tagging is optional. Consumers of an untagged adapter can still mock it by class: mockAdapter(MyQueueAdapter, ...). But tagging is the better DX, especially for factories that fan out into multiple concrete classes based on their arguments, so it's the recommended default for every published adapter.

See the testing guide for the consumer-side API.

Supporting merged options

If your adapter has options that users might want to set once for the entire context (connection strings, timeouts, credentials), implement MergedOptions<T>. This lets users register defaults via a plugin while still allowing per-adapter overrides.

import { type MergedOptions, type CraftContext } from '@routecraft/routecraft'

const MY_OPTIONS = Symbol.for('acme.adapter.my-adapter.options')

declare module '@routecraft/routecraft' {
  interface StoreRegistry {
    [MY_OPTIONS]: Partial<MyOptions>
  }
}

class MyAdapter implements Destination<unknown>, MergedOptions<MyOptions> {
  readonly adapterId = 'acme.adapter.my-adapter'
  public options: Partial<MyOptions>

  constructor(options?: Partial<MyOptions>) {
    this.options = options ?? {}
  }

  mergedOptions(context: CraftContext): MyOptions {
    const store = context.getStore(MY_OPTIONS) as Partial<MyOptions> | undefined
    return { ...store, ...this.options }
  }

  async send(exchange) {
    const opts = this.mergedOptions(exchange.context)
    // ...
  }
}

Then ship a companion plugin so users have a typed, discoverable API:

export function myAdapterPlugin(defaults: Partial<MyOptions>): CraftPlugin {
  return {
    apply(ctx) { ctx.setStore(MY_OPTIONS, defaults) },
  }
}

See the Merged Options guide for the full walkthrough and design rationale.

Sharing state between adapters

Adapters can use the context store to share state, read global configuration set by plugins, or maintain connections across exchanges. See Plugins for how to populate the context store at startup.

Related