Steps

import { env } from "@health-samurai/interbox";
import { defineLookup, defineState, defineStep, mllpSource, pipeline, postgresStore, skip, webhookSource } from "@health-samurai/interbox/chain";
import type { Batch, RowContext, StepContext } from "@health-samurai/interbox/chain";
import { fromMsh7, hl7v2Parser } from "@health-samurai/interbox/hl7v2";
import { aidbox, fhirItems, skipUnchanged } from "@health-samurai/interbox/aidbox";

A step is a function that takes a Batch of rows and returns a Batch. A pipeline runs its steps in the order they are declared, on both sides of its one .store():

• Before the store, the steps run once for each input — an MLLP frame, an HTTP request body — inside the transaction that stores it, before the sending side is answered. Every row they produce becomes a unit at the store.
• After the store, the steps run on a take: up to the store's batch units (100 by default) taken together. Every row they produce becomes a sender item of its unit, or, through a bridge, a unit of the next pipeline.

See Steps, store and senders for the shape of a pipeline, and Units and State for what a unit is and how it moves.

defineStep

function defineStep<A, B>(name: string, fn: StepFn<A, B>): NamedStep<A, B>;

type StepFn<A, B> = (rows: Batch<A>, ctx: StepContext) => Batch<B> | Promise<Batch<B>>;

interface NamedStep<A, B> {
  readonly name: string;
  readonly fn: StepFn<A, B>;
}

.step() takes a NamedStep or a plain StepFn. The name is what a failed unit records as its error_step, so the dashboard can say where it failed:

A position shifts when steps are reordered, so name any step you will want to find in an error.

const uppercase = defineStep("uppercase", (rows: Batch<string>) => rows.map((text) => text.toUpperCase()));

pipeline("lab")
  .source(webhookSource({ port: env("LAB_PORT") }))
  .store(postgresStore({ codec: "text" }))
  .step(uppercase)
  .sender(toLab);

A step returns the Batch that its own Batch's methods made. A Batch remembers which unit each row belongs to, and only its methods keep that link. An array, or a Batch from another step's run, is refused with a step must return the Batch that the methods of its own run's Batch made.

A throw from the step function itself, rather than from one row's function, has no row to blame. Before the store it refuses the input; after the store it fails every unit of the take, as a throw of the same kind would fail one unit (see Failures).

StepContext

The second argument of a step, one per input before the store and one per take after it.

RowContext

The second argument of every function you pass to a row method (map, flatMap, filter, skip, notReady and the four with… methods).

A RowContext has no state and no sender. A row function that needs them uses the step's own context:

const seen = defineState<{ at: string }>("lab.seen");

const markSeen = defineStep("mark-seen", (rows: Batch<LabResult>, ctx: StepContext) =>
  rows.map(async (row) => {
    ctx.state(seen).set(row.accession, { at: new Date().toISOString() });
    return row;
  }),
);

Batch

interface Batch<T> {
  map<U>(fn: (row: T, ctx: RowContext) => U | Promise<U>): Batch<U>;
  flatMap<U>(fn: (row: T, ctx: RowContext) => Children<U> | Promise<Children<U>>): Batch<U>;
  filter(fn: (row: T, ctx: RowContext) => boolean | Promise<boolean>): Batch<T>;
  skip(fn: (row: T, ctx: RowContext) => string | undefined | Promise<string | undefined>): Batch<T>;
  notReady(fn: (row: T, ctx: RowContext) => number | undefined | Promise<number | undefined>): Batch<T>;
  mapAll<U>(fn: (rows: T[], ctx: StepContext) => Promise<(U | Error)[]>): Batch<U>;
  withAttrs(fn: (row: T, ctx: RowContext) => Attrs | Promise<Attrs>): Batch<T>;
  withOrder(fn: (row: T, ctx: RowContext) => Date | Promise<Date>): Batch<T>;
  withSearchText(fn: (row: T, ctx: RowContext) => string | Promise<string>): Batch<T>;
  withMeta(fn: (row: T, ctx: RowContext) => ItemMeta | Promise<ItemMeta>): Batch<T>;
}

type Children<U> = Iterable<U | Skip> | AsyncIterable<U | Skip>;
type Attrs = Record<string, unknown>;
type ItemMeta = { key?: string } & Record<string, unknown>;

A method records what to do and runs nothing. The engine pulls the rows when the step has returned:

• Before the store, one row at a time. A row passes every remaining step before the next one is read, so a file split by flatMap streams through without being held whole. Nothing runs in parallel inside one input.
• After the store, method by method over the take. Each method runs over all the take's rows, up to 16 at a time, before the next method starts. Do not rely on the order of units within a take.

undefined from skip's or notReady's function keeps the row as it is. Any other answer from notReady's function must be a number of milliseconds from 0 to MAX_NOT_READY_MS, 24 hours; a negative, NaN, infinite or longer delay, or a value that is not a number, throws.

skip() elements in a flatMap

function skip(reason: string, attrs?: Attrs): Skip;

A flatMap may yield skip(reason, attrs) in place of a row. Before the store it records a skipped unit carrying the parent's attrs with attrs merged over them, so one bad line of a file skips that line and the rest of the file is stored:

const lines = defineStep("lines", (files: Batch<string>) =>
  files.flatMap(function* (file) {
    for (const [n, line] of file.split("\n").entries()) {
      if (line.trim() === "") continue;
      try {
        yield JSON.parse(line) as Record<string, unknown>;
      } catch {
        yield skip("bad json", { line: String(n + 1) });
      }
    }
  }),
);

After the store a skip() element is an item skip, counted on its unit. Item skips, and the reason of a unit skipped whole, are written into the unit's attrs under _skipped, as { "<reason>": <count> }.

Attrs, order and search text

• Attrs merge shallowly. A key you set replaces the value an earlier step set for it.
• Rows of one unit share its attrs. After a flatMap split, two items setting the same key leave the last value; set unit attrs before the split.
• Attrs are untyped. A misspelled attr name compiles, and a filter on it finds nothing. Filters compare an attr as text, so give a filtered attr a string value.
• When they are written. Before the store, with the unit. After the store, when the unit leaves the store as processed, sent or skipped. A unit that fails or waits keeps the attrs it had at the store.
• withOrder sets the time of the event the unit records. Without it a unit's order_at is the time it was created. It orders the queued versions of one resource, see Units and State.
• withSearchText sets the text the dashboard's search reads. It is searchable only in a pipeline that declares search: "trgm" on its store.

withMeta and item keys

withMeta gives an item what its sender reads. key is the item's identity at the destination, such as Patient/123: the engine keeps it in a column of its own, never sends two versions of one key at once, and sends them in order. The rest of the meta is stored beside the item. For Aidbox, fhirItems() writes it for you.

mapAll

mapAll is for one call that covers many rows, such as a bulk request to a service. It gets the rows' values and the step's StepContext, and returns one result per row, in the same order.

• An Error in a row's place fails only that row's unit (after the store) or the input (before the store).
• A throw, or a result whose length differs from the rows', fails every unit in the call after the store, and the input before it.

Failures

A failure acts on the row's unit. After the store:

A fifth failed attempt ends the unit at error, with engine/attempts_exhausted after a throw and engine/step_timeout after a deadline. The message of engine/attempts_exhausted is the last error's, after its kind when it had one, such as system/epic_unreachable: ….

Every retry, automatic or from the dashboard, starts at the store and runs every step after it again. A unit at error records the step that failed in error_step. See Errors for domainError and its kinds.

Before the store there is no unit to fail. Any throw, a domainError included, refuses the whole input: nothing of it is stored, MLLP answers AE and HTTP answers 500, and the sending side retries. A skip is not a failure: it records a skipped unit, and the input is answered as stored.

Deadlines

ctx.signal also fires when a stop's grace period ends. Pass it to anything that can take long, such as a fetch, so abandoned work stops. A late result of abandoned work is discarded.

State: defineState and ctx.state

function defineState<V>(namespace: string): StateHandle<V>;

interface State<V> {
  get(key: string): Promise<V | null>;
  get(keys: readonly string[]): Promise<Map<string, V>>;
  set(key: string, value: V): void;
  set(entries: Iterable<readonly [string, V]>): void;
  del(keys: string | readonly string[]): void;
}

State is a small durable key/value store, scoped to the pipeline: each pipeline has its own copy of every namespace, and no pipeline reads another's. Declare a handle once, at module level, and import it wherever a step needs it:

const lastSeen = defineState<{ orderAt: string }>("adt.last-seen");

const newerOnly = defineStep("newer-only", (msgs: Batch<Admit>, ctx: StepContext) =>
  msgs.skip(async (msg) => {
    const before = await ctx.state(lastSeen).get(msg.patientId);
    if (before !== null && before.orderAt >= msg.sentAt) return "older than what we have";
    ctx.state(lastSeen).set(msg.patientId, { orderAt: msg.sentAt });
    return undefined;
  }),
);

• Namespace names use lowercase Latin letters, digits, - and ., start with a letter, and are at most 46 characters. defineState throws on any other. The name is what the rows are stored under: renaming it leaves the old rows behind.
• Values are JSON. A key holding the NUL character or a lone UTF-16 surrogate, or a value JSON cannot hold, makes set or del throw.
• A many-key get is one query and answers a Map without the keys it did not find. Single-key gets made in the same tick are read together.
• A get after a set or del in the same work reads the buffered write. Other work sees only committed values.
• No time to live. A namespace that would grow forever is cleaned with del.

Writes are buffered, and commit with the work they belong to:

A set or del made after the step's work has finished, from a promise no one awaited, is dropped without throwing, and the engine logs an error naming the pipeline, the namespace and the key.

Reading units: ctx.units

interface UnitsReader {
  count(filter?: UnitsFilter): Promise<number>;
  find(query?: UnitsQuery): Promise<UnitRow[]>;
}

interface UnitsFilter {
  readonly attrs?: Readonly<Record<string, string>>;
  readonly status?: UnitStatus | readonly UnitStatus[];
}

interface UnitsQuery extends UnitsFilter {
  readonly limit?: number;               // 1 to 1000; 100 by default
  readonly order?: "newest" | "oldest";  // by unit id; "newest" by default
}

interface UnitRow {
  readonly id: string;
  readonly status: UnitStatus;
  readonly attrs: Readonly<Record<string, unknown>>;
  readonly createdAt: string;            // ISO 8601
  readonly orderAt: string;              // ISO 8601
  readonly errorStep?: string;           // present only on a unit at "error"
  readonly errorKind?: string;
  readonly errorMessage?: string;
}

type UnitStatus = "received" | "processed" | "sent" | "skipped" | "error" | "deferred";

ctx.units reads the units of the step's own pipeline, and no other:

const priorResults = defineStep("prior-results", (msgs: Batch<Oru>, ctx: StepContext) =>
  msgs.withAttrs(async (msg) => ({
    prior: String(await ctx.units.count({ attrs: { patient_id: msg.patientId }, status: "sent" })),
  })),
);

• Filters. Each attr matches by equality on its text, status matches any one given, and every condition must hold. An attr value must be a string; anything else throws.
• Indexes. A filter on an attr the store declares in index reads that index. Any other attr is a scan of the pipeline's units.
• Limits. find answers at most limit rows. A limit outside 1 to 1000, or an order other than the two, throws.
• Before the store, a call runs in the input's own transaction, so it sees the units that input has written so far.
• After the store, each call is a short read-only query of its own, under a statement timeout of what is left of the take's deadline. Once ctx.signal has fired, a call throws.
• No text search. ctx.units does not read search_text.
• A query per row adds up. A count inside map is one query per unit; for a whole take, use mapAll and query once.
• A Retry can see different units. The answer depends on what is in the database when the step runs.

Reference lookups: defineLookup and ctx.lookup

function defineLookup<Row = LookupRow>(name: string, spec: LookupSpec): LookupHandle<Row>;

interface LookupSpec {
  readonly table: string;               // read as reference.<table>
  readonly key: string;                 // a text column with a unique index of its own
  readonly columns: readonly string[];  // what each row answers; the key is not added
}

interface LookupReader<Row> {
  get(keys: readonly string[]): Promise<Map<string, Row>>;
}

type LookupRow = Readonly<Record<string, unknown>>;

class NotReady extends Error {
  constructor(message: string, afterMs?: number);
  readonly afterMs: number;
}

const MAX_NOT_READY_MS = 86_400_000;

function isNotReady(err: unknown): err is NotReady;

A lookup reads rows of a reference dataset, such as a provider registry or a code table, by key. Declare a handle once, at module level, export it from the workspace entry in LookupRegistry, and read through ctx.lookup:

const providers = defineLookup<{ name: string; taxonomy: string | null }>("providers", {
  table: "npi_registry",
  key: "npi",
  columns: ["name", "taxonomy"],
});

export const LookupRegistry = { providers };

const providerName = defineStep("provider-name", (claims: Batch<Claim>) =>
  claims.withAttrs(async (claim, ctx) => {
    const found = await ctx.lookup(providers).get([claim.npi]);
    return { provider: found.get(claim.npi)?.name ?? "unknown" };
  }),
);

• Registered handles only. The workspace entry, src/index.ts, exports LookupRegistry beside ChainRegistry: a plain object whose values are the handles its steps use, as defineLookup returned them. A Map, or a value that is not such a handle, fails the workspace load with an error naming the entry. A step may pass only a handle the registry holds, the very object, not another declared alike; any other throws an error naming the handle and LookupRegistry.
• The reference schema, and no other. table always means reference.<table>, whatever the connection's search_path. The engine's own tables and the workspace's inbound tables hold the messages pipelines carry, and a lookup must not become a way to read them, so reference data has a schema of its own. Rows are read from reference.<table> alone, as FROM ONLY reads them: never from a table that inherits from it, in whatever schema, so a partitioned table, whose rows live in its partitions, is never ready. The engine does not create the schema or its tables: create and load them yourself, readable by the role the engine connects as.
• Names. The lookup's name, table, key and every column use lowercase Latin letters, digits and _, start with a letter or _, and are at most 63 bytes, the most Postgres keeps of a name. defineLookup throws on any other. Names are always quoted, so a column called order is read as it is, and there is no *: a lookup reads the columns it lists.
• The key is a text column with a unique index on that column alone: valid, not partial, not on an expression, with no second key column, in the column's own collation and with the default operator class for text. A primary key on it is such an index; an index built COLLATE "C" or with text_pattern_ops on a column of another collation is not. get(keys) answers a Map holding only the keys found. A key given twice is answered once, a key holding the NUL character or a lone UTF-16 surrogate matches nothing, and a key that is not a string throws an error naming its index.
• Any number of keys. The engine reads them 1000 to a query and merges the answers. A call per row is a round trip per row: for a whole take, gather the keys in mapAll and call once.
• Values are JSON, as Postgres's row_to_json writes them: a date is its ISO string, a timestamptz an ISO string with the offset of the session's time zone, a char(n) is padded to n, a text comes back unchanged, and a bigint or numeric is a JavaScript number, a numeric NaN the string "NaN". A number past 2^53, or one with more than 15 significant digits, can come back rounded, and nothing reports it. Keep identifiers, such as account or member numbers, in text columns.

Where a lookup runs

• After the store, each call is a read-only transaction of its own, under a statement timeout of what is left of the take's deadline and a lock timeout of 2 seconds. Once ctx.signal has fired, a call throws.
• Before the store, a call runs in the input's own transaction, inside a savepoint with a lock timeout of 2 seconds, and the savepoint is rolled back when the call ends. A failed lookup never leaves the input's transaction unusable, and the table is not kept locked for the rest of the input. The input's own writes, and its commit, wait until no call is running, so a call a step did not await cannot roll them back.

Not ready

A table is ready when reference.<table> exists, holds at least one row, has its key as above, and has every listed column. Until then a call throws NotReady, and so does a call that waits on a lock past its 2-second lock timeout:

No table is checked when the workspace loads, so a table not loaded yet never stops the engine. The first call of a handle in each worker checks the table, and a call checks it again whenever Postgres answers from a different table under that name, as after a swap. Whether the table is empty is read on every call.

A step may catch NotReady to go on without the rows. Tell it with isNotReady(error), not instanceof: the workspace bundle carries its own copy of the class, and the error comes from the engine's. A step may also throw a NotReady of its own, with a delay in milliseconds, and the engine treats it the same way, waiting at least one second. The delay must be from 0 to MAX_NOT_READY_MS, 24 hours: the constructor throws on any other.

Loading a new version

Load it beside the table in use, then swap the two by rename:

CREATE TABLE reference.npi_registry__loading (LIKE reference.npi_registry INCLUDING ALL);
COPY reference.npi_registry__loading FROM STDIN;
ANALYZE reference.npi_registry__loading;
BEGIN;
ALTER TABLE reference.npi_registry RENAME TO npi_registry__old;
ALTER TABLE reference.npi_registry__loading RENAME TO npi_registry;
COMMIT;
DROP TABLE reference.npi_registry__old;

The load runs before the swap's transaction, so the renames hold their locks only for a moment, and no lookup sees the table half-loaded or missing. A lookup made during the swap waits for its commit and then reads the new table. One that waits past its 2-second lock timeout is not ready: its unit waits, or its input is refused.

ctx.sender.queued

ctx.sender.queued(keys: readonly string[]): Promise<Map<string, Readonly<Record<string, unknown>>>>;

For each key that has items waiting in this pipeline's sender queue, the meta of its newest queued version, newest by the items' order_at, then unit, then item. A key with nothing queued is not in the map. skipUnchanged() uses it to compare a resource with the version already waiting to go out.

ctx.webhook

ctx.webhook(eventType: `custom:${string}`, payload: unknown): void;

On a row function's context. Emits a custom event to the pipeline's webhooks that subscribe to it in .webhook({ on }). The payload must be JSON, or the call throws.

• Before the store, the event is written with the input's commit.
• After the store, it is written with the unit's result when the unit closes processed, sent or skipped, and dropped when the unit fails or waits.

See Webhooks for subscriptions and delivery.

ctx.source and provenanceId

After the store, ctx.source names the unit a row belongs to:

interface RowSource {
  readonly id: string;          // the unit's id
  readonly pipeline: string;
  readonly receivedAt: string;  // the unit's creation time, ISO 8601
}

Prefer receivedAt or ctx.unit.createdAt over new Date() for anything you put in an output. They do not move when the unit is retried, so a Retry produces the same output, and skipUnchanged() can skip what is already delivered.

To record in the destination which unit produced a set of resources, add a Provenance whose id comes from the unit:

import { provenanceId } from "@health-samurai/interbox";

function provenanceOf(ctx: RowContext, targets: readonly Resource[]): Provenance {
  const unit = ctx.unit;
  if (unit === undefined) throw new Error("provenance is built after the store");
  return {
    resourceType: "Provenance",
    id: provenanceId(unit),
    recorded: unit.createdAt.toISOString(),
    agent: [{ who: { display: "Interbox" } }],
    target: targets.map((r) => ({ reference: `${r.resourceType}/${r.id}` })),
  };
}

provenanceId(unit) is ib-<unit id>-<created time in base-36 milliseconds>, at most 32 characters. A Retry of the unit gives the same id, so its Provenance is updated rather than duplicated.

Terminology

translate(map: string, code: string): Promise<ConceptMapMatch | undefined>;
translateFor(context: ConceptMapContext, code: string): Promise<ConceptMapMatch | undefined>;
conceptMapFor(context: ConceptMapContext): string;

interface ConceptMapContext {
  type: string;
  sendingApplication: string;
  sendingFacility: string;
}

interface ConceptMapMatch {
  targetCode?: string;      // absent when the equivalence is unmatched or disjoint
  targetDisplay?: string;
  targetSystem?: string;
  equivalence: string;
  sourceDisplay?: string;
}

The lookups read the concept maps under src/terminology/concept-maps/ in the workspace. translate looks a code up in the map with that id; undefined means the map has no entry for the code, or there is no such map. translateFor looks it up in the map bound to a sender's context, and conceptMapFor names that map: the bound one, or, when none is bound, a name derived from the context, hl7v2-<application>-<facility>-<type>. A lookup that misses reads the map files again when they changed since they were last loaded.

const loincOf = defineStep("loinc-of", (codes: Batch<string>) =>
  codes.withAttrs(async (code, ctx) => ({ loinc: (await ctx.translate("lab-codes", code))?.targetCode ?? null })),
);

Sources: mllpSource and webhookSource

import { mllpSource, webhookSource } from "@health-samurai/interbox/chain";

function mllpSource(config: MllpSourceConfig): ChainSource<string>;
function webhookSource(config: WebhookSourceConfig): ChainSource<string>;

interface MllpSourceConfig {
  port: Num;
  host?: Str;
  maxBytes?: number;
  idleTimeoutSeconds?: number;
}

interface WebhookSourceConfig {
  port: Num;
  secret?: Str;
  maxBytes?: number;
}

A source listens on a port of its own and turns each input into one row, a string decoded as UTF-8: an MLLP frame, or the body of an HTTP POST. The steps before the store run on that row, and the sending side is answered once the input commits or is refused. A pipeline fed by another pipeline has no source of its own; it reads fromPipeline.

port, host and secret may be env() references. The engine checks every field when it loads the workspace, so an unset variable with no default, or a value out of range, refuses the load. It reads them again when the source starts.

mllpSource

• Each frame is one input, answered AA once it commits, and AE with the reason when it is refused (see Failures). The frames of one connection are answered in the order they arrived.
• A frame over maxBytes gets no answer: the connection is closed, and the refusal is counted as too_large (see Errors).
• On a stop, the source takes no new frame. A frame whose steps have not started when the stop begins is neither stored nor answered, and idle connections are closed. An input already in its steps has until the stop's grace period ends to commit and get its AA; after that it is abandoned, nothing of it is stored, no answer goes out, and its connection is closed. The sender resends.

webhookSource

Each POST body, on any path, is one input. The source answers:

• A client that disconnects abandons its input. The ctx.signal of the steps before the store fires, nothing of the input is stored, and it is not counted as a refusal.
• On a stop, the source takes no new request: one that still reaches it is answered 503, without running a step. An input already in its steps has until the stop's grace period ends to commit and get its 200; after that it is abandoned, nothing of it is stored, and it is answered 500. The sending side resends either.

HL7v2: hl7v2Parser() and fromMsh7

import { fromMsh7, hl7v2Parser } from "@health-samurai/interbox/hl7v2";

function hl7v2Parser(options?: { skipZSegments?: boolean }): NamedStep<string, HL7v2Message>;
function fromMsh7(msg: HL7v2Message, timeZone: string): Date;

hl7v2Parser() is the step hl7v2.parse. It parses each raw message, and sets these attrs on its unit:

A message that does not parse fails its unit with parse/missing_msh when the text has segment lines, and parse/not_hl7 when it has none. skipZSegments: true drops every Z segment. The parser does not set order_at.

fromMsh7(msg, timeZone) reads MSH-7 as an instant. A time without an offset is read in timeZone, an IANA zone such as America/New_York, because only you know the sender's clock. It throws field/missing_msh7, field/malformed_msh7 or field/unknown_time_zone, so the unit ends at error rather than retrying. Use it to order a feed's messages by their own time:

pipeline("adt")
  .source(mllpSource({ port: env("MLLP_PORT") }))
  .store(postgresStore({ codec: "text" }))
  .step(hl7v2Parser())
  .step((msgs) => msgs.withOrder((m) => fromMsh7(m, "America/New_York")))
  .step(adtToFhir)
  .step(fhirItems())
  .step(skipUnchanged())
  .sender(toAidbox);

Storing the raw message and parsing after the store, as above, lets a Retry after a parser fix parse the original again. Parsing before the store stores the parsed form instead, and a parse failure then refuses the input with AE.

Aidbox: fhirItems(), skipUnchanged() and aidbox()

import { aidbox, fhirItems, skipUnchanged } from "@health-samurai/interbox/aidbox";

The mapping from your input to FHIR resources is a step of your own, usually a flatMap after the store that turns each unit into its resources. The SDK ships no HL7v2-to-FHIR mapping step. adtToFhir above stands for yours:

const adtToFhir = defineStep("adt-to-fhir", (msgs: Batch<HL7v2Message>) =>
  msgs.flatMap((msg, ctx) => {
    const patient = patientOf(msg);
    const encounter = encounterOf(msg, patient);
    return [patient, encounter, provenanceOf(ctx, [patient, encounter])];
  }),
);

fhirItems()

function fhirItems(): NamedStep<Resource, FhirItem>;

The step aidbox.fhir-items. It gives each resource the item meta the Aidbox sender reads: key (Type/id), refs (the Type/id references the resource holds) and hash (a hash of its content, ignoring meta and key order). A resource with no string resourceType or id, or whose key holds the NUL character or a lone surrogate, fails its unit with mapper/exception.

Put it after every step that changes a resource, because the hash it stores is what skipUnchanged() compares. aidbox() takes only FhirItems, so a pipeline that leaves it out does not compile.

skipUnchanged()

function skipUnchanged(): NamedStep<FhirItem, FhirItem>;

The step aidbox.skip-unchanged. It skips, with the reason unchanged, a resource identical to the newest version of its key already queued in this pipeline, or, when none is queued, to the version this pipeline last delivered. A unit whose resources are all unchanged closes sent without sending anything, so a resend of the same message, or a Retry, costs nothing at Aidbox.

It cannot see a version queued by another unit of the same take: a resource that goes A → B → A within one take can end at B.

aidbox()

function aidbox(cfg: AidboxConfig): Sender<FhirItem>;

interface AidboxConfig {
  readonly id?: string;
  readonly url: Str;
  readonly auth: { kind: "basic"; user: Str; password: Str } | { kind: "bearer"; token: Str };
  readonly batch?: number;
  readonly maxRetries?: number;    // default 10
  readonly retryDelayMs?: number;  // default 30 000
  readonly runTimeoutMs?: number;  // default 300 000
  readonly validateFirst?: boolean;
}

const toAidbox = aidbox({
  url: env("AIDBOX_URL"),
  auth: { kind: "basic", user: env("AIDBOX_CLIENT_ID"), password: env("AIDBOX_CLIENT_SECRET") },
});

A sender that writes fhirItems() output to <url>/fhir in batch bundles, each seed with the resources it references that are still queued.

How it answers for each resource:

A resource whose content matches what this pipeline last delivered is answered delivered without being sent. A resource this pipeline has not delivered before is a PUT; one that changed since it was delivered is a PATCH. To remove an element from a resource delivered before, map it as null: a PATCH removes it, and the first send, a PUT, drops nulls. A PATCH that Aidbox answers 404 or 410, because the resource was deleted there, is sent again at once as a PUT in a request of its own. A wait has no limit: a unit whose resource waits for a reference that never comes stays processed, and is closed from the dashboard (see Operating Pipelines).

These environment variables tune every aidbox() sender in the engine:

contentHash and extractRefs, the two functions behind fhirItems(), are exported from /aidbox as well.

fingerprints

const fingerprints: StateHandle<{ hash: string }>;

The state handle of the namespace aidbox.fingerprints: for each resource aidbox() delivered, keyed Type/id, the content hash of the version Aidbox holds. Like all state it is per pipeline. aidbox() writes it after each delivery and reads it to choose between PUT, PATCH and skipping; skipUnchanged() reads it; and the admin library's deliveredKeys, deliveredSummary and forgetDelivered list, count and clear it (see Admin Scripts). A step may read it with ctx.state(fingerprints). Leave the writes to the sender: a wrong hash turns the next send of that resource into a PUT that replaces it, or skips a change. To force a resource out again, use forgetDelivered. The 1.x import fills it from 1.x's resource hashes (see Upgrading from 1.x).