feat(batch): Batch processing wrapper function (#1605)

* Refactored some types, added function wrapper and base test

* Added record check and tests, renamed factories

* Refactored type check logic in function

* Refactor test to remove error ignore

Author: erikayao93

Diff for: packages/batch/src/BasePartialBatchProcessor.ts

@@ -9,14 +9,14 @@ import {
   DEFAULT_RESPONSE,
   EventSourceDataClassTypes,
   EventType,
-  ItemIdentifier,
-  BatchResponse,
+  PartialItemFailures,
+  PartialItemFailureResponse,
 } from '.';
 
 abstract class BasePartialBatchProcessor extends BasePartialProcessor {
   public COLLECTOR_MAPPING;
 
-  public batchResponse: BatchResponse;
+  public batchResponse: PartialItemFailureResponse;
 
   public eventType: keyof typeof EventType;
 
@@ -52,16 +52,16 @@ abstract class BasePartialBatchProcessor extends BasePartialProcessor {
       );
     }
 
-    const messages: ItemIdentifier[] = this.getMessagesToReport();
+    const messages: PartialItemFailures[] = this.getMessagesToReport();
     this.batchResponse = { batchItemFailures: messages };
   }
 
   /**
    * Collects identifiers of failed items for a DynamoDB stream
    * @returns list of identifiers for failed items
    */
-  public collectDynamoDBFailures(): ItemIdentifier[] {
-    const failures: ItemIdentifier[] = [];
+  public collectDynamoDBFailures(): PartialItemFailures[] {
+    const failures: PartialItemFailures[] = [];
 
     for (const msg of this.failureMessages) {
       const msgId = (msg as DynamoDBRecord).dynamodb?.SequenceNumber;
@@ -77,8 +77,8 @@ abstract class BasePartialBatchProcessor extends BasePartialProcessor {
    * Collects identifiers of failed items for a Kinesis stream
    * @returns list of identifiers for failed items
    */
-  public collectKinesisFailures(): ItemIdentifier[] {
-    const failures: ItemIdentifier[] = [];
+  public collectKinesisFailures(): PartialItemFailures[] {
+    const failures: PartialItemFailures[] = [];
 
     for (const msg of this.failureMessages) {
       const msgId = (msg as KinesisStreamRecord).kinesis.sequenceNumber;
@@ -92,8 +92,8 @@ abstract class BasePartialBatchProcessor extends BasePartialProcessor {
    * Collects identifiers of failed items for an SQS batch
    * @returns list of identifiers for failed items
    */
-  public collectSqsFailures(): ItemIdentifier[] {
-    const failures: ItemIdentifier[] = [];
+  public collectSqsFailures(): PartialItemFailures[] {
+    const failures: PartialItemFailures[] = [];
 
     for (const msg of this.failureMessages) {
       const msgId = (msg as SQSRecord).messageId;
@@ -115,7 +115,7 @@ abstract class BasePartialBatchProcessor extends BasePartialProcessor {
    * Collects identifiers for failed batch items
    * @returns formatted messages to use in batch deletion
    */
-  public getMessagesToReport(): ItemIdentifier[] {
+  public getMessagesToReport(): PartialItemFailures[] {
     return this.COLLECTOR_MAPPING[this.eventType]();
   }
 
@@ -146,7 +146,7 @@ abstract class BasePartialBatchProcessor extends BasePartialProcessor {
   /**
    * @returns Batch items that failed processing, if any
    */
-  public response(): BatchResponse {
+  public response(): PartialItemFailureResponse {
     return this.batchResponse;
   }
 

Diff for: packages/batch/src/constants.ts

@@ -2,15 +2,15 @@
  * Constants for batch processor classes
  */
 import { DynamoDBRecord, KinesisStreamRecord, SQSRecord } from 'aws-lambda';
-import type { BatchResponse, EventSourceDataClassTypes } from '.';
+import type { PartialItemFailureResponse, EventSourceDataClassTypes } from '.';
 
 const EventType = {
   SQS: 'SQS',
   KinesisDataStreams: 'KinesisDataStreams',
   DynamoDBStreams: 'DynamoDBStreams',
 } as const;
 
-const DEFAULT_RESPONSE: BatchResponse = {
+const DEFAULT_RESPONSE: PartialItemFailureResponse = {
   batchItemFailures: [],
 };
 

Diff for: packages/batch/src/index.ts

@@ -3,3 +3,5 @@ export * from './errors';
 export * from './types';
 export * from './BasePartialProcessor';
 export * from './BasePartialBatchProcessor';
+export * from './BatchProcessor';
+export * from './processPartialResponse';

Diff for: packages/batch/src/processPartialResponse.ts

@@ -0,0 +1,30 @@
+import {
+  BasePartialBatchProcessor,
+  BaseRecord,
+  EventType,
+  PartialItemFailureResponse,
+} from '.';
+
+const processPartialResponse = async (
+  event: { Records: BaseRecord[] },
+  recordHandler: CallableFunction,
+  processor: BasePartialBatchProcessor
+): Promise<PartialItemFailureResponse> => {
+  if (!event.Records) {
+    const eventTypes: string = Object.values(EventType).toString();
+    throw new Error(
+      'Failed to convert event to record batch for processing.\nPlease ensure batch event is a valid ' +
+        eventTypes +
+        ' event.'
+    );
+  }
+
+  const records = event['Records'];
+
+  processor.register(records, recordHandler);
+  await processor.process();
+
+  return processor.response();
+};
+
+export { processPartialResponse };

Diff for: packages/batch/src/types.ts

@@ -17,15 +17,15 @@ type SuccessResponse = [string, ResultType, EventSourceDataClassTypes];
 
 type FailureResponse = [string, string, EventSourceDataClassTypes];
 
-type ItemIdentifier = { [key: string]: string };
-type BatchResponse = { [key: string]: ItemIdentifier[] };
+type PartialItemFailures = { itemIdentifier: string };
+type PartialItemFailureResponse = { batchItemFailures: PartialItemFailures[] };
 
 export type {
   BaseRecord,
   EventSourceDataClassTypes,
   ResultType,
   SuccessResponse,
   FailureResponse,
-  ItemIdentifier,
-  BatchResponse,
+  PartialItemFailures,
+  PartialItemFailureResponse,
 };

Diff for: packages/batch/tests/helpers/factories.ts

@@ -2,7 +2,7 @@ import { DynamoDBRecord, KinesisStreamRecord, SQSRecord } from 'aws-lambda';
 import { randomInt } from 'crypto';
 import { v4 } from 'uuid';
 
-const sqsEventFactory = (body: string): SQSRecord => {
+const sqsRecordFactory = (body: string): SQSRecord => {
   return {
     messageId: v4(),
     receiptHandle: 'AQEBwJnKyrHigUMZj6rYigCgxlaS3SLy0a',
@@ -21,7 +21,7 @@ const sqsEventFactory = (body: string): SQSRecord => {
   };
 };
 
-const kinesisEventFactory = (body: string): KinesisStreamRecord => {
+const kinesisRecordFactory = (body: string): KinesisStreamRecord => {
   let seq = '';
   for (let i = 0; i < 52; i++) {
     seq = seq + randomInt(10);
@@ -46,7 +46,7 @@ const kinesisEventFactory = (body: string): KinesisStreamRecord => {
   };
 };
 
-const dynamodbEventFactory = (body: string): DynamoDBRecord => {
+const dynamodbRecordFactory = (body: string): DynamoDBRecord => {
   let seq = '';
   for (let i = 0; i < 10; i++) {
     seq = seq + randomInt(10);
@@ -69,4 +69,4 @@ const dynamodbEventFactory = (body: string): DynamoDBRecord => {
   };
 };
 
-export { sqsEventFactory, kinesisEventFactory, dynamodbEventFactory };
+export { sqsRecordFactory, kinesisRecordFactory, dynamodbRecordFactory };