Event Sourcing · fullstack.org.ua

Просунутий23 хв· Fullstack Team

Event Sourcing

Зберігаємо події замість поточного стану, відновлюємо агрегати та зважуємо переваги й ризики аудиту.

Що таке Event Sourcing

Event Sourcing — це підхід, у якому система зберігає не поточний стан сутності, а послідовність подій, що призвели до цього стану.

Наприклад, замість збереження такого запису:

json

{
  "orderId": "order-42",
  "status": "PAID",
  "total": 1500
}

json

[
  {
    "type": "OrderCreated",
    "data": {
      "orderId": "order-42",
      "total": 1500
    }
  },
  {
    "type": "PaymentReceived",
    "data": {
      "orderId": "order-42",
      "amount": 1500
    }
  }
]

json

{
  "eventId": "event-1001",
  "aggregateId": "order-42",
  "aggregateType": "Order",
  "type": "PaymentReceived",
  "version": 2,
  "occurredAt": "2026-09-02T10:30:00.000Z",
  "data": {
    "amount": 1500
  },
  "metadata": {
    "correlationId": "request-77",
    "causationId": "command-12"
  }
}

javascript

function initialOrderState() {
  return {
    id: null,
    status: "NEW",
    total: 0,
    paidAmount: 0,
    version: 0
  };
}

function applyOrderEvent(state, event) {
  switch (event.type) {
    case "OrderCreated":
      return {
        ...state,
        id: event.data.orderId,
        total: event.data.total,
        status: "CREATED",
        version: event.version
      };

    case "PaymentReceived":
      return {
        ...state,
        paidAmount: state.paidAmount + event.data.amount,
        status: "PAID",
        version: event.version
      };

    case "OrderShipped":
      return {
        ...state,
        status: "SHIPPED",
        version: event.version
      };

    default:
      throw new Error(`Невідомий тип події: ${event.type}`);
  }
}

function rebuildOrder(events) {
  return events.reduce(applyOrderEvent, initialOrderState());
}

javascript

function handleOrderCommand(state, command) {
  switch (command.type) {
    case "CreateOrder":
      if (state.id !== null) {
        throw new Error("Замовлення вже існує");
      }

      if (!Number.isInteger(command.total) || command.total <= 0) {
        throw new Error("Сума замовлення має бути додатним цілим числом");
      }

      return [
        {
          type: "OrderCreated",
          data: {
            orderId: command.orderId,
            total: command.total
          }
        }
      ];

    case "PayOrder":
      if (state.id === null) {
        throw new Error("Замовлення не існує");
      }

      if (state.status !== "CREATED") {
        throw new Error("Замовлення не можна оплатити в поточному стані");
      }

      if (command.amount !== state.total) {
        throw new Error("Сума платежу не відповідає сумі замовлення");
      }

      return [
        {
          type: "PaymentReceived",
          data: {
            amount: command.amount
          }
        }
      ];

    case "ShipOrder":
      if (state.status !== "PAID") {
        throw new Error("Відправити можна лише оплачене замовлення");
      }

      return [
        {
          type: "OrderShipped",
          data: {}
        }
      ];

    default:
      throw new Error(`Невідома команда: ${command.type}`);
  }
}

javascript

class ConcurrencyError extends Error {}

class InMemoryEventStore {
  constructor() {
    this.streams = new Map();
  }

  load(aggregateId) {
    return this.streams.get(aggregateId) ?? [];
  }

  append(aggregateId, events, expectedVersion) {
    const currentEvents = this.load(aggregateId);
    const actualVersion = currentEvents.length;

    if (actualVersion !== expectedVersion) {
      throw new ConcurrencyError(
        `Конфлікт версій: очікувалась ${expectedVersion}, фактично ${actualVersion}`
      );
    }

    const storedEvents = events.map((event, index) => ({
      eventId: `${aggregateId}-${actualVersion + index + 1}`,
      aggregateId,
      aggregateType: "Order",
      type: event.type,
      data: event.data,
      version: actualVersion + index + 1,
      occurredAt: new Date().toISOString()
    }));

    this.streams.set(aggregateId, [
      ...currentEvents,
      ...storedEvents
    ]);

    return storedEvents;
  }
}

function executeOrderCommand(store, aggregateId, command) {
  const history = store.load(aggregateId);
  const state = rebuildOrder(history);
  const newEvents = handleOrderCommand(state, command);

  return store.append(aggregateId, newEvents, state.version);
}

const store = new InMemoryEventStore();

executeOrderCommand(store, "order-42", {
  type: "CreateOrder",
  orderId: "order-42",
  total: 1500
});

executeOrderCommand(store, "order-42", {
  type: "PayOrder",
  amount: 1500
});

executeOrderCommand(store, "order-42", {
  type: "ShipOrder"
});

const events = store.load("order-42");
const finalState = rebuildOrder(events);

console.log(events.map(({ type, version }) => ({ type, version })));
console.log(finalState);

javascript

class OrderListProjection {
  constructor() {
    this.orders = new Map();
  }

  handle(event) {
    switch (event.type) {
      case "OrderCreated":
        this.orders.set(event.aggregateId, {
          id: event.aggregateId,
          total: event.data.total,
          status: "CREATED"
        });
        break;

      case "PaymentReceived": {
        const order = this.orders.get(event.aggregateId);
        if (order) {
          order.status = "PAID";
        }
        break;
      }

      case "OrderShipped": {
        const order = this.orders.get(event.aggregateId);
        if (order) {
          order.status = "SHIPPED";
        }
        break;
      }
    }
  }
}

const projection = new OrderListProjection();

for (const event of store.load("order-42")) {
  projection.handle(event);
}

console.log(projection.orders.get("order-42"));

javascript

class IdempotentProjection {
  constructor() {
    this.processedEventIds = new Set();
    this.orders = new Map();
  }

  handle(event) {
    if (this.processedEventIds.has(event.eventId)) {
      return;
    }

    // Оновлення read-моделі має виконуватися узгоджено з фіксацією eventId.
    if (event.type === "OrderCreated") {
      this.orders.set(event.aggregateId, {
        id: event.aggregateId,
        total: event.data.total,
        status: "CREATED"
      });
    }

    this.processedEventIds.add(event.eventId);
  }
}

Event Sourcing

Що таке Event Sourcing

Стан, команди та події

Стан

Команда

Подія

Event Store

Основні поля

Відновлення агрегату

Команди та генерація подій

Конкурентний запис і optimistic concurrency

Події та проєкції

Ідемпотентність обробників

Snapshots

Версіювання подій

Аудит і відтворюваність

Виправлення помилкових подій

Переваги Event Sourcing

Повна історія

Відтворюваність

Повторна побудова проєкцій

Розділення запису та читання

Зручна інтеграція

Ризики та обмеження

Зростання складності

Eventual consistency

Висока вартість помилок у схемі

Обмеження конфіденційності

Проблеми з побічними ефектами

Не кожен домен потребує Event Sourcing

Типові помилки

Зберігання поточного стану замість подій

Події як команди

Редагування старих подій

Відсутність optimistic concurrency

Побічні ефекти в apply

Події без достатніх даних

Ідемпотентність лише на словах

Відсутність плану для нових версій

Практичний алгоритм проєктування

Підсумок

Побічні ефекти в `apply`