fix(assetlibraryhistory): utilize kinesis stream to batch assetlibrary history updates

ISSUES CLOSED: #87

Author: Tony Sherman

source/packages/services/assetlibraryhistory/infrastructure/cfn-assetLibraryHistory.yml

@@ -135,11 +135,13 @@ Resources:
     Properties:
       TopicRulePayload:
         Actions:
-          - Lambda:
-              FunctionArn: !Sub 'arn:aws:lambda:${AWS::Region}:${AWS::AccountId}:function:${EventsLambdaFunction}'
+          - Kinesis:
+              PartitionKey: ${topic(4)}
+              RoleArn: !GetAtt LambdaExecutionRole.Arn
+              StreamName: !Ref EventsKinesisStream
         Description: 'Saves AssetLibrary configuration changes'
         AwsIotSqlVersion: '2016-03-23'
-        RuleDisabled: 'false'
+        RuleDisabled: false
         Sql: !Sub "SELECT * FROM '${AssetLibraryEventsTopic}'"
 
   LambdaInvocationPermission:
@@ -285,6 +287,13 @@ Resources:
         ReadCapacityUnits: '5'
         WriteCapacityUnits: '5'
 
+  EventsKinesisStream:
+    Type: AWS::Kinesis::Stream
+    Properties:
+      Name: !Sub 'cdf-assetlibrary-history-stream-${Environment}'
+      StreamModeDetails:
+        StreamMode: 'on-demand'
+
   EventsLambdaFunction:
     Type: AWS::Serverless::Function
     Properties:
@@ -301,6 +310,15 @@ Resources:
           APP_CONFIG: !Ref ApplicationConfigurationOverride
           AWS_DYNAMODB_TABLE_EVENTS: !Ref HistoryTable
       Tracing: Active
+      Events:
+        Stream:
+          Type: Kinesis
+          Properties:
+            Stream: !GetAtt EventsKinesisStream.Arn
+            BatchSize: 100 # What should this be?
+            MaximumBatchingWindowInSeconds: 30
+            StartingPosition: LATEST
+            MaximumRetryAttempts: 1
 
   RESTLambdaFunction:
     Type: AWS::Serverless::Function

source/packages/services/assetlibraryhistory/src/lambda_iot_rule.ts

@@ -15,17 +15,18 @@ import 'reflect-metadata';
 import { logger } from '@awssolutions/simple-cdf-logger';
 import { container } from './di/inversify.config';
 import { TYPES } from './di/types';
-import { EventModel } from './events/events.models';
 import { EventsService } from './events/events.service';
 
 const eventsService: EventsService = container.get(TYPES.EventsService);
 
-exports.iot_rule_handler = async (event: EventModel, _context: unknown) => {
+exports.iot_rule_handler = async (event: any, _context: unknown) => {
     logger.debug(`events.service create: in: event: ${JSON.stringify(event)}`);
 
-    // TODO validation
-
-    await eventsService.create(event);
-
-    logger.debug('events.service create: exit:');
+    for (const record of event.Records) {
+        // Kinesis record data is base64 encoded
+        const payload = JSON.parse(Buffer.from(record.kinesis.data, 'base64').toString('ascii'));
+        // TODO validation
+        await eventsService.create(payload);
+        logger.debug('events.service create: exit:');
+    }
 };