Helios Integration Engine: Architectural Discussion

[angela-helios]

Introduction

A FHIR server is only as useful as the data inside it. The persistence layer discussion and the bulk export discussion cover how HFS stores data and how data leaves it in volume. This document is about the step before both: how data from the outside world, in whatever shape it arrives, becomes FHIR resources in HFS, reliably, with a record of what happened to every row, and with someone told when it stops arriving.

That is the job of the Helios Integration Engine (HIE). It is a separate service that sits between the systems that produce data (a laboratory information system dropping files in a bucket, a database, another FHIR server) and HFS. It is also what the client applications built on HFS call into to see where their data came from and whether it is fresh.

Like the discussions linked above, this is an architectural strategy document rather than a specification. It explains the building blocks, the technology choices we propose for each, and the Rust trait designs that will shape the engine. It is informed by two things: a hands-on comparison of two integration engines (Apache Camel and the Open Integration Engine, the open-source fork of Mirth Connect), and a working prototype in this repository that already takes a CSV from a folder, or an NDJSON export and Synthea Bundles from a folder or a bucket, to resources in HFS through the full error, retry and alerting path.

Who should read this? Anyone who has had to get a hospital's data into a FHIR server and keep it flowing: integration analysts, data engineers, and the people who get paged when the Tuesday file has seven columns instead of six. Feedback is very welcome.

A note on scope. This document covers getting data into HFS. Outbound integration (sending FHIR data to other systems) is a separate problem. Visual channel authoring, no-code data preparation and a catalog of hundreds of SaaS connectors, which commercial integration platforms sell, are explicitly not goals.

---

The Lay of the Land: What an Integration Engine Is

Every integration product, from a 2007 open-source engine to a cloud provider's managed service, decomposes into the same handful of blocks. We built the same lab-results flow in two of them and broke it on purpose; the blocks, and what each product did with them, were:

Block	What it does	What we saw
Source (connector, listener, reader)	Notices that new data exists and reads it	Both engines poll a folder; both choked on a UTF-16 file, and one retried it forever
Queue	Holds work durably between steps, so an outage or a crash loses nothing	Every mature engine has a persistent queue. The one without it sent a good file to an error folder when the destination was down, and nobody re-sent it
Transform	Turns a source record into FHIR	Text templating let a bad value reach the server; a scripted transform that builds objects rejected it with the exact reason before sending
Delivery (sender, destination)	Writes to the target, here HFS	Re-sending a file doubled the data in both engines. A row that kept failing re-wrote its good neighbours every half second: 20 copies in 10 seconds
Monitoring	Per-record history: what arrived, what was sent, what the target answered	The single most useful feature either engine had was a per-message history with a reprocess button
Alerting	Telling a person	Neither engine could say "nothing has arrived from the lab since Tuesday"

Three conclusions shape everything below.

Idempotency is the first requirement, not a feature. Sources re-send files. Operators re-run jobs. Networks deliver twice. Unless every resource carries a stable identity and delivery is an update on it, the FHIR server fills with duplicates, and the analysis built on top of it is wrong in ways nobody notices.

Data errors and transient errors are different problems. A non-numeric value, an unknown code, a 400 from HFS: no retry will fix these. They go to a dead letter with the file, the line and the reason, and wait for a person. An unreachable HFS, a 503, a timeout: these are retried with backoff and never dropped. An engine that treats both the same either multiplies writes or stops good data.

The queue is not something to build. Mature integration engines carry a heavy persistent-queue capability, and rebuilding one is not where the value is. But "queue" means two different things at two scales, and the right answer differs for each (see the deployment section).

---

The Essential Flow

The shape of every feed, regardless of source or target, is the same. Following one record, a line in a lab CSV, through the prototype:

1. Arrival. The source is polled (or notifies). A new object appears: a file in a folder, an object in a bucket.
2. Admission. The object's content hash is checked against what was already received; an identical object is recorded as a duplicate and not reprocessed. The object is decoded (UTF-8, with or without a byte-order mark, and UTF-16) and split into records: CSV rows, NDJSON lines, or the entries of a FHIR Bundle. Each record is stored with the object's name, hash and its own line number, in state received. From this point the engine's store, not the source, is the source of truth: a crash or an HFS outage loses nothing.
3. Transform. Each record is handed to the feed's transform, which returns zero or more FHIR resources or rejects the record with a reason. A rejected record goes to rejected with that reason. A transformed one goes to pending. Data that already is FHIR passes through unchanged.
4. Admission to delivery. The engine checks only what it relies on: a resourceType, and an identity: the resource's own id, or an identifier with system and value. Without either there is no idempotency, so the record is rejected.
5. Delivery. Pending records are sent to HFS. The two delivery paths are described below. A resource with an id is written to {type}/{id}; one without is written as a conditional update on its identifier. Either way the first delivery creates and any later one updates.
6. Disposition. A 2xx marks the record sent with HFS's response (location, version). A 4xx marks it rejected with the OperationOutcome text. A connection failure, 429 or 5xx leaves it pending with a next-attempt time that doubles each try up to a ceiling, and raises one outage alert after a configured number of attempts and one recovery alert when delivery resumes.
7. Completion. When every record of an object is sent or rejected, the object is done and is acknowledged at its source (archived). If any records were rejected, one alert names the object, the count and the first reason.
8. Freshness. Independently of any object, each feed declares how often it expects data. When nothing has arrived for longer than that, one silence alert is raised, and not repeated until data flows again.

A rejected record can be put back to received, optionally with a corrected row; the original row is kept alongside the correction. This is the reprocess button.

stateDiagram-v2
  [*] --> received: record stored
  [*] --> rejected: malformed row
  received --> pending: transform produced resources
  received --> rejected: transform rejected it
  pending --> sent: HFS 2xx (REST path)
  pending --> rejected: HFS 4xx
  pending --> pending: unreachable, 429, 5xx (backoff)
  pending --> submitted: manifest posted ($bulk-submit path)
  submitted --> sent: receipt success
  submitted --> rejected: receipt error
  submitted --> pending: file failed, resubmitted
  rejected --> received: reprocess

Loading

---

The First Use Case: A Bucket into HFS

The concrete goal for the first version is narrow on purpose: a data set lands in an S3 bucket, and it is ingested into HFS with confidence. "With confidence" means every one of the properties above holds: nothing lost, nothing duplicated, every rejection explained, and a person told.

Two delivery paths exist, and the choice is a per-feed setting rather than an architectural commitment.

REST, batch Bundles. The engine builds batch Bundles of PUT entries (PUT Patient/{id}, or PUT Observation?identifier=system|value) and posts them to [base]/. HFS supports both inside Bundles today; the prototype verified 201 on first delivery and 200 on re-delivery. A batch, unlike a transaction, returns a status per entry, so one bad row does not block its neighbours. This path suits feeds of hundreds to tens of thousands of records at a time, anything that needs an answer per record, and everything that goes through a transform.

$bulk-submit. HFS implements the Bulk Data Submit operation as the Data Consumer: a client posts $bulk-submit with a manifest URL, and HFS fetches the manifest and its NDJSON files, ingests them through its bulk engine, and publishes a status manifest with a receipt per line. The first real test case will almost certainly be an NDJSON export landing in a bucket, and for that case the engine does not rewrite the files: it admits each object as above, reads it into records for the history, and then hands HFS the source objects themselves. That path has enough moving parts to deserve its own section, next.

---

NDJSON in a Bucket: The $bulk-submit Path

The picture: an export lands in a bucket as one NDJSON file per resource type. HIE admits each object, reads it into records, and then a BulkSubmitDeliverer writes a manifest.json whose output[] entries point at the source objects through pre-signed URLs, posts $bulk-submit, and maps the receipts back to the records. HFS does the ingestion; HIE keeps the history, the dispositions, the alerts and the archive.

The flow for one drop:

1. Admission. Objects are listed, hashed and read into records exactly as on the REST path. They stay at their keys.
2. Pre-flight. Each record gets the engine's cheap checks in Rust, with no Python involved: is it a JSON object, does it have a resourceType, does it have an id. The result is recorded as a prediction on the record, not a disposition.
3. Manifest. The deliverer writes manifest.json with one output entry per object: type, a pre-signed GET URL for the object, and the line count. requiresAccessToken is false, since the URL is the credential. The manifest itself is written to the engine's own prefix in the same bucket, or served by the engine over HTTP in a single-instance deployment.
4. Kick-off. POST [hfs]/$bulk-submit with submitter = the feed id, submissionId = the drop id (below), manifestUrl, fhirBaseUrl = the engine's own base, and submissionStatus = in-progress. Every record of the drop moves to submitted.
5. Receipts. The engine polls the status URL. HFS publishes one receipt per line of each file: the line number, the resource type and id, whether it was created, updated or left unchanged, or the OperationOutcome when it failed. Receipts are mapped to records by (object, line), which is exactly how the records were stored. Success marks sent, an error marks rejected with the outcome text: the same two dispositions as the REST path, from the same fields.
6. Close-out and archive. When the manifest is terminal, the engine posts the status-only completed kick-off, acknowledges the objects at the source, and raises the per-object alerts as usual.

This is where the open questions sit.

(a) Pass-through records skip the transform and the engine's identity check, so the receipts are the only per-record disposition. Is that acceptable? Yes, with two provisions. First, the pre-flight in step 2 runs on every record anyway: it costs nothing, it needs no Python, and it lets the engine's API show "line 412 has no id" before HFS has even been asked, while HFS's receipt, when it arrives, is authoritative and overwrites the prediction. Second, a receipt carries everything the REST path's disposition carries (type, id, created or updated, outcome, OperationOutcome), keyed by the line number the engine already has, so the per-record history is as complete on this path as on the other. What a pass-through feed gives up is only the chance to fix a record before HFS sees it, and a feed that wants that declares a transform and takes the rewrite path below.

(b) HFS upserts by id on this path, so pass-through only works when the source resources carry stable ids. Agreed, and it should be an explicit feed setting rather than something guessed per object:

ids:	Behaviour
source (default)	Every line is expected to carry an id. The manifest points at the source objects. A line without an id is predicted rejected, and HFS's receipt confirms it.
derive	The engine writes a copy of each object in which every line gets id = UUIDv5 over (feed, identifier.system, identifier.value), and the manifest points at the copy. A line with neither id nor identifier is rejected before the copy is written and left out of it.

Mixing the two inside one feed is not offered: either the source owns the ids or the engine does. With source, HFS's replace import mode is the right one (upsert by id), and HFS_BULK_SUBMIT_SKIP_UNCHANGED makes a replayed export leave versions untouched.

(c) Acknowledge has to move to after HFS has fetched the file. Yes: on this path an object is acknowledged only when its submission is terminal and its receipts are mapped (step 6), not when its records are stored. Two consequences follow. The source listing will keep returning an object that is in flight, so admission distinguishes "this key, this hash, already admitted and in flight" (skipped silently) from "a new key with content already seen" (recorded as a duplicate). And the pre-signed URL's lifetime must cover the time HFS may take to reach the file; it is a feed setting with a generous default (24 hours). If HFS reports that a file could not be fetched or read to its end, the manifest comes back failed, the records of that file that received no receipt return to pending, and the engine posts a replacement manifest (replacesManifestUrl) with fresh URLs for those files only. The entries HFS committed before the break keep their receipts and are not re-sent.

(d) Batch boundary: one manifest per object, or per drop? Per drop, with a cap. An export arrives as a set of files that belong together (Patient.ndjson, Encounter.ndjson, Observation.ndjson), and that set is what the operator thinks of as "today's data". The drop is the set of objects admitted in one scan pass, split into manifests of at most manifest_objects_max objects. Its submissionId is derived from the feed and the sorted (key, hash) pairs of its objects, so re-posting the same drop is idempotent: HFS answers 409 for a submission it has already closed and the engine treats that as "already done". One manifest per object was the alternative: the receipt mapping is marginally simpler, but a nightly export of twenty files would open twenty submissions against HFS's per-tenant cap on concurrent submissions (four by default, 429 beyond), and the operator would have twenty statuses to watch for one arrival. Failure granularity is not what decides this: within a manifest, HFS reports per line, and a truncated file fails only that file.

What stays the same across both delivery paths: the records, their states, the API, the alerts and the archive. What differs is confined to the deliverer, which on this path is asynchronous: it hands work to HFS and collects the answer later. The trait surface below reflects that.

---

The Architectural Tensions

Business keys come from the source, or not at all. Idempotency needs a key that identifies the same real-world event every time it is sent: an accession number plus a test code, an encounter identifier, a message control id, or the stable id an exporting system already assigned. The prototype's example key for the lab CSV (patient|code|date) is a placeholder that would merge two same-day results. Every feed definition must say where its key comes from, and a transform that cannot produce one must reject the record rather than invent one.

Transform isolation against throughput. A transform written by an analyst can hang, crash or loop. Running it in the engine's process means it can take the engine down; running it in a separate process costs a round trip per record. The prototype runs a long-lived Python worker per feed and exchanges one JSON line per record; three Synthea patients (3,550 resources) went through it to HFS in 17 seconds on a laptop, and the protocol can carry batches when that is not enough.

One Python runtime, or a shared one. Transforms are Python. The analytics side of the product also runs Python. Making this one shared service that both halves call into is worth pursuing: the engine's transform protocol is already a request/response over JSON, so the worker can be a subprocess today and an HTTP or queue-backed service tomorrow without changing the trait. The decision can wait until the shared service exists; the engine should not block on it.

Run it yourself against run it managed. HFS made a choice that worked: the same trait surface serves an embedded single-instance deployment and a fleet, and the operator picks. The engine should do the same. The risk of spending years on thin layers over every cloud product is real; the defence is to keep the set of pluggable points small (source, store, notifier, delivery) and to ship one embedded implementation and one managed implementation of each, no more.

How deep to go. What we will not build: visual channel design, drag-and-drop mapping, no-code data preparation, a catalog of SaaS connectors, CDC replication, stream processing. Those are products. What we will build is the thin, reliable path from a handful of source types to HFS, with the operational properties above. Anyone who needs Kafka in the middle can put it there; the engine reads from a queue as readily as from a bucket.

---

Single-Instance vs Multi-Instance

Single instance: zero configuration

One process. Sources are polled. The record store is an embedded SQLite file in WAL mode, which is both the queue and the history. Transform workers are subprocesses. Delivery happens in-process; on the $bulk-submit path the engine serves manifests itself. The API and metrics are served by the same process. This is the prototype, and it is what every developer, every demo and every single-VM customer runs.

Multi instance: shared state, work pool

Several engine instances, possibly a separate worker tier. The record store is PostgreSQL, which is also the planned store for the rest of the product. The queue is a table: workers claim records with SELECT ... FOR UPDATE SKIP LOCKED under a lease with a heartbeat, the same claim strategy the bulk export discussion proposes for HFS, and for the same reason. Object sources are read from S3, and manifests are written there. Arrival can be notified instead of polled: an S3 event to SQS or EventBridge wakes a worker, with polling kept as the safety net. The Python runtime may be the shared service.

A table-as-queue in PostgreSQL carries a lab's volumes comfortably and keeps the queue and the history in one transaction, which a separate broker cannot. SQS fits as the notification channel, where at-least-once delivery and a visibility timeout are exactly what is wanted, rather than as the record queue. Kafka and NATS are not proposed: the workloads here are files and batches, not streams, and the operational cost is not justified by the first several customers.

The point, as with HFS, is that the handler code does not change between the two. The traits below are designed so that the embedded and the managed implementations are the only things that differ.

---

The Recommendation

Concern	Embedded default	Managed variant	Why
Source	Local folder	S3 bucket	object_store, already a dependency of HFS, covers S3, S3-compatible stores and an in-memory store for tests with one interface
Arrival	Polling with a file-age guard	S3 event to SQS, polling as fallback	Notification cuts latency; polling catches what notification missed
Record store and queue	SQLite, WAL	PostgreSQL, SKIP LOCKED leases	One transaction for queue and history; same claim strategy as HFS bulk jobs
Transform runtime	Python subprocess per feed, JSON lines	Shared Python service over the same protocol; a serverless function is possible since the protocol is plain JSON	Isolation now, consolidation later, no trait change
Delivery	REST batch Bundles	$bulk-submit pointing at the source objects	Per-feed choice by volume and latency
Identity	OIDC client credentials to HFS; bearer validation on the engine's API	Same	Reuse helios-auth; the engine must not be an anonymous writer
Monitoring	Prometheus /metrics, OTLP traces	Same, scraped by whatever watches HFS	One monitoring stack for the whole deployment
Alerting	Stored alerts plus a webhook (Slack, Teams, a pager)	An alerting product behind the webhook	Routing, escalation and on-call belong to a product and to the identity provider

What is deliberately left out of the table: a message broker between steps, a workflow orchestrator, and a visual designer.

---

Designing the Rust Traits

The prototype implements these as concrete types in one crate. This section proposes the trait surface that lets each concrete type be replaced. Names are proposals.

Types

/// A configured feed: one source, one reader, an optional transform, one delivery path.
pub struct FeedId(pub String);

/// One unit of input: a file, an object, a message. Immutable once admitted.
pub struct SourceObject {
    pub feed: FeedId,
    /// Stable within the source: a path, an object key, a message id.
    pub key: String,
    pub bytes: Vec<u8>,
    pub sha256: [u8; 32],
    pub received_at: DateTime<Utc>,
}

/// One record cut from a source object, with enough context to be named
/// to a human ("line 3 of labs.csv", "entry 12 of patient.json").
pub struct SourceRecord {
    pub object_id: ObjectId,
    pub line: u64,
    pub raw: serde_json::Value,
}

pub enum RecordState {
    Received,
    Pending { attempts: u32, next_attempt_at: DateTime<Utc> },
    /// Handed to HFS through a manifest; waiting for its receipt.
    Submitted { submission: SubmissionId },
    Sent { response: serde_json::Value },
    Rejected { kind: RejectionKind, reason: String, response: Option<serde_json::Value> },
}

pub enum RejectionKind {
    /// Could not be cut into fields: wrong column count, undecodable, not a JSON object.
    Parse,
    /// The transform said the data is wrong.
    Data,
    /// The transform itself failed; kept with its trace.
    Transform,
    /// The engine's own check: no resourceType, no id and no identifier.
    Validation,
    /// HFS answered 4xx for this entry, or its receipt reported an error.
    Server,
}

/// What a transform makes of one record.
pub enum TransformOutcome {
    Resources(Vec<serde_json::Value>),
    Rejected { kind: RejectionKind, reason: String },
}

/// What makes a resource's delivery idempotent.
pub enum DeliveryKey {
    Id { resource_type: String, id: String },
    Identifier { resource_type: String, system: String, value: String },
}

/// Why a delivery attempt did not succeed as a whole.
pub enum DeliveryError {
    /// Try again later: unreachable, timeout, 429, 5xx.
    Transient(String),
    /// The target refused the request itself.
    Refused { status: u16, body: serde_json::Value },
}

Source

A source lists what is new, fetches it, and acknowledges it once the engine is done with it. Polling and notification are both expressed through wait: the folder implementation sleeps for the poll interval, the S3 implementation waits on an SQS receive with the poll interval as a timeout. For the $bulk-submit path the source also mints a URL that HFS can fetch an object from.

#[async_trait]
pub trait Source: Send + Sync {
    /// Blocks until there may be new objects, or the interval elapses.
    async fn wait(&self) -> Result<()>;

    /// Keys of objects not yet acknowledged, oldest first. Must be safe to
    /// call repeatedly: an object listed twice is admitted once by its hash.
    async fn list(&self) -> Result<Vec<String>>;

    async fn fetch(&self, key: &str) -> Result<SourceObject>;

    /// A URL from which a third party (HFS) can read the object for `ttl`.
    /// A pre-signed GET on S3; a URL served by the engine for a folder.
    async fn share(&self, key: &str, ttl: Duration) -> Result<Url>;

    /// The engine is done with the object. Move it out of the way
    /// (archive prefix, processed folder, message delete).
    async fn acknowledge(&self, key: &str) -> Result<()>;
}

Implementations: FolderSource and S3Source (both in the prototype, the latter through object_store), later FhirSource (another FHIR server, driven by $export or history) and DatabaseSource.

Reader

Cutting an object into records is independent of where the object came from. A reader never fails an object because of one row.

pub trait Reader: Send + Sync {
    fn read(&self, object: &SourceObject) -> Result<Vec<ReadRecord>>;
}

pub struct ReadRecord {
    pub line: u64,
    pub raw: serde_json::Value,
    /// Set when the row could not be cut: it is stored and rejected, and
    /// the rest of the object proceeds.
    pub parse_error: Option<String>,
}

Implementations in the prototype: CsvReader, NdjsonReader, and BundleReader, which yields one record per entry and rewrites urn:uuid references between entries to Type/id so the entries can be delivered independently. Later: Hl7v2Reader.

RecordStore

The store is the queue, the history and the coordination point. Everything the API serves comes from it, and everything a worker does goes through it. The multi-instance implementation adds leases; the single-instance one satisfies them trivially.

#[async_trait]
pub trait RecordStore: Send + Sync {
    async fn find_object_by_hash(&self, feed: &FeedId, sha256: &[u8; 32]) -> Result<Option<ObjectRef>>;
    async fn admit_object(&self, object: &SourceObject, records: &[ReadRecord]) -> Result<ObjectId>;

    /// Claims up to `limit` records in `Received`, for this worker, under a lease.
    async fn claim_received(&self, feed: &FeedId, worker: &WorkerId, limit: usize) -> Result<Vec<SourceRecord>>;
    /// Claims up to `limit` records in `Pending` whose next attempt is due.
    async fn claim_due(&self, feed: &FeedId, worker: &WorkerId, limit: usize) -> Result<Vec<PendingRecord>>;
    async fn heartbeat(&self, worker: &WorkerId) -> Result<()>;

    async fn set_state(&self, record: RecordId, state: RecordState) -> Result<()>;
    /// Back to `Received`, keeping the original row when a correction is supplied.
    async fn reprocess(&self, record: RecordId, corrected: Option<serde_json::Value>) -> Result<bool>;

    /// Submissions in flight, for the deliverer to poll.
    async fn open_submissions(&self, feed: &FeedId) -> Result<Vec<SubmissionId>>;
    /// Records of a submission, by (object, line), for mapping receipts.
    async fn submitted_records(&self, submission: &SubmissionId) -> Result<Vec<(ObjectId, u64, RecordId)>>;

    /// Objects whose records are all terminal, marked done and returned once.
    async fn complete_objects(&self, feed: &FeedId) -> Result<Vec<ObjectSummary>>;
    async fn last_received_at(&self, feed: &FeedId) -> Result<Option<DateTime<Utc>>>;

    async fn record_alert(&self, alert: &Alert) -> Result<AlertId>;
    async fn last_alert_at(&self, feed: &FeedId, kind: AlertKind) -> Result<Option<DateTime<Utc>>>;
}

Implementations: SqliteStore (prototype), PostgresStore (leases via SKIP LOCKED, the same shape as BulkExportStorage claims in HFS).

Transformer

A transform is a function from one record to resources, run somewhere else. The trait hides where. A transformer failing (as opposed to rejecting a record) is a transient condition: the record stays claimed and is retried. A pass-through feed has no transformer; its records go from Received to Pending after the engine's own check.

#[async_trait]
pub trait Transformer: Send + Sync {
    async fn transform(&self, record: &SourceRecord, context: &TransformContext) -> Result<TransformOutcome>;
}

pub struct TransformContext {
    pub feed: FeedId,
    pub object_key: String,
    pub line: u64,
}

Implementations: PythonSubprocess (prototype; JSON lines over stdin/stdout, restarted on failure), PythonService (the shared runtime over HTTP, same messages), later FhirMappingLanguage.

Deliverer

Delivery takes a batch of records with their resources and keys. The REST path answers at once; the $bulk-submit path answers later, so the trait has two steps and the runner polls open submissions on every pass.

#[async_trait]
pub trait Deliverer: Send + Sync {
    /// Either a disposition per item now, or a submission to poll.
    async fn deliver(&self, batch: &[DeliveryItem]) -> Result<Delivery, DeliveryError>;

    /// Dispositions that have arrived for an open submission, by (object, line).
    /// `None` while the submission is still running.
    async fn poll(&self, submission: &SubmissionId) -> Result<Option<SubmissionResult>, DeliveryError>;
}

pub struct DeliveryItem {
    pub record: RecordId,
    pub object: ObjectId,
    pub line: u64,
    pub key: DeliveryKey,
    pub resource: serde_json::Value,
}

pub enum Delivery {
    Done(Vec<Disposition>),
    InFlight(SubmissionId),
}

pub struct SubmissionResult {
    pub dispositions: Vec<((ObjectId, u64), Disposition)>,
    /// Objects HFS could not read to their end; their unreceipted records go back to `Pending`.
    pub failed_objects: Vec<ObjectId>,
}

pub enum Disposition {
    Sent(serde_json::Value),
    Rejected { status: u16, body: serde_json::Value },
    Transient { status: u16, body: serde_json::Value },
}

Implementations: RestBatchDeliverer (prototype: PUT entries in a batch Bundle; a refused Bundle is re-sent one item at a time to isolate the bad one; poll is never called), BulkSubmitDeliverer (groups a drop's objects into manifests that point at the source objects or at the engine's rewritten copies, posts $bulk-submit, and maps HFS's per-line receipts back to records).

AlertSink

#[async_trait]
pub trait AlertSink: Send + Sync {
    async fn raise(&self, alert: &Alert) -> Result<()>;
}

pub enum AlertKind { RejectedRecords, UnreadableObject, DeliveryOutage, DeliveryRecovered, Silence }

The engine decides when (once per condition, with recovery) and what (feed, object, count, first reason). The sink decides where: the log, a webhook, later a product. Who is on call is not the engine's concern.

The runner

One FeedRunner per feed ties the traits together in a loop whose steps each pick up what the previous pass left: wait on the source, admit new objects, transform claimed records, deliver due records, poll open submissions, complete and acknowledge objects, check silence. A step that fails is simply retried on the next pass, which is what makes the loop safe to restart at any point. In a multi-instance deployment several runners for the same feed coexist because every claim goes through the store's lease.

---

The API

The engine's HTTP API is what a client application calls to render a data-sources page, and what an operator uses to find and fix a rejected row. The shape below is what the prototype serves; the exact needs of the client application are still being gathered, and this section will grow with them.

Resource	Purpose
GET /api/feeds	Each feed with its freshness (ok, silent, no_data_yet), counts by state, last object, last alert
GET /api/feeds/{id}/objects	Objects received, duplicates marked, per-object sent and rejected counts
GET /api/records?feed=&status=&object=	Records, filterable
GET /api/records/{id}	One record: object, line, original and corrected row, resources sent, target response or receipt
POST /api/records/{id}/reprocess	Back to the start, optionally with a corrected row
GET /api/alerts	Alerts, newest first
GET /metrics	Prometheus: records by state per feed, last-received timestamp per feed, alerts by kind

The API is protected by the same bearer-token validation HFS uses, from helios-auth.

---

Configuration: What Operators Will Touch

An engine configuration names HFS, the store, the listen address, the transform runtime and the alert webhook. A feed is one file:

id: lab-export
name: Lab export
source:
  type: s3                      # or folder
  bucket: lab-feeds
  prefix: export/
  notify: sqs://lab-feeds-arrivals   # optional; polling otherwise
format: ndjson                  # csv, ndjson, or fhir_bundle
transform: ~                    # a Python file, or none for pass-through
delivery:
  method: bulk-submit           # or rest
  ids: source                   # or derive
  manifest_objects_max: 50
  url_ttl: 24h
  initial_backoff: 2s
  max_backoff: 5m
  alert_after_attempts: 5
expect_every: 24h

Feeds and transforms are files under version control. Mapping rules are agreements with a data source; they deserve history and review. A UI that edits them should do so through an API that writes these files, not around them.

Environment variables follow the HFS convention: HIE_CONFIG, HIE_LOG, HIE_RUNTIME_DIR, and later HIE_STORE_URL, HIE_AUTH_*.

---

Authorization

The engine is a client of HFS and a server to its own API, and both sides use the identity provider.

• To HFS: a confidential client with its own credentials (SMART Backend Services, client_credentials with private_key_jwt), requesting system/*.write or the narrower scopes a feed needs, plus system/bulk-submit on the bulk path. Tokens are cached and refreshed per feed. On the bulk path HFS reads the files through pre-signed URLs and needs no credential of its own for them; HFS redacts the signed query string wherever it reports a file URL.
• Its own API: bearer tokens validated against the identity provider's JWKS, reusing helios-auth. Reading feed status and reprocessing a record are different permissions.

The engine never runs as an anonymous writer, and a feed's identity in HFS audit events should be the feed's client, not a shared service account.

---

What's Not in Scope (Yet)

• HL7v2 input. Most laboratory interfaces speak ORU over MLLP. It is the next reader and source after S3, and the one place where the engines we tried have a genuine head start.
• Code translation inside transforms through HTS (ConceptMap/$translate). The example transform carries a local code table; the real one calls HTS.
• Patient matching. A source identifier (an MRN) must become a Patient reference. This is its own design problem.
• Profile validation before delivery through HFS $validate.
• FHIR Mapping Language as a second transform runtime.
• Outbound integration, CDC, streaming, visual authoring.

---

Proposed Next Steps

1. Run the S3 source against a real bucket with an NDJSON export, on the REST path that already works, and confirm the archive and duplicate behaviour there.
2. Build BulkSubmitDeliverer as described above: share on the source, manifests per drop, the submitted state, receipt mapping by (object, line), replacement manifests for failed files, and acknowledgement after close-out. Measure the two paths on the same export.
3. Gather the client application's needs for the API and extend the API section.
4. Extract the traits above from the prototype's concrete types; keep the end-to-end tests green through the refactor.
5. Authentication. Client credentials to HFS; bearer validation on the API.
6. PostgresStore with leases, with a testcontainers integration test that runs two runners against one feed.
7. S3 event notification through SQS, with polling as fallback.
8. Decide the shared Python runtime with the analytics side, and add PythonService behind the same protocol if it goes that way.
9. HL7v2 reader and MLLP source.

---

Closing Thoughts

The products in this space are large because they try to be the whole pipeline for every industry. What a FHIR platform needs from an integration engine is narrower and harder to get right: a short, dependable path from a few kinds of source into the server, where nothing is lost, nothing is duplicated, every rejection can be explained to the person who owns the data, and silence is noticed.

The trait surface proposed here is small on purpose: a source, a reader, a store, a transformer, a deliverer, an alert sink. Each has one embedded implementation and one managed implementation, and the runner does not know which it holds. That is the same bet HFS made with its persistence layer, and it is what lets the same engine run on a laptop against SQLite and in a fleet against PostgreSQL, S3 and a shared transform service without rewriting the loop in the middle.

Thank you for reading. Feedback, objections and better names are all welcome.

• Angela

[angela-helios]

The API, assessed against the client UI's design

The client UI now has a decided design for its data section: three modules, Sources, Transforms and Monitoring, plus a five-step "New source" flow. This comment maps what those screens need onto what the engine serves today, and lists the gaps. It replaces the placeholder in "The API" above. The vocabulary below is the UI's; the engine's names are in parentheses where they differ.

Vocabulary

UI	Engine today	Note
Source: one connection, with a chain Connector → Transform → Load → FHIR server	Feed	A feed is a source. The chain's four columns are the feed's source type, transform, delivery method and target.
Connector: FHIR server, database, or files (SFTP, uploaded, CSV / HL7 v2 / NDJSON)	Folder, S3	Two of the three kinds do not exist yet.
Transform: a named Python program, versioned, shared between sources	A Python file named by the feed	No registry, no versions, no "used by".
Load: one write into the FHIR server (a pull, a file, the first load)	Object (file)	Close for file sources; a pulled source needs a load per pull.
Failed row with its reason and the transform line	Rejected record with its reason	The transform line is missing.
Alert: a rule in words ("nothing arrives for 2 hours"), who is notified, how; fires and clears itself	Fixed alert kinds raised once per condition	Rules are not configurable; recipients and channels are not modelled.

What the screens need, and what exists

Sources, the page of one source. The chain, the facts (last received, usually, brings, connected by whom and when, if it stops), Today (rows read, resources written, failed, next pull), by day for 14 days, last 7 days as a table, and the state (receiving, pulled every N minutes, nightly file, first load running, late, paused, not connected).

• Exists: GET /api/feeds gives id, name, source, freshness, counts by state, last received, last alert. GET /api/feeds/{id}/files gives each object with sent and rejected counts.
• Missing: per-day aggregates (GET /api/feeds/{id}/stats?days=14), a "today" summary with the next scheduled pull and the last load's duration, "brings" (the resource types a source produced, derivable from the records), "connected by whom and when" (needs source metadata the engine does not keep), and the richer state vocabulary (paused, first load running, not connected).

Loads. Every load with when, what, method, rows read, written, failed, how long it took; failures grouped by reason with the transform line, sample rows, Retry the N rows, Open the transform with this row.

• Exists: objects with counts; GET /api/records?file_id=&status=rejected with reason per row; POST /api/records/{id}/reprocess one row at a time; the original and corrected row.
• Missing: duration per load; failures grouped by reason (GET /api/loads/{id}/failures grouping rejected records by reason, with a sample); bulk retry by load or by reason; the transform line in a rejection (the Python runtime can record the line that raised DataError); a load for a pull, not only for a file.

New source. Step 1 the connector kind; step 2 connect with Test connection and What to read (tables, row count, newest timestamp, the column that marks new rows); step 3 pick a transform and Try it on five real rows; step 4 the load method with one recommended, the first load in bulk, the target server; step 5 review and Start. Then Settings (the same steps with Change), Pause, Check now.

• Exists: nothing over the API. Feeds are YAML files under version control (decision 4).
• Missing: source definitions as an API resource (POST /api/sources, PUT, pause, resume, check-now), with the engine writing the versioned files so decision 4 still holds; POST /api/sources/test-connection per connector kind, returning the evidence the screen keeps (reached in N ms, N rows, newest at T); discovery of what to read (tables, columns, new-row marker) for the database connector; a first load that runs in bulk and then switches to incremental, with progress (N of M rows, time left) for the band on the page.

Transforms. A list (name, version, from what rows to which resources, used by which source or "ready"), the code, versions explained in words with who, when and which source used each, Try it on real rows (what each row became, what the transform did), Save version, New transform from empty, a copy or a template; opened from a failed load with the failing line marked and the row ready in Try it.

• Exists: transforms as files; the Python worker that runs them.
• Missing: a transform registry (GET /api/transforms, versions, used-by); POST /api/transforms/{id}/try that runs a version on supplied or sampled rows without loading anything, returning the resources and the rejection reason per row; save a version (the engine writes the file and a version record); a load records the transform version it ran with.

Monitoring. Status: a "needs you" list (late sources, failed rows), last 24 hours (loads, rows read, failed rows, alerts sent), all sources in one table. Alerts: rules per source as sentences (nothing for 2 hours; more than 1% of a load's rows fail; the file is not there by 06:00; the FHIR server cannot be reached), who is notified, in the app and by email, a default rule born with every source, and what was sent in the last 7 days with when it cleared.

• Exists: silence, rejected-records, delivery-outage and recovery alerts, raised once and stored; a webhook.
• Missing: alert rules as data (/api/alert-rules: condition, duration or threshold, source, recipients, channels), evaluated by the engine; the cleared event with its reason; recipients resolved through the identity provider; in-app delivery (the UI's side, fed by the engine's alert stream) and email (a sink behind the webhook, or a mail sink); the status aggregate (GET /api/monitoring/status).

The assistant. The UI's assistant reads sources, loads, failures, transforms and status, and explains; it never connects, starts, pauses, retries or saves. Every one of its tools (describe_source, impact_of_source, explain_failed_rows, recommend_load, monitoring_status, describe_transform) is a read over the API above, so the API is the contract and the assistant's MCP server wraps it. impact_of_source (which analyses and hooks depend on a source) is the UI's own data joined on the source id; the engine only needs to expose stable ids and what a source brings.

What this changes in the design

1. Sources become API resources with a persisted definition, not only files. The engine owns the definition, keeps it under version control as today, and serves it. This is the largest addition and it is on the path of every screen.
2. Two more connector kinds, pulled rather than dropped: a FHIR server (incremental by _lastUpdated or $export) and a database (incremental by a column). Each pull is a load.
3. Transforms get a registry, versions and a dry run. The dry run reuses the worker; the line that raised a rejection is recorded.
4. Alerts become rules, with the four conditions the UI names, recipients and channels. The engine evaluates and records fire and clear; delivery stays pluggable.
5. Aggregates per source and overall (today, by day, last 24 hours, status) are served by the engine from its own store, so the UI never queries the database.
6. The first load is the $bulk-submit path with progress; the incremental loads are the REST path, per source, which is what step 4's "recommended" method encodes.

Authorization is unchanged: bearer tokens on the API, validated as HFS validates them; roles for who may connect, start or pause a source are an open item on the UI's side too.

[angela-helios]

Correction on the $bulk-submit receipts, after implementing the path

The section above says HFS publishes one receipt per line with the line number. That is what HFS keeps internally (BulkEntryResult.line_number), but it is not what it publishes. The status manifest links two kinds of receipt file: the output files hold one {"reference": "Type/id"} line per resource HFS stored, grouped by resource type; the outcome files hold one OperationOutcome per line HFS could not store, with the error text and no input file or line number.

So the engine maps receipts to records by Type/id, which every pass-through record carries, and that answers (a) a little differently than written:

• a record whose Type/id appears in an output file is sent, with the reference as its response;
• a record with no reference is rejected with "no receipt from HFS", plus the first error text HFS published, without a certain attribution to that line;
• the engine's pre-flight (a JSON object, a resourceType, an id) names the line for the failures it can predict before HFS runs, which in practice are most of them.

HFS publishing the input file URL and line number on its error OperationOutcome (as an extension on the issue) would close the gap; that is filed on the HFS side. The implementation is in the repo (src/bulk.rs, decision 21), with an end-to-end test where a stand-in HFS fetches the manifest and the file from the engine and publishes receipts.