Проекции и CQRS на STM: чистый фолд, forkIO, снимок в TVar

Сцена · журнал плохо отвечает на вопросы

Журнал событий отлично хранит историю и отвратительно отвечает на вопросы. «Кто сейчас в номерах?» это простейший запрос администратора, и в event-sourced системе он неожиданно дорогой: пришлось бы прочитать все стримы всех броней, свернуть каждый в состояние и отфильтровать заселённые.

Ответ на это не «давайте всё-таки хранить состояние», а разделение ролей. Модель для записи (Decider плюс журнал) оптимизирована под правила и согласованность. Модель для чтения строится отдельно, под конкретный вопрос, и живёт ровно столько, сколько этот вопрос кому-то нужен. Разделение называется CQRS, и это последний кирпич трека.

Сегодня строим read-модель occupancy: кто в каком номере сейчас живёт. И строим её так, как это принято в Haskell: чистый фолд отдельно, конкурентная обвязка отдельно.

Раздел 1 · Проекция это фолд, а не подписчик

Слово «проекция» звучит инфраструктурно, но по сути это функция:

apply :: state -> event -> state

Ровно та же форма, что у evolve из урока про Decider. Разница в назначении: evolve строит состояние агрегата для принятия решений, apply строит ответ на вопрос для показа. evolve работает с одним стримом, apply со всеми сразу.

Начнём с типов read-модели. Строка это ровно то, что нужно на экране ресепшен, и ни поля больше:

-- | Строка read-model: бронь, комната, гость.
data OccupancyRow = OccupancyRow
  { reservationId :: ReservationId
  , room :: RoomNumber
  , guest :: GuestEmail
  }
  deriving stock (Eq, Show)

type Occupancy = Map ReservationId OccupancyRow

Дальше возникает тонкость, из-за которой проекция почти никогда не бывает одной строчкой. Заселение приходит событием GuestCheckedIn, а в нём лежит только id брони: ни комнаты, ни гостя. Они были в ReservationPlaced, которое случилось раньше. Значит, проекции нужна собственная память о том, что происходило до:

-- | Снимок проекции: материализованный read-model плюс забронированные, но ещё
-- не заселённые брони (pending). pending нужен, чтобы при GuestCheckedIn
-- достать комнату и гостя, заданных в ReservationPlaced.
data OccupancySnapshot = OccupancySnapshot
  { occupancy :: Occupancy
  , pending :: Map ReservationId OccupancyRow
  }
  deriving stock (Eq, Show)

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

Раздел 2 · Чистый фолд

Сам шаг проекции это обычная тотальная функция без единого эффекта:

apply :: OccupancySnapshot -> AllEventsItem -> OccupancySnapshot
apply snapshot item =
  let sid = aeiStreamId item
   in case aeiEvent item of
        ReservationPlaced {room, guest} ->
          let row = OccupancyRow {reservationId = sid, room = room, guest = guest}
           in snapshot {pending = Map.insert sid row (pending snapshot)}
        GuestCheckedIn {} ->
          case Map.lookup sid (pending snapshot) of
            Nothing -> snapshot
            Just row ->
              snapshot
                { pending = Map.delete sid (pending snapshot)
                , occupancy = Map.insert sid row (occupancy snapshot)
                }
        GuestCheckedOut {} -> remove sid snapshot
        ReservationCancelled {} -> remove sid snapshot
  where
    remove sid s =
      s
        { occupancy = Map.delete sid (occupancy s)
        , pending = Map.delete sid (pending s)
        }

Обрати внимание на ветку GuestCheckedIn с Nothing: событие пришло, а брони в pending нет. В нормальном журнале так не бывает, но проекция обязана пережить и это. Правило read-side то же, что для evolve: не падать никогда. Упавшая проекция это не потерянная запись, а остановившийся экран у администратора, и чинить её придётся руками в проде.

Заметь ещё, что apply ничего не знает ни про STM, ни про каналы, ни про базу. Поэтому она тестируется списком событий в одну строку, без асинхронности:

it "GuestCheckedIn переносит из pending в occupancy" $ do
  let s = foldl apply emptySnapshot [item placed, item checkedIn]
  Map.size (Occ.occupancy s) `shouldBe` 1
  Map.size (Occ.pending s) `shouldBe` 0
  fmap (unRoomNumber . Occ.room) (Map.lookup resId (Occ.occupancy s)) `shouldBe` Just "101"

Разделение «чистый фолд плюс тонкая обвязка» это главный приём урока. Всё сложное в проекции это её логика, и она обязана быть проверяемой без потоков.

Раздел 3 · Живой поток: forkIO поверх подписки

Теперь обвязка. У нас есть broadcast-канал event store из прошлого урока, значит нужен кто-то, кто будет его читать и складывать результат туда, где его увидит читатель:

runOccupancyProjection :: TChan AllEventsItem -> IO (TVar Occupancy, ThreadId)
runOccupancyProjection chan = do
  view <- newTVarIO Map.empty
  tid <- forkIO (loop view emptySnapshot)
  pure (view, tid)
  where
    -- Бесконечный цикл с аккумулятором: одно событие, один шаг фолда,
    -- публикация свежего снимка occupancy.
    loop view snapshot = do
      item <- atomically (readTChan chan)
      let snapshot' = apply snapshot item
      atomically (writeTVar view (occupancy snapshot'))
      loop view snapshot'

Двенадцать строк, и в них три решения.

Первое: аккумулятор живёт в аргументе цикла, а не в TVar. Наружу публикуется только occupancy, готовый к показу, а служебный pending остаётся внутри потока. Read-модель отдаёт ответ на вопрос, а не свои внутренности.

Второе: forkIO порождает зелёный поток, а не системный. Отдельный поток на проекцию в Haskell стоит примерно ничего, и потому естественная архитектура это один поток на каждую read-модель.

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