Event Sourcing (опционально): когда это реально нужно

Event Sourcing: когда это реально нужно

Event Sourcing - это когда источник правды - события, а текущее состояние - результат их проигрывания. Это мощно, но дорого по сложности. Перекликается с domain events и event-driven архитектурой.

События в append-only логе и replay для получения текущего состояния

Обычная БД vs Event Sourcing

Обычная БД (state-based):            Event Sourcing:
──────────────────────────           ──────────────────────────
Хранит текущее состояние             Хранит ВСЕ события
UPDATE balance = 500                 BalanceCreated{amount: 0}
                                     MoneyDeposited{amount: 300}
                                     MoneyDeposited{amount: 400}
                                     MoneyWithdrawn{amount: 200}
                                     → replay → balance = 500

Потеряна история «почему 500»        Полная история: кто, когда, что

Event Store: хранилище событий

В Event Sourcing все события записываются в append-only хранилище:

// Событие - иммутабельный факт
type Event struct {
    ID            string    `json:"id"`
    AggregateID   string    `json:"aggregate_id"`
    AggregateType string    `json:"aggregate_type"`
    Type          string    `json:"type"`
    Version       int       `json:"version"`
    Data          []byte    `json:"data"`
    CreatedAt     time.Time `json:"created_at"`
}

// Event Store - append-only
type EventStore interface {
    Append(ctx context.Context, aggregateID string, events []Event, expectedVersion int) error
    Load(ctx context.Context, aggregateID string) ([]Event, error)
}

<?php
declare(strict_types=1);

// Событие - иммутабельный факт
final readonly class StoredEvent
{
    public function __construct(
        public string $id,
        public string $aggregateId,
        public string $aggregateType,
        public string $type,
        public int $version,
        public string $data,            // JSON-payload
        public DateTimeImmutable $createdAt,
    ) {}
}

// Event Store - append-only
interface EventStore
{
    /** @param StoredEvent[] $events */
    public function append(string $aggregateId, array $events, int $expectedVersion): void;

    /** @return StoredEvent[] */
    public function load(string $aggregateId): array;
}

SQL-схема для Event Store:

CREATE TABLE events (
    id             UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    aggregate_id   TEXT NOT NULL,
    aggregate_type TEXT NOT NULL,
    event_type     TEXT NOT NULL,
    version        INT NOT NULL,
    data           JSONB NOT NULL,
    created_at     TIMESTAMPTZ NOT NULL DEFAULT now(),
    UNIQUE (aggregate_id, version) - оптимистичная блокировка
);

CREATE INDEX idx_events_aggregate ON events(aggregate_id, version);

Агрегат: восстановление из событий

type UserProgress struct {
    UserID          int64
    CompletedMap    map[int64]time.Time // lessonID → completedAt
    TotalCompleted  int
    Version         int
}

// Восстанавливаем состояние, проигрывая события
func (p *UserProgress) Apply(event Event) {
    switch event.Type {
    case "lesson.completed":
        var data LessonCompletedData
        json.Unmarshal(event.Data, &data)
        p.CompletedMap[data.LessonID] = data.CompletedAt
        p.TotalCompleted++
    case "progress.reset":
        p.CompletedMap = make(map[int64]time.Time)
        p.TotalCompleted = 0
    }
    p.Version = event.Version
}

// Загружаем агрегат из Event Store
func LoadProgress(ctx context.Context, store EventStore, userID int64) (*UserProgress, error) {
    events, err := store.Load(ctx, fmt.Sprintf("user-progress-%d", userID))
    if err != nil {
        return nil, err
    }
    progress := &UserProgress{
        UserID:       userID,
        CompletedMap: make(map[int64]time.Time),
    }
    for _, evt := range events {
        progress.Apply(evt)
    }
    return progress, nil
}

<?php
declare(strict_types=1);

final class UserProgress
{
    /** @var array<int, DateTimeImmutable> lessonId => completedAt */
    private array $completedMap = [];
    private int $totalCompleted = 0;
    private int $version = 0;

    public function __construct(
        public readonly int $userId,
    ) {}

    // Восстанавливаем состояние, проигрывая события
    public function apply(StoredEvent $event): void
    {
        $data = json_decode($event->data, true, flags: JSON_THROW_ON_ERROR);

        match ($event->type) {
            'lesson.completed' => $this->onLessonCompleted($data),
            'progress.reset'   => $this->onReset(),
            default            => null,
        };

        $this->version = $event->version;
    }

    private function onLessonCompleted(array $data): void
    {
        $this->completedMap[$data['lesson_id']] = new DateTimeImmutable($data['completed_at']);
        $this->totalCompleted++;
    }

    private function onReset(): void
    {
        $this->completedMap = [];
        $this->totalCompleted = 0;
    }
}

// Загружаем агрегат из Event Store
final class ProgressLoader
{
    public function __construct(
        private readonly EventStore $store,
    ) {}

    public function load(int $userId): UserProgress
    {
        $progress = new UserProgress($userId);
        foreach ($this->store->load("user-progress-$userId") as $evt) {
            $progress->apply($evt);
        }
        return $progress;
    }
}

Snapshots: оптимизация загрузки

Если у агрегата тысячи событий, проигрывать их каждый раз дорого. Snapshot фиксирует состояние на определённой версии:

Snapshot фиксирует состояние, дальше Apply вызывается только для новых событий

Events: [1] [2] [3] ... [1000] [Snapshot v1000] [1001] [1002]
                                      ↑
                               Загружаем snapshot,
                               проигрываем только 1001 и 1002

type Snapshot struct {
    AggregateID string `json:"aggregate_id"`
    Version     int    `json:"version"`
    State       []byte `json:"state"`  // сериализованный агрегат
}

func LoadWithSnapshot(ctx context.Context, store EventStore, snapshotRepo SnapshotRepo, id string) (*UserProgress, error) {
    snap, _ := snapshotRepo.GetLatest(ctx, id)

    var progress UserProgress
    startVersion := 0

    if snap != nil {
        json.Unmarshal(snap.State, &progress)
        startVersion = snap.Version
    }

    events, err := store.LoadFrom(ctx, id, startVersion+1)
    if err != nil {
        return nil, err
    }

    for _, evt := range events {
        progress.Apply(evt)
    }
    return &progress, nil
}

<?php
declare(strict_types=1);

final readonly class Snapshot
{
    public function __construct(
        public string $aggregateId,
        public int $version,
        public string $state, // сериализованный агрегат (JSON)
    ) {}
}

final class ProgressLoaderWithSnapshot
{
    public function __construct(
        private readonly EventStore $store,
        private readonly SnapshotRepository $snapshots,
    ) {}

    public function load(string $aggregateId): UserProgress
    {
        $snap = $this->snapshots->getLatest($aggregateId);

        $progress = $snap !== null
            ? UserProgress::fromState($snap->state)
            : new UserProgress(userId: (int) substr($aggregateId, strrpos($aggregateId, '-') + 1));

        $startVersion = $snap?->version ?? 0;

        foreach ($this->store->loadFrom($aggregateId, $startVersion + 1) as $evt) {
            $progress->apply($evt);
        }
        return $progress;
    }
}
Простое правило: каждые N событий (например, 100). Или когда загрузка агрегата превышает допустимый порог по времени. Не нужно снапшотить каждый раз.

Event Versioning

Со временем формат событий меняется. Нельзя менять старые - они иммутабельны. Решение - версионирование:

// v1: начальная версия
type LessonCompletedV1 struct {
    LessonID int64 `json:"lesson_id"`
}

// v2: добавили поле
type LessonCompletedV2 struct {
    LessonID  int64         `json:"lesson_id"`
    Duration  time.Duration `json:"duration"`
    Score     int           `json:"score"`
}

// Upcaster: преобразует v1 → v2
func UpcastLessonCompleted(data []byte, version int) (LessonCompletedV2, error) {
    switch version {
    case 1:
        var v1 LessonCompletedV1
        json.Unmarshal(data, &v1)
        return LessonCompletedV2{LessonID: v1.LessonID, Duration: 0, Score: 0}, nil
    case 2:
        var v2 LessonCompletedV2
        json.Unmarshal(data, &v2)
        return v2, nil
    default:
        return LessonCompletedV2{}, fmt.Errorf("unknown version: %d", version)
    }
}

<?php
declare(strict_types=1);

// v1: начальная версия
final readonly class LessonCompletedV1
{
    public function __construct(
        public int $lessonId,
    ) {}
}

// v2: добавили поля
final readonly class LessonCompletedV2
{
    public function __construct(
        public int $lessonId,
        public int $durationSeconds,
        public int $score,
    ) {}
}

// Upcaster: преобразует v1 → v2
final class LessonCompletedUpcaster
{
    public function upcast(string $rawJson, int $version): LessonCompletedV2
    {
        $data = json_decode($rawJson, true, flags: JSON_THROW_ON_ERROR);

        return match ($version) {
            1 => new LessonCompletedV2(
                lessonId: $data['lesson_id'],
                durationSeconds: 0,
                score: 0,
            ),
            2 => new LessonCompletedV2(
                lessonId: $data['lesson_id'],
                durationSeconds: $data['duration_seconds'],
                score: $data['score'],
            ),
            default => throw new RuntimeException("unknown version: $version"),
        };
    }
}

Projections: из событий в Read Model

Projection (проекция) подписывается на поток событий и строит read model:

type ProgressProjection struct {
    readDB *sql.DB
}

func (p *ProgressProjection) Handle(ctx context.Context, event Event) error {
    switch event.Type {
    case "lesson.completed":
        var data LessonCompletedV2
        // upcasting...
        _, err := p.readDB.ExecContext(ctx,
            `INSERT INTO track_lessons_view (user_id, lesson_id, completed, completed_at)
             VALUES ($1, $2, true, $3)
             ON CONFLICT (user_id, lesson_id) DO UPDATE SET completed = true, completed_at = $3`,
            data.UserID, data.LessonID, event.CreatedAt)
        return err
    }
    return nil
}

<?php
declare(strict_types=1);

final class ProgressProjection
{
    public function __construct(
        private readonly Connection $readDb,
        private readonly LessonCompletedUpcaster $upcaster,
    ) {}

    public function handle(StoredEvent $event): void
    {
        if ($event->type !== 'lesson.completed') {
            return;
        }

        $data = $this->upcaster->upcast($event->data, $event->version);

        $this->readDb->executeStatement(
            'INSERT INTO track_lessons_view (user_id, lesson_id, completed, completed_at)
             VALUES (:user_id, :lesson_id, TRUE, :completed_at)
             ON CONFLICT (user_id, lesson_id) DO UPDATE
                SET completed = TRUE, completed_at = EXCLUDED.completed_at',
            [
                'user_id'      => (int) $event->aggregateId, // упрощённо
                'lesson_id'    => $data->lessonId,
                'completed_at' => $event->createdAt->format(DATE_ATOM),
            ],
        );
    }
}

Проекцию можно пересобрать: сбрасываем read model и проигрываем все события заново.

Если у тебя обычная платформа обучения, чаще всего хватит: - обычной БД как source of truth - [outbox](../event-driven/07-outbox-inbox.md) для надёжной публикации событий - [read models](./03-read-model.md) для быстрого чтения

Event Sourcing оправдан, когда важна полная история изменений, аудит, replay или сложный биллинг.

Сравнение подходов

Критерий            State-based      Event Sourcing
──────────────      ──────────       ──────────────
Сложность           Низкая           Высокая
История изменений   Потеряна         Полная
Аудит               Нужен отдельно   Встроен
Производительность  Быстрая запись   Быстрое чтение (проекции)
Масштабирование     Вертикальное     Горизонтальное
Отладка             Смотришь БД      Проигрываешь события
Миграции            ALTER TABLE      Event versioning

Мини-задание

  • Ответь: тебе нужна история «почему стало так» или достаточно текущего состояния?
  • Набросай 3-5 событий для одного агрегата в своём проекте (например, UserProgress)
  • Напиши функцию Apply(), которая восстанавливает состояние из этих событий
  • Подумай: при каком количестве событий понадобится snapshot?

Зарегистрируйтесь бесплатно, чтобы пройти квиз, решить задание с автопроверкой и вести прогресс.