/files

fileDropSource reads files that a sending system drops into SFTP, Amazon S3, Google Cloud Storage or Azure Blob Storage, and the steps in @health-samurai/interbox/files turn them into records. The concepts are in Reading Files from Storage.

fileDropSource

Reads files that a sending system drops into an SFTP directory or a cloud storage bucket, and hands each file to the pipeline's first step. Reading Files from Storage explains feeds, deliveries and readiness; this section lists the settings. The older 1.x sftpSource (see /builtins) reads one SFTP directory without deliveries.

import { env } from "@health-samurai/interbox";
import { fileDropSource, s3 } from "@health-samurai/interbox/files";

fileDropSource({
  transport: s3({ bucket: env("BUCKET"), prefix: "clinical" }),
});

The source, its transports and the steps below come from @health-samurai/interbox/files.

transport takes one of sftp(…), s3(…), gcs(…) or azure(…); their fields are on those pages.

With read: "file", Interbox reads files of up to 64 MiB as stored and skips a larger one (reason over maxFileBytes, listed at <version>; every skipped file's reason ends with the version it was listed at, or with unlisted when the latest listing no longer had it).

With .store(…) right after the source, Interbox stores each file whole, up to 16 MiB, and skips a larger one (reason over maxBytes). It stores the file as text with codec: "text", skipping a file that is not UTF-8 (reason not UTF-8), or as bytes with codec: "bytes". It refuses codec: "json" there. Both limits are fixed, and fileDropSource has no setting for them; the wholeFile() step has its own maxBytes. A file in another encoding goes through a step that reads it, such as csvRows() with encoding, or is stored as bytes.

group

group gets each file's path in the feed and returns the name of the delivery the file belongs to, or null to leave the file alone. A flat folder of exports told apart by a prefix, such as 12345_Patient.ndjson and 12345_Observation.ndjson, with the marker file written as 12345/_READY:

group: (path) => /^(\d+)[_/]/.exec(path)?.[1] ?? null

The marker file must map to its delivery too: here [_/] lets 12345/_READY join 12345. A marker file that lists its files can name them as 12345_Patient.ndjson: in a group delivery, a name that is not beside the marker file is looked for from the top of the feed.

A file at the top level of the feed joins whichever delivery group names for it, and that delivery is ready by the same rules as a folder. A file that group maps to its own path is a delivery of its own and needs no marker file.

Keep group stable: when it changes, the deliveries already in the feed get new names, and Interbox reads them again. A file for which group throws, or returns an empty name or a name with /, is reported and not read.

order

order gets a delivery name and returns the key Interbox sorts deliveries by. For Census10142026.csv, dated month, day, year:

order: (name) => name.replace(/^\D*(\d{2})(\d{2})(\d{4}).*$/, "$3$1$2")

A delivery for which order throws or returns an empty key is not read: Interbox rejects it with the reason order failed: <the error's message> or order failed: empty key.

latestOnly

With latestOnly: true, Interbox groups deliveries into series and reads only the newest of each. A delivery's series is its name with every digit removed, so ED_202610140600.csv and ED_202610141800.csv are one series, ED_.csv, and every name that is all digits, such as 20261014, is in one series too. Pass a function to name the series yourself:

latestOnly: (name) => name.split("_")[0]

A delivery for which that function throws or returns an empty name is rejected with the reason series failed: <the error's message> or series failed: empty series. While one delivery of a series is being read, no other delivery of it is read or rejected superseded by <name> until that one ends. A delivery of the series that is ready meanwhile is reported once, reason held behind <name>, which is still reading, and is read or rejected once that one ends.

latestOnly and ordered cannot be set together.

startAfter

startAfter names a delivery. Interbox never reads that delivery or any delivery that sorts before it, in the same order ordered uses: by name, or by the key order returns.

startAfter: "20261014T060000Z"

• The first time Interbox lists such a delivery, it marks it rejected for good and opens none of its files. Removing or changing startAfter later does not make it readable.
• The deliveries listed in the feed when a startAfter value first applies are closed without a report. One that appears later is reported once, reason named before startAfter <startAfter>.
• A delivery that was waiting and is no longer listed is not closed by startAfter. It is abandoned after waitFor.abandon, as without startAfter.
• With group, the value is a name group returns.
• A delivery Interbox has already started reading, or has finished with, is left as it is.
• A closed delivery has not been read: with ordered it does not hold back later deliveries, and with latestOnly it does not make a newer snapshot older.
• Interbox refuses to start when startAfter is empty, contains /, or is a name the feed's order function throws on or returns an empty key for.

Moving a feed from another reader says how to choose the value.

read: "delivery"

With read: "delivery", Interbox copies every data file of a delivery, as stored, into Postgres and commits the delivery as one unit. The unit lists the files. A step after the store opens them and reads them as it needs: a header file and a lines file that together make one record, one provider's rows spread over several files, or a snapshot to compare with the one before it.

pipeline("providers")
  .source(fileDropSource({
    transport: s3({ bucket: env("BUCKET"), prefix: "providers" }),
    read: "delivery",
    latestOnly: true,
    pollTerm: "45m",
  }))
  .store(postgresStore({ codec: "json" }))
  .step(buildProviders)
  .sender(intoPipeline("practitioners"));

Here buildProviders reads the delivery's files (see Reading stored files) and gives one row per provider, and intoPipeline("practitioners") stores each row as a unit of a second pipeline, whose source is fromPipeline("providers").

• Everything before reading is the same. Folders, group, the marker file, expect, order, ordered, latestOnly and startAfter decide which deliveries are read and when, as with read: "file".
• No step before the store. Interbox copies the files as they arrive, .gz and .zip files included, so the store comes right after the source and uses codec: "json". Interbox refuses to start a pipeline with a step before the store, a store with another codec, or deliveryEnd(), whoever wrote the step. The check when the workspace loads sees step names only and leaves these rules to the start.
• No size limit. read: "delivery" limits neither the bytes of a delivery nor those of a file. The files take space in Postgres until the retention deletes them, see Retention.
• Stored whole or skipped whole. Interbox streams the files into Postgres, then commits the unit in one short transaction. When it cannot read one file, it skips the whole delivery with a reason, such as file members.csv: version_changed, and keeps no file of it. A delivery is stored at most once: when a read stops part-way, the next poll reads it again from its first file.
• Time to copy. A delivery has to be copied within one poll, whose length is pollTerm (default 15m, from 6m to 24h). Set it above the time your largest delivery takes to copy. A delivery whose copy runs out of time three polls in a row is skipped, reason not read within <seconds>s. pollTerm is also how long the feed waits after an engine stops during a poll before another engine takes it over.
• Changing read on a running feed. A delivery that Interbox began reading file by file and had not finished when you set read: "delivery" is skipped, reason read partly file by file; the records it already stored stay. A delivery stored whole and then read under read: "file" is not read again.
• No reports. The source gives no input for a report file or a skipped delivery. A skipped delivery shows in the feed's state and in the log, with its reason.
• Time to process. A step after the store runs within the unit's 60 second deadline, which you cannot change yet. Keep a step that builds many rows quick.
• Rows stream out. Interbox writes the rows a step gives as it gives them, so a delivery of any number of records takes the same memory. mapAll after the store gets them in parts of up to 1000 rows, so work that needs the whole delivery at once belongs in the step that reads the files: hold the small table in a map and stream the large file against it, as in Reading stored files, keep the keys already given in a set to drop duplicates, and compare with the last delivery per row, through stored state or fingerprints.

The value of the delivery's unit is a StoredDelivery: { kind: "delivery", delivery, files }, where each file has id, name, version, size (as stored), sha256 and, when the marker file gives one, count. The files are in name order.

Retention

Interbox deletes a delivery's stored bytes after the pipeline's retention (pipeline(name, { retention }), default "7d", "forever" keeps them) has passed since its unit became sent or skipped. The unit and its file list stay, and opening a deleted file fails with an error that names the deletion. A unit in error or deferred keeps its files, so Retry works. A copy cut short leaves no stored file behind; after a crash, Interbox deletes the incomplete files about 25 hours after the copy began (the longest pollTerm plus an hour).

Reading stored files

A step after the store opens a file of the delivery with ctx.files.open(file), which gives a ReadableStream<Uint8Array> that Interbox reads from Postgres slice by slice. The built-in steps unpack(), csvRows(), ndjson(), lines() and hl7v2Messages() also take a file as their first argument and then read that file, with the rules and line numbers of the step:

Each takes the file as { name, body }, where body is the stream, and reads it whatever its name: the name filter of the step does not apply. options are those of the step; for csvRows() it can be a function of the file. A call with a file gives the records, not a step, so .step(csvRows(file)) is a type error. A file that cannot be read as a whole, such as a corrupt archive, throws an error that names it, with the cause. Read a body to its end before you open the next.

This step reads a small table into a map, then streams the larger file against it and checks that every attachment a claim names is in the delivery:

import { defineStep } from "@health-samurai/interbox/chain";
import type { Batch, StepContext } from "@health-samurai/interbox/chain";
import { csvRows, type StoredDelivery } from "@health-samurai/interbox/files";

const buildClaims = defineStep("buildClaims", (rows: Batch<StoredDelivery>, ctx: StepContext) =>
  rows.flatMap(async function* (delivery) {
    const open = (name: string) => {
      const file = delivery.files.find((f) => f.name === name)!;
      return { name: file.name, body: ctx.files.open(file) };
    };

    const providers = new Map<string, string>();
    for await (const row of csvRows(open("providers.csv"))) {
      if (row.kind === "record") providers.set(row.record.id!, row.record.name!);
    }

    const present = new Set(delivery.files.map((f) => f.name));
    for await (const row of csvRows(open("claims.csv.gz"))) {
      if (row.kind !== "record") continue;
      const { id, provider, attachment } = row.record;
      if (!providers.has(provider!)) throw new Error(`claims.csv line ${row.line}: unknown provider ${provider}`);
      if (!present.has(attachment!)) throw new Error(`claims.csv line ${row.line}: ${attachment} is not in the delivery`);
      yield { id, provider: providers.get(provider!) };
    }
  }));

Steps that split a file

The first steps of a file feed's pipeline turn each file into records. A step skips a record it cannot read and goes on. It skips the whole file when the file is not its kind, or when the file as a whole cannot be read, as the last column lists.

The text in parentheses is the skip's reason, as the pipeline's units show it; N is a size in bytes or a field count.

In a file or zip entry that was not gzip-compressed, ndjson(), lines() and csvRows() also skip a last line or row with no line break, reason unterminated: the file may have been cut short. csvRows() skips a last row that ends inside an open quote, unterminated, in a gzip file too. A gzip file carries its own check: when a gzip file fails it, or a zip entry cannot be decompressed, the step reading the file skips it, reason corrupt archive: and what is wrong with it, and keeps the records it read before. NUL bytes padding a gzip file after its compressed data are ignored; any other bytes there are corrupt archive. The deliveryEnd shape skips keep a line or row of the file from passing for the record that ends a delivery.

ndjson() gives each record as an NdjsonRecord, a Record<string, unknown>, and csvRows() as a CsvRecord. A mapping that expects a FHIR type casts the record:

import type { Bundle, Resource } from "@health-samurai/interbox/fhir/4.0.1";
import type { NdjsonRecord } from "@health-samurai/interbox/files";

function bundleToResources(row: NdjsonRecord): Resource[] {
  const bundle = row as unknown as Bundle;
  return (bundle.entry ?? []).flatMap((entry) => (entry.resource === undefined ? [] : [entry.resource]));
}

Each record carries the attrs kind ("file"), delivery (the delivery's name), file (its name), version (the file's version as stored) and line. A record from a zip archive also carries entry, the entry's path in the archive. A skip a step makes while reading carries the same attrs and stage: "read". line is the line number for ndjson() and lines(), the row's first line for csvRows() (the header is line 1), and the message's position in the batch for hl7v2Messages(). A step that takes a size limit accepts maxRecordBytes (or maxBytes for wholeFile()) to change it. hl7v2Messages() also takes encoding, UTF-8 by default; with "latin1" it reads Windows-1252.

csvRows()

csvRows() splits delimited text into rows. It reads UTF-8 with a header row by default, and takes:

• delimiter: "," by default; ";" for a European Excel export, "|" for a pipe-delimited extract, "\t" for a tab-separated one;
• header: false for a file without a header row, with columns naming the columns, else they are named c1, c2 and so on, one per field of the first row;
• trim: true to strip the spaces that pad fixed-looking columns, including blanks before an opening quote and after a closing one, which without trim are a bare quote;
• quote: false for a file that never quotes fields, so a stray " in the data is an ordinary character;
• encoding: for a file not in UTF-8, such as "windows-1252". A file that starts with a byte order mark is read in the encoding the mark names.

When one feed mixes formats, pass a function that picks the options per file:

csvRows((file) => (file.name.endsWith(".txt") ? { delimiter: "|", trim: true, quote: false } : {}))

The function gets the file's name, delivery, version and size. For a zip entry, name and size are the entry's own and version is the archive's. size is left out where the archive does not declare it: for a gzip file, and for a zip entry written with a data descriptor.

deliveryEnd()

A step that records how each delivery of a file feed was read. Put it right after .store(…). Once Interbox has read or skipped the last file of a delivery, deliveryEnd() gives it one verdict: complete, partial, empty or abandoned (see What happens to a delivery). The verdict shows up as one more unit of the pipeline, with the attrs kind: "delivery-end" and delivery, skipped with the verdict as its reason, so it appears wherever the pipeline's units do. The first verdict that applies is given:

A reading step is unpack(), ndjson(), lines(), csvRows(), hl7v2Messages() or wholeFile(). What it skips is lost to the reading: an unreadable record, a file it cannot open. A step of your own that skips or filters out records it does not want, before the store or after it, does not make the delivery partial, and neither does one that skips or filters out a whole file between unpack() and the reading step: such a file counts no record and is not compared with its count. Two pipelines that read one prefix and each skip the other's records or files both judge a whole delivery complete. A delivery whose records were all read and then skipped by your own steps is complete, not empty; one whose every file your own steps dropped before reading is empty.

Interbox counts what a reading step reads on the file input itself. A step of your own between unpack() and the reading step that changes the input must spread it, { ...file, name: … }, to keep that count.

A file's records are what the reading step read from it, skipped ones included, since a marker's count includes every record written; a step of your own that skips, drops or splits records after it does not change them. A partial verdict caused by a count goes on to name each such file, up to 100, with both counts: partial: clinical_00002.ndjson.gz: count 12417, records 12400. deliveryEnd() passes every other row on unchanged.

A file whose records no reading step counted and that stored at least one unit, such as one whose body a step of your own reads, or an input rebuilt without a spread, is judged as if every skip were lost to the reading, with its stored and skipped units as its records. One that stored no unit is treated as a file your own steps dropped before reading.

An archive unpack() opens and finds empty is a file read with no record: a count the marker gives it is compared, so an empty zip named with count: 5 makes the delivery partial.

deliveryEnd() does not scan the store. When Interbox commits a file, it records in the same transaction how many records the file stored and skipped, how many of the skips the reading made, and how many records the reading step read, or that the whole file was skipped. deliveryEnd() judges the delivery from those counts alone, so its cost grows with the delivery's files, not with the pipeline's units, and the store needs no index for it.

A pipeline with deliveryEnd() must also:

• start with unpack();
• store records with codec: "json";
• have deliveryEnd() as the first step after .store(…).

Interbox refuses to start a pipeline that breaks one of these. It recognises unpack() by a tag on its function, and deliveryEnd() by its name, so do not give a step of your own the name deliveryEnd. A step of your own before the store must also pass the delivery's end record through unchanged, which Interbox cannot check: wrap it in records(fn), which does.

Without deliveryEnd(), Interbox reads deliveries the same way and gives no verdict.

Put mapping steps after deliveryEnd(). records(fn) after deliveryEnd() gives only what fn gives, so fhirItems() can follow it. ndjson() and csvRows() keep the delivery's end record in their output type, and records(fn) keeps it in its own when its input has it, so a pipeline that maps ndjson() or csvRows() records to FHIR after the store needs deliveryEnd() before the mapping:

.step(unpack())
.step(csvRows())
.store(postgresStore({ codec: "json" }))
.step(deliveryEnd())
.step(records(rowToResources))
.step(fhirItems())

lines(), hl7v2Messages() and wholeFile() give no end record, so a pipeline built on them maps after the store with or without deliveryEnd().