Как связан паттерн Event Sourcing с паттерном CQRS?

Как связан паттерн Event Sourcing с паттерном CQRS

Event Sourcing и CQRS - это два архитектурных паттерна, которые часто используются вместе для построения сложных, масштабируемых систем. Хотя они независимы, вместе они создают мощную архитектуру.

Event Sourcing - что это?

Event Sourcing - это паттерн, где все изменения состояния приложения сохраняются как последовательность неизменяемых событий. Вместо сохранения текущего состояния, сохраняем историю всех событий.

// ТРАДИЦИОННЫЙ ПОДХОД
class BankAccount {
  constructor() {
    this.balance = 0; // Только текущее состояние
  }
  
  deposit(amount) {
    this.balance += amount; // Перезаписываем состояние
  }
  
  withdraw(amount) {
    this.balance -= amount; // История потеряна
  }
}

const account = new BankAccount();
account.deposit(1000);
account.withdraw(500);
console.log(account.balance); // 500
// Мы не знаем как мы пришли к этому состоянию!

// EVENT SOURCING ПОДХОД
class BankAccountWithEventSourcing {
  constructor() {
    this.events = []; // История всех событий
    this.balance = 0; // Вычисляется из событий
  }
  
  deposit(amount) {
    const event = {
      type: 'MoneyDeposited',
      amount: amount,
      timestamp: new Date(),
    };
    this.events.push(event);
    this.applyEvent(event);
  }
  
  withdraw(amount) {
    const event = {
      type: 'MoneyWithdrawn',
      amount: amount,
      timestamp: new Date(),
    };
    this.events.push(event);
    this.applyEvent(event);
  }
  
  applyEvent(event) {
    switch (event.type) {
      case 'MoneyDeposited':
        this.balance += event.amount;
        break;
      case 'MoneyWithdrawn':
        this.balance -= event.amount;
        break;
    }
  }
  
  getBalance() {
    return this.balance;
  }
  
  getHistory() {
    return this.events; // Полная история!
  }
  
  // Можем вернуться в прошлое
  getBalanceAt(timestamp) {
    let balance = 0;
    for (const event of this.events) {
      if (new Date(event.timestamp) > timestamp) break;
      if (event.type === 'MoneyDeposited') balance += event.amount;
      if (event.type === 'MoneyWithdrawn') balance -= event.amount;
    }
    return balance;
  }
}

const account = new BankAccountWithEventSourcing();
account.deposit(1000);
account.withdraw(500);

console.log(account.getBalance()); // 500
console.log(account.getHistory()); // Полная история операций
console.log(account.getBalanceAt(firstOperationTime)); // 1000

CQRS - что это?

CQRS (Command Query Responsibility Segregation) - это паттерн, разделяющий операции на:

• Commands (команды) - операции которые изменяют состояние
• Queries (запросы) - операции которые читают состояние

// CQRS ПОДХОД

// КОМАНДЫ - изменяют состояние
class Commands {
  async depositMoney(accountId, amount) {
    const command = new DepositMoneyCommand(accountId, amount);
    return await this.commandHandler.handle(command);
  }
  
  async withdrawMoney(accountId, amount) {
    const command = new WithdrawMoneyCommand(accountId, amount);
    return await this.commandHandler.handle(command);
  }
}

// ЗАПРОСЫ - только читают состояние
class Queries {
  async getBalance(accountId) {
    return await this.queryRepository.getBalance(accountId);
  }
  
  async getTransactionHistory(accountId) {
    return await this.queryRepository.getTransactionHistory(accountId);
  }
  
  async getTotalTransfers(accountId) {
    return await this.queryRepository.getTotalTransfers(accountId);
  }
}

// Command Handler обрабатывает команду и генерирует события
class DepositMoneyCommandHandler {
  async handle(command) {
    // Загружаем агрегат из Event Store
    const account = await eventStore.getAggregate(command.accountId);
    
    // Применяем команду
    account.deposit(command.amount);
    
    // Сохраняем события
    await eventStore.saveEvents(command.accountId, account.getUncommittedEvents());
    
    // Публикуем события для обновления читающей БД
    await eventBus.publish(account.getUncommittedEvents());
    
    return { success: true };
  }
}

Как они работают вместе

// ПОЛНАЯ АРХИТЕКТУРА: Event Sourcing + CQRS

// 1. WRITE SIDE (Command Side)
// - Принимает команды
// - Генерирует события через Event Sourcing
// - Сохраняет события в Event Store
// - Публикует события для других систем

class WriteModel {
  async processCommand(command) {
    // Загружаем историю событий
    const aggregate = await this.eventStore.getAggregate(command.aggregateId);
    
    // Применяем команду к агрегату
    aggregate.handle(command);
    
    // Получаем новые события
    const events = aggregate.getUncommittedEvents();
    
    // Сохраняем в Event Store
    await this.eventStore.appendEvents(command.aggregateId, events);
    
    // Публикуем события
    await this.eventBus.publish(events);
    
    return { success: true };
  }
}

// 2. EVENT STORE
// - Хранит все события в порядке их возникновения
// - Является source of truth
// - Полная аудит-база

class EventStore {
  async appendEvents(aggregateId, events) {
    // Сохраняем в БД
    for (const event of events) {
      await db.insert('events', {
        aggregateId,
        type: event.type,
        data: JSON.stringify(event.data),
        timestamp: new Date(),
        version: event.version,
      });
    }
  }
  
  async getAggregate(aggregateId) {
    // Загружаем все события для агрегата
    const events = await db.query(
      'SELECT * FROM events WHERE aggregateId = ? ORDER BY version',
      [aggregateId]
    );
    
    // Восстанавливаем состояние из событий
    const aggregate = new Aggregate();
    for (const event of events) {
      aggregate.apply(event);
    }
    
    return aggregate;
  }
}

// 3. READ SIDE (Query Side)
// - Слушает события
// - Обновляет оптимизированную читающую БД
// - Может иметь несколько проекций для разных запросов

class ReadModel {
  async onAccountCreated(event) {
    await this.readDb.insert('accounts', {
      id: event.aggregateId,
      createdAt: event.timestamp,
      balance: 0,
      transactionCount: 0,
    });
  }
  
  async onMoneyDeposited(event) {
    // Обновляем читающую БД оптимизированную для быстрых запросов
    await this.readDb.query(
      'UPDATE accounts SET balance = balance + ? WHERE id = ?',
      [event.amount, event.aggregateId]
    );
  }
  
  async onMoneyWithdrawn(event) {
    await this.readDb.query(
      'UPDATE accounts SET balance = balance - ? WHERE id = ?',
      [event.amount, event.aggregateId]
    );
  }
}

class QueryHandler {
  async getBalance(accountId) {
    // Читаем из оптимизированной БД (очень быстро)
    return await this.readDb.query(
      'SELECT balance FROM accounts WHERE id = ?',
      [accountId]
    );
  }
  
  async getTransactionHistory(accountId) {
    // Может быть другая проекция
    return await this.readDb.query(
      'SELECT * FROM transactions WHERE accountId = ? ORDER BY timestamp DESC',
      [accountId]
    );
  }
}

Преимущества Event Sourcing + CQRS

// 1. ПОЛНАЯ АУДИТ-БАЗА
// Мы знаем все что произошло и когда
const allEvents = await eventStore.getAllEvents(accountId);
console.log(allEvents); // Полная история

// 2. TIME TRAVEL - можем посмотреть состояние в прошлом
const stateAt2024 = await eventStore.getAggregateAtTime(
  accountId, 
  new Date('2024-01-01')
);

// 3. МАСШТАБИРУЕМОСТЬ
// Write и Read стороны могут масштабироваться независимо
// Write: One database
// Read: Many databases with different optimizations

// 4. БОЛЕЕ ИНФОРМАТИВНЫЕ ДАННЫЕ
// Можем анализировать события для insights
const totalTransfers = await eventStore.countEvents(
  { type: 'MoneyTransferred', toAccountId: accountId }
);

// 5. ОБРАБОТКА РАСХОЖДЕНИЙ (Eventual Consistency)
// Read сторона может быть немного отстающей, это OK
setTimeout(async () => {
  const latestBalance = await queryHandler.getBalance(accountId);
  console.log('Eventually consistent:', latestBalance);
}, 100);

Практический пример: E-commerce система

// Order Aggregate с Event Sourcing
class Order {
  constructor(orderId) {
    this.id = orderId;
    this.events = [];
    this.items = [];
    this.status = 'Created';
  }
  
  // КОМАНДЫ
  addItem(itemId, quantity) {
    this.raiseEvent(new ItemAddedToOrder(this.id, itemId, quantity));
  }
  
  confirmOrder() {
    this.raiseEvent(new OrderConfirmed(this.id, new Date()));
  }
  
  shipOrder(trackingNumber) {
    this.raiseEvent(new OrderShipped(this.id, trackingNumber, new Date()));
  }
  
  // Создание события
  raiseEvent(event) {
    this.applyEvent(event);
    this.events.push(event);
  }
  
  // Применение события к состоянию
  applyEvent(event) {
    if (event instanceof ItemAddedToOrder) {
      this.items.push({
        itemId: event.itemId,
        quantity: event.quantity,
      });
    } else if (event instanceof OrderConfirmed) {
      this.status = 'Confirmed';
    } else if (event instanceof OrderShipped) {
      this.status = 'Shipped';
      this.trackingNumber = event.trackingNumber;
    }
  }
  
  getUncommittedEvents() {
    return this.events;
  }
}

// Event Classes
class ItemAddedToOrder {
  constructor(orderId, itemId, quantity) {
    this.type = 'ItemAddedToOrder';
    this.orderId = orderId;
    this.itemId = itemId;
    this.quantity = quantity;
    this.timestamp = new Date();
  }
}

class OrderConfirmed {
  constructor(orderId, confirmedAt) {
    this.type = 'OrderConfirmed';
    this.orderId = orderId;
    this.confirmedAt = confirmedAt;
  }
}

class OrderShipped {
  constructor(orderId, trackingNumber, shippedAt) {
    this.type = 'OrderShipped';
    this.orderId = orderId;
    this.trackingNumber = trackingNumber;
    this.shippedAt = shippedAt;
  }
}

// КОМАНДЫ
class CreateOrderCommand {
  constructor(orderId, customerId) {
    this.orderId = orderId;
    this.customerId = customerId;
  }
}

// ЗАПРОСЫ
class GetOrderStatusQuery {
  constructor(orderId) {
    this.orderId = orderId;
  }
}

// Command Handler
class CreateOrderCommandHandler {
  async handle(command) {
    const order = new Order(command.orderId);
    // Команды обрабатываются в другом месте
    await eventStore.saveEvents(command.orderId, order.getUncommittedEvents());
  }
}

// Query Handler
class GetOrderStatusQueryHandler {
  async handle(query) {
    return await readDb.query(
      'SELECT status, itemCount FROM orders WHERE id = ?',
      [query.orderId]
    );
  }
}

Ключевые различия

Ключевые моменты

1. Event Sourcing сохраняет ВСЕ события (источник истины)
2. CQRS разделяет операции на команды и запросы
3. Вместе они дают: масштабируемость, аудит, time-travel, eventual consistency
4. Event Store - основа Event Sourcing
5. Read Models - оптимизированы для разных запросов
6. Write Model - обрабатывает команды и создаёт события
7. Eventual Consistency - read сторона может быть немного отстающей
8. Отлично работают для сложных доменов (финансы, e-commerce)

Event Sourcing + CQRS - это мощная архитектура для построения масштабируемых, отказоустойчивых систем с полной аудит-базой.