Documentation / Développement / Guides pratiques / Contribute an event source and binding

Contribute an event source and binding

This recipe adds event-driven ingestion to a plugin: a source mechanism that reads from an external system (a polled FHIR server, an HL7 socket…) and a declarative binding that projects each raw record into a domain event that starts a workflow. You author both; an operator later declares an inbound connection of your source on their instance.

Ingestion is split into three independently contributable concerns (libs/shared/src/plugin/manifest.ts):

  • eventSources — a generic mechanism. It takes the parameters of a connection plus a list of opaque feeds and emits raw records. It knows nothing about domain events. The host makes it an inbound connection type.
  • eventBindings — the declarative, reusable link: source + query + event + map. This is where a raw record becomes a domain event.
  • The instance — declares an inbound connection whose type is the source: its parameters and the bindings it enables. That side is the operator's; see Connect an event source and activate bindings.

For the why behind this seam, read The open event model.

Before you start

  • A plugin scaffold you can edit (see Scaffold a plugin).
  • The domain event type you intend to emit must be a declared event of a loaded plugin: every declared event is received by Événement reçu, and by the triggers that reference it with event: (see the plugin manifest reference). The binding's event must match that string exactly. Otherwise the host drops the binding at load and its feed never starts: « binding ignoré : événement « » non déclaré par les plugins chargés. Charger le plugin qui le déclare. » for a type no loaded plugin declares, « binding ignoré : « » est l'adresse du déclencheur épinglé « ». Viser un événement déclaré. » for the address of a trigger with its own address (pinnedRouting, such as the raw message of a reception). When the event is declared, its contract (contributes.events of the plugin that owns it) lists the fields your map produces and their types: read it on the Plugins page (Declared events) or in GET /api/catalog (eventTypes). To declare a new event, see Declare an event.
  • Familiarity with the projection grammar (MapExpr reference).

1. Write the source mechanism

Add a module that default-exports a SourceFactory (@meridian/shared/plugin/source). The factory receives { config, connection, feeds, emit, log } and returns a SourceHandle with start() / stop(). The host calls it once per enabled inbound connection of the source:

import type { SourceFactory, SourceFeed } from "@meridian/shared/plugin/source";

const fhirPoll: SourceFactory = ({ config, feeds, emit, log }) => {
  const intervalMs = Number(config.intervalSeconds ?? 15) * 1000;
  let timer: ReturnType<typeof setTimeout> | undefined;

  const tick = async () => {
    for (const feed of feeds) {
      // feed.query is the OPAQUE query descriptor authored by the binding.
      const records = await pollOnce(config, feed.query);
      for (const rec of records) {
        // emit(feedName, stable recordId, rawData)
        await emit(feed.name, rec.stableId, rec.raw);
      }
    }
    timer = setTimeout(() => void tick(), intervalMs);
  };

  return {
    start() { log(`polling every ${intervalMs / 1000}s`); void tick(); },
    stop() { if (timer) clearTimeout(timer); },
  };
};
export default fhirPoll;

Hold to the contract:

  • config is the parameters of the inbound connection (interval, page size…): the values the operator set, with the defaults your params declare (see below). Never hard-code an endpoint here.
  • connection, for a source that declares connection: { type }, resolves the outbound connection the operator named with the connection parameter, or the default connection of the type, at each call. That parameter is not in config.
  • feeds is one SourceFeed ({ name, query }) per binding enabled on the connection. Each query is a record you defined in the binding; the mechanism interprets it however it likes.
  • emit(feed, recordId, data) hands a raw record to the host's reception, which stores it in the inbox of the connection (message id <feed>:<recordId>) and delivers it to the binding of the feed. The recordId must be stable for the same logical record: a record emitted again with the same feed and recordId is a duplicate, even when its content changed, and is not delivered again. The event's idempotency key derives from it too (eventId = <connection id>:<feed>:<recordId>, apps/api/src/server/source-records.ts), and each published workflow that receives the event type starts a run keyed <eventId>:<slug>. Choose whether an update re-emits (a new recordId) or not, per your mechanism's semantics. A new recordId is a new event, which every workflow receiving the type gets, including workflows published since.
  • emit resolves once the host has stored the record. Processing happens at delivery, afterwards: a record whose processing fails (Restate ingress unreachable, for instance) is delivered again by the reception and does not concern your cursor. The retry starts only the runs that are still missing.
  • emit rejects when the host could not store the record (application database unreachable, connection disabled meanwhile). Treat the record as not delivered: keep it for the next poll instead of moving your cursor past it. Re-emitting a stored record is harmless: it is a duplicate.
  • A cursor on a timestamp must not skip records that share one. Read oldest first, move the cursor past a record only once its emit resolved, and resume from the cursor's instant inclusively, skipping the records of that instant already emitted: a strict > cursor loses a record written later in the same instant, or separated from the others of its instant by a page limit. The instant is the precision at which the server compares the timestamp (Blaze: the second, for _lastUpdated). A timestamp older than the previous one (a server clock that stepped back) calls for moving the cursor back, not for skipping the record.
  • The mechanism does no mapping to domain events — that is the binding's job (step 2).

The reference fhir-poll mechanism lives at external-plugins/fhir/sources/poll.ts, with its paging logic in external-plugins/fhir/lib/poll-page.ts, if you want a full example: a per-feed cursor that moves with each emitted record and survives resources sharing a lastUpdated second or a server clock that steps back, pages followed up to maxPages per poll and resumed right after, next links mapped onto the configured base URL through the base the server advertises, a start overlap window re-read after each restart (startOverlapSeconds), and the dedupe/since conventions read off feed.query.

Register the module in plugin.yaml, with the parameters a connection of the source takes:

contributes:
  eventSources:
    - name: fhir-poll
      module: ./sources/poll.ts
      label: Interrogation FHIR
      connection: { type: fhir-r4 }
      params:
        - { name: intervalSeconds, label: Intervalle (s), type: number, min: 1, default: 15 }
        - { name: count, label: Ressources par page, type: integer, min: 1, default: 50 }
      description: Polls a FHIR server and emits raw resources.
  • name is also the type of the inbound connections of the source, and label the name of the source in the console's connection form.
  • params has the shape of a connection type's parameters (Connection types). The host checks the values of a connection against them, applies their defaults and passes the result as config; the console builds the connection form from them. Without params, the parameters reach config unchecked.
  • connection: { type } gives the source an outbound connection. The host adds a parameter connection in front of yours, shown in the console as a picker of the connections of that type. A manifest that declares connection declares sdk: "^0.1.6".

export defaults to default; set it if you export the factory under another name (libs/shared/src/plugin/manifest.ts).

2. Declare a binding that starts a run

A binding names a feed (name), targets a mechanism (source), carries the opaque query the mechanism will run, and projects the raw record into a domain event via map:

contributes:
  eventBindings:
    - name: fhir-creatinine          # = the feed name handed to the mechanism
      source: fhir-poll
      event: BioResultReceived        # a declared event, or the type a trigger claims
      query: { resource: Observation, params: { code: "http://loinc.org|2160-0" } }
      map:
        patientId:   { path: subject.reference, ref: true }    # "Patient/x" → "x"
        encounterId: { path: encounter.reference, ref: true }
        analyte:     { const: creatinine }                     # literal
        value:       valueQuantity.value                       # dotted path (shorthand)
        loincCode:   { jsonpath: "$.code.coding[?(@.system=='http://loinc.org')].code" }
      require: [patientId, encounterId, value]
  • name is the feed name: it is what appears as feed.name in your mechanism and what an operator enables on a connection. Keep it unique and descriptive.
  • query is free-form (defaults to {}). Its shape is a private contract between this binding and the mechanism it targets — the FHIR poller reads resource/params, another mechanism would read something else.
  • event must equal a declared event of a loaded plugin; otherwise the host drops the binding at load (« binding ignoré : … », component plugins, with the cause) and its feed never starts. An emitted event is rejected when no published workflow receives the type or, under EVENT_CONTRACTS=enforce, when it does not conform to the contract of its declared type, and filtered (no run) when workflows receive it but none of their filters keeps it; each outcome is counted per source, feed and event type.
  • map is a record of field name → MapExpr. Each value is evaluated against the raw record; the results plus type (= event) and a generated eventId become the domain event. A map that throws counts as a rejected generation and does not crash the source.
  • require lists fields of the projected event that must be present. A record whose listed fields are absent — missing, null, a string empty once trimmed, or [] for a field the contract declares as a list — becomes no event: it is counted rejected and logged at debug level (« enregistrement filtré (champs exigés vides) »). List there the contract's required keys, and any field without which the workflow has nothing to work on: a query often returns records that are not events for the workflow (a FHIR thread entry without patient or content, for instance).

When event is a declared event, the host checks the binding once every plugin is loaded, and meridian-plugin validate and build do the same when the declaring plugin is in your dependsOn closure:

  • every map key must be a contract key — a declared field, the idField of one of the event's context kinds, or occurredAt;
  • every require name must be a contract key;
  • every required field, and the id of every required context kind, must be a map key.

A mismatch is a load warning (« binding hors contrat »); the binding still loads, and meridian-plugin build stops before writing dist/ when the built plugin itself has the problem. Coverage in map does not guarantee a value: a key present in map still yields no value when the record lacks the path. Under EVENT_CONTRACTS=enforce, such a record becomes no event: the source filters on the binding's effective require, its require plus the contract's required fields and required-kind ids, computed once every plugin is loaded. Under observe, the default, a record that passes the declared require becomes an event, starts runs without the value and is counted (meridian.events.contract_missing); list the contract's required keys in require so that it starts no run in any mode. Produce values of the declared types without relying on conversion — a Decimal field is a JSON number, not numeric text — and test the projection on complete resources.

The MapExpr grammar (full detail and edge cases in the MapExpr reference):

Form Meaning
"a.b.0.c" dotted path (shorthand)
{ path, ref: true } dotted path, then strip a FHIR reference ("Patient/x" → "x")
{ jsonpath, all?: true } JSONPath (filters, wildcards); first match unless all
{ const } literal value

The grammar is declarative and code-free (contract in libs/shared/src/plugin/mapping.ts; evaluation in libs/engine-core/src/engine/event-mapping.ts), so a binding travels between instances unchanged. For mapping FHIR resources specifically, see the FHIR mapping reference (apps/ehr-lab/docs/resources/reference/fhir-mapping.md).

When the platform writes to the resource you poll

A workflow can write to the resource your binding reads: a notification written to the record comes back as CommunicationReceived through the fhir-communication-received binding. Every published workflow on that event then starts on the platform's own write. Mark such writes so that a field of the event tells them apart — the platform's FHIR adapter writes Communications with sender.display = "meridian", and the binding maps sender.display to the event's sender field — and declare a defaultSelector on the event (event.sender != "meridian" on CommunicationReceived, see Declare an event). An event with no field that marks the write, such as ConditionRecorded for a condition the platform adds to the record, cannot be filtered this way.

3. Variant — resolve a validation instead of starting a run

A binding may instead observe an external decision and use it to resume a suspended run (a human validation completed in another system) rather than start a new one. Replace event + map with a resolve block. A binding must declare exactly one of event or resolve (libs/shared/src/plugin/manifest.ts):

contributes:
  eventBindings:
    - name: fhir-validation-decided
      source: fhir-poll
      query: { resource: Task, params: { code: "urn:ehr-lab:validation|decision", status: "completed,rejected" }, since: epoch }
      resolve:
        run:      { jsonpath: "$.identifier[?(@.system=='urn:ehr-lab:validation-run')].value" }      # the run key
        pending:  { jsonpath: "$.identifier[?(@.system=='urn:ehr-lab:validation-pending')].value" }  # the decided validation
        decision: status                                                                             # accepted|rejected
        by:       { path: owner.reference, ref: true }                                               # optional author

The run key comes back from where the platform wrote it: the FHIR adapter publishes each pending validation as a Task whose identifier urn:ehr-lab:validation-run carries the run id. A domain run id has the form <eventId>:<slug> and contains :, so store it in a field that accepts that character — an identifier value, not a FHIR resource id.

run, decision, and optional pending and by are each a MapExpr. On each emit the engine maps them and calls the host's resolution path (opResolve), which is idempotent — re-observing a decision that was already applied is a benign no-op. decision is normalised: anything other than rejected is treated as accepted.

pending projects the id of the validation the decision is about — the pendingId the host publishes with the validation request when the run suspends (PendingValidationRecord, @meridian/shared/domain/validation); the FHIR adapter writes it as the Task identifier urn:ehr-lab:validation-pending. Declare it whenever your feed can be read again after a decision was applied (since: epoch, a restart overlap window) and a run can wait on several validations in turn: the host then applies a decision only to the validation it names, answers 409 (a duplicate) for any other, and rejects a record that does not carry it. A manifest that uses pending declares sdk: "^0.1.2": an older host would drop the key without a word and correlate by run only. This branch carries no FHIR-specific notion; it is the generic "resume a durable validation" seam. See Add human validation and Run and resolve for the run-side of that lifecycle.

4. Build and ship

Bundle the plugin with the CLI (esbuild):

meridian-plugin build

This first runs the declared-event checks of step 2, then compiles your source module and behaviors into dist/; an external plugin is loaded from there. Details in Build and bundle. To distribute it, publish to a registry — see Publish to the registry.

5. Hand off to the operator

You do not configure an endpoint. Document, for the operator, the source name, each binding name, and the parameters your source declares. On their instance manifest they declare an inbound connection whose type is the source name, or create it in the console:

connections:
  fhir-poll:
    type: fhir-poll
    connection: entrepot          # an outbound fhir-r4 connection; absent: the default of the type
    intervalSeconds: 15
    bindings: [fhir-creatinine]   # absent: every binding of the source

The rest of the operator recipe — enabling bindings, verifying feeds fire — is Connect an event source and activate bindings; the manifest field reference is the instance manifest.

97 documents8 sectionssource : /docs · généré au build