1195 move EC from API to lambdas

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

@@ -31,6 +31,7 @@ export async function checkStaleEnhancedCoverage(cxIds: string[]): Promise<void>
  cxId,
  patientIds: patientsOfCx.map(p => p.id),
  cqLinkStatus: "linked",
+ context: "checkStaleEnhancedCoverage",
  });
  }
 }

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

@@ -17,26 +17,46 @@ export const completeEnhancedCoverage = async ({
  cxId,
  patientIds,
  cqLinkStatus,
+ startedAt,
+ context,
 }: {
  cxId: string;
  patientIds: string[];
  cqLinkStatus: CQLinkStatus;
+ startedAt?: number;
+ /**
+ * Context/operation in which this function is being called
+ */
+ context?: string;
 }): Promise<void> => {
- const { log } = out(`EC completer - cx ${cxId}`);
- log(
- `Completing EC for ${patientIds.length} patients, to status: ${cqLinkStatus}, and triggering doc queries`
- );
+ const startedAtLocal = Date.now();
+ const { log } = out(`EC completer - cx ${cxId}, ctxt ${context ?? "n/a"}`);
+ 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 = startedAt ? Date.now() - startedAt : undefined;
+ const durationMin = duration ? dayjs.duration(duration).asMinutes() : undefined;
+ const durationLocal = Date.now() - startedAtLocal;
+ const durationLocalMin = dayjs.duration(durationLocal).asMinutes();
+ log(
+ `Done, total duration: ${duration} ms / ${durationMin} min ` +
+ `(just to complete: ${durationLocal} ms / ${durationLocalMin} min)`
+ );
+ }
 };
 
 /**
@@ -49,17 +69,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-api-local.ts

@@ -2,7 +2,7 @@ import { CoverageEnhancementParams } from "@metriport/core/external/commonwell/c
 import { CoverageEnhancerLocal } from "@metriport/core/external/commonwell/cq-bridge/coverage-enhancer-local";
 import { CommonWellManagementAPI } from "@metriport/core/external/commonwell/management/api";
 import { out } from "@metriport/core/util/log";
-import { sleep } from "@metriport/core/util/sleep";
+import { sleep } from "@metriport/shared";
 import dayjs from "dayjs";
 import duration from "dayjs/plugin/duration";
 import { capture } from "../../../shared/notifications";
@@ -46,10 +46,12 @@ export class CoverageEnhancerApiLocal extends CoverageEnhancerLocal {
  log(`Giving some time for patients to be updated @ CW... (${waitTime} ms)`);
  await sleep(waitTime);
 
- await completeEnhancedCoverage({ cxId, patientIds, cqLinkStatus: "linked" });
-
- const duration = Date.now() - startedAt;
- const durationMin = dayjs.duration(duration).asMinutes();
- log(`Done, total time: ${duration} ms / ${durationMin} min`);
+ await completeEnhancedCoverage({
+ cxId,
+ patientIds,
+ cqLinkStatus: "linked",
+ startedAt,
+ context: "enhanceCoverage",
+ });
  }
 }

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

@@ -1,19 +1,24 @@
 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 { PatientLoaderLocal } from "../patient-loader-local";
 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,
+ new PatientLoaderLocal()
+ );
+ }
 
  const cwManagementUrl = Config.getCWManagementUrl();
  if (!cwManagementUrl) {

packages/api/src/routes/medical/internal-patient.ts

@@ -404,6 +404,7 @@ const completeEnhancedCoverageSchema = z.object({
  cxId: uuidSchema,
  patientIds: uuidSchema.array(),
  cqLinkStatus: cqLinkStatusSchema,
+ startedAt: z.number().nullish(),
 });
 
 /** ---------------------------------------------------------------------------
@@ -420,13 +421,16 @@ const completeEnhancedCoverageSchema = z.object({
 router.post(
  "/enhance-coverage/completed",
  asyncHandler(async (req: Request, res: Response) => {
- const { cxId, patientIds, cqLinkStatus } = completeEnhancedCoverageSchema.parse(req.body);
+ const { cxId, patientIds, cqLinkStatus, startedAt } = completeEnhancedCoverageSchema.parse(
+ req.body
+ );
 
  // intentionally async, no need to wait for it
  completeEnhancedCoverage({
  cxId,
  patientIds,
  cqLinkStatus,
+ startedAt: startedAt ?? undefined,
  }).catch(error => {
  console.log(
  `Failed to set cqLinkStatus for patients ${patientIds.join(", ")} - ${errorToString(error)}`

packages/core/src/domain/patient/patient-updater-metriport-api.ts

@@ -5,7 +5,7 @@ import { PatientUpdater } from "./patient-updater";
 
 dayjs.extend(duration);
 
-const UPDATE_TIMEOUT = dayjs.duration({ minutes: 2 });
+const UPDATE_TIMEOUT = dayjs.duration({ minutes: 3 });
 
 /**
  * Implementation of the PatientUpdater that calls the Metriport API

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,47 +31,53 @@ 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, startedAt);
+
+ 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[]) {
+ private async sendEnhancedCoverageDone(cxId: string, patientIds: string[], startedAt: number) {
  const payload: Input = {
  cxId,
  patientIds,
  done: true,
+ startedAt,
  };
  await this.sendMessageToQueue(cxId, payload);
  }

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

@@ -2,7 +2,7 @@ import { PatientLoader } from "../../../domain/patient/patient-loader";
 import { CQOrgHydrated, getOrgChunksFromPos, getOrgsByPrio, OrgPrio } from "./get-orgs";
 
 // Try to keep it even to make testing easier
-export const defaultMaxOrgsToProcess = 350;
+export const defaultMaxOrgsToProcess = 2500;
 
 export type CoverageEnhancementParams = {
  cxId: string;

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

@@ -1,5 +1,10 @@
 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;
+ startedAt?: number; // when the process started
+ } & ChunkProgress);

packages/core/src/external/commonwell/management/api.ts

@@ -124,12 +124,14 @@ export class CommonWellManagementAPI {
  oid,
  careQualityOrgIds,
  timeout = DEFAULT_TIMEOUT_INCLUDE_LIST.asMilliseconds(),
+ dryRun = false,
  log = console.log,
  debug = emptyFunction,
  }: {
  oid: string;
  careQualityOrgIds: string[];
  timeout?: number;
+ dryRun?: boolean | undefined;
  log?: typeof console.log | undefined;
  debug?: typeof console.log | undefined;
  }): Promise<void> {
@@ -139,6 +141,11 @@ export class CommonWellManagementAPI {
  return;
  }
 
+ if (dryRun) {
+ log(`[DRY RUN] Would be posting to /IncludeList and updating cookies, skipping...`);
+ return;
+ }
+
  log(`Posting to /IncludeList...`);
  const before = Date.now();
  const resp = await axios.post(