Data Acquisition Service

Overview

The Data Acquisition service is responsible for connecting and querying a tenant's endpoint for FHIR resources that are needed to evaluate patients for a measure. For Epic installations, Link Cloud is utilizing the Epic FHIR STU3 Patient List resource to inform which patients are currently admitted in the facility. While this is the current solution to acquiring the patient census, there are other means of patient acquisition being investigated (ADT V2, Bulk FHIR) to provide universal support across multiple EHR vendors.

In addition to FHIR endpoints, the service supports SFTP-based acquisition for tenants that drop extract files (e.g., Cerner CCL pipe-delimited extracts) onto a remote SFTP server rather than exposing a queryable FHIR API. See SFTP Data Acquisition below.

flowchart LR
nDataAcquisitionService_3BF82DDB["Service: Data Acquisition Service"]
  nPatientCensusScheduled_2383C0C6["Event: PatientCensusScheduled"] -->|consumed by| nDataAcquisitionService_3BF82DDB
  nDataAcquisitionRequested_4BBEB8F8["Command: DataAcquisitionRequested"] -->|handled by| nDataAcquisitionService_3BF82DDB
  nDataAcquisitionService_3BF82DDB -->|provides| nGetFacilityConfig_32C58A79["Query: GetFacilityConfig"]
  nDataAcquisitionService_3BF82DDB -->|provides| nGetServiceInfo_13F02E06["Query: GetServiceInfo"]
  subgraph nDataAccess_2A523756_domain["Domain: Data Access"]
    nDataAcquisitionRequested_4BBEB8F8
    nDataAcquisitionService_3BF82DDB
  end
  subgraph nReport_4D4FBBDA_domain["Domain: Report"]
    nPatientCensusScheduled_2383C0C6
  end
  subgraph nTenant_379CD514_domain["Domain: Tenant"]
    nGetFacilityConfig_32C58A79
    nGetServiceInfo_13F02E06
  end

Common Configurations

• Swagger
• Azure App Configuration
• Kafka Configuration
• Kafka Consumer Retry Configuration
• Service Registry Configuration
• CORS Configuration
• Token Service Configuration
• Service Authentication
• SQL Server Database Configuration
• Data Acquisition Service Configuration
• SFTP Acquisition Settings

Features and Functionality

Data Acquisition is a crucial step in the report generation pipeline, responsible for obtaining clinical data from external systems, such as FHIR R4 endpoints within electronic health record (EHR) systems. By systematically acquiring and managing data, it ensures that downstream processes like evaluation and reporting are equipped with the necessary information.

Key Roles of Data Acquisition

1. Census Acquisition:

• Retrieves recent patient admits from the EHR.
• Serves as the initial step for identifying patients relevant to quality measure evaluations.

2. Individual Patient Data Acquisition:

• Acquires detailed FHIR data for individual patients identified in the census.
• Includes all essential data elements required for initial measure evaluations.

3. Supplemental Data Acquisition:

• Retrieves additional data elements that may not be needed for initial evaluations but are desirable for complete submission and reporting.

Where Data Acquisition Fits in the Pipeline

Data acquisition plays a role in three distinct stages of the report generation pipeline:

1. Patient Identification:

• Acquiring a list of patients from the EHR to determine the cohort for evaluation.

2. Initial Data Collection and Evaluation:

• Obtaining the primary dataset needed for evaluating quality measures.
• Determines whether a patient qualifies for reporting based on initial measure criteria.

3. Supplemental Data Collection:

• Acquiring additional, non-essential data to enrich the submission.
• Completes the dataset for comprehensive evaluation and reporting.

Progressive Querying

To optimize data acquisition, the system employs a technique called Progressive Querying.

During progressive querying, data is acquired in stages to meet the evaluation pipeline's needs. It flows between services via Kafka topics/events, starting from Data Acquisition Worker Service to normalization, through initial evaluation to determine patient relevance, and back to acquire supplemental data, which is then normalized and re-evaluated.

This method minimizes the data retrieved from the EHR by acquiring only what is necessary at each stage of the pipeline:

1. Initial Querying: Focuses on essential data needed to evaluate measures and determine patient inclusion.
2. Supplemental Querying: Retrieves additional data elements after initial evaluations confirm patient relevance.
3. Final Evaluation: Combines initial and supplemental data for comprehensive measure evaluation and reporting.

Benefits of Progressive Querying

• Efficiency: Reduces the volume of data retrieved from the EHR, optimizing system performance.
• Precision: Focuses on acquiring only data that is needed for specific stages in the pipeline.
• Scalability: Supports large-scale operations by limiting unnecessary data transfers.

Bulk FHIR in Data Acquisition

Bulk FHIR is a mechanism under exploration for acquiring data efficiently. However, several limitations impact its general use in data acquisition workflows:

1. Challenges with Bulk FHIR:

• Most implementations lack sufficient support for acquiring specific patient data.
• Filtering returned data is often not robust enough.
• To align with the goal of acquiring only necessary data, the system does not currently implement Bulk FHIR for general initial or supplemental data acquisition.

2. Use Cases for Bulk FHIR:

• Patient Census Identification:
• Bulk FHIR is a viable solution for identifying the "census of patients," analogous to using the FHIR "List" endpoint.
• It can acquire "Patient" resources for a group of patients associated with a query, filter, or registry in the EHR.
• This use case is limited to identifying patients of interest and does not address broader data acquisition.

Configuration for Data Acquisition

Data acquisition is configurable per tenant, ensuring flexibility to accommodate diverse EHR systems and data requirements. Key configurable parameters include:

• Base FHIR URL:

  • For general data acquisition.
  • For FHIR List (patients of interest) retrieval.

• Authentication Information:

  • Configured per facility on the FhirQueryConfiguration (data queries) and FhirListConfiguration (patient census) entities. Each entity holds an optional Authentication block, managed via api/data/{facilityId}/{configType}/authentication (GET/POST/PUT/DELETE, IsLinkAdmin policy). configType is fhirQueryConfiguration or fhirQueryListConfiguration.
  • Secrets are never stored on the configuration row. Credential-bearing fields (ClientId, ClientSecret, UserName, Password, and CustomHeaders values) hold secret manager key names. Actual secrets are resolved at runtime via the ISecretManager interface against whichever backend the deployment is configured to use.
  • Successful token responses for Epic and OAuth flows are cached in-memory per FacilityId for the lifetime of expires_in.

  Supported AuthType values and their required fields:

  AuthType	Description	Required Fields
  Epic	Epic FHIR JWT client assertion (RS256). Issues a signed JWT to TokenUrl.	TokenUrl, Audience, ClientId (secret key), Key (RSA private key, PEM with escaped newlines)
  Basic	HTTP Basic Authentication (Authorization: Basic <base64(user:pass)>).	UserName (secret key), Password (secret key)
  OAuth	OAuth 2.0 client_credentials grant against TokenUrl.	TokenUrl, ClientId (secret key), ClientSecret (secret key), Scope
  CustomHeaders	Static custom headers injected on each request (header value resolved from secret manager).	CustomHeaders (non-empty Dictionary<string, string>: header name → secret key)
  OpenIDConnect	Reserved in the AuthType enum but not yet implemented (no service handler).	—

• Patient Census Retrieval:

  • FHIR List "id" or Bulk FHIR "Group ID" used for identifying the patient cohort.

• EHR Query Throttling/Limitations:

  • Configurable settings to respect EHR query limitations (e.g., maximum queries per minute).

Configuring Query Plans

Types of query plans:

• Daily
• Weekly
• Monthly
• Discharge

"Discharge" query plans are used when a patient is discharged from the hospital. This plan is triggered by a discharge event and is used to acquire data for the patient.

All other types of query plans are used to acquire data for patients who are currently in the hospital triggered by the end date/time of the scheduled report. The tenant's timezone is used for this so that if the reporting period ends at 12:59:59 PM, that represents 12:59:59 PM in the tenant's timezone, not UTC time.

All times are stored in UTC format. The tenant's time zone is configured with a valid value from IANA.

Initial / Supplemental Queries

The previously mentioned progressive query phases (initial, supplemental) are configurable through the query plan. Each configured phase can contain a list of FHIR resources that must be acquired from the configured endpoint.

Query Types

For each FHIR resource that must be queried, there are two main query types that are supported:

1. Parameter: Parameters that will be appended to the FHIR search for the configured resource.
2. Reference: Any FHIR references for the configured resource found in other acquired resources for that phase will be queried for.

Example Plan

Below is an example of a monthly query plan that's configured to acquire the following resources:

1. Initial Query Phase:

2. Supplemental Query Phase:

{
  "PlanName": "NHSNdQMAcuteCareHospitalInitialPopulation",
  "FacilityId": "st-marys-hospital",
  "EHRDescription": "",
  "LookBack": "P0D",
  "Type": "Monthly",
  "InitialQueries": {
    "0": {
      "ResourceType": "Encounter",
      "QueryConfigType": "Parameter",
      "Parameters": [
        {
          "ParameterType": "Variable",
          "Name": "patient",
          "Variable": 0,
          "Format": null
        },
        {
          "ParameterType": "Variable",
          "Name": "date",
          "Variable": 1,
          "Format": "ge{0}"
        },
        {
          "ParameterType": "Variable",
          "Name": "date",
          "Variable": 3,
          "Format": "le{0}"
        }
      ]
    },
    "1": {
      "QueryConfigType": "Reference",
      "ResourceType": "Location",
      "OperationType": "SearchPost",
      "Paged": 100
    }
  },
  "SupplementalQueries": {
    "0": {
      "QueryConfigType": "Parameter",
      "ResourceType": "MedicationRequest",
      "Parameters": [
        {
          "ParameterType": "Variable",
          "Name": "patient",
          "Variable": 0,
          "Format": null
        },
        {
          "ParameterType": "Variable",
          "Name": "authoredon",
          "Variable": 1,
          "Format": "ge{0}"
        },
        {
          "ParameterType": "Variable",
          "Name": "authoredon",
          "Variable": 3,
          "Format": "le{0}"
        },
        {
          "ParameterType": "Literal",
          "Name": "intent",
          "Literal": "order"
        }
      ]
    },
    "1": {
      "QueryConfigType": "Reference",
      "ResourceType": "Medication",
      "OperationType": "SearchPost",
      "Paged": 100
    }
  }
}

Configuring Census and Data Sources

Data sources (where the FHIR server is located and how to authenticate) are configured via "Query Configs". There is currently no association between a query plan and data source. Whenever data acquisition attempts to execute a query plan against a data source, it uses the FHIR server and authentication method specified by the "Query Config", for the specified facility/tenant.

TODO: Add details about how to authenticate against Epic, Cerner, Basic, and/or OAuth data sources.

Query Plans and Acquisition Logs

The Data Acquisition Service uses Query Plans to define the strategy for retrieving data. These plans are translated into Data Acquisition Logs, which represent discrete units of work to be executed by the worker service.

Acquisition Log Lifecycle

Each acquisition task follows a strictly managed state machine:

1. Scheduled: The log entry is created but not yet ready for execution. This usually happens when a Data Acquisition Requested event is received.
2. Ready: The system has determined that the log is eligible for execution. A Ready to Acquire event is produced for the worker.
3. Queued: The worker has received the Ready to Acquire event and successfully "claimed" the log.
4. InProgress: The worker is actively querying the FHIR endpoint and processing resources.
5. Completed: All resources for the log have been successfully acquired and normalized.
6. Failed: An error occurred during acquisition that exceeded the maximum retry attempts.

Log Creation Process

When a Data Acquisition Requested event is received, the service:

1. Retrieves the appropriate Query Plan for the facility and report type.
2. Identifies the target patients (either from the event itself or by creating a Census log).
3. Generates a set of DataAcquisitionLog entries for each required resource type and patient, assigned to the Initial phase.

Execution and Dependency Management

The system determines when a log should be executed based on its Status and QueryPhase:

• Phase-Based Execution: Logs are typically executed in phases. Initial phase logs are created first. Once initial data is acquired and evaluated by downstream services, they may trigger Supplemental acquisition by sending new Data Acquisition Requested events with a supplemental phase flag.
• Wait Logic: Supplemental logs are not created until the initial evaluation confirms the patient's relevance to the report. This prevents unnecessary data retrieval for patients who do not meet the report's criteria.
• Retry Mechanism: If a log fails due to transient issues (e.g., network timeout), it is incremented and returned to a Pending or Scheduled state for retry, up to a maximum of 5 attempts.

Query Plan Structure

A Query Plan relates to logs by defining the "blueprint" for their creation:

• ResourceType: The FHIR resource to be queried (e.g., Patient, Encounter, Observation).
• QueryConfigType:
  • Parameter: Appends specific FHIR search parameters (e.g., date=ge2024-01-01).
  • Reference: Instructs the system to find references to this resource type within other already acquired resources.
• QueryPhase: Categorizes the query into Initial or Supplemental.

For more details on how these logs are processed, see the Data Acquisition Worker Service.

Tail Messages

Data Acquisition emits a tail Resource Acquired event for each (patientId, reportTrackingId, queryType) tuple when all resources for the phase are produced. This tail sets acquisitionComplete = true and signals downstream services to proceed. See:

• Events → Resource Acquired
• Docs → Tail Messages: Patient Completion Signals

FHIR List Acquisition

Epic EHR deployments will have FHIR Lists available for Link to track recently admitted and discharged patients within a facility. Currently, there are six separate FHIR List Id's that track the time difference since the patient was admitted:

List Acquisition Processing Sequence

The following demonstrates the workflow when the Data Acquisition service consumes a PatientCensusScheduled event for an Epic EHR facility:

sequenceDiagram
    participant CS as Census Service
    participant SQL as SQL Server
    participant K as Kafka
    participant DAS as Data Acquisition Service
    participant W as Data Acquisition Worker Service
    participant EHR as EHR (FHIR)

    CS->>K: Produces PatientCensusScheduled
    K->>DAS: Consumes PatientCensusScheduled
    DAS->>SQL: Create Log 
    DAS->>K: Produce ReadyToAcquire
    K->>W: Consumes ReadyToAcquire
    W->>EHR: Request FHIR List
    EHR-->>W: FHIR List Response
    W->>K: Produce PatientsListAcquired
    K->>CS: Consume PatientsListAcquired
    CS->>CS: Construct PatientEvent payload
    CS->>SQL: Insert into Data Source (immutable)
    CS->>CS: Build Materialized View (current state)
    CS->>SQL: Insert into Materialized View Store
    CS->>K: Produce PatientEvent (if patient is discharged)

SFTP Data Acquisition

Some EHR deployments (notably Cerner via CCL extracts) cannot expose a queryable FHIR endpoint for census or resource acquisition. For these tenants, Data Acquisition supports pulling extract files from a per-facility SFTP server, parsing them into Link's domain model, and emitting Kafka events that drive the rest of the report generation pipeline.

SFTP acquisition is implemented as an independent acquisition channel from FHIR querying. It does not consume DataAcquisitionRequested or produce ReadyToAcquire — instead, a background worker polls the SftpAcquisitionLog table on a fixed interval, opens an SFTP session per facility, and dispatches files to a processor selected by the log's AcquisitionType and SubType.

When to Use SFTP Acquisition

Architecture

flowchart LR
    subgraph Tenant["Tenant Environment"]
        EHR[(EHR /<br/>Extract Job)]
        SFTP[(SFTP Server)]
    end

    subgraph Link["Link Cloud — Data Acquisition"]
        Cfg[SftpConfiguration<br/>per facility]
        Sched[PatientCensusService<br/>creates SftpAcquisitionLog]
        Worker[SftpAcquisitionService<br/>background poller]
        Handler[SftpAcquisitionHandler<br/>batch + parallelism]
        Proc[ISftpAcquisitionProcessor<br/>e.g. CernerCCLExtractProcessor]
        Parser[IFileParser&lt;T&gt;<br/>e.g. CernerCclExtractParser]
        Secrets[(Secret Manager<br/>sftp-credentials-&#123;orgId&#125;)]
        Log[(SftpAcquisitionLog)]
    end

    subgraph Downstream["Downstream"]
        Kafka{{Kafka<br/>CernerPatientsAcquired}}
        Census[Census Service<br/>CernerPatientsAcquiredListener]
    end

    EHR -->|drops extract files| SFTP
    Sched -->|on PatientCensusScheduled| Log
    Worker -->|every JobIntervalSeconds| Log
    Worker --> Handler
    Handler -->|reads creds| Secrets
    Handler -->|reads server config| Cfg
    Handler -->|opens session per facility| SFTP
    Handler --> Proc
    Proc --> Parser
    Proc -->|publishes per file| Kafka
    Proc -->|move/delete file| SFTP
    Proc -->|update status & benchmarks| Log
    Kafka --> Census

Acquisition Lifecycle

Each SftpAcquisitionLog follows its own state machine, distinct from the FHIR DataAcquisitionLog lifecycle described above.

stateDiagram-v2
    [*] --> Pending : Census schedule creates log

    Pending --> Processing : Worker claims log
    Processing --> Processed : File downloaded,<br/>parsed, event published
    Processing --> Failed : Transient error<br/>(network, parse, EHR)

    Failed --> Pending : Retry eligible<br/>(exponential backoff)
    Failed --> MaxRetriesReached : RetryAttempts &gt; MaxRetryAttempts

    Processed --> [*]
    MaxRetriesReached --> [*]

Retries use exponential backoff between BaseRetryDelaySeconds and MaxRetryDelaySeconds (e.g., 60s, 120s, 240s, 480s, ... capped at 3600s). The log's RetryAttempts field is incremented on each transition through Failed; once it exceeds MaxRetryAttempts, the log is parked in MaxRetriesReached and excluded from future polls.

Per-File Processing Sequence

sequenceDiagram
    participant W as SftpAcquisitionService
    participant H as SftpAcquisitionHandler
    participant S as SftpSession (Renci.SshNet)
    participant SFTP as SFTP Server
    participant P as Processor
    participant FP as IFileParser
    participant K as Kafka
    participant DB as SftpAcquisitionLog

    loop every JobIntervalSeconds (default 30s)
        W->>H: ExecuteAsync()
        H->>DB: Get pending + retry-eligible logs
        DB-->>H: logs grouped by FacilityId

        loop per facility (parallel up to MaxConcurrency)
            H->>S: Open session (Host, Port, creds)
            S->>SFTP: Connect + authenticate
            loop per log in facility batch
                H->>P: ProcessWithSessionAsync(log, session)
                P->>S: ListFilesAsync(remoteDir, pattern)
                S->>SFTP: LIST
                SFTP-->>S: matching files
                P->>S: DownloadFileAsync(file)
                S->>SFTP: GET
                SFTP-->>P: file stream
                P->>FP: ParseAsync(stream, config)
                FP-->>P: parsed records
                P->>K: Produce CernerPatientsAcquired
                alt ProcessedDirectory configured
                    P->>S: MoveFileAsync(processedDir)
                else RemoveAfterProcessing = true
                    P->>S: DeleteFileAsync()
                end
                P->>DB: Status = Processed (+ benchmarks)
            end
            H->>S: Dispose session
        end
    end

Configuration

SFTP acquisition is configured at three layers: service-wide settings, per-facility connection / acquisition definitions, and secret-managed credentials.

SFTP Acquisition Settings

Worker-level tuning, sourced from the SftpAcquisition configuration section in appsettings.json / Azure App Configuration. These apply to the SftpAcquisitionService background host.

Example:

"SftpAcquisition": {
  "EnableParallelProcessing": true,
  "ParallelProcessingThreshold": 5,
  "MaxConcurrency": 3,
  "MaxBatchSize": 10,
  "JobIntervalSeconds": 30,
  "MaxRetryAttempts": 3,
  "BaseRetryDelaySeconds": 60,
  "MaxRetryDelaySeconds": 3600
}

A separate SftpValidation section (bound to SftpValidationSettings) holds validation rules applied to incoming SftpConfiguration payloads via the configuration API.

Per-Facility SFTP Configuration

Each facility that uses SFTP acquisition has exactly one SftpConfiguration row, managed via the /api/data/sftp-configurations endpoints on the Data Acquisition service. This describes how to connect to the tenant's SFTP server and what files to look for.

Each SftpAcquisitionTypeConfiguration describes a specific class of file:

FileParsingConfiguration (used by ConfigurableDelimitedParser):

Credentials

SFTP credentials are never stored on the SftpConfiguration row. They are persisted via the ISecretManager interface in whichever backend the deployment is configured to use, under the key:

sftp-credentials-{organizationId}

The payload is an SftpCredentialsModel (Username, Password). The configuration API exposes endpoints to set, retrieve status (without revealing the password), and delete credentials independently of the connection metadata.

Connection Test

A dedicated test endpoint (/api/data/sftp-configurations/{facilityId}/test) opens a live SFTP session using the saved configuration and credentials, lists the configured RemoteDirectory, and reports success or the underlying error. Use this when onboarding a tenant or after rotating credentials.

Kafka Events

SFTP acquisition emits events on a separate topic from FHIR acquisition. The current processor produces:

• CernerPatientsAcquired — published by CernerCCLExtractProcessor once per successfully parsed census file. Consumed by the Census service's CernerPatientsAcquiredListener, which translates encounters into PatientEvent admit/discharge messages that feed the standard census workflow.

SFTP processing intentionally does not emit PatientListsAcquired or ReadyToAcquire; those events remain specific to the FHIR-driven acquisition path.

Observability

When EnableBenchmarking is set on an SftpConfiguration, the worker writes an SftpAcquisitionBenchmark entry per processed file containing per-stage durations (connection, listing, download, parse, publish) and item counts. These are stored on the SftpAcquisitionLog and are queryable via /api/data/sftp-logs. The Notes field on the log captures human-readable error messages when a run fails.

Logging

The Logging configuration defines the logging levels for different parts of the application.

Database Schema

OpenAPI Operations

Relationships

flowchart LR
nDataAcquisitionService_3BF82DDB["Service: Data Acquisition Service"]
  nPatientCensusScheduled_2383C0C6["Event: PatientCensusScheduled"] -->|consumed by| nDataAcquisitionService_3BF82DDB
  nDataAcquisitionRequested_4BBEB8F8["Command: DataAcquisitionRequested"] -->|handled by| nDataAcquisitionService_3BF82DDB
  nDataAcquisitionService_3BF82DDB -->|provides| nGetFacilityConfig_32C58A79["Query: GetFacilityConfig"]
  nDataAcquisitionService_3BF82DDB -->|provides| nGetServiceInfo_13F02E06["Query: GetServiceInfo"]
  subgraph nDataAccess_2A523756_domain["Domain: Data Access"]
    nDataAcquisitionRequested_4BBEB8F8
    nDataAcquisitionService_3BF82DDB
  end
  subgraph nReport_4D4FBBDA_domain["Domain: Report"]
    nPatientCensusScheduled_2383C0C6
  end
  subgraph nTenant_379CD514_domain["Domain: Tenant"]
    nGetFacilityConfig_32C58A79
    nGetServiceInfo_13F02E06
  end