Events: Event Schema

Events: Event Schema

Define and evolve event schemas using a schema registry with Avro, Protobuf, or JSON Schema.

When to Use

• Multiple services exchange events and must agree on the data structure
• You need to evolve event schemas without breaking existing consumers
• You're using Kafka and need schema validation at the broker level
• You want compile-time type safety for event payloads across services
• You're building a long-lived event-sourced system where events are stored permanently

Instructions

JSON Schema + Zod (TypeScript-native, no registry needed for small teams):

import { z } from 'zod';

// Define schemas with Zod — generates TypeScript types automatically
const OrderCreatedEventSchema = z.object({
  eventId: z.string().uuid(),
  eventType: z.literal('ORDER_CREATED'),
  schemaVersion: z.literal(1),
  aggregateId: z.string(),
  aggregateType: z.literal('Order'),
  occurredAt: z.string().datetime(),
  payload: z.object({
    orderId: z.string().uuid(),
    userId: z.string().uuid(),
    items: z.array(
      z.object({
        productId: z.string(),
        quantity: z.number().int().positive(),
        unitPrice: z.number().positive(),
      })
    ),
    totalAmount: z.number().positive(),
    currency: z.string().length(3), // ISO 4217
  }),
});

type OrderCreatedEvent = z.infer<typeof OrderCreatedEventSchema>;

// Envelope schema for all events
const EventEnvelopeSchema = z.object({
  eventId: z.string().uuid(),
  eventType: z.string(),
  schemaVersion: z.number().int().positive(),
  aggregateId: z.string(),
  occurredAt: z.string().datetime(),
  correlationId: z.string().uuid().optional(),
  causationId: z.string().uuid().optional(),
});

// Validate incoming events
function parseOrderEvent(raw: unknown): OrderCreatedEvent {
  return OrderCreatedEventSchema.parse(raw); // throws ZodError if invalid
}

// Create events with correct structure
function createOrderEvent(order: Order, correlationId?: string): OrderCreatedEvent {
  return {
    eventId: crypto.randomUUID(),
    eventType: 'ORDER_CREATED',
    schemaVersion: 1,
    aggregateId: order.id,
    aggregateType: 'Order',
    occurredAt: new Date().toISOString(),
    payload: {
      orderId: order.id,
      userId: order.userId,
      items: order.items,
      totalAmount: order.total,
      currency: 'USD',
    },
  };
}

Schema versioning strategy:

// Version 1
const UserCreatedV1 = z.object({
  schemaVersion: z.literal(1),
  userId: z.string(),
  email: z.string().email(),
});

// Version 2 — backward compatible (added optional field)
const UserCreatedV2 = z.object({
  schemaVersion: z.literal(2),
  userId: z.string(),
  email: z.string().email(),
  name: z.string().optional(), // new optional field — safe addition
});

// Discriminated union for versioned parsing
const UserCreatedEvent = z.discriminatedUnion('schemaVersion', [UserCreatedV1, UserCreatedV2]);

function parseUserCreated(raw: unknown) {
  const event = UserCreatedEvent.parse(raw);
  // Normalize to latest version
  if (event.schemaVersion === 1) {
    return { ...event, schemaVersion: 2 as const, name: undefined };
  }
  return event;
}

Event registry for type-safe dispatch:

// Central event registry
type EventRegistry = {
  'order.created': OrderCreatedEvent;
  'order.shipped': OrderShippedEvent;
  'user.created': UserCreatedEvent;
  'payment.failed': PaymentFailedEvent;
};

type EventType = keyof EventRegistry;

// Schema map for validation
const eventSchemas: { [K in EventType]: z.ZodType<EventRegistry[K]> } = {
  'order.created': OrderCreatedEventSchema,
  'order.shipped': OrderShippedEventSchema,
  'user.created': UserCreatedEventSchema,
  'payment.failed': PaymentFailedEventSchema,
};

function validateEvent<T extends EventType>(type: T, raw: unknown): EventRegistry[T] {
  return eventSchemas[type].parse(raw) as EventRegistry[T];
}

Details

Schema compatibility rules (Avro / Confluent conventions apply to JSON Schema too):

Change	Backward Compatible	Forward Compatible
Add optional field	Yes	Yes
Add required field	No	Yes
Remove optional field	Yes	No
Remove required field	No	No
Rename field	No	No (use aliases)
Change field type	No	No

Safe evolution pattern: Never remove or rename fields. Add fields as optional. Bump schemaVersion. Keep old schemas in the registry for producers still on v1.

Confluent Schema Registry (for Kafka): Stores Avro/Protobuf/JSON schemas. Producers register schemas; consumers validate against them. Enforces compatibility rules at publish time — prevents breaking changes from reaching consumers.

Anti-patterns:

• Using any or untyped JSON as event payloads — schema drift becomes undetectable
• Breaking schema changes without a version bump — consumers parse incorrectly in silence
• Storing schema versions only in docs, not in the event envelope — impossible to know how to parse at runtime

Event ID for idempotency: Always include eventId (UUID). Consumers use it to deduplicate redelivered events.

Source

microservices.io/patterns/data/event-sourcing.html

Process

1. Read the instructions and examples in this document.
2. Apply the patterns to your implementation, adapting to your specific context.
3. Verify your implementation against the details and edge cases listed above.

Harness Integration

• Type: knowledge — this skill is a reference document, not a procedural workflow.
• No tools or state — consumed as context by other skills and agents.

Success Criteria

• The patterns described in this document are applied correctly in the implementation.
• Edge cases and anti-patterns listed in this document are avoided.