eventium-core 0.6.1 → 0.7.0

raw patch · 6 files changed

+71/−16 lines, 6 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.md +23/−0
• eventium-core.cabal +1/−1
• src/Eventium/CommandHandler.hs +1/−1
• src/Eventium/ProcessManager.hs +32/−0
• src/Eventium/ProjectionCache/Cache.hs +11/−11
• src/Eventium/ProjectionCache/Types.hs +3/−3

@@ -1,5 +1,28 @@ # eventium-core Changelog +## 0.7.0++### Breaking changes++- `ProjectionCache` type parameters reordered from+  `ProjectionCache key position encoded m` to `ProjectionCache key position m encoded`+  (and the `VersionedProjectionCache` / `GlobalProjectionCache` synonyms from+  `<encoded> <m>` to `<m> <encoded>`), so the monad sits before the payload —+  consistent with `EventStoreReader` / `EventStoreWriter` and the rest of the+  library. Update any explicit `ProjectionCache` / `*ProjectionCache` type+  signatures accordingly; behaviour is unchanged.++### Additions++- `cachedProcessManagerEventHandler` -- like `processManagerEventHandler`, but+  reads and advances the process manager's global projection through a+  `GlobalProjectionCache` instead of replaying the entire global stream on every+  event. Each event folds only the events written since the last snapshot, so+  write-path latency no longer grows with the size of the event log. Correctness+  matches the uncached handler when the cache commits atomically with the write+  (e.g. a SQL-backed cache in the write transaction). Generic over event,+  command, state, and backend.+ ## 0.6.1  - Raise `base` lower bound to `>= 4.20` (GHC 9.10) to match the supported toolchain. Fixes a Hackage build failure where `foldl'` was not in scope on older `base` versions.

@@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack  name:           eventium-core-version:        0.6.1+version:        0.7.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,

@@ -78,7 +78,7 @@   (Monad m) =>   VersionedEventStoreWriter m event ->   VersionedEventStoreReader m event ->-  VersionedProjectionCache state m ->+  VersionedProjectionCache m state ->   CommandHandler state event command err ->   UUID ->   command ->

@@ -22,6 +22,7 @@     fireAndForgetDispatcher,     runProcessManagerEffects,     processManagerEventHandler,+    cachedProcessManagerEventHandler,   ) where @@ -30,6 +31,8 @@ import Data.Text (Text) import Eventium.EventHandler (EventHandler (..)) import Eventium.Projection+import Eventium.ProjectionCache.Cache (getLatestGlobalProjectionWithCache)+import Eventium.ProjectionCache.Types (GlobalProjectionCache, ProjectionCache (..)) import Eventium.Store.Class (GlobalEventStoreReader, VersionedStreamEvent) import Eventium.Store.Types (MetadataEnricher) import Eventium.UUID@@ -147,5 +150,34 @@ processManagerEventHandler pm globalReader dispatcher = EventHandler $ \event -> do   let globalProj = globalStreamProjection pm.projection   sp <- getLatestStreamProjection globalReader globalProj+  let effects = pm.react sp.state event+  runProcessManagerEffects dispatcher effects++-- | Like 'processManagerEventHandler', but reads and advances the process+-- manager's global projection through a 'GlobalProjectionCache' instead of+-- replaying the entire global stream on every event.+--+-- For each event it loads the last snapshot and folds only the events written+-- since it (via 'getLatestGlobalProjectionWithCache'), then persists the+-- advanced snapshot. Cost is O(events since the snapshot) per call rather than+-- O(total store size), so write-path latency no longer grows with the event+-- log. Wire the same 'GlobalProjectionCache' into a startup catch-up if you want+-- to avoid a one-time full fold on the first event after the cache is empty.+--+-- Correctness matches the uncached handler when the cache commits atomically+-- with the write (e.g. a SQL-backed cache in the write transaction): the+-- snapshot advances iff the events do. Generic over event, command, state and+-- backend — the 'GlobalProjectionCache' abstracts persistence.+cachedProcessManagerEventHandler ::+  (Monad m) =>+  ProcessManager state event command ->+  GlobalEventStoreReader m event ->+  GlobalProjectionCache m state ->+  CommandDispatcher m command ->+  EventHandler m (VersionedStreamEvent event)+cachedProcessManagerEventHandler pm globalReader cache dispatcher = EventHandler $ \event -> do+  let globalProj = globalStreamProjection pm.projection+  sp <- getLatestGlobalProjectionWithCache globalReader cache globalProj+  cache.storeSnapshot () sp.position sp.state   let effects = pm.react sp.state event   runProcessManagerEffects dispatcher effects

@@ -23,8 +23,8 @@ -- cache in another 'Monad' while forgetting the original 'Monad'. runProjectionCacheUsing ::   (forall a. mstore a -> m a) ->-  ProjectionCache key position encoded mstore ->-  ProjectionCache key position encoded m+  ProjectionCache key position mstore encoded ->+  ProjectionCache key position m encoded runProjectionCacheUsing runCache pc =   ProjectionCache     { storeSnapshot = \uuid version st -> runCache $ pc.storeSnapshot uuid version st,@@ -37,8 +37,8 @@ codecProjectionCache ::   (Monad m) =>   Codec state encoded ->-  ProjectionCache key position encoded m ->-  ProjectionCache key position state m+  ProjectionCache key position m encoded ->+  ProjectionCache key position m state codecProjectionCache codec pc =   ProjectionCache storeSnapshot' loadSnapshot'   where@@ -52,7 +52,7 @@ getLatestVersionedProjectionWithCache ::   (Monad m) =>   VersionedEventStoreReader m event ->-  VersionedProjectionCache state m ->+  VersionedProjectionCache m state ->   VersionedStreamProjection state event ->   m (VersionedStreamProjection state event) getLatestVersionedProjectionWithCache store cache proj =@@ -63,7 +63,7 @@ getLatestGlobalProjectionWithCache ::   (Monad m) =>   GlobalEventStoreReader m event ->-  GlobalProjectionCache state m ->+  GlobalProjectionCache m state ->   GlobalStreamProjection state event ->   m (GlobalStreamProjection state event) getLatestGlobalProjectionWithCache store cache proj =@@ -71,7 +71,7 @@  getLatestProjectionWithCache' ::   (Monad m, Ord position) =>-  ProjectionCache key position state m ->+  ProjectionCache key position m state ->   StreamProjection projKey position state event ->   key ->   m (StreamProjection projKey position state event)@@ -92,7 +92,7 @@ updateVersionedProjectionCache ::   (Monad m) =>   VersionedEventStoreReader m event ->-  VersionedProjectionCache state m ->+  VersionedProjectionCache m state ->   VersionedStreamProjection state event ->   m () updateVersionedProjectionCache reader cache proj = do@@ -103,7 +103,7 @@ updateGlobalProjectionCache ::   (Monad m) =>   GlobalEventStoreReader m event ->-  GlobalProjectionCache state m ->+  GlobalProjectionCache m state ->   GlobalStreamProjection state event ->   m () updateGlobalProjectionCache reader cache proj = do@@ -117,7 +117,7 @@ snapshotEventHandler ::   (Monad m) =>   VersionedEventStoreReader m event ->-  VersionedProjectionCache state m ->+  VersionedProjectionCache m state ->   Projection state event ->   EventHandler m (VersionedStreamEvent event) snapshotEventHandler reader cache proj =@@ -132,7 +132,7 @@ snapshotGlobalEventHandler ::   (Monad m) =>   GlobalEventStoreReader m event ->-  GlobalProjectionCache state m ->+  GlobalProjectionCache m state ->   Projection state (VersionedStreamEvent event) ->   EventHandler m (GlobalStreamEvent event) snapshotGlobalEventHandler reader cache proj =

@@ -20,7 +20,7 @@ -- The @key@ and @position@ type parameters are polymorphic so we can abstract -- over a cache for individual event streams, and a cache for globally ordered -- streams.-data ProjectionCache key position encoded m+data ProjectionCache key position m encoded   = ProjectionCache   { -- | Stores the state for a projection at a given @key@ and @position@.     -- This is pretty unsafe, because there is no guarantee what is stored is@@ -32,7 +32,7 @@   }  -- | Type synonym for a 'ProjectionCache' used on individual event streams.-type VersionedProjectionCache encoded m = ProjectionCache UUID EventVersion encoded m+type VersionedProjectionCache m encoded = ProjectionCache UUID EventVersion m encoded  -- | Type synonym for a 'ProjectionCache' that is used in conjunction with a -- 'GlobalStreamEventStore'. The key is fixed to @()@ (singleton) because@@ -41,4 +41,4 @@ -- The original @eventful@ library kept the key polymorphic -- (@GloballyOrderedProjectionCache key serialized m@), but that is redundant -- when each cache instance is already scoped by name.-type GlobalProjectionCache encoded m = ProjectionCache () SequenceNumber encoded m+type GlobalProjectionCache m encoded = ProjectionCache () SequenceNumber m encoded