packages feed

eventium-core 0.3.2 → 0.4.0

raw patch · 7 files changed

+225/−27 lines, 7 files

Files

eventium-core.cabal view
@@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack  name:           eventium-core-version:        0.3.2+version:        0.4.0 synopsis:       Core module for eventium description:    Eventium-core provides the core abstractions and utilities for building event sourcing systems in Haskell.                 It includes event store interfaces, command handlers, projections, event handlers, event publishers,
src/Eventium/CommandHandler.hs view
@@ -91,9 +91,14 @@       result <- writer.storeEvents uuid (ExactPosition sp.position) events       case result of         Left writeErr -> return $ Left (ConcurrencyConflict writeErr)-        Right endVersion -> do-          let newState = foldl' cmdHandler.projection.eventHandler sp.state events-          cache.storeSnapshot uuid endVersion newState+        Right writeResult -> do+          -- Refresh the snapshot only when events were actually written; the+          -- end version is the last written event's per-stream version.+          case lastVersion writeResult of+            Nothing -> return ()+            Just endV -> do+              let newState = foldl' cmdHandler.projection.eventHandler sp.state events+              cache.storeSnapshot uuid endV newState           return $ Right events  -- | Use a pair of 'Codec's to wrap a 'CommandHandler' with event type
src/Eventium/EventPublisher.hs view
@@ -1,4 +1,5 @@ {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TupleSections #-}  -- | Defines an event publisher abstraction that decouples event notification -- from event storage.@@ -9,14 +10,23 @@ -- Use 'publishingEventStoreWriter' to wrap an 'EventStoreWriter' so that -- events are automatically published after a successful write. module Eventium.EventPublisher-  ( EventPublisher (..),+  ( -- * Per-stream (versioned) publishing+    EventPublisher (..),     synchronousPublisher,     publishingEventStoreWriter,     publishingTaggedCodecEventStoreWriter,++    -- * Global publishing (real 'SequenceNumber's)+    GlobalEventPublisher (..),+    synchronousGlobalPublisher,+    globalToVersionedHandler,+    publishingGlobalEventStoreWriter,+    publishingGlobalTaggedCodecEventStoreWriter,   ) where  import Control.Exception (throw)+import Data.Functor.Contravariant (contramap) import Eventium.Codec (Codec (..), DecodeError (..)) import Eventium.EventHandler import Eventium.Store.Class@@ -56,11 +66,10 @@     result <- write uuid expectedPos events     case result of       Left err -> return $ Left err-      Right endVersion -> do-        let startVersion = endVersion - fromIntegral (length events) + 1-            versionedEvents = zipWith (\v e -> StreamEvent uuid v (emptyMetadata "") e) [startVersion ..] events+      Right wr -> do+        let versionedEvents = zipWith (\(v, _) e -> StreamEvent uuid v (emptyMetadata "") e) wr events         publish uuid versionedEvents-        return $ Right endVersion+        return $ Right wr  -- | Like 'publishingEventStoreWriter' but for writers that accept -- @TaggedEvent encoded@. Each tagged event's payload is decoded through the@@ -80,17 +89,114 @@     result <- write uuid expectedPos taggedEvents     case result of       Left err -> return $ Left err-      Right endVersion -> do-        let startVersion = endVersion - fromIntegral (length taggedEvents) + 1-            versionedEvents =+      Right wr -> do+        let versionedEvents =               zipWith-                ( \v (TaggedEvent meta enc) ->+                ( \(v, _) (TaggedEvent meta enc) ->                     let event = case codec.decode enc of                           Just e -> e                           Nothing -> throw $ DecodeError "publishingTaggedCodecEventStoreWriter" "Failed to decode tagged event payload"                      in StreamEvent uuid v meta event                 )-                [startVersion ..]+                wr                 taggedEvents         publish uuid versionedEvents-        return $ Right endVersion+        return $ Right wr++-- -----------------------------------------------------------------------------+-- Global publishing+-- -----------------------------------------------------------------------------++-- | A 'GlobalEventPublisher' pushes 'GlobalStreamEvent's — events carrying their+-- real global 'SequenceNumber' — to consumers after a successful write.+--+-- Unlike 'EventPublisher' (which only knows per-stream versions), this uses the+-- 'EventWriteResult' to deliver the true global positions, so a synchronous+-- subscriber (e.g. a 'Eventium.ReadModel.ReadModel' driven in the write+-- transaction) sees the same positions the global reader would.+newtype GlobalEventPublisher m event = GlobalEventPublisher+  { publishGlobalEvents :: [GlobalStreamEvent event] -> m ()+  }++-- | Fan out to both publishers, in order.+instance (Applicative m) => Semigroup (GlobalEventPublisher m event) where+  GlobalEventPublisher p1 <> GlobalEventPublisher p2 =+    GlobalEventPublisher $ \events -> p1 events *> p2 events++instance (Applicative m) => Monoid (GlobalEventPublisher m event) where+  mempty = GlobalEventPublisher $ \_ -> pure ()++-- | A 'GlobalEventPublisher' that synchronously delivers each global event to an+-- 'EventHandler'.+synchronousGlobalPublisher ::+  (Monad m) =>+  EventHandler m (GlobalStreamEvent event) ->+  GlobalEventPublisher m event+synchronousGlobalPublisher handler = GlobalEventPublisher (handleEvents handler)++-- | Lift a per-stream ('VersionedStreamEvent') handler to consume+-- 'GlobalStreamEvent's, by projecting to the inner versioned event. Lets+-- existing versioned consumers (process managers, loggers) compose into a+-- 'GlobalEventPublisher'.+globalToVersionedHandler ::+  EventHandler m (VersionedStreamEvent event) ->+  EventHandler m (GlobalStreamEvent event)+globalToVersionedHandler = contramap (.payload)++-- | Build a 'GlobalStreamEvent' for each written event from the 'EventWriteResult'+-- (its per-stream version + assigned global position) and its payload.+toGlobalStreamEvents ::+  UUID ->+  EventWriteResult ->+  (a -> (EventMetadata, event)) ->+  [a] ->+  [GlobalStreamEvent event]+toGlobalStreamEvents uuid wr unpack =+  zipWith+    ( \(v, gseq) a ->+        let (meta, event) = unpack a+         in StreamEvent () gseq meta (StreamEvent uuid v meta event)+    )+    wr++-- | Like 'publishingEventStoreWriter' but publishes 'GlobalStreamEvent's with+-- real global positions via a 'GlobalEventPublisher'.+publishingGlobalEventStoreWriter ::+  (Monad m) =>+  VersionedEventStoreWriter m event ->+  GlobalEventPublisher m event ->+  VersionedEventStoreWriter m event+publishingGlobalEventStoreWriter (EventStoreWriter write) (GlobalEventPublisher publish) =+  EventStoreWriter $ \uuid expectedPos events -> do+    result <- write uuid expectedPos events+    case result of+      Left err -> return $ Left err+      Right wr -> do+        publish (toGlobalStreamEvents uuid wr (emptyMetadata "",) events)+        return $ Right wr++-- | Like 'publishingTaggedCodecEventStoreWriter' but publishes+-- 'GlobalStreamEvent's with real global positions. Each tagged payload is+-- decoded through the 'Codec'; the tagged metadata is preserved on both the+-- inner versioned event and the outer global event (mirroring the global+-- reader).+--+-- Throws 'DecodeError' if any event fails to decode.+publishingGlobalTaggedCodecEventStoreWriter ::+  (Monad m) =>+  Codec event encoded ->+  VersionedEventStoreWriter m (TaggedEvent encoded) ->+  GlobalEventPublisher m event ->+  VersionedEventStoreWriter m (TaggedEvent encoded)+publishingGlobalTaggedCodecEventStoreWriter codec (EventStoreWriter write) (GlobalEventPublisher publish) =+  EventStoreWriter $ \uuid expectedPos taggedEvents -> do+    result <- write uuid expectedPos taggedEvents+    case result of+      Left err -> return $ Left err+      Right wr -> do+        let decodeTagged (TaggedEvent meta enc) =+              case codec.decode enc of+                Just e -> (meta, e)+                Nothing -> throw $ DecodeError "publishingGlobalTaggedCodecEventStoreWriter" "Failed to decode tagged event payload"+        publish (toGlobalStreamEvents uuid wr decodeTagged taggedEvents)+        return $ Right wr
src/Eventium/ReadModel.hs view
@@ -9,8 +9,10 @@ module Eventium.ReadModel   ( ReadModel (..),     runReadModel,+    catchUpReadModel,     rebuildReadModel,     combineReadModels,+    readModelPublisher,   ) where @@ -20,6 +22,7 @@ import Data.Foldable (traverse_) import qualified Data.List.NonEmpty as NE import Eventium.EventHandler+import Eventium.EventPublisher (GlobalEventPublisher (..)) import Eventium.EventSubscription (CheckpointStore (..), PollingIntervalMillis) import Eventium.Store.Class @@ -30,7 +33,8 @@ -- * 'initialize' — idempotent setup (run migrations, create tables) -- * 'eventHandler' — processes global stream events, writes to user-defined storage -- * 'checkpointStore' — tracks the last processed 'SequenceNumber'--- * 'reset' — drop view data and reset checkpoint (for full rebuilds)+-- * 'reset' — drop the view's data (tables). The checkpoint is reset by+--   'rebuildReadModel', so 'reset' need only clear user-owned storage. data ReadModel m event = ReadModel   { initialize :: m (),     eventHandler :: EventHandler m (GlobalStreamEvent event),@@ -50,15 +54,20 @@   rm.initialize   forever $ pollReadModelOnce globalReader pollIntervalMs rm --- | Reset the read model and replay all events from the beginning.--- Returns after processing all currently available events.-rebuildReadModel ::+-- | Bring a read model up to date from its current checkpoint, without+-- resetting it. Initializes, then replays @checkpoint+1 → latest@ and returns+-- once all currently-available events are processed.+--+-- This is the one-shot startup/backfill counterpart to 'runReadModel' (which+-- polls forever) and 'rebuildReadModel' (which resets first). Safe to run on a+-- persistent read model at boot: it never wipes durable state, and replays+-- nothing when the checkpoint is already current.+catchUpReadModel ::   (Monad m) =>   GlobalEventStoreReader m event ->   ReadModel m event ->   m ()-rebuildReadModel globalReader rm = do-  rm.reset+catchUpReadModel globalReader rm = do   rm.initialize   replayAll   where@@ -72,6 +81,20 @@           rm.checkpointStore.saveCheckpoint (NE.last ne).position           replayAll +-- | Reset the read model and replay all events from the beginning.+-- Returns after processing all currently available events.+rebuildReadModel ::+  (Monad m) =>+  GlobalEventStoreReader m event ->+  ReadModel m event ->+  m ()+rebuildReadModel globalReader rm = do+  rm.reset+  -- Reset the checkpoint we own, so the user's 'reset' only has to drop view+  -- data; then replay from the start.+  rm.checkpointStore.saveCheckpoint 0+  catchUpReadModel globalReader rm+ -- | Combine multiple read models into one. Events are fanned out to all -- handlers. Initialize and reset run all sub-models. --@@ -94,6 +117,25 @@           },       reset = traverse_ (.reset) rms     }++-- | Drive a 'ReadModel' synchronously from a write's published global events:+-- apply its handler to the batch and advance its checkpoint to the last event's+-- global position. The synchronous counterpart to 'runReadModel' — wire it into+-- a 'publishingGlobalEventStoreWriter' / 'publishingGlobalTaggedCodecEventStoreWriter'+-- to project the read model in the write transaction (strong consistency).+--+-- The same 'ReadModel' value can therefore be run either asynchronously+-- ('runReadModel') or synchronously (here). 'initialize'/'reset'/'rebuildReadModel'+-- remain the lifecycle/backfill path.+readModelPublisher ::+  (Monad m) =>+  ReadModel m event ->+  GlobalEventPublisher m event+readModelPublisher rm = GlobalEventPublisher $ \events -> do+  handleEvents rm.eventHandler events+  case NE.nonEmpty events of+    Nothing -> return ()+    Just ne -> rm.checkpointStore.saveCheckpoint (NE.last ne).position  pollReadModelOnce ::   (MonadIO m) =>
src/Eventium/Store/Class.hs view
@@ -59,7 +59,7 @@ -- | An 'EventStoreWriter' is a function to write some events of type @event@ -- to an event store in some monad @m@. newtype EventStoreWriter key position m event-  = EventStoreWriter {storeEvents :: key -> ExpectedPosition position -> [event] -> m (Either (EventWriteError position) EventVersion)}+  = EventStoreWriter {storeEvents :: key -> ExpectedPosition position -> [event] -> m (Either (EventWriteError position) EventWriteResult)}  instance Contravariant (EventStoreWriter key position m) where   contramap f (EventStoreWriter writer) = EventStoreWriter $ \key expectedPos -> writer key expectedPos . fmap f@@ -72,11 +72,11 @@ transactionalExpectedWriteHelper ::   (Monad m, Ord position, Num position) =>   (key -> m position) ->-  (key -> [event] -> m EventVersion) ->+  (key -> [event] -> m EventWriteResult) ->   key ->   ExpectedPosition position ->   [event] ->-  m (Either (EventWriteError position) EventVersion)+  m (Either (EventWriteError position) EventWriteResult) transactionalExpectedWriteHelper getLatestVersion' storeEvents' key expected =   go expected getLatestVersion' storeEvents' key   where@@ -89,10 +89,10 @@   (Monad m) =>   Maybe (position -> Bool) ->   (key -> m position) ->-  (key -> [event] -> m EventVersion) ->+  (key -> [event] -> m EventWriteResult) ->   key ->   [event] ->-  m (Either (EventWriteError position) EventVersion)+  m (Either (EventWriteError position) EventWriteResult) transactionalExpectedWriteHelper' Nothing _ storeEvents' uuid events =   storeEvents' uuid events <&> Right transactionalExpectedWriteHelper' (Just f) getLatestVersion' storeEvents' uuid events = do
src/Eventium/Store/Types.hs view
@@ -22,6 +22,14 @@     ExpectedPosition (..),     EventWriteError (..), +    -- * Write result+    WrittenEventPosition,+    EventWriteResult,+    versions,+    globalPositions,+    lastVersion,+    lastPosition,+     -- * Utility types     EventVersion (..),     SequenceNumber (..),@@ -29,6 +37,7 @@ where  import Data.Aeson+import qualified Data.List.NonEmpty as NE import Data.Text (Text) import Data.Time (UTCTime) import Eventium.UUID@@ -106,6 +115,41 @@ newtype EventWriteError position   = EventStreamNotAtExpectedVersion position   deriving (Show, Eq)++-- | A written event's assigned per-stream 'EventVersion' paired with the global+-- 'SequenceNumber' the store gave it. Pairing version and position per event+-- (rather than returning two parallel lists) keeps them structurally in sync.+type WrittenEventPosition = (EventVersion, SequenceNumber)++-- | The outcome of a successful write: one '(EventVersion, SequenceNumber)' per+-- event written, in write order (so the stream's resulting end version is+-- @fst (last writeResult)@).+--+-- Exposing the assigned global positions lets a synchronous subscriber publish+-- 'GlobalStreamEvent's with real positions (and advance a 'CheckpointStore' in+-- the write transaction), rather than fabricating a position. The positions are+-- assigned under the store's global-ordering guarantee (the PostgreSQL exclusive+-- lock / SQLite single-writer / memory single-append), so they match what the+-- global reader derives for the same events.+type EventWriteResult = [WrittenEventPosition]++-- | All per-stream versions assigned by a write, in write order.+versions :: EventWriteResult -> [EventVersion]+versions = map fst++-- | All global positions assigned by a write, in write order.+globalPositions :: EventWriteResult -> [SequenceNumber]+globalPositions = map snd++-- | The per-stream end version (the last written event's version), or 'Nothing'+-- when no events were written.+lastVersion :: EventWriteResult -> Maybe EventVersion+lastVersion = fmap (fst . NE.last) . NE.nonEmpty++-- | The last (highest) global position assigned, or 'Nothing' when no events+-- were written. Handy for advancing a 'Eventium.EventSubscription.CheckpointStore'.+lastPosition :: EventWriteResult -> Maybe SequenceNumber+lastPosition = fmap (snd . NE.last) . NE.nonEmpty  -- | Event versions are a strictly increasing series of integers for each -- projection. They allow us to order the events when they are replayed, and
tests/Eventium/CommandDispatcherSpec.hs view
@@ -49,8 +49,9 @@         let events = map (.payload) taggedEvents             startVersion = fromIntegral (length existing)             versioned = zipWith (\i e -> StreamEvent uuid i (emptyMetadata "") e) [startVersion ..] events+            poss = take (length events) [SequenceNumber (length existing + 1) ..]         modifyIORef eventsRef (++ versioned)-        pure (Right (startVersion + fromIntegral (length events) - 1))+        pure (Right (zip [startVersion ..] poss))       reader = EventStoreReader $ \query -> do         allEvts <- readIORef eventsRef         pure $ filterByQuery query allEvts