build: move EC from API to lambdas

Ref: metriport/metriport-internal#1195

packages/api/src/external/commonwell/cq-bridge/coverage-enhancement-complete.ts

@@ -22,21 +22,29 @@ export const completeEnhancedCoverage = async ({
  patientIds: string[];
  cqLinkStatus: CQLinkStatus;
 }): Promise<void> => {
+ const startedAt = Date.now();
  const { log } = out(`EC completer - cx ${cxId}`);
- log(
- `Completing EC for ${patientIds.length} patients, to status: ${cqLinkStatus}, and triggering doc queries`
- );
+ try {
+ log(
+ `Completing EC for ${patientIds.length} patients, to status: ${cqLinkStatus}, ` +
+ `and triggering doc queries - patients: ${patientIds.join(", ")}`
+ );
 
- // Promise that will be executed for each patient
- const completeECForPatient = async (patientId: string): Promise<void> => {
- const { patient, updated } = await setCQLinkStatus({ cxId, patientId, cqLinkStatus });
- if (!updated) return; // if the status was already set don't do anything else
- if (cqLinkStatus === "linked") await finishEnhancedCoverage(patient, log);
- };
+  // Promise that will be executed for each patient
+  const completeECForPatient = async (patientId: string): Promise<void> => {
+  const { patient, updated } = await setCQLinkStatus({ cxId, patientId, cqLinkStatus });
+  if (!updated) return; // if the status was already set don't do anything else
+  if (cqLinkStatus === "linked") await finishEnhancedCoverage(patient, log);
+  };
 
- await executeAsynchronously(patientIds, completeECForPatient, {
- numberOfParallelExecutions: PARALLEL_UPDATES,
- });
+ await executeAsynchronously(patientIds, completeECForPatient, {
+ numberOfParallelExecutions: PARALLEL_UPDATES,
+ });
+ } finally {
+ const duration = Date.now() - startedAt;
+ const durationMin = dayjs.duration(duration).asMinutes();
+ log(`Done, duration: ${duration} ms / ${durationMin} min`);
+ }
 };
 
 /**
@@ -49,17 +57,10 @@ async function finishEnhancedCoverage(patient: Patient, log = console.log): Prom
  throw new MetriportError(`Patient ${patient.id} has no facility`);
  }
 
- const startedAt = Date.now();
- try {
- const triggerDocRefs = new TriggerAndQueryDocRefsLocal();
- await triggerDocRefs.queryDocsForPatient({
- cxId: patient.cxId,
- patientId: patient.id,
- triggerWHNotificationsToCx,
- });
- } finally {
- const duration = Date.now() - startedAt;
- const durationMin = dayjs.duration(duration).asMinutes();
- log(`Done DQ, duration: ${duration} ms / ${durationMin} min`);
- }
+ const triggerDocRefs = new TriggerAndQueryDocRefsLocal();
+ await triggerDocRefs.queryDocsForPatient({
+ cxId: patient.cxId,
+ patientId: patient.id,
+ triggerWHNotificationsToCx,
+ });
 }

packages/api/src/external/commonwell/cq-bridge/coverage-enhancement-init.ts

@@ -3,7 +3,7 @@ import { MetriportError } from "@metriport/core/util/error/metriport-error";
 import { groupBy } from "lodash";
 import { getOrganizationOrFail } from "../../../command/medical/organization/get-organization";
 import { getPatients } from "../../../command/medical/patient/get-patient";
-import { getPatientsToEnhanceCoverage } from "./coverage-enhancement-get-patients";
+import { PatientToLink, getPatientsToEnhanceCoverage } from "./coverage-enhancement-get-patients";
 import { makeCoverageEnhancer } from "./coverage-enhancer-factory";
 import { setCQLinkStatus } from "./cq-link-status";
 
@@ -25,18 +25,7 @@ export async function initEnhancedCoverage(
  return;
  }
 
- const getPatientsToProcess = async () => {
- if (patientIds && patientIds.length > 0) {
- const cxId = cxIds[0];
- if (cxIds.length != 1 || !cxId) {
- throw new MetriportError(`Exacly one cxId must be set when patientIds are present`);
- }
- return getPatients({ cxId, patientIds });
- }
- return getPatientsToEnhanceCoverage(cxIds);
- };
-
- const patients = await getPatientsToProcess();
+ const patients = await getPatientsToProcess(cxIds, patientIds);
  const patientsByCx = groupBy(patients, "cxId");
 
  const entries = Object.entries(patientsByCx);
@@ -65,3 +54,17 @@ export async function initEnhancedCoverage(
  });
  }
 }
+
+async function getPatientsToProcess(
+ cxIds: string[],
+ patientIds?: string[]
+): Promise<PatientToLink[]> {
+ if (patientIds && patientIds.length > 0) {
+ const cxId = cxIds[0];
+ if (cxIds.length != 1 || !cxId) {
+ throw new MetriportError(`Exactly one cxId must be set when patientIds are present`);
+ }
+ return getPatients({ cxId, patientIds });
+ }
+ return getPatientsToEnhanceCoverage(cxIds);
+}

packages/api/src/external/commonwell/cq-bridge/coverage-enhancer-factory.ts

@@ -1,19 +1,19 @@
 import { CookieManagerOnSecrets } from "@metriport/core/domain/auth/cookie-management/cookie-manager-on-secrets";
 import { CoverageEnhancer } from "@metriport/core/external/commonwell/cq-bridge/coverage-enhancer";
-// import { CoverageEnhancerCloud } from "@metriport/core/external/commonwell/cq-bridge/coverage-enhancer-cloud";
+import { CoverageEnhancerCloud } from "@metriport/core/external/commonwell/cq-bridge/coverage-enhancer-cloud";
 import { CommonWellManagementAPI } from "@metriport/core/external/commonwell/management/api";
 import { Config } from "../../../shared/config";
 import { CoverageEnhancerApiLocal } from "./coverage-enhancer-api-local";
 
 export function makeCoverageEnhancer(): CoverageEnhancer | undefined {
- // if (Config.isCloudEnv()) {
- //  const cwPatientLinkQueueUrl = Config.getCWPatientLinkQueueUrl();
- //  if (!cwPatientLinkQueueUrl) {
- //  console.log(`Could not return a CoverageEnhancer, mising cwPatientLinkQueueUrl`);
- //  return undefined;
- //  }
- //  return new CoverageEnhancerCloud(Config.getAWSRegion(), cwPatientLinkQueueUrl);
- // }
+ if (Config.isCloudEnv()) {
+ const cwPatientLinkQueueUrl = Config.getCWPatientLinkQueueUrl();
+ if (!cwPatientLinkQueueUrl) {
+ console.log(`Could not return a CoverageEnhancer, mising cwPatientLinkQueueUrl`);
+ return undefined;
+ }
+ return new CoverageEnhancerCloud(Config.getAWSRegion(), cwPatientLinkQueueUrl);
+ }
 
  const cwManagementUrl = Config.getCWManagementUrl();
  if (!cwManagementUrl) {

packages/core/src/external/commonwell/cq-bridge/coverage-enhancer-cloud.ts

@@ -1,8 +1,14 @@
+import dayjs from "dayjs";
+import duration from "dayjs/plugin/duration";
 import { PatientLoader } from "../../../domain/patient/patient-loader";
 import { sha256 } from "../../../util/hash";
+import { out } from "../../../util/log";
 import { SQSClient } from "../../aws/sqs";
+import { LinkPatientsCommand } from "../management/link-patients";
 import { CoverageEnhancementParams, CoverageEnhancer } from "./coverage-enhancer";
-import { Input } from "./cq-link-patients";
+import { ChunkProgress, Input } from "./cq-link-patients";
+
+dayjs.extend(duration);
 
 /**
  * Implementation of the Enhanced Coverage flow with the logic running on AWS lambdas.
@@ -25,40 +31,45 @@ export class CoverageEnhancerCloud extends CoverageEnhancer {
  patientIds,
  fromOrgChunkPos = 0,
  }: CoverageEnhancementParams) {
- const { chunks } = await this.getCarequalityOrgs({ cxId, patientIds, fromOrgChunkPos });
-
- // Each chunk of CQ orgs
- for (const cqOrgList of chunks) {
- await this.sendEnhancedCoverageByCxAndChunk({
+ const startedAt = Date.now();
+ const { log } = out(`EC - cloud - cx ${cxId}`);
+ try {
+ const { total, chunks } = await this.getCarequalityOrgs({
  cxId,
- orgOID,
  patientIds,
- cqOrgIds: cqOrgList.map(o => o.id),
+ fromOrgChunkPos,
  });
+
+ log(`CQ orgs: ${total}, chunks: ${chunks.length}/${chunks.length + fromOrgChunkPos}`);
+ log(`patients: ${patientIds.join(", ")}`);
+
+ // Each chunk of CQ orgs
+ for (const [i, cqOrgList] of chunks.entries()) {
+ await this.sendEnhancedCoverageByCxAndChunk({
+ cxId,
+ cxOrgOID: orgOID,
+ patientIds,
+ cqOrgIds: cqOrgList.map(o => o.id),
+ chunkIndex: i,
+ chunkTotal: chunks.length,
+ });
+ }
+ } finally {
+ await this.sendEnhancedCoverageDone(cxId, patientIds);
+
+ const duration = Date.now() - startedAt;
+ const durationMin = dayjs.duration(duration).asMinutes();
+ log(`Time to send SQS messages: ${duration} ms / ${durationMin} min`);
  }
- await this.sendEnhancedCoverageDone(cxId, patientIds);
  }
 
  // for each patientId, send a message to SQS with the patientId and the orgChunks
- private async sendEnhancedCoverageByCxAndChunk({
- cxId,
- orgOID,
- patientIds,
- cqOrgIds,
- }: {
- cxId: string;
- orgOID: string;
- patientIds: string[];
- cqOrgIds: string[];
- }) {
+ private async sendEnhancedCoverageByCxAndChunk(params: LinkPatientsCommand & ChunkProgress) {
  const payload: Input = {
- cxId,
- cxOrgOID: orgOID,
- patientIds,
- cqOrgIds,
+ ...params,
  done: false,
  };
- await this.sendMessageToQueue(cxId, payload);
+ await this.sendMessageToQueue(params.cxId, payload);
  }
 
  private async sendEnhancedCoverageDone(cxId: string, patientIds: string[]) {

packages/core/src/external/commonwell/cq-bridge/cq-link-patients.ts

@@ -1,5 +1,7 @@
 import { LinkPatientsCommand } from "../management/link-patients";
 
+export type ChunkProgress = { chunkIndex?: number | undefined; chunkTotal?: number | undefined };
+
 export type Input =
- | (LinkPatientsCommand & { done: false })
- | (Pick<LinkPatientsCommand, "cxId" | "patientIds"> & { done: true });
+ | (LinkPatientsCommand & { done: false } & ChunkProgress)
+ | (Pick<LinkPatientsCommand, "cxId" | "patientIds"> & { done: true } & ChunkProgress);

packages/infra/lib/api-stack.ts

@@ -353,7 +353,7 @@ export class APIStack extends Stack {
  bucket: generalBucket,
  alarmSnsAction: slackNotification?.alarmAction,
  });
- // const cqLinkPatientQueue = cwEnhancedQueryQueues?.linkPatientQueue;
+ const linkPatientsQueue = cwEnhancedQueryQueues?.linkPatientsQueue;
  const cookieStore = cwEnhancedQueryQueues?.cookieStore;
 
  //-------------------------------------------
@@ -392,7 +392,7 @@ export class APIStack extends Stack {
  configId: appConfigConfigId,
  cxsWithEnhancedCoverageFeatureFlag,
  },
- // cqLinkPatientQueue
+ linkPatientsQueue,
  cookieStore
  );
 
@@ -480,8 +480,7 @@ export class APIStack extends Stack {
  alarmSnsAction: slackNotification?.alarmAction,
  });
 
- // cqLinkPatientQueue &&
- cookieStore &&
+ if (cookieStore && linkPatientsQueue) {
  cwEnhancedCoverageConnector.setupLambdas({
  stack: this,
  vpc: this.vpc,
@@ -491,9 +490,10 @@ export class APIStack extends Stack {
  apiAddress: apiLoadBalancerAddress,
  bucket: generalBucket,
  alarmSnsAction: slackNotification?.alarmAction,
- // linkPatientQueue: cqLinkPatientQueue,
+ linkPatientsQueue,
  cookieStore,
  });
+ }
 
  //-------------------------------------------
  // API Gateway

packages/infra/lib/api-stack/api-service.ts

@@ -56,7 +56,7 @@ export function createAPIService(
  configId: string;
  cxsWithEnhancedCoverageFeatureFlag: string;
  },
- // cqLinkPatientQueue: IQueue | undefined
+ linkPatientsQueue: IQueue | undefined,
  cookieStore: secret.ISecret | undefined
 ): {
  cluster: ecs.Cluster;
@@ -163,9 +163,9 @@ export function createAPIService(
  APPCONFIG_CONFIGURATION_ID: appConfigEnvVars.configId,
  CXS_WITH_ENHANCED_COVERAGE_FEATURE_FLAG:
  appConfigEnvVars.cxsWithEnhancedCoverageFeatureFlag,
- // ...(cqLinkPatientQueue && {
- // CW_CQ_PATIENT_LINK_QUEUE_URL: cqLinkPatientQueue.queueUrl,
- // }),
+ ...(linkPatientsQueue && {
+  CW_CQ_PATIENT_LINK_QUEUE_URL: linkPatientsQueue.queueUrl,
+ }),
  ...(coverageEnhancementConfig && {
  CW_MANAGEMENT_URL: coverageEnhancementConfig.managementUrl,
  }),
@@ -216,12 +216,12 @@ export function createAPIService(
  queue: sidechainFHIRConverterDLQ,
  resource: fargateService.service.taskDefinition.taskRole,
  });
- // cqLinkPatientQueue &&
- //  provideAccessToQueue({
- //  accessType: "send",
- //  queue: cqLinkPatientQueue,
- //  resource: fargateService.service.taskDefinition.taskRole,
- //  });
+ linkPatientsQueue &&
+ provideAccessToQueue({
+ accessType: "send",
+ queue: linkPatientsQueue,
+ resource: fargateService.service.taskDefinition.taskRole,
+ });
  if (cookieStore) {
  cookieStore.grantRead(fargateService.service.taskDefinition.taskRole);
  cookieStore.grantWrite(fargateService.service.taskDefinition.taskRole);