feat: add support for FHIR Subscriptions (#573)

* feat: add GSI to Resource DDB Table (#533)

* feat: Add data validation for subscription (#543)

* fix: remove _subsciptionStatus from export result field (#555)

* feat: sns, sqs, dlq for Subscriptions  (#554)

* feat: Rest hook Lambda (#558)

* feat: subscriptionReaper (#557)

* feat: add subscriptionsMatcher Lambda (#559)

* test: Add Subscriptions test infrastructure/helper (#569)

* fix: update unit tests for subscription reaper (#567)

* test: add subscriptions env vars in gh actions (#572)

* fix: encrypt logs for new Lambda fns (#574)

* test: add Subscription reaper tests (#575)

* feat: emit end to end latency metric from rest-hook Lambda (#570)

* test: add tests for tenant isolation of subscriptions (#577)

* feat: add DLQ for matcher Lambda (#576)

* test: add end to end tests for subscriptions (#578)

* perf: partial failures for restHook Lambda (#579)

* docs: add Subscription docs (#582)

Co-authored-by: Sukeerth Vegaraju <[email protected]>
Co-authored-by: zheyanyu <[email protected]>
Co-authored-by: Yanyu Zheng <[email protected]>
Co-authored-by: brndhpkn <[email protected]>

Author: 5 people

.eslintrc.js

@@ -25,6 +25,11 @@ module.exports = {
         'no-empty-function': 'off',
         '@typescript-eslint/no-empty-function': 'error',
         'import/no-extraneous-dependencies': ['error', { devDependencies: ['**/*.test.ts', 'integration-tests/*'] }],
+        // @types/aws-lambda is special since aws-lambda is not the name of a package that we take as a dependency.
+        // Making eslint recognize it would require several additional plugins and it's not worth setting it up right now.
+        // See https://github.com/typescript-eslint/typescript-eslint/issues/1624
+        // eslint-disable-next-line import/no-unresolved
+        'import/no-unresolved': ['error', { ignore: ['aws-lambda'] }],
         'no-shadow': 'off', // replaced by ts-eslint rule below
         '@typescript-eslint/no-shadow': 'error',
     },

.github/workflows/deploy.yaml

@@ -80,12 +80,14 @@ jobs:
       - name: Install npm dependencies
         run: yarn install
       - name: Download US Core IG
-        # NOTE if updateing the IG version. Please see update implementationGuides.test.ts test too.
+        # NOTE if updating the IG version. Please see update implementationGuides.test.ts test too.
         run: |
           mkdir -p implementationGuides
           curl http://hl7.org/fhir/us/core/STU3.1.1/package.tgz | tar xz -C implementationGuides
       - name: Compile IGs
         run: yarn run compile-igs
+      - name: Setup allowList for Subscriptions integ tests
+        run: cp integration-tests/infrastructure/allowList-integTests.ts src/subscriptions/allowList.ts
       - name: Install serverless
         run: npm install -g [email protected]
       - name: Deploy Hapi validator
@@ -125,12 +127,18 @@ jobs:
             serviceUrlSecretName: SERVICE_URL
             cognitoClientIdSecretName: COGNITO_CLIENT_ID
             apiKeySecretName: API_KEY
+            subscriptionsNotificationsTableSecretName: SUBSCRIPTIONS_NOTIFICATIONS_TABLE
+            subscriptionsEndpointSecretName: SUBSCRIPTIONS_ENDPOINT
+            subscriptionsApiKeySecretName: SUBSCRIPTIONS_API_KEY
           - enableMultiTenancy: true
             region: us-west-1
             serviceUrlSuffix: /tenant/tenant1
             serviceUrlSecretName: MULTITENANCY_SERVICE_URL
             cognitoClientIdSecretName: MULTITENANCY_COGNITO_CLIENT_ID
             apiKeySecretName: MULTITENANCY_API_KEY
+            subscriptionsNotificationsTableSecretName: MULTITENANCY_SUBSCRIPTIONS_NOTIFICATIONS_TABLE
+            subscriptionsEndpointSecretName: MULTITENANCY_SUBSCRIPTIONS_ENDPOINT
+            subscriptionsApiKeySecretName: MULTITENANCY_SUBSCRIPTIONS_API_KEY
     steps:
       - uses: actions/checkout@v2
         with:
@@ -195,6 +203,10 @@ jobs:
           COGNITO_USERNAME_PRACTITIONER_ANOTHER_TENANT: ${{ secrets.COGNITO_USERNAME_PRACTITIONER_ANOTHER_TENANT }}
           COGNITO_PASSWORD: ${{ secrets.COGNITO_PASSWORD }}
           MULTI_TENANCY_ENABLED: ${{ matrix.enableMultiTenancy }}
+          SUBSCRIPTIONS_ENABLED: 'true'
+          SUBSCRIPTIONS_NOTIFICATIONS_TABLE: ${{ secrets.[matrix.subscriptionsNotificationsTableSecretName] }}
+          SUBSCRIPTIONS_ENDPOINT: ${{ secrets.[matrix.subscriptionsEndpointSecretName] }}
+          SUBSCRIPTIONS_API_KEY: ${{ secrets.[matrix.subscriptionsApiKeySecretName] }}
         run: yarn int-test
 
   merge-develop-to-mainline:

.gitignore

@@ -12,7 +12,6 @@ dist
 .idea
 yarn-error.log
 
-
 auditLogMover/.serverless
 auditLogMover/node_modules
 

README.md

@@ -40,7 +40,9 @@ git clone https://github.com/awslabs/fhir-works-on-aws-deployment.git
 
 If you intend to use FHIR Implementation Guides read the [Using Implementation Guides](./USING_IMPLEMENTATION_GUIDES.md) documentation first. 
 
-If you intend to do a multi-tenant deployment read the [Using Multi-Tenancy](./USING_MULTI_TENANCY.md) documentation first. 
+If you intend to do a multi-tenant deployment read the [Using Multi-Tenancy](./USING_MULTI_TENANCY.md) documentation first.
+
+If you intend to use FHIR Subscriptions read the [Using Subscriptions](./USING_SUBSCRIPTIONS.md) documentation first.
 
 ## Architecture
 

USING_SUBSCRIPTIONS.md

@@ -0,0 +1,104 @@
+# Subscriptions
+
+The FHIR Subscription resource is used to define a push-based subscription from a server to another system. 
+Once a subscription is registered with the server, the server checks every resource that is created or updated, 
+and if the resource matches the given criteria, it sends a message on the defined channel so that another system can take an appropriate action.
+
+FHIR Works on AWS implements Subscriptions v4.0.1: https://www.hl7.org/fhir/R4/subscription.html
+
+![Architecture diagram](resources/FWoA-subscriptions.svg)
+
+## Getting Started
+
+1. As an additional security measure, all destination endpoints must be allow-listed before notifications can be delivered to them. 
+Update [src/subscriptions/allowList.ts](src/subscriptions/allowList.ts) to configure your allow-list. 
+
+
+2. Use the `enableSubscriptions` option when deploying the stack:
+
+   ```bash
+   serverless deploy --enableSubscriptions true
+   ```
+
+
+**Note**  
+Enabling subscriptions incurs a cost even if there are no active subscriptions. It is recommended to only enable it if you intend to use it.
+
+## Creating Subscriptions
+
+A Subscription is a FHIR resource. Use the REST API to create, update or delete Subscriptions. 
+Refer to the [FHIR documentation](https://www.hl7.org/fhir/R4/subscription.html#resource) for the details of the Subscription resource.
+
+Create Subscription example:
+```
+POST <API_URL>/Subscription 
+{
+  "resourceType": "Subscription",
+  "status": "requested",
+  "end": "2022-01-01T00:00:00Z",
+  "reason": "Monitor new neonatal function",
+  "criteria": "Observation?code=http://loinc.org|1975-2",
+  "channel": {
+    "type": "rest-hook",
+    "endpoint": "https://my-endpoint.com/on-result",
+    "payload": "application/fhir+json"
+  }
+}
+```
+
+After the example Subscription is created, whenever an Observation is created or updated that matches the `criteria`, 
+a notification will be sent to `https://my-endpoint.com/on-result`.  
+
+Consider the following when working with Subscriptions:
+
+* Subscriptions start sending notifications within 1 minute of being created.
+* Notifications are delivered at-least-once and with best-effort ordering.
+
+## Supported Features
+
+Currently the only supported channel is **REST Hook**.
+
+If a Subscription has an `end` date, it is automatically deleted on that date.
+
+FWoA supports 2 types of notifications
+
+- **Empty notification**
+
+   This kind of notification occurs for Subscriptions without a `channel.payload` defined. Example:
+   ```json
+   {
+     "resourceType": "Subscription",
+     "criteria": "Observation?name=http://loinc.org|1975-2",
+     "channel": {
+       "type": "rest-hook",
+       "endpoint": "https://my-endpoint.com/on-result"
+     }
+   }
+   ```
+   When a matching Observation is created/updated, FWoA Sends a POST request with an **empty body** to:
+   ```
+   POST https://my-endpoint.com/on-result
+   ```
+
+- **Id-only notification**
+
+   This kind of notification occurs for Subscriptions with `channel.payload` set to `application/fhir+json`. Example:
+   ```json
+   {
+     "resourceType": "Subscription",
+     "criteria": "Observation?name=http://loinc.org|1975-2",
+     "channel": {
+       "type": "rest-hook", 
+       "payload": "application/fhir+json",
+       "endpoint": "https://my-endpoint.com/on-result"
+     }
+   }
+   ```
+   When a matching Observation is created/updated, FWoA Sends a PUT request with an **empty body** to:
+   ```
+   PUT https://my-endpoint.com/on-result/Observation/<matching ObservationId>
+   ```
+   **Note**  
+   The Id-only notifications differ slightly from the FHIR spec.
+   The spec indicates that the entire matching FHIR resource is sent in JSON format, but we chose to only send the Id since
+   sending the entire resource poses a security risk.

bulkExport/glueScripts/export-script.py

@@ -216,7 +216,7 @@ def get_transitive_references(resource, transitive_reference_map, server_url):
 
 # Drop fields that are not needed
 print('Dropping fields that are not needed')
-data_source_cleaned_dyn_frame = DropFields.apply(frame = filtered_dates_resource_dyn_frame, paths = ['documentStatus', 'lockEndTs', 'vid', '_references', '_tenantId', '_id'])
+data_source_cleaned_dyn_frame = DropFields.apply(frame = filtered_dates_resource_dyn_frame, paths = ['documentStatus', 'lockEndTs', 'vid', '_references', '_tenantId', '_id', '_subscriptionStatus'])
 
 def add_dup_resource_type(record):
     record["resourceTypeDup"] = record["resourceType"]

cloudformation/subscriptions.yaml

@@ -0,0 +1,143 @@
+#
+#  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+#  SPDX-License-Identifier: Apache-2.0
+#
+
+Resources:
+  SubscriptionsKey:
+    Type: 'AWS::KMS::Key'
+    Properties:
+      Description: Encryption key for rest hook queue that can be used by SNS
+      EnableKeyRotation: true
+      KeyPolicy:
+        Statement:
+          - Effect: Allow
+            Principal:
+              Service: 'sns.amazonaws.com'
+            Action:
+              - 'kms:Decrypt'
+              - 'kms:GenerateDataKey*'
+            Resource: '*'
+          - Sid: Allow administration of the key
+            Effect: Allow
+            Principal:
+              AWS: !Join ['', ['arn:aws:iam::', !Ref AWS::AccountId, ':root']]
+            Action:
+              - 'kms:*'
+            Resource: '*'
+
+  RestHookQueue:
+    Type: AWS::SQS::Queue
+    Properties:
+      KmsMasterKeyId: !Ref SubscriptionsKey
+      RedrivePolicy:
+        deadLetterTargetArn: !GetAtt RestHookDLQ.Arn
+        maxReceiveCount: 2
+
+  RestHookDLQ:
+    Type: AWS::SQS::Queue
+    Properties:
+      MessageRetentionPeriod: 1209600 # 14 days in seconds
+      KmsMasterKeyId: 'alias/aws/sqs'
+
+  RestHookQueuePolicy:
+    Type: AWS::SQS::QueuePolicy
+    Properties:
+      Queues: [!Ref RestHookQueue]
+      PolicyDocument:
+        Statement:
+          - Effect: Deny
+            Action:
+              - SQS:*
+            Resource:
+              - !GetAtt RestHookQueue.Arn
+            Principal: '*'
+            Condition:
+              Bool:
+                'aws:SecureTransport': false
+          - Effect: Allow
+            Action:
+              - SQS:SendMessage
+            Resource:
+              - !GetAtt RestHookQueue.Arn
+            Principal:
+              Service: 'sns.amazonaws.com'
+            Condition:
+              ArnEquals:
+                aws:SourceArn: !Ref SubscriptionsTopic
+
+  RestHookDLQPolicy:
+    Type: AWS::SQS::QueuePolicy
+    Properties:
+      Queues: [!Ref RestHookDLQ]
+      PolicyDocument:
+        Statement:
+          - Effect: Deny
+            Action:
+              - SQS:*
+            Resource:
+              - !GetAtt RestHookDLQ.Arn
+            Principal: '*'
+            Condition:
+              Bool:
+                'aws:SecureTransport': false
+
+  SubscriptionsTopic:
+    Type: AWS::SNS::Topic
+    Properties:
+      TopicName: 'SubscriptionsTopic'
+      KmsMasterKeyId: !Ref SubscriptionsKey
+
+  RestHookSubscription:
+    Type: 'AWS::SNS::Subscription'
+    Properties:
+      TopicArn: !Ref SubscriptionsTopic
+      Endpoint: !GetAtt RestHookQueue.Arn
+      Protocol: sqs
+      FilterPolicy:
+        channelType:
+          - 'rest-hook'
+
+  RestHookLambdaRole:
+    Type: AWS::IAM::Role
+    Metadata:
+      cfn_nag:
+        rules_to_suppress:
+          - id: W11
+            reason: '* only applies to X-Ray statement which does not define a group or sampling-rule'
+    Properties:
+      AssumeRolePolicyDocument:
+        Version: '2012-10-17'
+        Statement:
+          - Effect: 'Allow'
+            Principal:
+              Service: 'lambda.amazonaws.com'
+            Action: 'sts:AssumeRole'
+      Policies:
+        - PolicyName: 'restHookLambdaPolicy'
+          PolicyDocument:
+            Version: '2012-10-17'
+            Statement:
+              - Effect: Allow
+                Action:
+                  - logs:CreateLogStream
+                  - logs:CreateLogGroup
+                  - logs:PutLogEvents
+                Resource: !Sub 'arn:${AWS::Partition}:logs:${AWS::Region}:*:*'
+              - Effect: Allow
+                Action:
+                  - 'xray:PutTraceSegments'
+                  - 'xray:PutTelemetryRecords'
+                Resource:
+                  - '*'
+              - Effect: Allow
+                Action:
+                  - 'kms:Decrypt'
+                Resource:
+                  - !GetAtt SubscriptionsKey.Arn
+              - Effect: Allow
+                Action:
+                  - 'sqs:DeleteMessage'
+                  - 'sqs:ReceiveMessage'
+                  - 'sqs:GetQueueAttributes'
+                Resource: !GetAtt RestHookQueue.Arn

integration-tests/SubscriptionsHelper.ts

@@ -0,0 +1,47 @@
+/*
+ *  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *  SPDX-License-Identifier: Apache-2.0
+ *
+ */
+import * as AWS from 'aws-sdk';
+import { DynamoDB } from 'aws-sdk';
+
+export interface SubscriptionNotification {
+    httpMethod: string;
+    path: string;
+    body?: string | null;
+    headers?: string[];
+}
+
+// eslint-disable-next-line import/prefer-default-export
+export class SubscriptionsHelper {
+    private readonly notificationsTableName: string;
+
+    private readonly dynamodbClient: DynamoDB;
+
+    constructor(notificationsTableName: string) {
+        this.notificationsTableName = notificationsTableName;
+        this.dynamodbClient = new AWS.DynamoDB();
+    }
+
+    /**
+     * Gets all notifications received for a given path.
+     * @param path - The path where the notifications were sent. It is recommended to use unique paths for each test run (e.g. by appending an uui to it)
+     */
+    async getNotifications(path: string): Promise<SubscriptionNotification[]> {
+        const { Items } = await this.dynamodbClient
+            .query({
+                TableName: this.notificationsTableName,
+                KeyConditionExpression: '#path = :pathValue',
+                ExpressionAttributeNames: { '#path': 'path' },
+                ExpressionAttributeValues: { ':pathValue': { S: path } },
+            })
+            .promise();
+
+        if (Items === undefined) {
+            return [];
+        }
+
+        return Items.map((item) => AWS.DynamoDB.Converter.unmarshall(item) as SubscriptionNotification);
+    }
+}