diff --git a/eventium-memory.cabal b/eventium-memory.cabal --- a/eventium-memory.cabal +++ b/eventium-memory.cabal @@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack name: eventium-memory -version: 0.3.2 +version: 0.4.0 synopsis: In-memory implementations for eventium description: Eventium-memory provides in-memory implementations of event stores and projection caches for the Eventium event sourcing framework. This package is ideal for development, testing, and prototyping @@ -49,7 +49,7 @@ build-depends: base >=4.9 && <5 , containers >=0.6 && <0.8 - , eventium-core >=0.3.0 && <0.4.0 + , eventium-core >=0.4.0 && <0.5.0 , mtl >=2.2 && <2.4 , safe ==0.3.* , stm ==2.5.* @@ -89,7 +89,7 @@ HUnit , base >=4.9 && <5 , containers >=0.6 && <0.8 - , eventium-core >=0.3.0 && <0.4.0 + , eventium-core >=0.4.0 && <0.5.0 , eventium-testkit , hspec , mtl >=2.2 && <2.4 diff --git a/src/Eventium/Store/Memory.hs b/src/Eventium/Store/Memory.hs --- a/src/Eventium/Store/Memory.hs +++ b/src/Eventium/Store/Memory.hs @@ -191,23 +191,29 @@ (StartQueryAt startSeq) -> startSeq storeEventMap :: - EventMap event -> UUID -> [event] -> (EventMap event, EventVersion) + EventMap event -> UUID -> [event] -> (EventMap event, EventWriteResult) storeEventMap store@(EventMap uuidMap globalEvents) uuid events = let versStart = latestEventVersion store uuid - streamEvents = zipWith (\v e -> StreamEvent uuid v (emptyMetadata "") e) [versStart + 1 ..] events + vers = take (length events) [versStart + 1 ..] + streamEvents = zipWith (\v e -> StreamEvent uuid v (emptyMetadata "") e) vers events newMap = Map.insertWith (flip (><)) uuid (Seq.fromList streamEvents) uuidMap globalEvents' = globalEvents >< Seq.fromList streamEvents - in (EventMap newMap globalEvents', versStart + EventVersion (length events)) + -- The global reader numbers events 1-based by their position in + -- 'globalEvents'; mirror that from the length before appending. + globals = take (length events) [SequenceNumber (Seq.length globalEvents + 1) ..] + in (EventMap newMap globalEvents', zip vers globals) storeEventMapTagged :: - EventMap event -> UUID -> [TaggedEvent event] -> (EventMap event, EventVersion) + EventMap event -> UUID -> [TaggedEvent event] -> (EventMap event, EventWriteResult) storeEventMapTagged store@(EventMap uuidMap globalEvents) uuid taggedEvents = let versStart = latestEventVersion store uuid + vers = take (length taggedEvents) [versStart + 1 ..] streamEvents = zipWith (\v (TaggedEvent meta e) -> StreamEvent uuid v meta e) - [versStart + 1 ..] + vers taggedEvents newMap = Map.insertWith (flip (><)) uuid (Seq.fromList streamEvents) uuidMap globalEvents' = globalEvents >< Seq.fromList streamEvents - in (EventMap newMap globalEvents', versStart + EventVersion (length taggedEvents)) + globals = take (length taggedEvents) [SequenceNumber (Seq.length globalEvents + 1) ..] + in (EventMap newMap globalEvents', zip vers globals) diff --git a/tests/Eventium/ReadModelSpec.hs b/tests/Eventium/ReadModelSpec.hs --- a/tests/Eventium/ReadModelSpec.hs +++ b/tests/Eventium/ReadModelSpec.hs @@ -180,3 +180,107 @@ rB <- readTVarIO resetB rA `shouldBe` True rB `shouldBe` True + + describe "readModelPublisher (synchronous driver)" $ do + it "applies the handler and advances the checkpoint during the write" $ do + eventTVar <- eventMapTVar + sumRef <- newTVarIO (0 :: Int) + checkpointRef <- newTVarIO (0 :: SequenceNumber) + let baseWriter = runEventStoreWriterUsing atomically (tvarEventStoreWriter eventTVar) + globalReader = runEventStoreReaderUsing atomically (tvarGlobalEventStoreReader eventTVar) + rm = + ReadModel + { initialize = pure (), + eventHandler = EventHandler $ \globalEvent -> + case globalEvent.payload.payload of + Added n -> atomically $ modifyTVar' sumRef (+ n) + _ -> return (), + checkpointStore = + CheckpointStore + { getCheckpoint = readTVarIO checkpointRef, + saveCheckpoint = atomically . writeTVar checkpointRef + }, + reset = pure () + } + -- Same ReadModel, driven synchronously in the write path. + writer = publishingGlobalEventStoreWriter baseWriter (readModelPublisher rm) + + _ <- writer.storeEvents (uuidFromInteger 1) NoStream [Added 1, Added 2] + _ <- writer.storeEvents (uuidFromInteger 2) NoStream [Added 10] + + -- Handler ran synchronously during the writes. + readTVarIO sumRef `shouldReturn` 13 + -- Checkpoint advanced in-line to the last assigned global position. + readTVarIO checkpointRef `shouldReturn` 3 + -- Nothing is left for an async catch-up: the checkpoint is already current. + newEvents <- globalReader.getEvents (eventsStartingAt () 4) + length newEvents `shouldBe` 0 + + describe "catchUpReadModel" $ do + it "brings a model current without resetting, and is a no-op when current" $ do + eventTVar <- eventMapTVar + let writer = runEventStoreWriterUsing atomically (tvarEventStoreWriter eventTVar) + globalReader = runEventStoreReaderUsing atomically (tvarGlobalEventStoreReader eventTVar) + _ <- writer.storeEvents (uuidFromInteger 1) NoStream [Added 1, Added 2] + _ <- writer.storeEvents (uuidFromInteger 2) NoStream [Added 10] + + sumRef <- newTVarIO (0 :: Int) + checkpointRef <- newTVarIO (0 :: SequenceNumber) + resetCount <- newTVarIO (0 :: Int) + let rm = + ReadModel + { initialize = pure (), + eventHandler = EventHandler $ \globalEvent -> + case globalEvent.payload.payload of + Added n -> atomically $ modifyTVar' sumRef (+ n) + _ -> return (), + checkpointStore = + CheckpointStore + { getCheckpoint = readTVarIO checkpointRef, + saveCheckpoint = atomically . writeTVar checkpointRef + }, + reset = atomically $ modifyTVar' resetCount (+ 1) + } + + catchUpReadModel globalReader rm + readTVarIO sumRef `shouldReturn` 13 + readTVarIO checkpointRef `shouldReturn` 3 + -- Did not reset, and a second catch-up applies nothing new. + catchUpReadModel globalReader rm + readTVarIO sumRef `shouldReturn` 13 + readTVarIO resetCount `shouldReturn` 0 + + describe "rebuildReadModel resets the checkpoint itself" $ do + it "replays from scratch even when reset clears only view data" $ do + eventTVar <- eventMapTVar + let writer = runEventStoreWriterUsing atomically (tvarEventStoreWriter eventTVar) + globalReader = runEventStoreReaderUsing atomically (tvarGlobalEventStoreReader eventTVar) + _ <- writer.storeEvents (uuidFromInteger 1) NoStream [Added 1, Added 2] + + sumRef <- newTVarIO (0 :: Int) + checkpointRef <- newTVarIO (0 :: SequenceNumber) + let rm = + ReadModel + { initialize = pure (), + eventHandler = EventHandler $ \globalEvent -> + case globalEvent.payload.payload of + Added n -> atomically $ modifyTVar' sumRef (+ n) + _ -> return (), + checkpointStore = + CheckpointStore + { getCheckpoint = readTVarIO checkpointRef, + saveCheckpoint = atomically . writeTVar checkpointRef + }, + -- Clears view DATA only — deliberately does NOT touch the checkpoint. + reset = atomically $ writeTVar sumRef 0 + } + + -- Advance the checkpoint as if previously caught up. + catchUpReadModel globalReader rm + readTVarIO sumRef `shouldReturn` 3 + readTVarIO checkpointRef `shouldReturn` 2 + + -- Rebuild must still replay from the start: eventium resets the checkpoint + -- even though reset left it at 2. + rebuildReadModel globalReader rm + readTVarIO sumRef `shouldReturn` 3