FHIR Pending Lab Results Sentinel

id: fhir-pending-lab-results-sentinel
namespace: company.team
description: |
  On a schedule, read-only search a FHIR R4 server for lab reports that are
  still registered, partial or preliminary long after the specimen was
  collected (nothing is created, updated or deleted), and classify the
  backlog as HEALTHY, WARNING, CRITICAL or UNREACHABLE. Log quiet runs and
  page Slack when results are stuck before they reach clinicians.

inputs:
  - id: fhir_base_url
    type: STRING
    defaults: "https://fhir.example.org/fhir"
    displayName: "FHIR Base URL"
    description: "FHIR R4 service base URL, without a trailing slash (the URL that
      /metadata is served under)."
  - id: category
    type: STRING
    defaults: "LAB"
    displayName: "Report Category"
    description: "DiagnosticReport category code to watch (HL7 v2-0074), for example
      LAB, MB (microbiology) or HM (hematology)."
  - id: stale_hours
    type: INT
    defaults: 24
    displayName: "Stale After (hours)"
    description: "A report that is still registered, partial or preliminary this
      many hours after its clinically relevant time (specimen collection) counts
      as stale."
  - id: critical_hours
    type: INT
    defaults: 72
    displayName: "Critical After (hours)"
    description: "If any report is still not final this many hours after collection,
      the check is classified CRITICAL."
  - id: critical_stale_count
    type: INT
    defaults: 10
    displayName: "Critical Stale Count"
    description: "At or above this many stale reports, the check is classified
      CRITICAL even if none has reached critical_hours."

variables:
  pending_search: "{{ inputs.fhir_base_url
    }}/DiagnosticReport?status=registered,partial,preliminary&category={{
    inputs.category }}"

triggers:
  - id: hourly_results_check
    type: io.kestra.plugin.core.trigger.Schedule
    description: Check the results backlog every hour. Shipped disabled; enable it
      once fhir_base_url and the FHIR_ACCESS_TOKEN secret point at a real server
      and a manual run has been verified.
    cron: "15 * * * *"
    disabled: true
    inputs:
      fhir_base_url: "https://fhir.example.org/fhir"

tasks:
  - id: read_capability_statement
    type: io.kestra.plugin.core.http.Request
    description: Read the server's CapabilityStatement from /metadata, limited with
      _elements to the FHIR version, software and kind (a full
      CapabilityStatement can be several megabytes), to confirm the endpoint is
      a reachable FHIR server before trusting any search result. This is a GET;
      nothing is changed.
    uri: "{{ inputs.fhir_base_url }}/metadata?_elements=fhirVersion,software,kind"
    method: GET
    headers:
      Accept: application/fhir+json
      Authorization: "Bearer {{ secret('FHIR_ACCESS_TOKEN') }}"
      Cache-Control: no-cache
    options:
      allowFailed: true
      timeout:
        connectTimeout: PT10S
        readIdleTimeout: PT30S
    retry:
      type: constant
      interval: PT15S
      maxAttempts: 3
    allowFailure: true

  - id: search_pending_reports
    type: io.kestra.plugin.core.http.Request
    description: Search DiagnosticReport for reports in the category that are not
      final yet, sorted by clinically relevant date, returning the accurate
      total and only the oldest report.
    runIf: "{{ outputs.read_capability_statement.code is defined and
      outputs.read_capability_statement.code == 200 and
      (outputs.read_capability_statement.body | trim | startsWith('{')) }}"
    uri: "{{ render(vars.pending_search)
      }}&_sort=date&_count=1&_total=accurate&_elements=code,status,effective"
    method: GET
    headers:
      Accept: application/fhir+json
      Authorization: "Bearer {{ secret('FHIR_ACCESS_TOKEN') }}"
      Cache-Control: no-cache
    options:
      allowFailed: true
      timeout:
        connectTimeout: PT10S
        readIdleTimeout: PT60S
    retry:
      type: constant
      interval: PT15S
      maxAttempts: 3
    allowFailure: true

  - id: count_stale_reports
    type: io.kestra.plugin.core.http.Request
    description: Count the same reports whose clinically relevant date is older than
      stale_hours, with _summary=count so no clinical data is returned.
    runIf: "{{ outputs.read_capability_statement.code is defined and
      outputs.read_capability_statement.code == 200 and
      (outputs.read_capability_statement.body | trim | startsWith('{')) }}"
    uri: "{{ render(vars.pending_search) }}&date=lt{{ now() |
      dateAdd(-inputs.stale_hours, 'HOURS') | date(\"yyyy-MM-dd'T'HH:mm:ss'Z'\",
      timeZone='UTC') }}&_summary=count"
    method: GET
    headers:
      Accept: application/fhir+json
      Authorization: "Bearer {{ secret('FHIR_ACCESS_TOKEN') }}"
      Cache-Control: no-cache
    options:
      allowFailed: true
      timeout:
        connectTimeout: PT10S
        readIdleTimeout: PT60S
    retry:
      type: constant
      interval: PT15S
      maxAttempts: 3
    allowFailure: true

  - id: count_critical_reports
    type: io.kestra.plugin.core.http.Request
    description: Count the reports whose clinically relevant date is older than
      critical_hours, with _summary=count.
    runIf: "{{ outputs.read_capability_statement.code is defined and
      outputs.read_capability_statement.code == 200 and
      (outputs.read_capability_statement.body | trim | startsWith('{')) }}"
    uri: "{{ render(vars.pending_search) }}&date=lt{{ now() |
      dateAdd(-inputs.critical_hours, 'HOURS') |
      date(\"yyyy-MM-dd'T'HH:mm:ss'Z'\", timeZone='UTC') }}&_summary=count"
    method: GET
    headers:
      Accept: application/fhir+json
      Authorization: "Bearer {{ secret('FHIR_ACCESS_TOKEN') }}"
      Cache-Control: no-cache
    options:
      allowFailed: true
      timeout:
        connectTimeout: PT10S
        readIdleTimeout: PT60S
    retry:
      type: constant
      interval: PT15S
      maxAttempts: 3
    allowFailure: true

  - id: classify_results_backlog
    type: io.kestra.plugin.core.debug.Return
    description: Combine the four responses in one jq query. Anything other than a
      CapabilityStatement from /metadata and a searchset Bundle with a total
      from each search (a missing response, a non-200 code, a body that is not
      JSON such as an HTML login page, an OperationOutcome) becomes UNREACHABLE
      instead of looking like an empty backlog.
    format: >-
      {{ (outputs.read_capability_statement.code is not defined or
      outputs.read_capability_statement.code != 200 or
      outputs.search_pending_reports is not defined or
      outputs.search_pending_reports.code is not defined or
      outputs.search_pending_reports.code != 200 or outputs.count_stale_reports
      is not defined or outputs.count_stale_reports.code is not defined or
      outputs.count_stale_reports.code != 200 or outputs.count_critical_reports
      is not defined or outputs.count_critical_reports.code is not defined or
      outputs.count_critical_reports.code != 200 or not
      (outputs.read_capability_statement.body | trim | startsWith('{')) or not
      (outputs.search_pending_reports.body | trim | startsWith('{')) or not
      (outputs.count_stale_reports.body | trim | startsWith('{')) or not
      (outputs.count_critical_reports.body | trim | startsWith('{'))) ?
      '{"status":"UNREACHABLE","pending":0,"stale":0,"critical":0,"oldest":"none"}'
      : ({'m': fromJson(outputs.read_capability_statement.body), 'p':
      fromJson(outputs.search_pending_reports.body), 's':
      fromJson(outputs.count_stale_reports.body), 'c':
      fromJson(outputs.count_critical_reports.body)} | jq(' if .m.resourceType
      != "CapabilityStatement" or .p.resourceType != "Bundle" or .s.resourceType
      != "Bundle" or .c.resourceType != "Bundle" or (.p.total | type) !=
      "number" or (.s.total | type) != "number" or (.c.total | type) != "number"
      then {status: "UNREACHABLE", pending: 0, stale: 0, critical: 0, oldest:
      "none"} else (.p.entry[0].resource // null) as $o | {fhir_version:
      .m.fhirVersion, pending: .p.total, stale: .s.total, critical: .c.total,
      oldest: (if $o == null then "none" else (($o.code.text //
      $o.code.coding[0].display // $o.code.coding[0].code // "report") + " (" +
      $o.status + ", collected " + ($o.effectiveDateTime //
      $o.effectivePeriod.start // "unknown") + ", DiagnosticReport/" + $o.id +
      ")") end), status: (if .c.total > 0 or .s.total >= ' ~
      inputs.critical_stale_count ~ ' then "CRITICAL" elif .s.total > 0 then
      "WARNING" else "HEALTHY" end)} end ') | first | toJson) }}

  - id: route_by_backlog
    type: io.kestra.plugin.core.flow.Switch
    description: Route to a quiet log when no report is stale, an informational
      Slack message when some are, or a Slack page when reports are critically
      late or the FHIR server could not be read.
    value: "{{ fromJson(outputs.classify_results_backlog.value).status }}"
    cases:
      "HEALTHY":
        - id: log_healthy
          type: io.kestra.plugin.core.log.Log
          description: Log a quiet confirmation when no pending report is older than
            stale_hours.
          message: "FHIR {{ inputs.category }} results healthy: {{
            fromJson(outputs.classify_results_backlog.value).pending }}
            report(s) not final yet, none older than {{ inputs.stale_hours }}
            hour(s)."
      "WARNING":
        - id: alert_warning_backlog
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Post an informational Slack message when some reports are still not
            final after stale_hours.
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {
              "text": "FHIR {{ inputs.category }} results WARNING: {{ fromJson(outputs.classify_results_backlog.value).stale }} of {{ fromJson(outputs.classify_results_backlog.value).pending }} unfinished report(s) are older than {{ inputs.stale_hours }} hour(s). Oldest: {{ fromJson(outputs.classify_results_backlog.value).oldest }}. Execution {{ execution.id }}."
            }
      "CRITICAL":
        - id: alert_critical_backlog
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Page Slack when reports are still not final after critical_hours,
            or too many are stale, which usually means an interface, analyzer or
            validation queue is stuck.
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {
              "text": "FHIR {{ inputs.category }} results CRITICAL: {{ fromJson(outputs.classify_results_backlog.value).critical }} report(s) still not final after {{ inputs.critical_hours }} hour(s), {{ fromJson(outputs.classify_results_backlog.value).stale }} after {{ inputs.stale_hours }} hour(s). Oldest: {{ fromJson(outputs.classify_results_backlog.value).oldest }}. Results may not be reaching clinicians. Execution {{ execution.id }}."
            }
      "UNREACHABLE":
        - id: alert_unreachable_server
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Page Slack when the FHIR server could not be read, did not answer
            like a FHIR server, or rejected the search.
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {
              "text": "FHIR {{ inputs.category }} results UNREACHABLE: {{ inputs.fhir_base_url }} did not return a CapabilityStatement and searchset Bundles (server down, expired access token, wrong base URL, or search rejected). The results backlog is unknown. Execution {{ execution.id }}."
            }
    defaults:
      - id: log_unclassified
        type: io.kestra.plugin.core.log.Log
        description: Log any classification outside the four known values, so a change
          in the jq query never fails silently.
        message: "FHIR pending lab results sentinel completed with an unexpected
          classification: {{ outputs.classify_results_backlog.value }}."

  - id: log_audit_result
    type: io.kestra.plugin.core.log.Log
    description: Record one audit line for every run, whichever branch fired, with
      the HTTP codes and the classification, so the check's own history stays
      queryable even if Slack delivery fails.
    message: "FHIR pending lab results audit: server={{ inputs.fhir_base_url }}
      category={{ inputs.category }} metadata_http={{
      outputs.read_capability_statement.code ?? 'n/a' }} pending_http={{
      outputs.search_pending_reports.code ?? 'n/a' }} stale_http={{
      outputs.count_stale_reports.code ?? 'n/a' }} critical_http={{
      outputs.count_critical_reports.code ?? 'n/a' }} result={{
      outputs.classify_results_backlog.value }} execution={{ execution.id }}."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert Slack when the sentinel flow itself fails outright, so a
      broken check is never mistaken for an empty results backlog.
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {
        "text": "FHIR pending lab results sentinel flow {{ flow.id }} FAILED (execution {{ execution.id }}). The {{ inputs.category }} results backlog at {{ inputs.fhir_base_url }} is unknown; check the task logs and the FHIR server manually."
      }

outputs:
  - id: status
    type: STRING
    value: "{{ outputs.classify_results_backlog is defined ?
      fromJson(outputs.classify_results_backlog.value).status : 'UNREACHABLE'
      }}"
  - id: backlog
    type: JSON
    value: "{{ outputs.classify_results_backlog is defined ?
      outputs.classify_results_backlog.value : '{\"status\":\"UNREACHABLE\"}'
      }}"