Event store на Postgres: STM, postgresql-simple, оптимистичная блокировка

Сцена · Decider ничего не помнит

Decider из прошлого урока умеет решать: даёт события на команду и сворачивает стрим в состояние. Но помнить он не умеет. Перезапусти процесс, и все брони исчезнут, потому что жили они в списке в памяти теста.

Сегодня добавляем память. Не таблицу reservations с колонкой status, которую мы перезаписываем при каждом изменении, а append-only журнал событий. Состояние в нём не хранится: оно выводится. Строка в таблице означает «вот этот факт случился», и её никто никогда не меняет.

У такого выбора есть цена, и её честно стоит назвать. Чтение агрегата становится дороже (нужно прочитать весь стрим), схема событий превращается в контракт навсегда, а удаление персональных данных требует отдельной техники. Взамен ты получаешь полный аудит без единой строчки дополнительного кода и возможность построить новую проекцию по данным, которые собраны год назад.

Карта урока

1. Контракт EventStore: две функции, стрим и версия.
2. Оптимистичная блокировка: почему две команды на одну бронь не затрут друг друга.
3. In-memory реализация на STM плюс трансляция событий подписчикам.
4. Схема Postgres и адаптер на postgresql-simple, который собирается без живой базы.

Раздел 1 · Контракт EventStore

Один агрегат это один стрим. Контракт хранилища сводится к двум операциям: прочитать стрим целиком и дописать в него события. Как и Decider, описываем его записью функций, а не классом типов:

-- | Срез стрима: его длина (версия) и события по порядку.
data StreamSlice = StreamSlice
  { sliceVersion :: Word32
  , sliceEvents :: [ReservationEvent]
  }
  deriving stock (Eq, Show)

-- | Контракт хранилища. loadStream читает стрим целиком (пустой это version=0).
-- appendStream дописывает события при условии, что версия стрима равна
-- ожидаемой, иначе ConcurrencyConflict.
data EventStore = EventStore
  { loadStream :: ReservationId -> IO StreamSlice
  , appendStream :: ReservationId -> Word32 -> [ReservationEvent] -> IO (Either DomainError ())
  }

Две реализации, in-memory и Postgres, будут разными значениями одного типа. Это удобно: тест домена берёт первое, прод второе, и подмена не требует ни моков, ни классов, ни тайпклассовой магии.

Обрати внимание на типы возврата. loadStream возвращает IO StreamSlice без Either: чтение либо отдаёт стрим, либо падает соединением, и промежуточного доменного исхода тут нет. appendStream возвращает IO (Either DomainError ()), потому что у записи есть честный доменный отказ, и он называется конфликтом версий.

Раздел 2 · Оптимистичная блокировка

Представь двух администраторов, которые одновременно нажали «заселить» на одной броне. Оба прочитали стрим, оба увидели Reserved, оба получили от decide событие GuestCheckedIn. Если обе записи пройдут, в журнале окажутся два заселения одного гостя, и любая проекция на этом сломается.

Решение это оптимистичная блокировка. appendStream принимает ожидаемую версию, и если текущая длина стрима ей не равна, запись отвергается с ConcurrencyConflict. Первая команда запишется, вторая получит конфликт, перечитает свежий стрим и перерешает уже на актуальном состоянии, где заселять, возможно, уже некого.

Ключевое отличие от пессимистичной блокировки: мы ничего не держим. Между чтением и записью транзакция не открыта, база не блокирует строки, администратор может думать сколько угодно. Мы лишь требуем, чтобы мир не изменился к моменту записи, и честно сообщаем, если он изменился.

Раздел 3 · In-memory store на STM

Первая реализация живёт в памяти. Она нужна не только для тестов: на ней видно, что весь контракт умещается в тридцать строк, а вся сложность Postgres-версии это работа с базой, а не с моделью.

Состояние это TVar с картой стримов, плюс broadcast-канал, куда транслируется каждое записанное событие. На канале в следующем уроке поедет живая проекция:

data InMemoryEventStore = InMemoryEventStore
  { imStreams :: TVar (Map ReservationId [ReservationEvent])
  , imChan :: TChan AllEventsItem
  }

newInMemoryStore :: IO InMemoryEventStore
newInMemoryStore = do
  streams <- newTVarIO Map.empty
  chan <- newBroadcastTChanIO
  pure (InMemoryEventStore streams chan)

-- | Подписка через dupTChan: получает только то, что записано после момента
-- подписки, поэтому подписываться нужно до append-ов.
subscribeStore :: InMemoryEventStore -> IO (TChan AllEventsItem)
subscribeStore store = atomically (dupTChan (imChan store))

Сама запись целиком лежит в одной STM-транзакции:

append store stream expectedVersion newEvents = atomically $ do
  m <- readTVar (imStreams store)
  let existing = Map.findWithDefault [] stream m
      actual = fromIntegral (length existing)
  if actual /= expectedVersion
    then
      pure
        ( Left
            ConcurrencyConflict
              { reservationId = unReservationId stream
              , expected = expectedVersion
              , actual = actual
              }
        )
    else do
      -- Сначала фиксируем запись в карте, затем публикуем уже записанное.
      writeTVar (imStreams store) (Map.insert stream (existing <> newEvents) m)
      mapM_ (\ev -> writeTChan (imChan store) (AllEventsItem stream ev)) newEvents
      pure (Right ())

Здесь важен atomically, знакомый по уроку про STM. Проверка версии и запись выполняются как одна транзакция: между ними не может вклиниться другой поток. Без этого проверка была бы бессмысленной, ведь именно на зазоре между «прочитал версию» и «записал» и живёт гонка.

И ещё одна деталь, из-за которой этот код выглядит именно так: публикация в канал стоит после записи в карту, внутри той же транзакции. Подписчик не увидит событие, которого нет в сторе, а если транзакция откатится, не увидит ничего. В реализации на настоящей шине это свойство пришлось бы обеспечивать отдельно, и обычно оно теряется первым.

Раздел 4 · Схема Postgres

Таблица событий выглядит одинаково почти во всех проектах, потому что требований к ней ровно четыре: порядок, принадлежность стриму, полезная нагрузка и уникальность версии.

schemaDDL :: Query
schemaDDL =
  "create table if not exists events ( \
  \  id          bigserial primary key, \
  \  stream_id   text        not null, \
  \  version     integer     not null, \
  \  type        text        not null, \
  \  payload     jsonb       not null, \
  \  created_at  timestamptz not null default now(), \
  \  unique (stream_id, version) \
  \); \
  \create index if not exists events_stream_id_idx on events (stream_id, version); \
  \create index if not exists events_id_idx on events (id);"

Разберём по строкам. id bigserial задаёт глобальный порядок всех событий: по нему читают проекции, которым нужен весь журнал, а не один стрим. version это порядок внутри стрима. unique (stream_id, version) это та самая оптимистичная блокировка, но уже гарантией базы, а не проверкой в коде: занятую версию нельзя вставить дважды даже из двух разных процессов. payload jsonb хранит событие в том формате провода, который мы задали в прошлом уроке, поэтому по нему можно строить SQL-запросы и индексы, если однажды понадобится.

Тип Query у postgresql-simple это не Text: строка SQL не склеивается с данными, а параметры всегда подставляются через ?. Так инъекция становится не «запрещённой практикой», а невыразимой конструкцией, ровно в духе первого урока трека.