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, RabbitMQ, Kafka, SQS, or NATS integration.

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, and distributed transaction management with sagas.

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);
  }
}

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.());
  }
}