feat(parser): support EventBridge WITH_METADATA deliveries

packages/parser/src/envelopes/eventbridge-with-metadata.ts

@@ -0,0 +1,70 @@
+import type { ZodType } from 'zod';
+import { ParseError } from '../errors.js';
+import { EventBridgeSchema } from '../schemas/eventbridge.js';
+import { EventBridgeWithMetadataRecordSchema } from '../schemas/eventbridge-with-metadata.js';
+import type { ParsedResult } from '../types/index.js';
+import { envelopeDiscriminator } from './envelope.js';
+
+/**
+ * Validates a WITH_METADATA batch and extracts each record's parsed `Data.detail`.
+ *
+ * Returns details in input order. Invalid records fail the whole batch, with
+ * validation paths relative to the original input. Use the record schema
+ * directly for single-object delivery or to retain delivery metadata.
+ */
+const EventBridgeWithMetadataEnvelope = {
+  /**
+   * Identifies the envelope's array output.
+   * @hidden
+   */
+  [envelopeDiscriminator]: 'array' as const,
+
+  /**
+   * Parses every record's detail using the supplied schema.
+   *
+   * @param data - the batch to parse
+   * @param schema - the schema for each record's detail
+   */
+  parse<T>(data: unknown, schema: ZodType<T>): T[] {
+    try {
+      return EventBridgeWithMetadataRecordSchema.extend({
+        Data: EventBridgeSchema.extend({ detail: schema }),
+      })
+        .array()
+        .nonempty()
+        .parse(data)
+        .map((record) => record.Data.detail);
+    } catch (error) {
+      throw new ParseError(
+        'Failed to parse EventBridge WITH_METADATA envelope',
+        {
+          cause: error,
+        }
+      );
+    }
+  },
+
+  /**
+   * Parses a batch, returning the original input and error on failure.
+   *
+   * @param data - the batch to parse
+   * @param schema - the schema for each record's detail
+   */
+  safeParse<T>(data: unknown, schema: ZodType<T>): ParsedResult<unknown, T[]> {
+    try {
+      return {
+        success: true,
+        data: EventBridgeWithMetadataEnvelope.parse(data, schema),
+      };
+    } catch (error) {
+      return {
+        success: false,
+        // parse always wraps validation and transform failures in ParseError.
+        error: error as ParseError,
+        originalEvent: data,
+      };
+    }
+  },
+};
+
+export { EventBridgeWithMetadataEnvelope };

packages/parser/src/envelopes/index.ts

@@ -3,6 +3,7 @@ export { ApiGatewayV2Envelope } from './api-gatewayv2.js';
 export { CloudWatchEnvelope } from './cloudwatch.js';
 export { DynamoDBStreamEnvelope } from './dynamodb.js';
 export { EventBridgeEnvelope } from './eventbridge.js';
+export { EventBridgeWithMetadataEnvelope } from './eventbridge-with-metadata.js';
 export { KafkaEnvelope } from './kafka.js';
 export { KinesisEnvelope } from './kinesis.js';
 export { KinesisFirehoseEnvelope } from './kinesis-firehose.js';

packages/parser/src/schemas/eventbridge-with-metadata.ts

@@ -0,0 +1,51 @@
+import { z } from 'zod';
+import { EventBridgeSchema } from './eventbridge.js';
+
+/**
+ * Validates system metadata for an EventBridge WITH_METADATA delivery.
+ *
+ * Only `aws:DeliveryType` is required. All metadata values, including sequence
+ * numbers that can exceed JavaScript's safe integer range, remain strings.
+ *
+ * @see https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-custom-bus-addressing.html
+ */
+const EventBridgeWithMetadataSystemMetadataSchema = z.object({
+  'aws:DeliveryType': z.enum(['LIVE', 'REPLAY']),
+  ContentType: z.string().optional(),
+  EventGroupId: z.string().optional(),
+  DeduplicationId: z.string().optional(),
+  'aws:EventId': z.string().optional(),
+  'aws:IngestionTime': z.string().optional(),
+  'aws:SequenceNumber': z.string().optional(),
+  'aws:Source': z.string().optional(),
+  'aws:DetailType': z.string().optional(),
+  'aws:SchemaId': z.string().optional(),
+  'aws:RegistryType': z.string().optional(),
+});
+
+/**
+ * Validates one WITH_METADATA record containing a standard EventBridge event.
+ *
+ * Use this schema directly for single-object delivery with `MaxBatchSize: 1`.
+ * Arbitrary `PutRawEvents` payloads are outside this schema's scope.
+ */
+const EventBridgeWithMetadataRecordSchema = z.object({
+  Data: EventBridgeSchema,
+  Metadata: z.record(z.string(), z.string()).optional(),
+  SystemMetadata: EventBridgeWithMetadataSystemMetadataSchema,
+});
+
+/**
+ * Validates a nonempty batch of WITH_METADATA records.
+ *
+ * Extend {@link EventBridgeWithMetadataRecordSchema | `EventBridgeWithMetadataRecordSchema`}
+ * before creating an array schema to validate application details and retain metadata.
+ */
+const EventBridgeWithMetadataSchema =
+  EventBridgeWithMetadataRecordSchema.array().nonempty();
+
+export {
+  EventBridgeWithMetadataRecordSchema,
+  EventBridgeWithMetadataSchema,
+  EventBridgeWithMetadataSystemMetadataSchema,
+};

packages/parser/src/schemas/index.ts

@@ -58,6 +58,11 @@ export {
   DynamoDBStreamToKinesisRecord,
 } from './dynamodb.js';
 export { EventBridgeSchema } from './eventbridge.js';
+export {
+  EventBridgeWithMetadataRecordSchema,
+  EventBridgeWithMetadataSchema,
+  EventBridgeWithMetadataSystemMetadataSchema,
+} from './eventbridge-with-metadata.js';
 export {
   KafkaMskEventSchema,
   KafkaRecordSchema,

packages/parser/src/types/index.ts

@@ -39,6 +39,9 @@ export type {
   DynamoDBStreamRecord,
   DynamoDBStreamToKinesisRecordEvent,
   EventBridgeEvent,
+  EventBridgeWithMetadataEvent,
+  EventBridgeWithMetadataRecord,
+  EventBridgeWithMetadataSystemMetadata,
   KafkaMskEvent,
   KafkaRecord,
   KafkaSelfManagedEvent,

packages/parser/src/types/schema.ts

@@ -24,6 +24,9 @@ import type {
   DynamoDBStreamSchema,
   DynamoDBStreamToKinesisRecord,
   EventBridgeSchema,
+  EventBridgeWithMetadataRecordSchema,
+  EventBridgeWithMetadataSchema,
+  EventBridgeWithMetadataSystemMetadataSchema,
   KafkaMskEventSchema,
   KafkaRecordSchema,
   KafkaSelfManagedEventSchema,
@@ -122,6 +125,21 @@ type DynamoDBStreamToKinesisRecordEvent = z.infer<
 
 type EventBridgeEvent = z.infer<typeof EventBridgeSchema>;
 
+/** Represents the system metadata of a WITH_METADATA delivery. */
+type EventBridgeWithMetadataSystemMetadata = z.infer<
+  typeof EventBridgeWithMetadataSystemMetadataSchema
+>;
+
+/** Represents one WITH_METADATA record containing a standard EventBridge event. */
+type EventBridgeWithMetadataRecord = z.infer<
+  typeof EventBridgeWithMetadataRecordSchema
+>;
+
+/** Represents a nonempty batch of WITH_METADATA records. */
+type EventBridgeWithMetadataEvent = z.infer<
+  typeof EventBridgeWithMetadataSchema
+>;
+
 type KafkaSelfManagedEvent = z.infer<typeof KafkaSelfManagedEventSchema>;
 
 type KafkaRecord = z.infer<typeof KafkaRecordSchema>;
@@ -201,6 +219,9 @@ export type {
   DynamoDBStreamRecord,
   DynamoDBStreamToKinesisRecordEvent,
   EventBridgeEvent,
+  EventBridgeWithMetadataEvent,
+  EventBridgeWithMetadataRecord,
+  EventBridgeWithMetadataSystemMetadata,
   KafkaMskEvent,
   KafkaRecord,
   KafkaSelfManagedEvent,

packages/parser/tests/types/eventbridge-with-metadata.test-d.ts

@@ -0,0 +1,63 @@
+import { describe, expectTypeOf, it } from 'vitest';
+import { z } from 'zod';
+import { EventBridgeWithMetadataEnvelope } from '../../src/envelopes/index.js';
+import { parse } from '../../src/parser.js';
+import type {
+  EventBridgeWithMetadataRecordSchema,
+  EventBridgeWithMetadataSchema,
+} from '../../src/schemas/index.js';
+import type {
+  EventBridgeWithMetadataEvent,
+  EventBridgeWithMetadataRecord,
+  EventBridgeWithMetadataSystemMetadata,
+  ParsedResult,
+} from '../../src/types/index.js';
+import type { ParserOutput } from '../../src/types/parser.js';
+
+describe('Types: EventBridge WITH_METADATA', () => {
+  it('infers transformed detail arrays for parsing and safe parsing', () => {
+    // Prepare
+    const schema = z.object({ orderId: z.string().transform(Number) });
+    const event: unknown = [];
+
+    // Act
+    const result = parse(event, EventBridgeWithMetadataEnvelope, schema);
+    const safeResult = parse(
+      event,
+      EventBridgeWithMetadataEnvelope,
+      schema,
+      true
+    );
+
+    // Assess
+    expectTypeOf(result).toEqualTypeOf<{ orderId: number }[]>();
+    expectTypeOf(safeResult).toEqualTypeOf<
+      ParsedResult<unknown, { orderId: number }[]>
+    >();
+    expectTypeOf<
+      ParserOutput<typeof schema, typeof EventBridgeWithMetadataEnvelope>
+    >().toEqualTypeOf<{ orderId: number }[]>();
+  });
+
+  it('exports inferred record and batch types', () => {
+    // Assess
+    expectTypeOf<
+      z.infer<typeof EventBridgeWithMetadataRecordSchema>
+    >().toEqualTypeOf<EventBridgeWithMetadataRecord>();
+    expectTypeOf<
+      z.infer<typeof EventBridgeWithMetadataSchema>
+    >().toEqualTypeOf<EventBridgeWithMetadataEvent>();
+    expectTypeOf<EventBridgeWithMetadataEvent>().toEqualTypeOf<
+      EventBridgeWithMetadataRecord[]
+    >();
+    expectTypeOf<EventBridgeWithMetadataRecord['Metadata']>().toEqualTypeOf<
+      Record<string, string> | undefined
+    >();
+    expectTypeOf<
+      EventBridgeWithMetadataSystemMetadata['aws:SequenceNumber']
+    >().toEqualTypeOf<string | undefined>();
+    expectTypeOf<
+      EventBridgeWithMetadataSystemMetadata['aws:DeliveryType']
+    >().toEqualTypeOf<'LIVE' | 'REPLAY'>();
+  });
+});