Collect data

src/scripts/uploadPremadeBundles.js

@@ -4,7 +4,7 @@ const mongoUtil = require('../util/mongo');
 const { createResource } = require('../util/mongo.controller');
 const { createPatientGroupsPerMeasure } = require('../util/groupUtils');
 
-const ecqmContentR4Path = path.resolve(path.join(__dirname, '../../ecqm-content-r4-2021/bundles/measure/'));
+const ecqmContentR4Path = path.resolve(path.join(__dirname, '../../ecqm-content-qicore-2025/bundles/measure/'));
 
 // files containing EXM bundles of interest from specified directory
 const bundleFiles = [];
@@ -57,7 +57,7 @@ const getBundleFiles = (directory, searchPattern) => {
  * Uploads all the resources from the specified directory into the
  * database.
  *
- * TODO: Currently configured for ecqm-content-r4-2021 measure bundles,
+ * TODO: Currently configured for ecqm-content-qicore-2025 measure bundles,
  * but may want to expand to other measure bundle providers in the future.
  */
 async function main() {
@@ -80,18 +80,18 @@ async function main() {
       throw new Error('Provided directory not found.');
     }
 
-    // otherwise load from ecqm-content-r4-2021
+    // otherwise load from ecqm-content-qicore-2025
   } else {
     try {
       if (!searchPattern) {
         // default searchPattern to retrieve all filenames that begin with a capital letter and end with -bundle.json
         searchPattern = /^[A-Z].*-bundle.json$/;
       }
-      console.log(`Finding bundles in ecqm-content-r4-2021 repo at ${ecqmContentR4Path}.`);
+      console.log(`Finding bundles in ecqm-content-qicore-2025 repo at ${ecqmContentR4Path}.`);
       getEcqmBundleFiles(ecqmContentR4Path, searchPattern);
     } catch {
       throw new Error(
-        'ecqm-content-r4-2021 directory not found. Git clone the ecqm-content-r4-2021 repo into the root directory and run script again'
+        'ecqm-content-qicore-2025 directory not found. Git clone the ecqm-content-qicore-2025 repo into the root directory and run script again'
       );
     }
   }

src/server/app.js

@@ -1,7 +1,7 @@
 const fastify = require('fastify');
 const cors = require('@fastify/cors');
 
-const { bulkExport, patientBulkExport, groupBulkExport } = require('../services/export.service');
+const { bulkExport, patientBulkExport, groupBulkExport, collectData } = require('../services/export.service');
 const { checkBulkStatus, kickoffImport } = require('../services/bulkstatus.service');
 const { returnNDJsonContent } = require('../services/ndjson.service');
 const { groupSearchById, groupSearch, groupCreate, groupUpdate, groupRemove } = require('../services/group.service');
@@ -39,6 +39,10 @@ function build(opts) {
   app.post('/Patient', patientCreate);
   app.put('/Patient/:patientId', patientUpdate);
   app.delete('/Patient/:patientId', patientRemove);
+  app.get('/Measure/$collect-data', collectData);
+  app.post('/Measure/$collect-data', collectData);
+  app.get('/Measure/:measureId/$collect-data', collectData);
+  app.post('/Measure/:measureId/$collect-data', collectData);
 
   return app;
 }

src/services/export.service.js

@@ -6,6 +6,12 @@ const patientResourceTypes = Object.keys(patientAttributePaths);
 const { createOperationOutcome } = require('../util/errorUtils');
 const { verifyPatientsInGroup, actualizeGroup } = require('../util/groupUtils');
 const { gatherParams } = require('../util/serviceUtils');
+const _ = require('lodash');
+const {
+  createDataExchangeMeasureReport,
+  createPatientBundle,
+  findPatientResources
+} = require('../util/collectDataUtils');
 
 /**
  * Exports data from a FHIR server, whether or not it is associated with a patient.
@@ -417,4 +423,122 @@ async function validatePatientReferences(patientParam, reply) {
   return false;
 }
 
-module.exports = { bulkExport, patientBulkExport, groupBulkExport };
+/**
+ * Implements limited parameters for $collect-data according to https://hl7.org/fhir/us/davinci-deqm/STU5/OperationDefinition-collect-data.html
+ * Returns a set of bundles that have data of interest for the specified measures, organized by the specified subject
+ * @param {Object} request the request object passed in by the user
+ * @param {Object} reply the response object
+ */
+const collectData = async (request, reply) => {
+  const parameters = gatherParams(request.method, request.query, request.body, reply);
+
+  if (validateCollectDataParams(parameters, reply)) {
+    request.log.info('Measure >>> $collect-data');
+
+    const patientIds = [parameters.subject.split('Patient/')[1]];
+    // Check for measure resolution - errors if there are any issues with measures passed
+    const measureArr = Array.isArray(parameters.measureId) ? parameters.measureId : [parameters.measureId];
+    const measurePromises = measureArr.map(async id => {
+      const measure = await findResourceById(id, 'Measure');
+      if (!measure) {
+        reply.code(404).send(new Error(`Unable to find measure with measureId ${id}`));
+      }
+      return measure;
+    });
+    const measures = await Promise.all(measurePromises);
+
+    const bundles = await Promise.all(
+      patientIds.map(async id => {
+        const patient = await findResourceById(id, 'Patient');
+        const resourcesMRPairs = await Promise.all(
+          measures.map(async measure => {
+            const patientResources = await findPatientResources(patient, measure);
+            const measureReport = createDataExchangeMeasureReport(
+              measure,
+              {
+                start: parameters.periodStart,
+                end: parameters.periodEnd
+              },
+              id,
+              patientResources
+            );
+            return [patientResources, measureReport];
+          })
+        );
+        const [patientResourcesArray, measureReports] = _.unzip(resourcesMRPairs);
+        const uniqueResources = _.uniqBy(
+          patientResourcesArray.flat(),
+          resource => `${resource.resourceType}/${resource.id}`
+        );
+        return createPatientBundle(patient, uniqueResources, measureReports);
+      })
+    );
+
+    reply.code(200).send(bundles);
+  }
+};
+
+/**
+ * Checks that the parameters input to $collect-data are valid. Returns true if all the
+ * export params are valid, meaning no errors were thrown in the process.
+ * @param {Object} parameters object containing a combination of request parameters from request query and body
+ * @param {Object} reply the response object
+ */
+function validateCollectDataParams(parameters, reply) {
+  let unrecognizedParams = [];
+  Object.keys(parameters).forEach(param => {
+    if (
+      ![
+        'periodStart',
+        'periodEnd',
+        'measureId',
+        'measureIdentifier',
+        'measureUrl',
+        'measureResource',
+        'measure',
+        'subject',
+        'subjectGroup',
+        'practitioner',
+        'lastReceivedOn',
+        'organizationResource',
+        'organization',
+        'validateResources',
+        'dataEndpoint'
+      ].includes(param)
+    ) {
+      unrecognizedParams.push(param);
+    }
+  });
+  if (unrecognizedParams.length > 0) {
+    reply
+      .code(400)
+      .send(
+        createOperationOutcome(
+          `The following parameters are unrecognized by the server: ${unrecognizedParams.join(', ')}.`,
+          { issueCode: 400, severity: 'error' }
+        )
+      );
+    return false;
+  }
+
+  let unsupportedParams = [];
+  Object.keys(parameters).forEach(param => {
+    if (!['periodStart', 'periodEnd', 'measureId', 'subject'].includes(param)) {
+      unsupportedParams.push(param);
+    }
+  });
+  if (unsupportedParams.length > 0) {
+    reply
+      .code(501)
+      .send(
+        createOperationOutcome(
+          `The following parameters are not yet supported by the server: ${unsupportedParams.join(', ')}.`,
+          { issueCode: 501, severity: 'error' }
+        )
+      );
+    return false;
+  }
+  return true;
+}
+
+module.exports = { bulkExport, patientBulkExport, groupBulkExport, collectData };

src/util/collectDataUtils.js

@@ -0,0 +1,175 @@
+const { v4: uuidv4 } = require('uuid');
+const { patientAttributePaths } = require('fhir-spec-tools/build/data/patient-attribute-paths');
+const patientResourceTypes = Object.keys(patientAttributePaths);
+const { addTypeFilter, getDocuments } = require('./exportToNDJson');
+const _ = require('lodash');
+
+/**
+ * Creates a patient bundle resource. Creates using a patient resource and
+ * an array of the patient's associated resources
+ * @param {Object} patient FHIR Patient object
+ * @param {Array} resources array of resources associated with the patient
+ * @param {Array} measureReports array of MeasureReport resources specifying measure information for data exchange
+ * @returns {Object} a FHIR patient bundle resource
+ */
+function createPatientBundle(patient, resources, measureReports) {
+  const bundle = {
+    type: 'transaction',
+    resourceType: 'Bundle',
+    id: uuidv4(),
+    entry: []
+  };
+  resources.forEach(r => {
+    bundle.entry?.push({
+      resource: r,
+      request: {
+        method: 'PUT',
+        url: `${r.resourceType}/${r.id}`
+      },
+      fullUrl: r.fullUrl ?? `urn:uuid:${r.id}`
+    });
+  });
+
+  measureReports.forEach(measureReport => {
+    bundle.entry?.push({
+      resource: measureReport,
+      request: {
+        method: 'PUT',
+        url: `MeasureReport/${measureReport.id}`
+      },
+      fullUrl: measureReport.fullUrl ?? `urn:uuid:${measureReport.id}`
+    });
+  });
+
+  return bundle;
+}
+
+/**
+ * Creates a FHIR data exchange MeasureReport from measure and subject data
+ * https://hl7.org/fhir/us/davinci-deqm/STU5/StructureDefinition-datax-measurereport-deqm.html
+ * @param measure FHIR Measure
+ * @param measurementPeriod FHIR Period representing the measurement period
+ * @param subjectId the patient id the MeasureReport is associated with
+ * @param patientResources the patient resources of relevance to the passed measure
+ * @returns { fhir4.MeasureReport } a data exchange measure report used to send Measure-relevant data to a server
+ */
+function createDataExchangeMeasureReport(measure, measurementPeriod, subjectId, patientResources) {
+  return {
+    resourceType: 'MeasureReport',
+    id: uuidv4(),
+    measure: measure.url?.includes('|') ? measure.url : `${measure.url}|${measure.version}`, //canonical measure/version
+    period: measurementPeriod,
+    status: 'complete',
+    type: 'data-collection',
+    subject: { reference: `Patient/${subjectId}` },
+    date: new Date().toISOString(),
+    reporter: { reference: 'Organization/bulk-export-server' }, //TODO: do we need to send an organization resource?
+    meta: {
+      profile: ['http://hl7.org/fhir/us/davinci-deqm/StructureDefinition/datax-measurereport-deqm']
+    },
+    extension: [
+      {
+        url: 'http://hl7.org/fhir/us/davinci-deqm/StructureDefinition/extension-submitDataUpdateType',
+        valueCode: 'snapshot'
+      }
+    ],
+    evaluatedResource: patientResources?.map(r => {
+      return { reference: `${r.resourceType}/${r.id}` };
+    }),
+    contained: [{ resourceType: 'Organization', id: 'bulk-export-server' }]
+  };
+}
+
+/**
+ * Finds all resources related to the passed patient resource but limited by the data requirements on the passed measure
+ * @param {FHIR.Patient} patient fhir patient resource
+ * @param {FHIR.Measure} measure fhir measure resource
+ * @returns {Array} an array of filtered fhir resources related to the passed patient
+ */
+async function findPatientResources(patient, measure) {
+  const dataRequirements = measure.contained?.find(c => c.id === 'effective-data-requirements').dataRequirement;
+  const types = _.uniq(dataRequirements.map(dr => dr.type));
+  const [patientTypes, nonPatientTypes] = types.reduce(
+    ([patient, nonPatient], type) => {
+      (patientResourceTypes.includes(type) ? patient : nonPatient).push(type);
+      return [patient, nonPatient];
+    },
+    [[], []]
+  );
+  // nonPatientTypes can be in data requirements when they may be referenced from other resources that reference patients
+  // i.e. Patient references MedicationRequest references Medication could end up with Medication as non-patient type
+  // ignored for now, but should either get all (non-patient query) or be able to do more complex query creation
+  if (nonPatientTypes.length > 0) {
+    console.warn('Ignoring non-patient types found in data requirements:', nonPatientTypes);
+  }
+  const typeFilters = typeFiltersForMeasure(dataRequirements);
+  // create lookup objects for (1) _typeFilter queries that contain search parameters, and (2) _typeFilter
+  // queries that contain type:in/code:in/etc. queries
+  const searchParameterQueries = {};
+  const valueSetQueries = {};
+  if (typeFilters) {
+    addTypeFilter(typeFilters, searchParameterQueries, valueSetQueries);
+  }
+
+  // for each patient, collect resource documents from each export type collection
+  const typeDocs = patientTypes.map(async collectionName => {
+    return (
+      await getDocuments(collectionName, searchParameterQueries[collectionName], valueSetQueries[collectionName], [
+        patient.id
+      ])
+    ).document;
+  });
+
+  //flatten all type arrays into a single array
+  return (await Promise.all(typeDocs)).flat();
+}
+
+function typeFiltersForMeasure(dataRequirements) {
+  // create record resourcetype => [] of valid typeFilter query strings that will be separated by "&" and logically handled as ORs
+  const typeFilters = {};
+  dataRequirements.forEach(dr => {
+    //empty array is general _type query that overrides a more specific _typeFilter
+    if (typeFilters[dr.type]?.length === 0) return; // only handle data requirements with types for now
+
+    if (
+      dr.codeFilter?.some(cf => {
+        const hasVS = cf.path && cf.valueSet;
+        const hasCode = cf.path && cf.code && cf.code.every(coding => !!coding.code);
+        const hasNeither = !hasVS && !hasCode;
+        return hasNeither;
+      })
+    ) {
+      // if any codeFilter for a data requirement doesn't have sufficient information, default to getting all for that type (general _type query)
+      typeFilters[dr.type] = [];
+      return;
+    }
+    //all codefilters have a path and proper code and can be added to our typefilter array
+    const fhirQueries = dr.codeFilter?.map(cf => {
+      const hasVS = cf.path && cf.valueSet;
+      const hasCode = cf.path && cf.code && cf.code.every(coding => !!coding.code);
+      if (hasVS && hasCode) {
+        // default to valueset method for now, but this should probably be expanded to a full list of codes from the vs with additional
+        // codes appended (or separated into separate top-level queries where other included codeFilters are repeated
+        // Example: 'Procedure?code=1&category=3,4','Procedure?code=1&category:in=vs')
+        return `${cf.path}:in=${cf.valueSet}`;
+      } else if (hasVS) {
+        return `${cf.path}:in=${cf.valueSet}`;
+      } else {
+        // hasCode
+        // potential multiple codes are comma-separated to be OR'd for this path
+        return `${cf.path}=${cf.code?.map(coding => coding.code).join(',')}`;
+      }
+    });
+    const tfStr = `${dr.type}?${fhirQueries?.join('&')}`; //Example value: 'Procedure?code=1,2&category=3,4'
+    if (typeFilters[dr.type]) {
+      typeFilters[dr.type].push(tfStr);
+    } else {
+      typeFilters[dr.type] = [tfStr];
+    }
+  });
+
+  // array of typeFilters that should be treated as OR'd (empty array will be ignored and flattened)
+  return Object.values(typeFilters).flat();
+}
+
+module.exports = { createPatientBundle, createDataExchangeMeasureReport, findPatientResources };

src/util/exportToNDJson.js

@@ -428,7 +428,7 @@ const processVSTypeFilter = async function (valueSetQueries) {
         let vs = await findOneResourceWithQuery({ url: value }, 'ValueSet');
         // throw an error if we don't have the value set
         if (!vs) {
-          throw new Error('Value set was not found in the database');
+          throw new Error(`Value set with value ${value} was not found in the database`);
         }
         const vsResolved = getCodesFromValueSet(vs);
         // extract the property (i.e. code, type)
@@ -440,7 +440,9 @@ const processVSTypeFilter = async function (valueSetQueries) {
       await Promise.all(results);
     }
   }
-
+  if (queryArray.length === 0) {
+    return {};
+  }
   return {
     $or: queryArray
   };