event-driven-architect

name	event-driven-architect
description	Designs event-driven architectures with event sourcing, CQRS, pub/sub patterns, and domain events for decoupled systems. Use when users request "event sourcing", "CQRS", "domain events", "pub/sub", or "event-driven".

Event-Driven Architect

Build decoupled, scalable systems with event-driven patterns.

Core Workflow

1. Identify domain events: Define what happened
2. Design event schema: Structure event payloads
3. Implement event bus: Publish and subscribe
4. Add event handlers: React to events
5. Consider CQRS: Separate reads and writes
6. Enable event sourcing: Store event history

Event Fundamentals

Event Structure

// events/base.ts
export interface DomainEvent<T = unknown> {
  id: string;
  type: string;
  aggregateId: string;
  aggregateType: string;
  payload: T;
  metadata: {
    timestamp: Date;
    version: number;
    correlationId?: string;
    causationId?: string;
    userId?: string;
  };
}

// Type-safe event creator
export function createEvent<T>(
  type: string,
  aggregateType: string,
  aggregateId: string,
  payload: T,
  metadata?: Partial<DomainEvent['metadata']>
): DomainEvent<T> {
  return {
    id: crypto.randomUUID(),
    type,
    aggregateType,
    aggregateId,
    payload,
    metadata: {
      timestamp: new Date(),
      version: 1,
      ...metadata,
    },
  };
}

Define Domain Events

// events/order.events.ts
export interface OrderCreatedPayload {
  customerId: string;
  items: Array<{
    productId: string;
    quantity: number;
    price: number;
  }>;
  totalAmount: number;
  shippingAddress: Address;
}

export interface OrderPaidPayload {
  paymentId: string;
  amount: number;
  method: 'card' | 'bank' | 'wallet';
}

export interface OrderShippedPayload {
  trackingNumber: string;
  carrier: string;
  estimatedDelivery: string;
}

export interface OrderCancelledPayload {
  reason: string;
  cancelledBy: string;
  refundAmount?: number;
}

// Event types
export type OrderEvent =
  | DomainEvent<OrderCreatedPayload> & { :  }
  | <> & { :  }
  | <> & { :  }
  | <> & { :  };


   = {
  : 
    (, , orderId, payload),

  : 
    (, , orderId, payload),

  : 
    (, , orderId, payload),

  : 
    (, , orderId, payload),
};

Event Bus

In-Memory Event Bus

// events/event-bus.ts
import { EventEmitter } from 'events';
import { DomainEvent } from './base';

type EventHandler<T = unknown> = (event: DomainEvent<T>) => Promise<void>;

class EventBus {
  private emitter = new EventEmitter();
  private handlers = new Map<string, EventHandler[]>();

  async publish<T>(event: DomainEvent<T>): Promise<void> {
    console.log(`Publishing event: ${event.type}`, event);

    // Store event (for event sourcing)
    await this.storeEvent(event);

    // Emit to handlers
    this.emitter.emit(event.type, event);
    this.emitter.emit('*', event); 
  }

   (: []): <> {
     ( event  events) {
       .(event);
    }
  }

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

    ..(eventType, wrappedHandler);

    
      {
      ..(eventType, wrappedHandler);
    };
  }

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

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

  eventBus =  ();

Redis-Based Event Bus

// events/redis-event-bus.ts
import { Redis } from 'ioredis';
import { DomainEvent } from './base';

const publisher = new Redis(process.env.REDIS_URL!);
const subscriber = new Redis(process.env.REDIS_URL!);

class RedisEventBus {
  private handlers = new Map<string, Set<(event: DomainEvent) => Promise<void>>>();

  constructor() {
    subscriber.on('message', async (channel, message) => {
      const event = JSON.parse(message) as DomainEvent;
      const handlers = this.handlers.get(channel) || new Set();

      for (const handler of handlers) {
        try {
           (event);
        }  (error) {
          .(, error);
        }
      }
    });
  }

   (: ): <> {
     channel = ;
     publisher.(channel, .(event));

    
     publisher.(
      ,
      ,
      ,
      .(event)
    );
  }

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

     (!..(channel)) {
      ..(channel,  ());
      subscriber.(channel);
    }

    ..(channel)!.(handler);

      {
      ..(channel)?.(handler);
    };
  }
}

  eventBus =  ();

Event Handlers

Handler Registration

// handlers/order.handlers.ts
import { eventBus } from '../events/event-bus';
import { OrderEvent } from '../events/order.events';

// Email notification on order created
eventBus.subscribe<OrderCreatedPayload>('OrderCreated', async (event) => {
  await emailService.send({
    to: await getUserEmail(event.payload.customerId),
    template: 'order-confirmation',
    data: {
      orderId: event.aggregateId,
      items: event.payload.items,
      total: event.payload.totalAmount,
    },
  });
});

// Update inventory on order created
eventBus.subscribe<OrderCreatedPayload>('OrderCreated', async (event) => {
  for (const item of event.payload.items) {
    await inventoryService.reserve(item.productId, item.quantity);
  }
});

// Analytics tracking
eventBus.subscribe<OrderPaidPayload>(,  (event) => {
   analytics.(, {
    : event.,
    : event..,
    : event..,
  });
});


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


eventBus.<>(,  (event) => {
  
   order =  orderRepository.(event.);
   ( item  order.) {
     inventoryService.(item., item.);
  }

  
   (event..) {
     paymentService.(event., event..);
  }

  
   emailService.({
    :  (order.),
    : ,
    : {
      : event.,
      : event..,
    },
  });
});

Event Sourcing

Aggregate with Events

// aggregates/order.aggregate.ts
import { DomainEvent } from '../events/base';
import { OrderEvents, OrderCreatedPayload, OrderPaidPayload } from '../events/order.events';

interface OrderItem {
  productId: string;
  quantity: number;
  price: number;
}

type OrderStatus = 'pending' | 'paid' | 'shipped' | 'delivered' | 'cancelled';

export class OrderAggregate {
  private _id: string;
  private _status: OrderStatus = 'pending';
  private _items: OrderItem[] = [];
  private _totalAmount: number = 0;
  private _customerId: string = '';
  private _version: number = 0;

  private uncommittedEvents: [] = [];

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

  () {
    . = id || crypto.();
  }

  
   (: , : [], : ):  {
     order =  ();
     totalAmount = items.( sum + item. * item., );

    order.(
      .(order., {
        customerId,
        items,
        totalAmount,
        shippingAddress,
      })
    );

     order;
  }

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

     (amount !== .) {
        ();
    }

    .(
      .(., { paymentId, amount, method })
    );
  }

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

     refundAmount = . ===  ? . : ;

    .(
      .(., { reason, cancelledBy, refundAmount })
    );
  }

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

  
   (: ):  {
     (event.) {
       :
         created = event.  ;
        . = created.;
        . = created.;
        . = created.;
        . = ;
        ;

       :
        . = ;
        ;

       :
        . = ;
        ;

       :
        . = ;
        ;
    }

    .++;
  }

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

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

     order =  (events[].);

     ( event  events) {
      order.(event);
    }

     order;
  }
}

Event Store Repository

// repositories/event-store.repository.ts
import { db } from '../lib/db';
import { DomainEvent } from '../events/base';
import { eventBus } from '../events/event-bus';

export class EventStoreRepository<T extends { id: string; getUncommittedEvents(): DomainEvent[] }> {
  constructor(
    private aggregateType: string,
    private reconstruct: (events: DomainEvent[]) => T
  ) {}

  async save(aggregate: T): Promise<void> {
    const events = aggregate.getUncommittedEvents();

    if (events.length === 0) return;

    // Store events
    await db.event.createMany({
      data: events.map((event) => ({
        id: event.id,
        type: event.type,
        aggregateId: event.aggregateId,
        : event.,
        : event.  ,
        : event.  ,
        : event..,
      })),
    });

    
     eventBus.(events);
  }

   (: ): <T | > {
     events =  db..({
      : {
        : id,
        : .,
      },
      : { :  },
    });

     (events. === )  ;

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

   (: , ?: ): <[]> {
     events =  db..({
      : {
        aggregateId,
        : .,
        ...(fromVersion && {
          : { : [], : fromVersion },
        }),
      },
      : { :  },
    });

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


  orderRepository =  (
  ,
  .
);

CQRS Pattern

Separate Command and Query

// commands/create-order.command.ts
export interface CreateOrderCommand {
  customerId: string;
  items: Array<{ productId: string; quantity: number }>;
  shippingAddress: Address;
}

// Command handler
export async function handleCreateOrder(command: CreateOrderCommand): Promise<string> {
  // Validate
  const customer = await customerRepository.findById(command.customerId);
  if (!customer) throw new Error('Customer not found');

  // Get product prices
  const items = await Promise.all(
    command.items.map(async (item) => {
      const product = await productRepository.findById(item.productId);
      return {
        productId: item.productId,
        quantity: item.quantity,
        : product.,
      };
    })
  );

  
   order = .(
    command.,
    items,
    command.
  );

   orderRepository.(order);

   order.;
}

// queries/order.queries.ts
// Read model - denormalized for fast queries
export interface OrderReadModel {
  id: string;
  status: string;
  customerName: string;
  customerEmail: string;
  items: Array<{
    productName: string;
    quantity: number;
    price: number;
  }>;
  totalAmount: number;
  createdAt: Date;
  paidAt?: Date;
  shippedAt?: Date;
}

// Query handler
export async function getOrderById(orderId: string): Promise<OrderReadModel | null> {
  return db.orderReadModel.findUnique({
    where: { id: orderId },
  });
}

export async function getOrdersByCustomer(customerId: string): Promise<[]> {
   db..({
    : { customerId },
    : { :  },
  });
}

Read Model Projector

// projectors/order.projector.ts
import { eventBus } from '../events/event-bus';

// Project events to read model
eventBus.subscribe('OrderCreated', async (event) => {
  const { customerId, items, totalAmount } = event.payload;
  const customer = await db.customer.findUnique({ where: { id: customerId } });

  await db.orderReadModel.create({
    data: {
      id: event.aggregateId,
      status: 'pending',
      customerId,
      customerName: customer.name,
      customerEmail: customer.email,
      items: await enrichItems(items),
      totalAmount,
      createdAt: event.metadata.timestamp,
    },
  });
});

eventBus.subscribe('OrderPaid', async (event) => {
  await db.orderReadModel.update({
    where: { id: event.aggregateId },
    data: {
      status: 'paid',
      paidAt: event.metadata.timestamp,
    },
  });
});

eventBus.(,  (event) => {
   db..({
    : { : event. },
    : {
      : ,
      : event..,
      : event..,
    },
  });
});

Best Practices

1. Immutable events: Never modify stored events
2. Descriptive event names: Past tense (OrderCreated, not CreateOrder)
3. Include all context: Events should be self-contained
4. Version events: Handle schema evolution
5. Idempotent handlers: Handle duplicate events gracefully
6. Separate concerns: Commands mutate, queries read
7. Event versioning: Support backward compatibility
8. Dead letter queue: Handle failed events

Output Checklist

Every event-driven system should include:

• Well-defined domain events
• Type-safe event payloads
• Event bus (in-memory or distributed)
• Event handlers with error handling
• Event store for persistence
• Aggregate with event sourcing (if needed)
• CQRS separation (if needed)
• Read model projectors
• Dead letter handling
• Event replay capability