metriport · leite08 · May 31, 2025 · May 31, 2025 · May 31, 2025 · May 31, 2025
@@ -4,10 +4,6 @@ import { parseFhirBundle, SearchSetBundle } from "@metriport/shared/medical";
 import { createConsolidatedDataFilePath } from "../../domain/consolidated/filename";
 import { Patient } from "../../domain/patient";
 import { executeWithRetriesS3, returnUndefinedOn404, S3Utils } from "../../external/aws/s3";
-import {
-  bundleToString,
-  FhirResourceToText,
-} from "../../external/fhir/export/string/bundle-to-string";
 import { out } from "../../util";
 import { Config } from "../../util/config";
 import { processAsyncError } from "../../util/error/shared";
@@ -138,20 +134,3 @@ export async function getConsolidatedPatientDataAsync({
     .execute(payload)
     .catch(processAsyncError("Failed to get consolidated patient data async", undefined, true));
 }
-
-/**
- * Get consolidated patient data as a list of FHIR resources in string/text format.
- *
- * The output is focused on searching, not a complete representation of the FHIR resources.
- *
- * @param patient The patient to get the consolidated data for.
- * @returns The consolidated data as text.
- */
-export async function getConsolidatedAsText({
-  patient,
-}: {
-  patient: Patient;
-}): Promise<FhirResourceToText[]> {
-  const consolidated = await getConsolidatedPatientData({ patient });
-  return bundleToString(consolidated);
-}
@@ -1,9 +1,12 @@
+import { Bundle, Resource } from "@medplum/fhirtypes";
+import { MetriportError } from "@metriport/shared";
 import { timed } from "@metriport/shared/util/duration";
 import { Patient } from "../../../../domain/patient";
+import { normalize } from "../../../../external/fhir/consolidated/normalize";
 import { OpenSearchFhirIngestor } from "../../../../external/opensearch/fhir-ingestor";
 import { OnBulkItemError } from "../../../../external/opensearch/shared/bulk";
 import { capture, out } from "../../../../util";
-import { getConsolidatedPatientData } from "../../consolidated-get";
+import { getConsolidatedFile } from "../../consolidated-get";
 import { getConfigs } from "./fhir-config";
 
 /**
@@ -17,15 +20,20 @@ export async function ingestPatientConsolidated({
 }: {
   patient: Patient;
   onItemError?: OnBulkItemError;
-}) {
-  const { log } = out(`ingestLexical - cx ${patient.cxId}, pt ${patient.id}`);
+}): Promise<void> {
+  const { cxId, id: patientId } = patient;
+  const { log } = out(`ingestPatientConsolidated - cx ${cxId}, pt ${patientId}`);
 
   const ingestor = new OpenSearchFhirIngestor(getConfigs());
 
   log("Getting consolidated and cleaning up the index...");
   const [bundle] = await Promise.all([
-    timed(() => getConsolidatedPatientData({ patient }), "getConsolidatedPatientData", log),
-    ingestor.delete({ cxId: patient.cxId, patient
10000
Id: patient.id }),
+    timed(
+      () => getConsolidatedBundle({ cxId, patientId }),
+      "getConsolidatedBundleAndNotifyWhenMissing",
+      log
+    ),
+    ingestor.delete({ cxId, patientId }),
   ]);
 
   const resources =
@@ -39,14 +47,14 @@ export async function ingestPatientConsolidated({
   log("Done, calling ingestBulk...");
   const startedAt = Date.now();
   const errors = await ingestor.ingestBulk({
-    cxId: patient.cxId,
-    patientId: patient.id,
+    cxId,
+    patientId,
     resources,
     onItemError,
   });
   const elapsedTime = Date.now() - startedAt;
 
-  if (errors.size > 0) captureErrors({ cxId: patient.cxId, patientId: patient.id, errors, log });
+  if (errors.size > 0) captureErrors({ cxId, patientId, errors, log });
 
   log(`Ingested ${resources.length} resources in ${elapsedTime} ms`);
 }
@@ -68,3 +76,29 @@ function captureErrors({
     extra: { cxId, patientId, countPerErrorType: JSON.stringify(errorMapToObj, null, 2) },
   });
 }
+
+async function getConsolidatedBundle({
+  cxId,
+  patientId,
+}: {
+  cxId: string;
+  patientId: string;
+}): Promise<Bundle<Resource>> {
+  const { log } = out(`getConsolidatedBundle - cx ${cxId}, pt ${patientId}`);
+
+  const consolidated = await getConsolidatedFile({ cxId, patientId });
+
+  const bundle = consolidated.bundle;
+  if (!bundle) {
+    const bucket = consolidated.fileLocation;
+    const key = consolidated.fileName;
+    const msg = `No consolidated bundle found during ingestion in OS`;
+    log(`${msg} for patient ${patientId} w/ key ${key}, skipping ingestion`);
+    throw new MetriportError(msg, undefined, { cxId, patientId, key, bucket });
+  }
+
+  // TODO ENG-316 Remove this step when we implement normalization on consolidated
+  const normalizedBundle = await normalize({ cxId, patientId, bundle });
+
+  return normalizedBundle;
+}
@@ -1,6 +1,5 @@
 import {
   AllergyIntolerance,
-  Bundle,
   Communication,
   Composition,
   Condition,
@@ -119,30 +118,14 @@ export type FhirResourceToText = {
 };
 
 /**
- * Converts a FHIR Bundle to a list of string representations of its resources.
- * Focused on searching, so not a complete representation of the FHIR resources.
+ * Converts a FHIR resource to a string representation.
+ * Focused on searching, so not a complete representation of the FHIR resource.
  * Skips unsupported resources - @see isSupportedResource
  *
- * @param bundle - FHIR Bundle to convert
+ * @param resource - FHIR resource to convert
  * @param isDebug - Whether to include debug information in the output
- * @returns List of string representations of the resources in the bundle
+ * @returns String representation of the resource
  */
-export function bundleToString(bundle: Bundle, isDebug = false): FhirResourceToText[] {
-  if (!bundle.entry?.length) return [];
-
-  const resp = bundle.entry.flatMap(entry => {
-    const resource = entry.resource;
-    if (!resource || !resource.id) return [];
-    const text = resourceToString(resource, isDebug);
-    if (!text) return [];
-    return {
-      id: resource.id,
-      type: resource.resourceType,
-      text,
-    };
-  });
-  return resp;
-}
 
 export function resourceToString(resource: Resource, isDebug?: boolean): string | undefined {
   if (!isSupportedResource(resource)) return undefined;

@@ -4,7 +4,7 @@ import { buildCompleteBundleEntry, extractFhirTypesFromBundle } from "../shared/
 import { sortCodings } from "./coding";
 import { normalizeConditions } from "./resources/condition";
 import { normalizeCoverages } from "./resources/coverage";
-import { normalizeEncounters } from "./resources/encounter";
+import { filterInvalidEncounters } from "./resources/encounter";
 import { normalizeObservations } from "./resources/observation";
 
 /**
@@ -33,11 +33,11 @@ export function normalizeFhir(fhirBundle: Bundle<Resource>): Bundle<Resource> {
   const normalizedConditions = normalizeConditions(resourceArrays.conditions);
   resourceArrays.conditions = normalizedConditions;
 
-  const normalizedEncounters = normalizeEncounters(
+  const validEncounters = filterInvalidEncounters(
     resourceArrays.encounters,
     resourceArrays.locations
   );
-  resourceArrays.encounters = normalizedEncounters;
+  resourceArrays.encounters = validEncounters;
 
   normalizedBundle.entry = Object.entries(resourceArrays).flatMap(([, resources]) => {
     const entriesArray = Array.isArray(resources) ? resources : [resources];

@@ -3,8 +3,9 @@ import { cloneDeep } from "lodash";
 import { ICD_10_URL } from "../../../../util/constants";
 import { chronicityMap } from "../../shared/chronicity-map";
 import {
-  ChronicityExtension,
   buildChronicityExtension,
+  ChronicityExtension,
+  findChronicityExtension,
 } from "../../shared/extensions/chronicity-extension";
 
 export function normalizeConditions(conditions: Condition[]): Condition[] {
@@ -18,6 +19,10 @@ export function normalizeConditions(conditions: Condition[]): Condition[] {
       if (coding.code) {
         const chronicity = chronicityMap[coding.code.replace(".", "")];
         if (chronicity) {
+          const existingChronicityExtension = findChronicityExtension(updCondition.extension ?? []);
+          if (existingChronicityExtension) {
+            return;
+          }
           chronicityExtension = buildChronicityExtension(chronicity);
         }
       }

@@ -3,7 +3,10 @@ import { toTitleCase } from "@metriport/shared";
 import { getValidCodings } from "../../codeable-concept";
 import { buildReferenceFromStringRelative } from "../../shared/bundle";
 
-export function normalizeEncounters(encounters: Encounter[], locations: Location[]): Encounter[] {
+export function filterInvalidEncounters(
+  encounters: Encounter[],
+  locations: Location[]
+): Encounter[] {
   return encounters.flatMap(encounter => {
     const hasReason = hasEncounterReason(encounter);
     const hasLocation = hasEncounterLocation(encounter, locations);

@@ -214,7 +214,6 @@ export class LambdasNestedStack extends NestedStack {
       lambdaLayers: props.lambdaLayers,
       vpc: props.vpc,
       envType: props.config.environmentType,
-      fhirToBundleLambda: this.fhirToBundleLambda,
       fhirServerUrl: props.config.fhirServerUrl,
       bundleBucket: props.medicalDocumentsBucket,
       openSearchEndpoint: props.openSearch.endpoint,
@@ -232,7 +231,6 @@ export class LambdasNestedStack extends NestedStack {
       lambdaLayers: props.lambdaLayers,
       vpc: props.vpc,
       envType: props.config.environmentType,
-      fhirToBundleLambda: this.fhirToBundleLambda,
       bundleBucket: props.medicalDocumentsBucket,
       openSearchEndpoint: props.openSearch.endpoint,
       openSearchAuth: props.openSearch.auth,
@@ -707,7 +705,6 @@ export class LambdasNestedStack extends NestedStack {
     lambdaLayers: LambdaLayers;
     vpc: ec2.IVpc;
     envType: EnvType;
-    fhirToBundleLambda: lambda.Function;
     fhirServerUrl: string;
     bundleBucket: s3.IBucket;
     featureFlagsTable: dynamodb.Table;
@@ -726,7 +723,6 @@ export class LambdasNestedStack extends NestedStack {
       lambdaLayers,
       vpc,
       envType,
-      fhirToBundleLambda,
       fhirServerUrl,
       bundleBucket,
       featureFlagsTable,
@@ -756,7 +752,6 @@ export class LambdasNestedStack extends NestedStack {
         SEARCH_INDEX: openSearchDocumentsIndexName,
         CONSOLIDATED_SEARCH_INDEX: openSearchConsolidatedIndexName,
         CONSOLIDATED_INGESTION_INITIAL_DATE: consolidatedDataIngestionInitialDate,
-        FHIR_TO_BUNDLE_LAMBDA_NAME: fhirToBundleLambda.functionName,
         ...(sentryDsn ? { SENTRY_DSN: sentryDsn } : {}),
       },
       layers: [lambdaLayers.shared, lambdaLayers.langchain],
@@ -768,7 +763,6 @@ export class LambdasNestedStack extends NestedStack {
     });
 
     bundleBucket.grantReadWrite(theLambda);
-    fhirToBundleLambda.grantInvoke(theLambda);
     openSearchAuth.secret.grantRead(theLambda);
     featureFlagsTable.grantReadData(theLambda);
 
@@ -798,7 +792,6 @@ export class LambdasNestedStack extends NestedStack {
     lambdaLayers: LambdaLayers;
     vpc: ec2.IVpc;
     envType: EnvType;
-    fhirToBundleLambda: lambda.Function;
     bundleBucket: s3.IBucket;
     featureFlagsTable: dynamodb.Table;
     openSearchEndpoint: string;
@@ -815,7 +808,6 @@ export class LambdasNestedStack extends NestedStack {
       lambdaLayers,
       vpc,
       envType,
-      fhirToBundleLambda,
       bundleBucket,
       featureFlagsTable,
       openSearchEndpoint,
@@ -839,7 +831,6 @@ export class LambdasNestedStack extends NestedStack {
         SEARCH_PASSWORD_SECRET_ARN: openSearchAuth.secret.secretArn,
         SEARCH_INDEX: openSearchDocumentsIndexName,
         CONSOLIDATED_SEARCH_INDEX: openSearchConsolidatedIndexName,
-        FHIR_TO_BUNDLE_LAMBDA_NAME: fhirToBundleLambda.functionName,
         ...(sentryDsn ? { SENTRY_DSN: sentryDsn } : {}),
       },
       layers: [lambdaLayers.shared, lambdaLayers.langchain],
@@ -854,7 +845,6 @@ export class LambdasNestedStack extends NestedStack {
     );
 
     bundleBucket.grantReadWrite(theLambda);
-    fhirToBundleLambda.grantInvoke(theLambda);
     openSearchAuth.secret.grantRead(theLambda);
     featureFlagsTable.grantReadData(theLambda);
 

@@ -1,12 +1,10 @@
 import * as dotenv from "dotenv";
 dotenv.config();
 // keep that ^ on top
-import { MetriportMedicalApi } from "@metriport/api-sdk";
-import { getConsolidatedAsText } from "@metriport/core/command/consolidated/consolidated-get";
-import { getDomainFromDTO } from "@metriport/core/command/patient-loader-metriport-api";
+import { Bundle, Resource } from "@medplum/fhirtypes";
 import {
-  bundleToString,
   FhirResourceToText,
+  resourceToString,
 } from "@metriport/core/external/fhir/export/string/bundle-to-string";
 import { getFileContents } from "@metriport/core/util/fs";
 import { sleep } from "@metriport/core/util/sleep";
@@ -33,8 +31,6 @@ dayjs.extend(duration);
 
 const bundleFilePath: string | undefined = undefined;
 const patientId = getEnvVar("PATIENT_ID");
-const apiKey = getEnvVar("API_KEY");
-const apiUrl = getEnvVar("API_URL");
 const cxId = getEnvVar("CX_ID");
 
 const outputRootFolderName = `semantic-text-from-consolidated`;
@@ -53,18 +49,6 @@ async function main() {
     const bundleContents = getFileContents(bundleFilePath);
     const bundle = JSON.parse(bundleContents);
     resources = bundleToString(bundle);
-  } else {
-    if (!apiKey || !apiUrl || !patientId || !cxId) {
-      throw new Error("Environment variables must be set if bundleFilePath is not set");
-    }
-    console.log("Getting consolidated resources from the API...");
-    const metriportAPI = new MetriportMedicalApi(apiKey, {
-      baseAddress: apiUrl,
-      timeout: 120_000,
-    });
-    const patientDto = await metriportAPI.getPatient(patientId);
-    const patient = getDomainFromDTO(patientDto, cxId);
-    resources = await getConsolidatedAsText({ patient });
   }
 
   const headers = ["id", "type", "text"];
@@ -76,6 +60,32 @@ async function main() {
   console.log(`>>> Output file: ${outputFilePath}`);
 }
 
+/**
+ * Converts a FHIR Bundle to a list of string representations of its resources.
+ * Focused on searching, so not a complete representation of the FHIR resources.
+ * Skips unsupported resources - @see isSupportedResource
+ *
+ * @param bundle - FHIR Bundle to convert
+ * @param isDebug - Whether to include debug information in the output
+ * @returns List of string representations of the resources in the bundle
+ */
+function bundleToString(bundle: Bundle<Resource>, isDebug = false): FhirResourceToText[] {
+  if (!bundle.entry?.length) return [];
+
+  const resp = bundle.entry.flatMap(entry => {
+    const resource = entry.resource;
+    if (!resource || !resource.id) return [];
+    const text = resourceToString(resource, isDebug);
+    if (!text) return [];
+    return {
+      id: resource.id,
+      type: resource.resourceType,
+      text,
+    };
+  });
+  return resp;
+}
+
 if (require.main === module) {
   main();
 }