allow insert record

Author: masl2

lambdas/letter-updates-transformer/src/__tests__/letter-updates-transformer.test.ts

@@ -26,18 +26,18 @@ jest.mock("crypto", () => ({
   randomBytes: (size: number) => randomBytes[String(size)],
 }));
 
-const eventSource = "/data-plane/supplier-api/nhs-supplier-api-dev/main/letters";
+const eventSource =
+  "/data-plane/supplier-api/nhs-supplier-api-dev/main/letters";
 const mockedDeps: jest.Mocked<Deps> = {
-    snsClient: { send: jest.fn() } as unknown as SNSClient,
-    logger: { info: jest.fn(), error: jest.fn() } as unknown as pino.Logger,
-    env: {
-      EVENTPUB_SNS_TOPIC_ARN: "arn:aws:sns:region:account:topic",
-      EVENT_SOURCE: eventSource
-    } as unknown as EnvVars,
-  } as Deps;
+  snsClient: { send: jest.fn() } as unknown as SNSClient,
+  logger: { info: jest.fn(), error: jest.fn() } as unknown as pino.Logger,
+  env: {
+    EVENTPUB_SNS_TOPIC_ARN: "arn:aws:sns:region:account:topic",
+    EVENT_SOURCE: eventSource,
+  } as unknown as EnvVars,
+} as Deps;
 
 describe("letter-updates-transformer Lambda", () => {
-
   beforeEach(() => {
     jest.useFakeTimers();
   });
@@ -53,7 +53,9 @@ describe("letter-updates-transformer Lambda", () => {
       const newLetter = generateLetter("PRINTED");
       const expectedEntries = [
         expect.objectContaining({
-          Message: JSON.stringify(mapLetterToCloudEvent(newLetter, eventSource)),
+          Message: JSON.stringify(
+            mapLetterToCloudEvent(newLetter, eventSource),
+          ),
         }),
       ];
 
@@ -79,7 +81,9 @@ describe("letter-updates-transformer Lambda", () => {
       newLetter.reasonCode = "R1";
       const expectedEntries = [
         expect.objectContaining({
-          Message: JSON.stringify(mapLetterToCloudEvent(newLetter, eventSource)),
+          Message: JSON.stringify(
+            mapLetterToCloudEvent(newLetter, eventSource),
+          ),
         }),
       ];
 
@@ -106,7 +110,9 @@ describe("letter-updates-transformer Lambda", () => {
       newLetter.reasonCode = "R2";
       const expectedEntries = [
         expect.objectContaining({
-          Message: JSON.stringify(mapLetterToCloudEvent(newLetter, eventSource)),
+          Message: JSON.stringify(
+            mapLetterToCloudEvent(newLetter, eventSource),
+          ),
         }),
       ];
 
@@ -138,14 +144,28 @@ describe("letter-updates-transformer Lambda", () => {
       expect(mockedDeps.snsClient.send).not.toHaveBeenCalled();
     });
 
-    it("does not publish non-modify events", async () => {
+    it("publishes INSERT events", async () => {
       const handler = createHandler(mockedDeps);
       const newLetter = generateLetter("ACCEPTED");
+      const expectedEntries = [
+        expect.objectContaining({
+          Message: JSON.stringify(
+            mapLetterToCloudEvent(newLetter, eventSource),
+          ),
+        }),
+      ];
 
       const testData = generateKinesisEvent([generateInsertRecord(newLetter)]);
       await handler(testData, mockDeep<Context>(), jest.fn());
 
-      expect(mockedDeps.snsClient.send).not.toHaveBeenCalled();
+      expect(mockedDeps.snsClient.send).toHaveBeenCalledWith(
+        expect.objectContaining({
+          input: expect.objectContaining({
+            TopicArn: "arn:aws:sns:region:account:topic",
+            PublishBatchRequestEntries: expectedEntries,
+          }),
+        }),
+      );
     });
 
     it("does not publish invalid letter data", async () => {

lambdas/letter-updates-transformer/src/letter-updates-transformer.ts

@@ -21,14 +21,13 @@ export default function createHandler(deps: Deps): Handler<KinesisStreamEvent> {
   return async (streamEvent: KinesisStreamEvent) => {
     deps.logger.info({ description: "Received event", streamEvent });
 
-    const cloudEvents: LetterEvent[] = streamEvent.Records.map((record) =>
+    // Ensure logging by extracting all records first
+    const ddbRecords: DynamoDBRecord[] = streamEvent.Records.map((record) =>
       extractPayload(record, deps),
-    )
-      .filter((record) => record.eventName === "MODIFY")
-      .filter(
-        (record) =>
-          isChanged(record, "status") || isChanged(record, "reasonCode"),
-      )
+    );
+
+    const cloudEvents: LetterEvent[] = ddbRecords
+      .filter((record) => filterRecord(record, deps))
       .map((element) => extractNewLetter(element))
       .map((element) => mapLetterToCloudEvent(element, deps.env.EVENT_SOURCE));
 
@@ -50,6 +49,29 @@ export default function createHandler(deps: Deps): Handler<KinesisStreamEvent> {
   };
 }
 
+function filterRecord(record: DynamoDBRecord, deps: Deps): boolean {
+  let allowEvent = false;
+  if (record.eventName === "INSERT") {
+    allowEvent = true;
+  }
+
+  if (
+    record.eventName === "MODIFY" &&
+    (isChanged(record, "status") || isChanged(record, "reasonCode"))
+  ) {
+    allowEvent = true;
+  }
+
+  deps.logger.info({
+    description: "Filtering record",
+    eventName: record.eventName,
+    eventId: record.eventID,
+    allowEvent,
+  });
+
+  return allowEvent;
+}
+
 function extractPayload(
   record: KinesisStreamRecord,
   deps: Deps,

lambdas/letter-updates-transformer/src/mappers/letter-mapper.ts

@@ -5,7 +5,7 @@ import { LetterForEventPub } from "../types";
 
 export default function mapLetterToCloudEvent(
   letter: LetterForEventPub,
-  source: string
+  source: string,
 ): LetterEvent {
   const eventId = randomUUID();
   const dataschemaversion = eventSchemaPackage.version;
@@ -16,7 +16,7 @@ export default function mapLetterToCloudEvent(
     plane: "data",
     dataschema: `https://notify.nhs.uk/cloudevents/schemas/supplier-api/letter.${letter.status}.${dataschemaversion}.schema.json`,
     dataschemaversion,
-    source: source,
+    source,
     subject: `letter-origin/supplier-api/letter/${letter.id}`,
 
     data: {