Event-sourcing and projection

In our previous log, we have described an effect-based event sourcing system.

After giving it some thought, we can use the same idea for projections.

The basic idea behind this is to introduce an interceptor, like an interpreter, without effect consumption.

It should be placed right after event storage.

There are two ways to build it:

• Adding a new effect
• Creating the interceptor over PlainEventSourceEffect

The first option is heavy, while the second requires either a lot of plumbing (in polysemy, high over effect and high over interpreter do not mix) or dropping the high order effect.

We can start by simplifying the effect as follows:

data PlainEventSourceEffect (eventType :: Type) (m :: Type -> Type) (a :: Type) where
  FetchPlainEvents ::
    AggregateId ->
    PlainEventSourceEffect eventType m (Events eventType)
  StorePlainEvents ::
    AggregateId ->
    Events eventType ->
    PlainEventSourceEffect eventType m ()
  ListPlainAggregateIds ::
    Proxy eventType ->
    PlainEventSourceEffect eventType m [AggregateId]

To limit breakage, we can come up with the following helper, which is a substitution of the operation:

withPlainEvents ::
  forall eventType a r.
  (Member (PlainEventSourceEffect eventType) r) =>
  AggregateId ->
  (Events eventType -> Sem r (Events eventType, a)) ->
  Sem r a
withPlainEvents aggregateId f = do
  originalAggregateEvents <- fetchPlainEvents aggregateId
  (newEvents, x) <- f originalAggregateEvents
  storePlainEvents aggregateId newEvents
  return x

This leads to a drastic simplification of the in-memory interpreter, as follows:

type PlainMemoryStore eventType = Map.Map AggregateId (Events eventType)

runEventSourceMemory ::
  forall eventType r.
  InterpreterFor (PlainEventSourceEffect eventType) r
runEventSourceMemory =
  evalState @(PlainMemoryStore eventType) mempty
    . go
    . raiseUnder @(State (PlainMemoryStore eventType))
  where
    go :: InterpreterFor (PlainEventSourceEffect eventType) (State (PlainMemoryStore eventType) ': r)
    go =
      interpret $
        \case
          FetchPlainEvents aggregateId ->
            gets @(Map.Map AggregateId (Events eventType)) (Map.findWithDefault mempty aggregateId)
          StorePlainEvents aggregateId newEvents ->
            modify $ Map.alter (Just . maybe newEvents (<> newEvents)) aggregateId
          ListPlainAggregateIds _ -> do
            gets @(Map.Map AggregateId (Events eventType)) Map.keys

That being done, we can come up with the following interceptor:

withProjection ::
  forall eventType r a x.
  (AggregateId -> Events eventType -> Sem r x) ->
  Sem (PlainEventSourceEffect eventType ': r) a ->
  Sem (PlainEventSourceEffect eventType ': r) a
withProjection project =
  reinterpret $
    \case
      FetchPlainEvents aggregateId ->
        fetchPlainEvents aggregateId
      StorePlainEvents aggregateId newEvents -> do
        storePlainEvents aggregateId newEvents
        void $ raise $ project aggregateId newEvents
      ListPlainAggregateIds p -> do
        listPlainAggregateIds p

As stated in the introduction, we simply pass through the operations to other consumers, applying the callback after the storage.

As a reminder, have the following events:

data TourEvent
  = TourCreated {name :: Text, start :: SpaceTime, end :: SpaceTime, groupSize :: Int}
  | BookingRequested {id :: BookingRequestId, buyer :: PersonIdentity, groupSize :: Int}
  | BookingAccepted {id :: BookingRequestId}
  deriving stock (Eq, Show, Generic)
  deriving (FromJSON, ToJSON) via (Generically TourEvent)

A natural projection is to keep a list of up-to-date tours and bookings; we can start with the following domain:

data Tour = Tour
  { name :: Text,
    start :: SpaceTime,
    end :: SpaceTime,
    groupSize :: Int,
    bookings :: Bookings
  }
  deriving stock (Eq, Show)

type Bookings = Map.Map BookingRequestId Booking

updateTourBookings :: (Bookings -> Bookings) -> Tour -> Tour
updateTourBookings f Tour {..} =
  Tour {bookings = f bookings, ..}

data Booking = Booking
  { id :: BookingRequestId,
    buyer :: PersonIdentity,
    groupSize :: Int,
    accepted :: Bool
  }
  deriving stock (Eq, Show)

The next step is to come up with a projection, applying each event to update the structure, as follows:

type BookingsMemoryState = Map.Map AggregateId Tour

projectBookings :: (Members '[State BookingsMemoryState] r) => AggregateId -> Events TourEvent -> Sem r ()
projectBookings aggregateId =
  mapM_ $
    \case
      TourCreated {..} ->
        modify $
          Map.insert aggregateId Tour {bookings = mempty, ..}
      BookingRequested {..} ->
        modify $
          Map.adjust
            (updateTourBookings $ Map.insert id Booking {accepted = False, ..})
            aggregateId
      BookingAccepted {..} ->
        modify $
          Map.adjust
            (updateTourBookings $ Map.adjust (\Booking {..} -> Booking {accepted = True, ..}) id)
            aggregateId

We can finally draft a small test, as follows:

spec :: Spec
spec =
  describe "Tours" $ do
    it "Booking projection" $
      let bookings =
            run $
              execState @BookingsMemoryState mempty $
                runEventSourceMemory @TourEvent $
                  withProjection projectBookings $ do
                    let aggregateId = AggregateId "0"
                    -- Command 0
                    withPlainEvents aggregateId $ \_ ->
                      return ([TourCreated "Eiffle Tower" (SpaceTime "In front" $ read "2026-08-18 09:00:00 +0200") (SpaceTime "In front" $ read "2026-08-18 10:00:00 +0200") 12], ())
                    -- Command 1
                    withPlainEvents aggregateId $ \_ ->
                      return ([BookingRequested (BookingRequestId "0") (PersonIdentity "Marvin") 3], ())
                    -- Command 2
                    withPlainEvents aggregateId $ \_ ->
                      return ([BookingAccepted (BookingRequestId "0")], ())
       in bookings
            `shouldBe` Map.singleton
              (AggregateId "0")
              Tour
                { name = "Eiffle Tower",
                  start = SpaceTime "In front" $ read "2026-08-18 09:00:00 +0200",
                  end = SpaceTime "In front" $ read "2026-08-18 10:00:00 +0200",
                  groupSize = 12,
                  bookings = Map.singleton (BookingRequestId "0") $ Booking (BookingRequestId "0") (PersonIdentity "Marvin") 3 True
                }

Having a projection as an interceptor gives a lot of modularity, but it comes at the cost of ordering consumers:

• Start with interceptors (projections)
• Continue with the interpreter (consuming the event store)
• Consume the remaining effects