feat: enable unit testing subscription criteria (medplum#4581)

* feat: enable unit testing subscription criteria * Organized imports --------- Co-authored-by: Cody Ebberson <[email protected]>
dillonstreator · May 27, 2024 · 9f3bb56 · 9f3bb56
1 parent 41005df
commit 9f3bb56
Show file tree

Hide file tree

Showing 6 changed files with 191 additions and 99 deletions.
diff --git a/packages/core/src/subscriptions/index.test.ts b/packages/core/src/subscriptions/index.test.ts
@@ -1,6 +1,12 @@
-import { Bundle, Communication, Parameters, SubscriptionStatus } from '@medplum/fhirtypes';
+import { Bundle, Communication, Parameters, Subscription, SubscriptionStatus } from '@medplum/fhirtypes';
 import WS from 'jest-websocket-mock';
-import { RobustWebSocket, SubscriptionEmitter, SubscriptionEventMap, SubscriptionManager } from '.';
+import {
+ RobustWebSocket,
+ SubscriptionEmitter,
+ SubscriptionEventMap,
+ SubscriptionManager,
+ resourceMatchesSubscriptionCriteria,
+} from '.';
 import { MockMedplumClient } from '../client-test-utils';
 import { generateId } from '../crypto';
 import { OperationOutcomeError } from '../outcomes';
@@ -749,3 +755,53 @@ describe('SubscriptionManager', () => {
  });
  });
 });
+
+describe('resourceMatchesSubscriptionCriteria', () => {
+ it('should return true for a resource that matches the criteria', async () => {
+ const subscription: Subscription = {
+ resourceType: 'Subscription',
+ status: 'active',
+ reason: 'test subscription',
+ criteria: 'Communication',
+ channel: {
+ type: 'rest-hook',
+ endpoint: 'Bot/123',
+ },
+ extension: [
+ {
+ url: 'https://medplum.com/fhir/StructureDefinition/fhir-path-criteria-expression',
+ valueString: '%previous.status = "in-progress" and %current.status = "completed"',
+ },
+ {
+ url: 'https://medplum.com/fhir/StructureDefinition/subscription-supported-interaction',
+ valueCode: 'update',
+ },
+ ],
+ };
+
+ const result1 = await resourceMatchesSubscriptionCriteria({
+ resource: {
+ resourceType: 'Communication',
+ status: 'in-progress',
+ },
+ subscription,
+ context: { interaction: 'create' },
+ getPreviousResource: async () => undefined,
+ });
+ expect(result1).toBe(false);
+
+ const result2 = await resourceMatchesSubscriptionCriteria({
+ resource: {
+ resourceType: 'Communication',
+ status: 'completed',
+ },
+ subscription,
+ context: { interaction: 'update' },
+ getPreviousResource: async () => ({
+ resourceType: 'Communication',
+ status: 'in-progress',
+ }),
+ });
+ expect(result2).toBe(true);
+ });
+});
diff --git a/packages/core/src/subscriptions/index.ts b/packages/core/src/subscriptions/index.ts
@@ -1,8 +1,13 @@
-import { Bundle, Parameters, Subscription, SubscriptionStatus } from '@medplum/fhirtypes';
+import { Bundle, Parameters, Subscription, SubscriptionStatus, Resource } from '@medplum/fhirtypes';
 import { MedplumClient } from '../client';
 import { TypedEventTarget } from '../eventtarget';
 import { OperationOutcomeError, serverError, validationError } from '../outcomes';
-import { ProfileResource, getReferenceString, resolveId } from '../utils';
+import { ProfileResource, getExtension, getReferenceString, resolveId } from '../utils';
+import { Logger } from '../logger';
+import { matchesSearchRequest } from '../search/match';
+import { toTypedValue } from '../fhirpath/utils';
+import { evalFhirPathTyped } from '../fhirpath/parse';
+import { parseSearchRequest } from '../search/search';
 
 export type SubscriptionEventMap = {
  connect: { type: 'connect'; payload: { subscriptionId: string } };
@@ -355,3 +360,116 @@ export class SubscriptionManager {
  return this.masterSubEmitter;
  }
 }
+
+export type BackgroundJobInteraction = 'create' | 'update' | 'delete';
+
+export interface BackgroundJobContext {
+ interaction: BackgroundJobInteraction;
+}
+
+type ResourceMatchesSubscriptionCriteria = {
+ resource: Resource;
+ subscription: Subscription;
+ context: BackgroundJobContext;
+ logger?: Logger;
+ getPreviousResource: (currentResource: Resource) => Promise<Resource | undefined>;
+};
+
+export async function resourceMatchesSubscriptionCriteria({
+ resource,
+ subscription,
+ context,
+ getPreviousResource,
+ logger,
+}: ResourceMatchesSubscriptionCriteria): Promise<boolean> {
+ if (subscription.meta?.account && resource.meta?.account?.reference !== subscription.meta.account.reference) {
+ logger?.debug('Ignore resource in different account compartment');
+ return false;
+ }
+
+ if (!matchesChannelType(subscription, logger)) {
+ logger?.debug(`Ignore subscription without recognized channel type`);
+ return false;
+ }
+
+ const subscriptionCriteria = subscription.criteria;
+ if (!subscriptionCriteria) {
+ logger?.debug(`Ignore rest hook missing criteria`);
+ return false;
+ }
+
+ const searchRequest = parseSearchRequest(subscriptionCriteria);
+ if (resource.resourceType !== searchRequest.resourceType) {
+ logger?.debug(
+ `Ignore rest hook for different resourceType (wanted "${searchRequest.resourceType}", received "${resource.resourceType}")`
+ );
+ return false;
+ }
+
+ const fhirPathCriteria = await isFhirCriteriaMet(subscription, resource, getPreviousResource);
+ if (!fhirPathCriteria) {
+ logger?.debug(`Ignore rest hook for criteria returning false`);
+ return false;
+ }
+
+ const supportedInteractionExtension = getExtension(
+ subscription,
+ 'https://medplum.com/fhir/StructureDefinition/subscription-supported-interaction'
+ );
+ if (supportedInteractionExtension && supportedInteractionExtension.valueCode !== context.interaction) {
+ logger?.debug(
+ `Ignore rest hook for different interaction (wanted "${supportedInteractionExtension.valueCode}", received "${context.interaction}")`
+ );
+ return false;
+ }
+
+ return matchesSearchRequest(resource, searchRequest);
+}
+
+/**
+ * Returns true if the subscription channel type is ok to execute.
+ * @param subscription - The subscription resource.
+ * @param logger - The logger.
+ * @returns True if the subscription channel type is ok to execute.
+ */
+function matchesChannelType(subscription: Subscription, logger?: Logger): boolean {
+ const channelType = subscription.channel?.type;
+
+ if (channelType === 'rest-hook') {
+ const url = subscription.channel?.endpoint;
+ if (!url) {
+ logger?.debug(`Ignore rest-hook missing URL`);
+ return false;
+ }
+
+ return true;
+ }
+
+ if (channelType === 'websocket') {
+ return true;
+ }
+
+ return false;
+}
+
+export async function isFhirCriteriaMet(
+ subscription: Subscription,
+ currentResource: Resource,
+ getPreviousResource: (currentResource: Resource) => Promise<Resource | undefined>
+): Promise<boolean> {
+ const criteria = getExtension(
+ subscription,
+ 'https://medplum.com/fhir/StructureDefinition/fhir-path-criteria-expression'
+ );
+ if (!criteria?.valueString) {
+ return true;
+ }
+ const previous = await getPreviousResource(currentResource);
+ const evalInput = {
+ '%current': toTypedValue(currentResource),
+ '%previous': toTypedValue(previous ?? {}),
+ };
+ const evalValue = evalFhirPathTyped(criteria.valueString, [toTypedValue(currentResource)], evalInput);
+ console.log(evalValue);
+ return evalValue?.[0]?.value === true;
+}
diff --git a/packages/server/src/workers/context.ts b/packages/server/src/workers/context.ts
diff --git a/packages/server/src/workers/index.ts b/packages/server/src/workers/index.ts
@@ -1,7 +1,7 @@
 import { Resource } from '@medplum/fhirtypes';
+import { BackgroundJobContext } from '@medplum/core';
 import { MedplumServerConfig } from '../config';
 import { globalLogger } from '../logger';
-import { BackgroundJobContext } from './context';
 import { addCronJobs, closeCronWorker, initCronWorker } from './cron';
 import { addDownloadJobs, closeDownloadWorker, initDownloadWorker } from './download';
 import { addSubscriptionJobs, closeSubscriptionWorker, initSubscriptionWorker } from './subscription';

diff --git a/packages/server/src/workers/subscription.ts b/packages/server/src/workers/subscription.ts
@@ -1,5 +1,7 @@
 import {
  AccessPolicyInteraction,
+ BackgroundJobContext,
+ BackgroundJobInteraction,
  ContentType,
  OperationOutcomeError,
  Operator,
@@ -9,9 +11,8 @@ import {
  getReferenceString,
  isGone,
  isNotFound,
- matchesSearchRequest,
  normalizeOperationOutcome,
- parseSearchRequest,
+ resourceMatchesSubscriptionCriteria,
  satisfiedAccessPolicy,
  serverError,
  stringify,
@@ -30,8 +31,7 @@ import { getRedis } from '../redis';
 import { createSubEventNotification } from '../subscriptions/websockets';
 import { parseTraceparent } from '../traceparent';
 import { AuditEventOutcome } from '../util/auditevent';
-import { BackgroundJobContext, BackgroundJobInteraction } from './context';
-import { createAuditEvent, findProjectMembership, isFhirCriteriaMet, isJobSuccessful } from './utils';
+import { createAuditEvent, findProjectMembership, getPreviousResource, isJobSuccessful } from './utils';
 
 /**
  * The upper limit on the number of times a job can be retried.
@@ -270,73 +270,13 @@ async function matchesCriteria(
  context: BackgroundJobContext
 ): Promise<boolean> {
  const ctx = getRequestContext();
- if (subscription.meta?.account && resource.meta?.account?.reference !== subscription.meta.account.reference) {
- ctx.logger.debug('Ignore resource in different account compartment');
- return false;
- }
-
- if (!matchesChannelType(subscription)) {
- ctx.logger.debug(`Ignore subscription without recognized channel type`);
- return false;
- }
-
- const subscriptionCriteria = subscription.criteria;
- if (!subscriptionCriteria) {
- ctx.logger.debug(`Ignore rest hook missing criteria`);
- return false;
- }
-
- const searchRequest = parseSearchRequest(subscriptionCriteria);
- if (resource.resourceType !== searchRequest.resourceType) {
- ctx.logger.debug(
- `Ignore rest hook for different resourceType (wanted "${searchRequest.resourceType}", received "${resource.resourceType}")`
- );
- return false;
- }
-
- const fhirPathCriteria = await isFhirCriteriaMet(subscription, resource);
- if (!fhirPathCriteria) {
- ctx.logger.debug(`Ignore rest hook for criteria returning false`);
- return false;
- }
-
- const supportedInteractionExtension = getExtension(
+ return resourceMatchesSubscriptionCriteria({
+ resource,
  subscription,
- 'https://medplum.com/fhir/StructureDefinition/subscription-supported-interaction'
- );
- if (supportedInteractionExtension && supportedInteractionExtension.valueCode !== context.interaction) {
- ctx.logger.debug(
- `Ignore rest hook for different interaction (wanted "${supportedInteractionExtension.valueCode}", received "${context.interaction}")`
- );
- return false;
- }
-
- return matchesSearchRequest(resource, searchRequest);
-}
-
-/**
- * Returns true if the subscription channel type is ok to execute.
- * @param subscription - The subscription resource.
- * @returns True if the subscription channel type is ok to execute.
- */
-function matchesChannelType(subscription: Subscription): boolean {
- const channelType = subscription.channel?.type;
-
- if (channelType === 'rest-hook') {
- const url = subscription.channel?.endpoint;
- if (!url) {
- getLogger().debug(`Ignore rest-hook missing URL`);
- return false;
- }
-
- return true;
- }
-
- if (channelType === 'websocket') {
- return true;
- }
-
- return false;
+ context,
+ logger: ctx.logger,
+ getPreviousResource: getPreviousResource,
+ });
 }
 
 /**

diff --git a/packages/server/src/workers/utils.ts b/packages/server/src/workers/utils.ts
@@ -1,4 +1,4 @@
-import { createReference, evalFhirPathTyped, getExtension, isResource, Operator, toTypedValue } from '@medplum/core';
+import { createReference, getExtension, isResource, Operator } from '@medplum/core';
 import {
  AuditEvent,
  AuditEventEntity,
@@ -109,24 +109,7 @@ export function getAuditEventEntityRole(resource: Resource): Coding {
  }
 }
 
-export async function isFhirCriteriaMet(subscription: Subscription, currentResource: Resource): Promise<boolean> {
- const criteria = getExtension(
- subscription,
- 'https://medplum.com/fhir/StructureDefinition/fhir-path-criteria-expression'
- );
- if (!criteria?.valueString) {
- return true;
- }
- const previous = await getPreviousResource(currentResource);
- const evalInput = {
- '%current': toTypedValue(currentResource),
- '%previous': toTypedValue(previous ?? {}),
- };
- const evalValue = evalFhirPathTyped(criteria.valueString, [toTypedValue(currentResource)], evalInput);
- return evalValue?.[0]?.value === true;
-}
-
-async function getPreviousResource(currentResource: Resource): Promise<Resource | undefined> {
+export async function getPreviousResource(currentResource: Resource): Promise<Resource | undefined> {
  const systemRepo = getSystemRepo();
  const history = await systemRepo.readHistory(currentResource.resourceType, currentResource?.id as string);