event-driven

name	event-driven
description	Event-driven architecture patterns including message queues, pub/sub, event sourcing, CQRS, and sagas. Use when implementing async messaging, distributed transactions, event stores, command query separation, domain events, integration events, data streaming, choreography, orchestration, or integrating with RabbitMQ, Kafka, Apache Pulsar, AWS SQS, AWS SNS, NATS, event buses, or message brokers.
triggers	["event","message","messaging","pub/sub","pubsub","publish/subscribe","kafka","rabbitmq","sqs","sns","nats","pulsar","event sourcing","CQRS","saga","choreography","orchestration","event store","domain event","integration event","message queue","message broker","event bus","data streaming","stream processing","event-driven"]

Event-Driven Architecture

Overview

Event-driven architecture (EDA) enables loosely coupled, scalable systems by communicating through events rather than direct calls. This skill covers message queues, pub/sub patterns, event sourcing, CQRS, distributed transaction management with sagas, and data streaming with Kafka.

Available Agents

• senior-software-engineer (Opus) - Architecture design, pattern selection, distributed system design
• software-engineer (Sonnet) - Event handler implementation, consumer/producer code
• security-engineer (Opus) - Event security, authorization patterns, message encryption
• senior-infrastructure-engineer (Opus) - Message broker setup, Kafka clusters, queue configuration

Key Concepts

Message Queues

RabbitMQ Implementation:

import amqp, { Channel, Connection } from "amqplib";

interface QueueConfig {
  name: string;
  durable: boolean;
  deadLetterExchange?: string;
  messageTtl?: number;
  maxRetries?: number;
}

class RabbitMQClient {
  private connection: Connection | null = null;
  private channel: Channel | null = null;

  async connect(url: string): Promise<void> {
    this.connection = await amqp.connect(url);
    this.channel = await this.connection.createChannel();

    // Handle connection errors
    this.connection.on("error", (err) => {
      console.error("RabbitMQ connection error:", err);
      this.reconnect(url);
    });
  }

  async setupQueue(config: QueueConfig): Promise<void> {
    if (!this.channel) throw new Error("Not connected");

    const options: amqp.Options.AssertQueue = {
      durable: config.durable,
      arguments: {},
    };

    if (config.deadLetterExchange) {
      options.arguments!["x-dead-letter-exchange"] = config.deadLetterExchange;
    }
    if (config.messageTtl) {
      options.arguments!["x-message-ttl"] = config.messageTtl;
    }

    await this.channel.assertQueue(config.name, options);
  }

  async publish(
    queue: string,
    message: unknown,
    options?: PublishOptions
  ): Promise<void> {
    if (!this.channel) throw new Error("Not connected");

    const content = Buffer.from(JSON.stringify(message));
    const publishOptions: amqp.Options.Publish = {
      persistent: true,
      messageId: options?.messageId || crypto.randomUUID(),
      timestamp: Date.now(),
      headers: options?.headers,
    };

    this.channel.sendToQueue(queue, content, publishOptions);
  }

  async consume<T>(
    queue: string,
    handler: (
      message: T,
      ack: () => void,
      nack: (requeue?: boolean) => void
    ) => Promise<void>,
    options?: ConsumeOptions
  ): Promise<void> {
    if (!this.channel) throw new Error("Not connected");

    await this.channel.prefetch(options?.prefetch || 10);

    await this.channel.consume(queue, async (msg) => {
      if (!msg) return;

      try {
        const content: T = JSON.parse(msg.content.toString());
        const retryCount =
          (msg.properties.headers?.["x-retry-count"] as number) || 0;

        await handler(
          content,
          () => this.channel!.ack(msg),
          (requeue = false) => {
            if (requeue && retryCount < (options?.maxRetries || 3)) {
              // Requeue with incremented retry count
              this.channel!.nack(msg, false, false);
              this.publish(queue, content, {
                headers: { "x-retry-count": retryCount + 1 },
              });
            } else {
              this.channel!.nack(msg, false, false); // Send to DLQ
            }
          }
        );
      } catch (error) {
        console.error("Message processing error:", error);
        this.channel!.nack(msg, false, false);
      }
    });
  }
}

AWS SQS Implementation:

import {
  SQSClient,
  SendMessageCommand,
  ReceiveMessageCommand,
  DeleteMessageCommand,
} from "@aws-sdk/client-sqs";

interface SQSMessage<T> {
  id: string;
  body: T;
  receiptHandle: string;
  approximateReceiveCount: number;
}

class SQSQueue<T> {
  private client: SQSClient;
  private queueUrl: string;

  constructor(queueUrl: string, region: string = "us-east-1") {
    this.client = new SQSClient({ region });
    this.queueUrl = queueUrl;
  }

  async send(
    message: T,
    options?: { delaySeconds?: number; deduplicationId?: string }
  ): Promise<string> {
    const command = new SendMessageCommand({
      QueueUrl: this.,
      : .(message),
      : options?.,
      : options?.,
      : options?. ?  : ,
    });

     response =  ..(command);
     response.!;
  }

   (
    :  = ,
    :  = 
  ): <<T>[]> {
     command =  ({
      : .,
      : maxMessages,
      : waitTimeSeconds,
      : [],
    });

     response =  ..(command);

     (response. || []).( ({
      : msg.!,
      : .(msg.!)  T,
      : msg.!,
      : (
        msg.?. || 
      ),
    }));
  }

   (: ): <> {
     command =  ({
      : .,
      : receiptHandle,
    });
     ..(command);
  }

   (
    :  <>,
    ?: { ?: ; ?:  }
  ): <> {
     maxRetries = options?. || ;

     () {
       messages =  .();

       .(
        messages.( (msg) => {
           {
             (msg.);
             .(msg.);
          }  (error) {
            .(, error);

             (msg. >= maxRetries) {
              
              .();
            }
            
          }
        })
      );

       (messages. ===  && options?.) {
          ( (r, options.));
      }
    }
  }
}

Pub/Sub Patterns

Kafka Implementation:

import { Kafka, Producer, Consumer, EachMessagePayload } from "kafkajs";

interface Event<T = unknown> {
  id: string;
  type: string;
  timestamp: Date;
  source: string;
  data: T;
  metadata?: Record<string, string>;
}

class KafkaEventBus {
  private kafka: Kafka;
  private producer: Producer | null = null;
  private consumers: Map<string, Consumer> = new Map();

  constructor(config: { brokers: string[]; clientId: string }) {
    this.kafka = new Kafka({
      clientId: config.clientId,
      brokers: config.brokers,
    });
  }

  async (): <> {
    . = ..({
      : ,
      : ,
    });
     ..();
  }

   publish<T>(
    : ,
    : <<T>,  | >
  ): <> {
     (!.)   ();

     : <T> = {
      ...event,
      : crypto.(),
      :  (),
    };

     ..({
      topic,
      : [
        {
          :
            event. &&  event. ===  &&   event.
              ? ((event.  { :  }).)
              : fullEvent.,
          : .(fullEvent),
          : {
            : event.,
            : event.,
          },
        },
      ],
    });
  }

   subscribe<T>(
    : [],
    : ,
    :  <>,
    ?: { ?:  }
  ): <> {
     consumer = ..({ groupId });
     consumer.();

     ( topic  topics) {
       consumer.({
        topic,
        : options?.,
      });
    }

    ..(groupId, consumer);

     consumer.({
      :  ({
        topic,
        partition,
        message,
      }: ) => {
         {
           : <T> = .(message.!.());
           (event);
        }  (error) {
          .(
            ,
            error
          );
           error; 
        }
      },
    });
  }

   (): <> {
     .?.();
     ( consumer  ..()) {
       consumer.();
    }
  }
}


 eventBus =  ({
  : [],
  : ,
});

 eventBus.();


 eventBus.<>(, {
  : ,
  : ,
  : { : , : [], :  },
});


 eventBus.<>(
  [],
  ,
   (event) => {
     (event. === ) {
       (event.);
    }
  }
);

NATS Implementation:

import {
  connect,
  NatsConnection,
  StringCodec,
  JetStreamManager,
  JetStreamClient,
} from "nats";

class NATSEventBus {
  private nc: NatsConnection | null = null;
  private js: JetStreamClient | null = null;
  private sc = StringCodec();

  async connect(servers: string[]): Promise<void> {
    this.nc = await connect({ servers });

    // Setup JetStream for persistence
    const jsm = await this.nc.jetstreamManager();
    this.js = this.nc.jetstream();

    // Create stream if not exists
    try {
      await jsm.streams.add({
        name: "EVENTS",
        subjects: ["events.*"],
        : ,
        : ,
        :  *  *  *  * , 
      });
    }  (e) {
      
    }
  }

   (: , : ): <> {
     (!.)   ();

     ..(
      ,
      ..(.(data))
    );
  }

   (
    : ,
    : ,
    :  <>
  ): <> {
     (!.)   ();

     consumer =  ..
      .(, durableName)
      .( () => {
        
         jsm =  .!.();
         jsm..(, {
          : durableName,
          : ,
          : ,
          : ,
        });
         .!..(, durableName);
      });

     messages =  consumer.();

      ( msg  messages) {
       {
         data = .(..(msg.));
         (data);
        msg.();
      }  (error) {
        .(, error);
        msg.();
      }
    }
  }
}

Event Sourcing

interface DomainEvent {
  id: string;
  aggregateId: string;
  aggregateType: string;
  type: string;
  version: number;
  timestamp: Date;
  data: unknown;
  metadata: {
    userId?: string;
    correlationId?: string;
    causationId?: string;
  };
}

interface EventStore {
  append(events: DomainEvent[]): Promise<void>;
  getEvents(aggregateId: string, fromVersion?: number): Promise<DomainEvent[]>;
  getEventsByType(type: string, fromTimestamp?: Date): Promise<DomainEvent[]>;
}

// PostgreSQL Event Store
class PostgresEventStore implements EventStore {
  constructor(private pool: ) {}

   (: []): <> {
     client =  ..();

     {
       client.();

       ( event  events) {
        
         { rows } =  client.(
          ,
          [event.]
        );

         currentVersion = rows[]?. || ;
         (event. !== currentVersion + ) {
            (
            
          );
        }

         client.(
          ,
          [
            event.,
            event.,
            event.,
            event.,
            event.,
            event.,
            .(event.),
            .(event.),
          ]
        );
      }

       client.();

      
       ( event  events) {
         ..(event);
      }
    }  (error) {
       client.();
       error;
    }  {
      client.();
    }
  }

   (
    : ,
    :  = 
  ): <[]> {
     { rows } =  ..(
      ,
      [aggregateId, fromVersion]
    );

     rows.(.);
  }
}


   {
   : ;
   :  = ;
   : [] = [];

   ():  {
     .;
  }
   ():  {
     .;
  }

  () {
    . = id;
  }

   (
    : <
      ,
       |  |  |  | 
    >
  ):  {
     :  = {
      ...event,
      : crypto.(),
      : .,
      : ..,
      : . + ,
      :  (),
    };

    .(domainEvent);
    . = domainEvent.;
    ..(domainEvent);
  }

    (: ): ;

  (: []):  {
     ( event  events) {
      .(event);
      . = event.;
    }
  }

  (): [] {
     [....];
  }

  ():  {
    . = [];
  }
}


    {
   :
    | 
    | 
    | 
    | 
    |  = ;
   : [] = [];
   :  = ;

   (: , : , : []):  {
     order =  (id);
    order.({
      : ,
      : { customerId, items },
      : {},
    });
     order;
  }

  ():  {
     (. !== ) {
        ();
    }
    .({
      : ,
      : { :  () },
      : {},
    });
  }

  (: ):  {
     ([, , ].(.)) {
        ();
    }
    .({
      : ,
      : { reason, :  () },
      : {},
    });
  }

   (: ):  {
     (event.) {
       :
         data = event.  { : [] };
        . = data.;
        . = data..(
           sum + item. * item.,
          
        );
        . = ;
        ;
       :
        . = ;
        ;
       :
        . = ;
        ;
    }
  }
}

CQRS (Command Query Responsibility Segregation)

// Commands
interface Command {
  type: string;
  payload: unknown;
  metadata: {
    userId: string;
    correlationId: string;
    timestamp: Date;
  };
}

interface CommandHandler<T extends Command> {
  handle(command: T): Promise<void>;
}

// Command Bus
class CommandBus {
  private handlers: Map<string, CommandHandler<Command>> = new Map();

  register<T extends Command>(type: string, handler: CommandHandler<T>): void {
    this.handlers.set(type, handler as CommandHandler<Command>);
  }

  async dispatch(command: Command): Promise<void> {
     handler = ..(command.);
     (!handler) {
        ();
    }
     handler.(command);
  }
}


 <> {
  : ;
  : ;
}

 <  <>, > {
  (: ): <>;
}


  {
   : <, <<>, >> =
     ();

  register<  <>, >(
    : ,
    : <, >
  ):  {
    ..(, handler  <<>, >);
  }

   execute<>(: <>): <> {
     handler = ..(query.);
     (!handler) {
        ();
    }
     handler.(query)  <>;
  }
}


  {
  : ;
  : ;
  : ;
  : ;
  : <{
    : ;
    : ;
    : ;
    : ;
  }>;
  : ;
  : ;
  : ;
}

  {
  () {
    .();
  }

   ():  {
    ..(, ..());
    ..(, ..());
    ..(, ..());
  }

    (: ): <> {
     data = event.  ;

    
     customer =  ...(data.);

    
     items =  .(
      data..( (item) => {
         product =  ...(item.);
         {
          ...item,
          : product.,
        };
      })
    );

     ...({
      : event.,
      : data.,
      : customer.,
      : ,
      items,
      : items.( sum + i. * i., ),
      : event.,
      : event.,
    });
  }

    (: ): <> {
     ...(event., {
      : ,
      : event.,
    });
  }

    (: ): <> {
     ...(event., {
      : ,
      : event.,
    });
  }
}

Saga Pattern for Distributed Transactions

interface SagaStep<TData> {
  name: string;
  execute: (data: TData) => Promise<void>;
  compensate: (data: TData) => Promise<void>;
}

interface SagaDefinition<TData> {
  name: string;
  steps: SagaStep<TData>[];
}

interface SagaInstance {
  id: string;
  sagaName: string;
  data: unknown;
  currentStep: number;
  status: "running" | "completed" | "compensating" | "failed";
  completedSteps: string[];
  error?: string;
  startedAt: Date;
  updatedAt: Date;
}

class SagaOrchestrator {
  private sagas: Map<, <>> =  ();
   : ;

  register<>(: <>):  {
    ..(saga., saga  <>);
  }

   start<>(: , : ): <> {
     saga = ..(sagaName);
     (!saga)   ();

     :  = {
      : crypto.(),
      sagaName,
      data,
      : ,
      : ,
      : [],
      :  (),
      :  (),
    };

     ..(instance);
     .(instance, saga);

     instance.;
  }

    (
    : ,
    : <>
  ): <> {
     (instance. >= saga..) {
      instance. = ;
       ..(instance);
      ;
    }

     step = saga.[instance.];

     {
       step.(instance.);

      instance..(step.);
      instance.++;
      instance. =  ();
       ..(instance);

       .(instance, saga);
    }  (error) {
      instance. = ;
      instance. = error   ? error. : (error);
       ..(instance);

       .(instance, saga);
    }
  }

    (
    : ,
    : <>
  ): <> {
    
     ( i = instance.. - ; i >= ; i--) {
       stepName = instance.[i];
       step = saga..( s. === stepName);

       (step) {
         {
           step.(instance.);
        }  (error) {
          .(, error);
          
        }
      }
    }

    instance. = ;
    instance. =  ();
     ..(instance);
  }
}


  {
  : ;
  : ;
  : <{ : ; : ; :  }>;
  ?: ;
  ?: ;
}

 : <> = {
  : ,
  : [
    {
      : ,
      :  (data) => {
         inventoryService.(data.);
      },
      :  (data) => {
         inventoryService.(data.);
      },
    },
    {
      : ,
      :  (data) => {
         total = data..(
           sum + i. * i.,
          
        );
         payment =  paymentService.(data., total);
        data. = payment.;
      },
      :  (data) => {
         (data.) {
           paymentService.(data.);
        }
      },
    },
    {
      : ,
      :  (data) => {
         shipment =  shippingService.(
          data.,
          data.
        );
        data. = shipment.;
      },
      :  (data) => {
         (data.) {
           shippingService.(data.);
        }
      },
    },
    {
      : ,
      :  (data) => {
         orderService.(data.);
      },
      :  (data) => {
         orderService.(data., );
      },
    },
  ],
};

Idempotency and Exactly-Once Delivery

interface IdempotencyKey {
  key: string;
  response?: unknown;
  createdAt: Date;
  expiresAt: Date;
}

class IdempotencyService {
  constructor(private redis: Redis) {}

  async process<T>(
    key: string,
    operation: () => Promise<T>,
    ttlSeconds: number = 86400 // 24 hours
  ): Promise<T> {
    const lockKey = `idempotency:lock:${key}`;
    const dataKey = `idempotency:data:${key}`;

    // Try to acquire lock
    const locked = await this.redis.set(lockKey, "1", "EX", 30, "NX");

    if (!locked) {
      // Another process is handling this request, wait for result
      return this.waitForResult<T>(dataKey);
    }

    try {
      // Check if already processed
       existing =  ..(dataKey);
       (existing) {
         .(existing)  T;
      }

      
       result =  ();

      
       ..(dataKey, ttlSeconds, .(result));

       result;
    }  {
       ..(lockKey);
    }
  }

    waitForResult<T>(
    : ,
    :  = 
  ): <T> {
     startTime = .();

     (.() - startTime < maxWaitMs) {
       data =  ..(dataKey);
       (data) {
         .(data)  T;
      }
        ( (r, ));
    }

      ();
  }
}


 <T> {
  () {}

   (
    : ,
    :  <T>
  ): <{ : T; :  }> {
     dedupKey = ;

    
     existing =  ..(dedupKey);
     (existing) {
       { : .(existing)  T, :  };
    }

    
     result =  ();

    
     ..(
      dedupKey,
      .,
      .(result)
    );

     { result, :  };
  }
}

Dead Letter Queues

interface DeadLetterMessage {
  id: string;
  originalQueue: string;
  originalMessage: unknown;
  error: string;
  failedAt: Date;
  retryCount: number;
  lastRetryAt?: Date;
}

class DeadLetterQueueManager {
  constructor(
    private dlqStore: DLQStore,
    private originalQueue: MessageQueue
  ) {}

  async moveToDeadLetter(
    message: unknown,
    originalQueue: string,
    error: Error,
    retryCount: number
  ): Promise<void> {
    const dlqMessage: DeadLetterMessage = {
      id: crypto.randomUUID(),
      originalQueue,
      originalMessage: message,
      error: error.message,
      failedAt: new Date(),
      retryCount,
    };

    await this.dlqStore.(dlqMessage);

    
     dlqSize =  ..(originalQueue);
     (dlqSize > ) {
       ..({
        : ,
        : ,
      });
    }
  }

   (: ): <> {
     dlqMessage =  ..(messageId);
     (!dlqMessage)   ();

     {
       ..(
        dlqMessage.,
        dlqMessage.
      );
       ..(messageId);
    }  (error) {
      dlqMessage. =  ();
      dlqMessage.++;
       ..(dlqMessage);
       error;
    }
  }

   (: ): <{ : ; :  }> {
     messages =  ..(queue);
     success = ;
     failed = ;

     ( message  messages) {
       {
         .(message.);
        success++;
      }  {
        failed++;
      }
    }

     { success, failed };
  }

   (: , ?: ): <> {
     ..(queue, olderThan);
  }
}

Data Streaming with Kafka

Stream Processing:

import { Kafka, CompressionTypes } from "kafkajs";

interface StreamRecord<T> {
  key: string;
  value: T;
  timestamp: number;
  partition: number;
  offset: string;
}

class KafkaStreamProcessor {
  private kafka: Kafka;

  constructor(brokers: string[]) {
    this.kafka = new Kafka({
      clientId: "stream-processor",
      brokers,
    });
  }

  // Stateful stream aggregation
  async aggregateStream<TInput, TState>(
    inputTopic: string,
    outputTopic: string,
    groupId: string,
    initialState: TState,
    aggregator: (state: TState, record: TInput) => TState,
    windowMs: number = 
  ): <> {
     consumer = ..({ groupId });
     producer = ..({
      : .,
    });

     consumer.();
     producer.();
     consumer.({ : inputTopic });

     stateByKey =  <, >();
     windowTimers =  <, .>();

     consumer.({
      :  ({ message }) => {
         key = message.?.() || ;
         :  = .(message.!.());

        
         currentState = stateByKey.(key) || initialState;
         newState = (currentState, value);
        stateByKey.(key, newState);

        
         existingTimer = windowTimers.(key);
         (existingTimer) (existingTimer);

        
         timer = ( () => {
           finalState = stateByKey.(key);
           producer.({
            : outputTopic,
            : [
              {
                key,
                : .(finalState),
                : .().(),
              },
            ],
          });
          stateByKey.(key);
          windowTimers.(key);
        }, windowMs);

        windowTimers.(key, timer);
      },
    });
  }

  
   joinStreams<, , >(
    : ,
    : ,
    : ,
    : ,
    :  ,
    :  = 
  ): <> {
     consumer = ..({ groupId });
     producer = ..();

     consumer.();
     producer.();
     consumer.({ : [leftTopic, rightTopic] });

     leftCache =  <, { : ; :  }>();
     rightCache =  <, { : ; :  }>();

     consumer.({
      :  ({ topic, message }) => {
         key = message.?.() || ;
         timestamp = (message.);
         now = .();

        
        .(leftCache, now, windowMs);
        .(rightCache, now, windowMs);

         (topic === leftTopic) {
           :  = .(message.!.());
          leftCache.(key, { : leftData, timestamp });

          
           rightEntry = rightCache.(key);
           (
            rightEntry &&
            .(timestamp - rightEntry.) <= windowMs
          ) {
             result = (leftData, rightEntry.);
             producer.({
              : outputTopic,
              : [{ key, : .(result) }],
            });
          }
        }  {
           :  = .(message.!.());
          rightCache.(key, { : rightData, timestamp });

          
           leftEntry = leftCache.(key);
           (
            leftEntry &&
            .(timestamp - leftEntry.) <= windowMs
          ) {
             result = (leftEntry., rightData);
             producer.({
              : outputTopic,
              : [{ key, : .(result) }],
            });
          }
        }
      },
    });
  }

   cleanOldEntries<T>(
    : <, { : T; :  }>,
    : ,
    : 
  ):  {
     ( [key, entry]  cache.()) {
       (now - entry. > windowMs) {
        cache.(key);
      }
    }
  }
}


 <T> {
  () {}

  
  map<R>(:  R): <R> {
     outputTopic = ;
    .(outputTopic,  (record) => ({
      : record.,
      : (record.),
    }));
      <R>(., outputTopic);
  }

  
  (:  ): <T> {
     outputTopic = ;
    .(outputTopic,  (record) =>
      (record.) ? record : 
    );
      <T>(., outputTopic);
  }

  
  groupBy<K, V>(
    :  K,
    :  V,
    :  = 
  ): <V> {
     outputTopic = ;
     groups =  <, T[]>();
     timers =  <, .>();

    .(outputTopic,  (record, producer) => {
       key = ((record.));
       values = groups.(key) || [];
      values.(record.);
      groups.(key, values);

       existingTimer = timers.(key);
       (existingTimer) (existingTimer);

       timer = ( () => {
         groupValues = groups.(key) || [];
         result = ((record.), groupValues);
         producer.({
          : outputTopic,
          : [{ key, : .(result) }],
        });
        groups.(key);
        timers.(key);
      }, windowMs);

      timers.(key, timer);

       ; 
    });

      <V>(., outputTopic);
  }

    (
    : ,
    :  <{ : ; :  } | >
  ): <> {
     consumer = ..({
      : ,
    });
     producer = ..();

     consumer.();
     producer.();
     consumer.({ : . });

     consumer.({
      :  ({ message, partition }) => {
         : <T> = {
          : message.?.() || ,
          : .(message.!.()),
          : (message.),
          partition,
          : message.,
        };

         result =  (record, producer);
         (result) {
           producer.({
            : outputTopic,
            : [
              {
                : result.,
                : .(result.),
              },
            ],
          });
        }
      },
    });
  }
}

Event Sourcing Patterns

Snapshots for Performance:

interface Snapshot {
  aggregateId: string;
  version: number;
  state: unknown;
  timestamp: Date;
}

class SnapshotStore {
  constructor(private db: Database) {}

  async save(snapshot: Snapshot): Promise<void> {
    await this.db.snapshots.upsert({
      aggregateId: snapshot.aggregateId,
      version: snapshot.version,
      state: JSON.stringify(snapshot.state),
      timestamp: snapshot.timestamp,
    });
  }

  async getLatest(aggregateId: string): Promise<Snapshot | null> {
    const row = await this.db.snapshots.findOne(
      { aggregateId },
      { orderBy: { :  } }
    );
     row
      ? {
          : row.,
          : row.,
          : .(row.),
          : row.,
        }
      : ;
  }
}


     {
     = ; 

   (
    : ,
    : 
  ): <> {
    
     snapshot =  snapshotStore.(.);
     (snapshot) {
      .(snapshot.);
      . = snapshot.;

      
       events =  eventStore.(., snapshot.);
      .(events);
    }  {
      
       events =  eventStore.(.);
      .(events);
    }
  }

   (
    : ,
    : 
  ): <> {
     events = .();
     eventStore.(events);
    .();

    
     (. % . === ) {
       snapshotStore.({
        : .,
        : .,
        : .(),
        :  (),
      });
    }
  }

    (): ;
    (: ): ;
}

Event Upcasting (Schema Migration):

interface EventUpcaster {
  eventType: string;
  fromVersion: number;
  toVersion: number;
  upcast: (event: DomainEvent) => DomainEvent;
}

class EventStoreWithUpcasting implements EventStore {
  private upcasters: Map<string, EventUpcaster[]> = new Map();

  registerUpcaster(upcaster: EventUpcaster): void {
    const existing = this.upcasters.get(upcaster.eventType) || [];
    existing.push(upcaster);
    existing.sort((a, b) => a.fromVersion - b.fromVersion);
    this.upcasters.set(upcaster.eventType, existing);
  }

  async getEvents(
    aggregateId: string,
    fromVersion: number = 0
  ): Promise<[]> {
     rawEvents =  ..(
      aggregateId,
      fromVersion
    );

     rawEvents.( .(event));
  }

   (: ):  {
     upcasters = ..(event.) || [];
     currentEvent = event;

     ( upcaster  upcasters) {
       eventVersion = (currentEvent.  )?. || ;
       (eventVersion === upcaster.) {
        currentEvent = upcaster.(currentEvent);
      }
    }

     currentEvent;
  }
}


 :  = {
  : ,
  : ,
  : ,
  :  ({
    ...event,
    : {
      ...(event.  ),
      
      : (event.  ).,
      : (event.  ).,
      : ,
    },
  }),
};

Saga Patterns

Choreography vs Orchestration:

// CHOREOGRAPHY: Services react to events independently
class OrderService {
  async onOrderCreated(event: OrderCreatedEvent): Promise<void> {
    // Publish event, other services react
    await this.eventBus.publish("order.created", {
      orderId: event.orderId,
      customerId: event.customerId,
      items: event.items,
    });
  }
}

class InventoryService {
  constructor(private eventBus: EventBus) {
    // Listen and react to order events
    this.eventBus.subscribe("order.created", this.reserveInventory.bind(this));
  }

  private async reserveInventory(event: OrderCreatedEvent): Promise<void> {
    try {
      await this.reserve(event.items);
      
       ..(, {
        : event.,
      });
    }  (error) {
      
       ..(, {
        : event.,
        : error.,
      });
    }
  }
}

  {
  () {
    
    ..(
      ,
      ..()
    );
  }

    (: ): <> {
    
  }
}


  {
   (: ): <> {
     {
      
       ..(orderId);

      
       ..(orderId);

      
       ..(orderId);

      
       ..(orderId);
    }  (error) {
      
       .(orderId);
    }
  }

    (: ): <> {
    
     ..(orderId);
     ..(orderId);
     ..(orderId);
     ..(orderId);
  }
}

Saga State Machine:

type SagaState =
  | "STARTED"
  | "INVENTORY_RESERVED"
  | "PAYMENT_PROCESSED"
  | "SHIPPED"
  | "COMPLETED"
  | "COMPENSATING"
  | "FAILED";

interface SagaStateMachine<TData> {
  state: SagaState;
  data: TData;
  transitions: Map<SagaState, SagaTransition<TData>>;
}

interface SagaTransition<TData> {
  onEnter: (data: TData) => Promise<void>;
  onSuccess: SagaState;
  onFailure: SagaState;
  compensate?: (data: TData) => Promise<void>;
}

class StatefulSagaOrchestrator<TData> {
  async execute(saga: SagaStateMachine<TData>): Promise<void> {
    let currentState = saga.;

     (currentState !==  && currentState !== ) {
       transition = saga..(currentState);
       (!transition)
          ();

       {
         transition.(saga.);
        currentState = transition.;
        saga. = currentState;
         .(saga); 
      }  (error) {
        
         (transition.) {
           transition.(saga.);
        }
        currentState = transition.;
        saga. = currentState;
         .(saga);
      }
    }
  }

    (: <>): <> {
    
     ..({
      : (saga.  ).,
      : saga.,
      : saga.,
      :  (),
    });
  }
}

Best Practices

1. Event Design

   • Events should be immutable and represent facts
   • Use past tense naming (OrderCreated, not CreateOrder)
   • Include all necessary data; avoid references to mutable state
   • Version your events for schema evolution

2. Idempotency

   • Always design consumers to be idempotent
   • Use unique message IDs for deduplication
   • Store processing state to handle retries

3. Error Handling

   • Implement dead letter queues for failed messages
   • Set reasonable retry limits with exponential backoff
   • Monitor DLQ size and alert on growth

4. Ordering

   • Use partition keys for ordering guarantees in Kafka
   • Understand at-least-once vs exactly-once semantics
   • Design for out-of-order message handling when needed

5. Monitoring

   • Track message lag, processing time, and error rates
   • Set up alerts for consumer lag
   • Monitor event store growth and query performance

Examples

Complete Order Processing Flow

// 1. API receives order request
app.post("/orders", async (req, res) => {
  const command: CreateOrderCommand = {
    type: "CreateOrder",
    payload: req.body,
    metadata: {
      userId: req.user.id,
      correlationId: req.headers["x-correlation-id"] as string,
      timestamp: new Date(),
    },
  };

  await commandBus.dispatch(command);
  res.status(202).json({ message: "Order creation initiated" });
});

// 2. Command handler creates aggregate and persists events
class CreateOrderHandler implements CommandHandler<CreateOrderCommand> {
  async handle(command: CreateOrderCommand): Promise<void> {
    const order = Order.create(
      crypto.randomUUID(),
      command.payload.customerId,
      command.payload.
    );

     ..(order.());
  }
}