add more envelopes and tests, refactored utils to autocomplete event files

Author: am29d

packages/parser/src/envelopes/Envelope.ts

@@ -15,10 +15,13 @@ export abstract class Envelope {
     data: unknown,
     schema: T
   ): z.infer<T>[] {
-    if (typeof data !== 'object') {
-      throw new Error('Data must be an object');
-    }
-
-    return schema.parse(data);
+    if (typeof data === 'string') {
+      return schema.parse(JSON.parse(data));
+    } else if (typeof data === 'object') {
+      return schema.parse(data);
+    } else
+      throw new Error(
+        `Invalid data type for envelope. Expected string or object, got ${typeof data}`
+      );
   }
 }

packages/parser/src/envelopes/Envelopes.ts

@@ -9,6 +9,8 @@ import { LambdaFunctionUrlEnvelope } from './lambda.js';
 import { SnsEnvelope, SnsSqsEnvelope } from './sns.js';
 import { VpcLatticeEnvelope } from './vpc-lattice.js';
 import { VpcLatticeV2Envelope } from './vpc-latticev2.js';
+import { DynamoDBStreamEnvelope } from './dynamodb.js';
+import { KinesisEnvelope } from './kinesis.js';
 
 /**
  * A collection of envelopes to create new envelopes.
@@ -17,9 +19,11 @@ export class Envelopes {
   public static readonly API_GW_ENVELOPE = new ApiGatewayEnvelope();
   public static readonly API_GW_V2_ENVELOPE = new ApiGatwayV2Envelope();
   public static readonly CLOUDWATCH_ENVELOPE = new CloudWatchEnvelope();
+  public static readonly DYNAMO_DB_STREAM_ENVELOPE =
+    new DynamoDBStreamEnvelope();
   public static readonly EVENT_BRIDGE_ENVELOPE = new EventBridgeEnvelope();
   public static readonly KAFKA_ENVELOPE = new KafkaEnvelope();
-  public static readonly KINESIS_ENVELOPE = new KafkaEnvelope();
+  public static readonly KINESIS_ENVELOPE = new KinesisEnvelope();
   public static readonly KINESIS_FIREHOSE_ENVELOPE =
     new KinesisFirehoseEnvelope();
   public static readonly LAMBDA_FUCTION_URL_ENVELOPE =

packages/parser/src/envelopes/apigw.ts

@@ -1,6 +1,7 @@
 import { Envelope } from './Envelope.js';
 import { z, ZodSchema } from 'zod';
 import { APIGatewayProxyEventSchema } from '../schemas/apigw.js';
+import { type ApiGatewayProxyEvent } from '../types/schema.js';
 
 /**
  * API Gateway envelope to extract data within body key"
@@ -11,7 +12,11 @@ export class ApiGatewayEnvelope extends Envelope {
   }
 
   public parse<T extends ZodSchema>(data: unknown, schema: T): z.infer<T> {
-    const parsedEnvelope = APIGatewayProxyEventSchema.parse(data);
+    const parsedEnvelope: ApiGatewayProxyEvent =
+      APIGatewayProxyEventSchema.parse(data);
+    if (parsedEnvelope.body === undefined) {
+      throw new Error('Body field of API Gateway event is undefined');
+    }
 
     return this._parse(parsedEnvelope.body, schema);
   }

packages/parser/src/envelopes/apigwv2.ts

@@ -12,6 +12,9 @@ export class ApiGatwayV2Envelope extends Envelope {
 
   public parse<T extends ZodSchema>(data: unknown, schema: T): z.infer<T> {
     const parsedEnvelope = APIGatewayProxyEventV2Schema.parse(data);
+    if (parsedEnvelope.body === undefined) {
+      throw new Error('Body field of API Gateway V2 event is undefined');
+    }
 
     return this._parse(parsedEnvelope.body, schema);
   }

packages/parser/src/envelopes/cloudwatch.ts

@@ -20,7 +20,7 @@ export class CloudWatchEnvelope extends Envelope {
     const parsedEnvelope = CloudWatchLogsSchema.parse(data);
 
     return parsedEnvelope.awslogs.data.logEvents.map((record) => {
-      return this._parse(record, schema);
+      return this._parse(record.message, schema);
     });
   }
 }

packages/parser/src/envelopes/kafka.ts

@@ -33,8 +33,10 @@ export class KafkaEnvelope extends Envelope {
         ? KafkaMskEventSchema.parse(data)
         : KafkaSelfManagedEventSchema.parse(data);
 
-    return parsedEnvelope.records.data.map((record: KafkaRecord) => {
-      return this._parse(record.value, schema);
+    return Object.values(parsedEnvelope.records).map((topicRecord) => {
+      return topicRecord.map((record: KafkaRecord) => {
+        return this._parse(record.value, schema);
+      });
     });
   }
 }

packages/parser/src/envelopes/lambda.ts

@@ -13,6 +13,10 @@ export class LambdaFunctionUrlEnvelope extends Envelope {
   public parse<T extends ZodSchema>(data: unknown, schema: T): z.infer<T> {
     const parsedEnvelope = LambdaFunctionUrlSchema.parse(data);
 
-    return this.parse(parsedEnvelope.body, schema);
+    if (parsedEnvelope.body === undefined) {
+      throw new Error('Body field of Lambda function URL event is undefined');
+    }
+
+    return this._parse(parsedEnvelope.body, schema);
   }
 }

packages/parser/src/envelopes/sns.ts

@@ -1,6 +1,6 @@
 import { z, ZodSchema } from 'zod';
 import { Envelope } from './Envelope.js';
-import { SnsNotificationSchema, SnsSchema } from '../schemas/sns.js';
+import { SnsSchema, SnsSqsNotificationSchema } from '../schemas/sns.js';
 import { SqsSchema } from '../schemas/sqs.js';
 
 /**
@@ -46,9 +46,11 @@ export class SnsSqsEnvelope extends Envelope {
     const parsedEnvelope = SqsSchema.parse(data);
 
     return parsedEnvelope.Records.map((record) => {
-      const snsNotification = SnsNotificationSchema.parse(record.body);
+      const snsNotification = SnsSqsNotificationSchema.parse(
+        JSON.parse(record.body)
+      );
 
-      return this._parse(snsNotification, schema);
+      return this._parse(snsNotification.Message, schema);
     });
   }
 }

packages/parser/src/envelopes/sqs.ts

@@ -20,9 +20,7 @@ export class SqsEnvelope extends Envelope {
     const parsedEnvelope = SqsSchema.parse(data);
 
     return parsedEnvelope.Records.map((record) => {
-      const body = JSON.parse(record.body);
-
-      return this._parse(body, schema);
+      return this._parse(record.body, schema);
     });
   }
 }

packages/parser/src/middleware/parser.ts

@@ -3,9 +3,9 @@ import { MiddlewareObj } from '@middy/core';
 import { ZodSchema } from 'zod';
 import { Envelope } from '../envelopes/Envelope.js';
 
-interface ParserOptions {
-  schema: ZodSchema;
-  envelope?: Envelope;
+interface ParserOptions<S extends ZodSchema, E extends Envelope> {
+  schema: S;
+  envelope?: E;
 }
 
 /**
@@ -36,7 +36,9 @@ interface ParserOptions {
  *
  * @param options
  */
-const parser = (options: ParserOptions): MiddlewareObj => {
+const parser = <S extends ZodSchema, E extends Envelope>(
+  options: ParserOptions<S, E>
+): MiddlewareObj => {
   const before = (request: MiddyLikeRequest): void => {
     const { schema, envelope } = options;
     if (envelope) {
@@ -46,14 +48,8 @@ const parser = (options: ParserOptions): MiddlewareObj => {
     }
   };
 
-  const after = (_request: MiddyLikeRequest): void => {};
-
-  const onError = (_request: MiddyLikeRequest): void => {};
-
   return {
     before,
-    after,
-    onError,
   };
 };
 

packages/parser/src/schemas/cloudwatch.ts

@@ -30,15 +30,9 @@ const CloudWatchLogsSchema = z.object({
   }),
 });
 
-const extractCloudWatchLogFromEvent = (
-  data: string
-): z.infer<typeof CloudWatchLogsDecodeSchema> => {
-  return decompressRecordToJSON(data);
-};
-
 export {
   CloudWatchLogsSchema,
   CloudWatchLogsDecodeSchema,
   decompressRecordToJSON,
-  extractCloudWatchLogFromEvent,
+  CloudWatchLogEventSchema,
 };

packages/parser/src/schemas/kinesis.ts

@@ -1,13 +1,31 @@
 import { z } from 'zod';
+import { gunzipSync } from 'node:zlib';
 
 const KinesisDataStreamRecordPayload = z.object({
   kinesisSchemaVersion: z.string(),
   partitionKey: z.string(),
   sequenceNumber: z.string(),
   approximateArrivalTimestamp: z.number(),
-  data: z.string(),
+  data: z.string().transform((data) => {
+    const decompresed = decompress(data);
+    const decoded = Buffer.from(data, 'base64').toString('utf-8');
+    try {
+      // If data was not compressed, try to parse it as JSON otherwise it must be string
+      return decompresed === data ? JSON.parse(decoded) : decompresed;
+    } catch (e) {
+      return decoded;
+    }
+  }),
 });
 
+const decompress = (data: string): string => {
+  try {
+    return JSON.parse(gunzipSync(Buffer.from(data, 'base64')).toString('utf8'));
+  } catch (e) {
+    return data;
+  }
+};
+
 const KinesisDataStreamRecord = z.object({
   eventSource: z.literal('aws:kinesis'),
   eventVersion: z.string(),

packages/parser/src/schemas/sns.ts

@@ -9,16 +9,26 @@ const SnsNotificationSchema = z.object({
   Subject: z.string().optional(),
   TopicArn: z.string(),
   UnsubscribeUrl: z.string().url(),
+  UnsubscribeURL: z.string().url().optional(),
+  SigningCertUrl: z.string().url().optional(),
+  SigningCertURL: z.string().url().optional(),
   Type: z.literal('Notification'),
   MessageAttributes: z.record(z.string(), SnsMsgAttribute).optional(),
   Message: z.string(),
   MessageId: z.string(),
   Signature: z.string().optional(),
   SignatureVersion: z.string().optional(),
-  SigningCertUrl: z.string().url().optional(),
   Timestamp: z.string().datetime(),
 });
 
+const SnsSqsNotificationSchema = SnsNotificationSchema.extend({
+  UnsubscribeURL: z.string().optional(),
+  SigningCertURL: z.string().url().optional(),
+}).omit({
+  UnsubscribeUrl: true,
+  SigningCertUrl: true,
+});
+
 const SnsRecordSchema = z.object({
   EventSource: z.literal('aws:sns'),
   EventVersion: z.string(),
@@ -30,4 +40,10 @@ const SnsSchema = z.object({
   Records: z.array(SnsRecordSchema),
 });
 
-export { SnsSchema, SnsRecordSchema, SnsNotificationSchema, SnsMsgAttribute };
+export {
+  SnsSchema,
+  SnsRecordSchema,
+  SnsNotificationSchema,
+  SnsMsgAttribute,
+  SnsSqsNotificationSchema,
+};

packages/parser/src/types/schema.ts

@@ -0,0 +1,17 @@
+import { KafkaRecordSchema } from '../schemas/kafka.js';
+import { z } from 'zod';
+import {
+  KinesisDataStreamRecord,
+  KinesisDataStreamRecordPayload,
+} from '../schemas/kinesis.js';
+import { APIGatewayProxyEventSchema } from '../schemas/apigw.js';
+
+export type KafkaRecord = z.infer<typeof KafkaRecordSchema>;
+
+export type KinesisDataStreamRecord = z.infer<typeof KinesisDataStreamRecord>;
+
+export type KinesisDataStreamRecordPayload = z.infer<
+  typeof KinesisDataStreamRecordPayload
+>;
+
+export type ApiGatewayProxyEvent = z.infer<typeof APIGatewayProxyEventSchema>;

packages/parser/tests/unit/envelopes/apigwt.test.ts

@@ -0,0 +1,29 @@
+/**
+ * Test built in schema envelopes for api gateway
+ *
+ * @group unit/parser/envelopes
+ */
+
+import { generateMock } from '@anatine/zod-mock';
+import { Envelopes } from '../../../src/envelopes/Envelopes.js';
+import { TestEvents, TestSchema } from '../schema/utils.js';
+import { ApiGatewayProxyEvent } from '../../../src/types/schema.js';
+
+describe('ApigwEnvelope ', () => {
+  const envelope = Envelopes.API_GW_ENVELOPE;
+  it('should parse custom schema in envelope', () => {
+    const testCustomSchemaObject = generateMock(TestSchema);
+    const testEvent = TestEvents.apiGatewayProxyEvent as ApiGatewayProxyEvent;
+
+    testEvent.body = JSON.stringify(testCustomSchemaObject);
+    const resp = envelope.parse(testEvent, TestSchema);
+    expect(resp).toEqual(testCustomSchemaObject);
+  });
+
+  it('should throw no body provided', () => {
+    const testEvent = TestEvents.apiGatewayProxyEvent as ApiGatewayProxyEvent;
+    testEvent.body = undefined;
+
+    expect(() => envelope.parse(testEvent, TestSchema)).toThrow();
+  });
+});

packages/parser/tests/unit/envelopes/apigwv2.test.ts

@@ -0,0 +1,32 @@
+/**
+ * Test built in schema envelopes for api gateway v2
+ *
+ * @group unit/parser/envelopes
+ */
+
+import { Envelopes } from '../../../src/envelopes/Envelopes.js';
+import { TestEvents, TestSchema } from '../schema/utils.js';
+import { generateMock } from '@anatine/zod-mock';
+import { APIGatewayProxyEventV2 } from 'aws-lambda';
+
+describe('ApiGwV2Envelope ', () => {
+  const envelope = Envelopes.API_GW_V2_ENVELOPE;
+
+  it('should parse custom schema in envelope', () => {
+    const testEvent =
+      TestEvents.apiGatewayProxyV2Event as APIGatewayProxyEventV2;
+    const data = generateMock(TestSchema);
+
+    testEvent.body = JSON.stringify(data);
+
+    expect(envelope.parse(testEvent, TestSchema)).toEqual(data);
+  });
+
+  it('should throw when no body provided', () => {
+    const testEvent =
+      TestEvents.apiGatewayProxyV2Event as APIGatewayProxyEventV2;
+    testEvent.body = undefined;
+
+    expect(() => envelope.parse(testEvent, TestSchema)).toThrow();
+  });
+});