fix(parser): wrap envelope parse failures in ParseError

packages/parser/src/envelopes/cloudwatch.ts

@@ -2,7 +2,7 @@ import { ZodError, type ZodType, type z } from 'zod';
 import { ParseError } from '../errors.js';
 import { CloudWatchLogsSchema } from '../schemas/index.js';
 import type { ParsedResult } from '../types/index.js';
-import { envelopeDiscriminator, prefixIssuePaths } from './envelope.js';
+import { envelopeDiscriminator, parseOrThrow } from './envelope.js';
 
 /**
  * CloudWatch Envelope to extract messages from the `awslogs.data.logEvents` key.
@@ -14,26 +14,20 @@ export const CloudWatchEnvelope = {
    */
   [envelopeDiscriminator]: 'array' as const,
   parse<T>(data: unknown, schema: ZodType<T>): T[] {
-    const parsedEnvelope = CloudWatchLogsSchema.parse(data);
+    const parsedEnvelope = parseOrThrow(
+      CloudWatchLogsSchema,
+      data,
+      'Failed to parse CloudWatch Log envelope'
+    );
 
-    return parsedEnvelope.awslogs.data.logEvents.map((record, index) => {
-      try {
-        return schema.parse(record.message);
-      } catch (error) {
-        throw new ParseError(
-          `Failed to parse CloudWatch log event at index ${index}`,
-          {
-            cause: prefixIssuePaths(error, [
-              'awslogs',
-              'data',
-              'logEvents',
-              index,
-              'message',
-            ]),
-          }
-        );
-      }
-    });
+    return parsedEnvelope.awslogs.data.logEvents.map((record, index) =>
+      parseOrThrow(
+        schema,
+        record.message,
+        `Failed to parse CloudWatch log event at index ${index}`,
+        ['awslogs', 'data', 'logEvents', index, 'message']
+      )
+    );
   },
 
   safeParse<T>(data: unknown, schema: ZodType<T>): ParsedResult<unknown, T[]> {

packages/parser/src/envelopes/dynamodb.ts

@@ -3,7 +3,7 @@ import { ParseError } from '../errors.js';
 import { DynamoDBStreamSchema } from '../schemas/index.js';
 import type { DynamoDBStreamEnvelopeResponse } from '../types/envelope.js';
 import type { ParsedResult } from '../types/index.js';
-import { envelopeDiscriminator, prefixIssuePaths } from './envelope.js';
+import { envelopeDiscriminator, parseOrThrow } from './envelope.js';
 
 /**
  * DynamoDB Stream Envelope to extract data within NewImage/OldImage
@@ -21,29 +21,25 @@ export const DynamoDBStreamEnvelope = {
     data: unknown,
     schema: ZodType<T>
   ): DynamoDBStreamEnvelopeResponse<T>[] {
-    const parsedEnvelope = DynamoDBStreamSchema.parse(data);
+    const parsedEnvelope = parseOrThrow(
+      DynamoDBStreamSchema,
+      data,
+      'Failed to parse DynamoDB Stream envelope'
+    );
 
     const processImage = (
       image: unknown,
       imageType: 'NewImage' | 'OldImage',
       recordIndex: number
-    ) => {
-      try {
-        return image ? schema.parse(image) : undefined;
-      } catch (error) {
-        throw new ParseError(
-          `Failed to parse DynamoDB record at index ${recordIndex}`,
-          {
-            cause: prefixIssuePaths(error, [
-              'Records',
-              recordIndex,
-              'dynamodb',
-              imageType,
-            ]),
-          }
-        );
-      }
-    };
+    ) =>
+      image
+        ? parseOrThrow(
+            schema,
+            image,
+            `Failed to parse DynamoDB record at index ${recordIndex}`,
+            ['Records', recordIndex, 'dynamodb', imageType]
+          )
+        : undefined;
 
     return parsedEnvelope.Records.map((record, index) => ({
       NewImage: processImage(record.dynamodb.NewImage, 'NewImage', index),

packages/parser/src/envelopes/envelope.ts

@@ -1,4 +1,5 @@
-import { ZodError } from 'zod';
+import { ZodError, type ZodType } from 'zod';
+import { ParseError } from '../errors.js';
 
 /**
  * This is a discriminator to differentiate whether an envelope returns an array or an object
@@ -24,4 +25,30 @@ const prefixIssuePaths = (error: unknown, path: PropertyKey[]): unknown =>
       )
     : error;
 
-export { envelopeDiscriminator, prefixIssuePaths };
+/**
+ * Parses the data with the schema, and throws a `ParseError` with the given message when parsing fails.
+ *
+ * The original error is kept as the cause. When a path is given, the issue paths of a `ZodError` are
+ * prefixed with it through {@link prefixIssuePaths | `prefixIssuePaths`}.
+ *
+ * @param schema - the schema to parse the data with
+ * @param data - the data to parse
+ * @param message - the message of the `ParseError` thrown when parsing fails
+ * @param path - the path of the data inside the event
+ */
+const parseOrThrow = <T>(
+  schema: ZodType<T>,
+  data: unknown,
+  message: string,
+  path?: PropertyKey[]
+): T => {
+  try {
+    return schema.parse(data);
+  } catch (error) {
+    throw new ParseError(message, {
+      cause: path === undefined ? error : prefixIssuePaths(error, path),
+    });
+  }
+};
+
+export { envelopeDiscriminator, parseOrThrow, prefixIssuePaths };

packages/parser/src/envelopes/kafka.ts

@@ -4,8 +4,12 @@ import {
   KafkaMskEventSchema,
   KafkaSelfManagedEventSchema,
 } from '../schemas/kafka.js';
-import type { KafkaMskEvent, ParsedResult } from '../types/index.js';
-import { envelopeDiscriminator } from './envelope.js';
+import type {
+  KafkaMskEvent,
+  KafkaSelfManagedEvent,
+  ParsedResult,
+} from '../types/index.js';
+import { envelopeDiscriminator, parseOrThrow } from './envelope.js';
 
 /**
  * Get the event source from the data.
@@ -14,20 +18,17 @@ import { envelopeDiscriminator } from './envelope.js';
  *
  * @param data - The data to extract the event source from
  */
-const extractEventSource = (
-  data: unknown
-): 'aws:kafka' | 'SelfManagedKafka' => {
-  const verifiedData = z
-    .object({
+const extractEventSource = (data: unknown): 'aws:kafka' | 'SelfManagedKafka' =>
+  parseOrThrow(
+    z.object({
       eventSource: z.union([
         z.literal('aws:kafka'),
         z.literal('SelfManagedKafka'),
       ]),
-    })
-    .parse(data);
-
-  return verifiedData.eventSource;
-};
+    }),
+    data,
+    'Failed to parse Kafka envelope'
+  ).eventSource;
 
 /**
  * Kafka event envelope to extract data within body key
@@ -44,17 +45,27 @@ export const KafkaEnvelope = {
    */
   [envelopeDiscriminator]: 'array' as const,
   parse<T>(data: unknown, schema: ZodType<T>): z.infer<ZodType<T>>[] {
-    const eventSource = extractEventSource(data);
-
-    const parsedEnvelope =
-      eventSource === 'aws:kafka'
-        ? KafkaMskEventSchema.parse(data)
-        : KafkaSelfManagedEventSchema.parse(data);
+    const parsedEnvelope = parseOrThrow<KafkaMskEvent | KafkaSelfManagedEvent>(
+      extractEventSource(data) === 'aws:kafka'
+        ? KafkaMskEventSchema
+        : KafkaSelfManagedEventSchema,
+      data,
+      'Failed to parse Kafka envelope'
+    );
 
     const values: z.infer<ZodType<T>>[] = [];
-    for (const topicRecord of Object.values(parsedEnvelope.records)) {
-      for (const record of topicRecord) {
-        values.push(schema.parse(record.value));
+    for (const [topicKey, topicRecord] of Object.entries(
+      parsedEnvelope.records
+    )) {
+      for (const [index, record] of topicRecord.entries()) {
+        values.push(
+          parseOrThrow(
+            schema,
+            record.value,
+            `Failed to parse Kafka record at index ${index} of ${topicKey}`,
+            ['records', topicKey, index, 'value']
+          )
+        );
       }
     }
 

packages/parser/src/envelopes/kinesis-firehose.ts

@@ -2,7 +2,7 @@ import { ZodError, type ZodType, type z } from 'zod';
 import { ParseError } from '../errors.js';
 import { KinesisFirehoseSchema } from '../schemas/index.js';
 import type { ParsedResult } from '../types/index.js';
-import { envelopeDiscriminator, prefixIssuePaths } from './envelope.js';
+import { envelopeDiscriminator, parseOrThrow } from './envelope.js';
 
 /**
  * Kinesis Firehose Envelope to extract array of Records
@@ -23,29 +23,20 @@ export const KinesisFirehoseEnvelope = {
    */
   [envelopeDiscriminator]: 'array' as const,
   parse<T>(data: unknown, schema: ZodType<T>): T[] {
-    let parsedEnvelope: z.infer<typeof KinesisFirehoseSchema>;
-    try {
-      parsedEnvelope = KinesisFirehoseSchema.parse(data);
-    } catch (error) {
-      throw new ParseError('Failed to parse Kinesis Firehose envelope', {
-        cause: error as Error,
-      });
-    }
+    const parsedEnvelope = parseOrThrow(
+      KinesisFirehoseSchema,
+      data,
+      'Failed to parse Kinesis Firehose envelope'
+    );
 
-    return parsedEnvelope.records.map((record, recordIndex) => {
-      let parsedRecord: T;
-      try {
-        parsedRecord = schema.parse(record.data);
-      } catch (error) {
-        throw new ParseError(
-          `Failed to parse Kinesis Firehose record at index ${recordIndex}`,
-          {
-            cause: prefixIssuePaths(error, ['records', recordIndex, 'data']),
-          }
-        );
-      }
-      return parsedRecord;
-    });
+    return parsedEnvelope.records.map((record, recordIndex) =>
+      parseOrThrow(
+        schema,
+        record.data,
+        `Failed to parse Kinesis Firehose record at index ${recordIndex}`,
+        ['records', recordIndex, 'data']
+      )
+    );
   },
 
   safeParse<T>(data: unknown, schema: ZodType<T>): ParsedResult<unknown, T[]> {

packages/parser/src/envelopes/kinesis.ts

@@ -2,7 +2,7 @@ import { ZodError, type ZodType, type z } from 'zod';
 import { ParseError } from '../errors.js';
 import { KinesisDataStreamSchema } from '../schemas/kinesis.js';
 import type { ParsedResult } from '../types/index.js';
-import { envelopeDiscriminator, prefixIssuePaths } from './envelope.js';
+import { envelopeDiscriminator, parseOrThrow } from './envelope.js';
 
 /**
  * Kinesis Data Stream Envelope to extract array of Records
@@ -21,34 +21,20 @@ export const KinesisEnvelope = {
    */
   [envelopeDiscriminator]: 'array' as const,
   parse<T>(data: unknown, schema: ZodType<T>): T[] {
-    let parsedEnvelope: z.infer<typeof KinesisDataStreamSchema>;
-    try {
-      parsedEnvelope = KinesisDataStreamSchema.parse(data);
-    } catch (error) {
-      throw new ParseError('Failed to parse Kinesis Data Stream envelope', {
-        cause: error as Error,
-      });
-    }
+    const parsedEnvelope = parseOrThrow(
+      KinesisDataStreamSchema,
+      data,
+      'Failed to parse Kinesis Data Stream envelope'
+    );
 
-    return parsedEnvelope.Records.map((record, recordIndex) => {
-      let parsedRecord: T;
-      try {
-        parsedRecord = schema.parse(record.kinesis.data);
-      } catch (error) {
-        throw new ParseError(
-          `Failed to parse Kinesis Data Stream record at index ${recordIndex}`,
-          {
-            cause: prefixIssuePaths(error, [
-              'Records',
-              recordIndex,
-              'kinesis',
-              'data',
-            ]),
-          }
-        );
-      }
-      return parsedRecord;
-    });
+    return parsedEnvelope.Records.map((record, recordIndex) =>
+      parseOrThrow(
+        schema,
+        record.kinesis.data,
+        `Failed to parse Kinesis Data Stream record at index ${recordIndex}`,
+        ['Records', recordIndex, 'kinesis', 'data']
+      )
+    );
   },
 
   safeParse<T>(data: unknown, schema: ZodType<T>): ParsedResult<unknown, T[]> {

packages/parser/src/envelopes/sns-sqs.ts

@@ -3,7 +3,11 @@ import { ParseError } from '../errors.js';
 import { SnsSqsNotificationSchema } from '../schemas/sns.js';
 import { SqsSchema } from '../schemas/sqs.js';
 import type { ParsedResult, SnsSqsNotification } from '../types/index.js';
-import { envelopeDiscriminator, prefixIssuePaths } from './envelope.js';
+import {
+  envelopeDiscriminator,
+  parseOrThrow,
+  prefixIssuePaths,
+} from './envelope.js';
 
 const createError = (index: number, issues: z.core.$ZodIssue[]) => ({
   issues: issues.map((issue) => ({
@@ -57,14 +61,11 @@ export const SnsSqsEnvelope = {
    */
   [envelopeDiscriminator]: 'array' as const,
   parse<T>(data: unknown, schema: ZodType<T>): T[] {
-    let parsedEnvelope: z.infer<typeof SqsSchema>;
-    try {
-      parsedEnvelope = SqsSchema.parse(data);
-    } catch (error) {
-      throw new ParseError('Failed to parse SQS Envelope', {
-        cause: error as Error,
-      });
-    }
+    const parsedEnvelope = parseOrThrow(
+      SqsSchema,
+      data,
+      'Failed to parse SQS Envelope'
+    );
 
     return parsedEnvelope.Records.map((record, recordIndex) => {
       let body: unknown;