diff --git a/eventium-core.cabal b/eventium-core.cabal --- a/eventium-core.cabal +++ b/eventium-core.cabal @@ -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, diff --git a/src/Eventium/CommandHandler.hs b/src/Eventium/CommandHandler.hs --- a/src/Eventium/CommandHandler.hs +++ b/src/Eventium/CommandHandler.hs @@ -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 diff --git a/src/Eventium/EventPublisher.hs b/src/Eventium/EventPublisher.hs --- a/src/Eventium/EventPublisher.hs +++ b/src/Eventium/EventPublisher.hs @@ -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 diff --git a/src/Eventium/ReadModel.hs b/src/Eventium/ReadModel.hs --- a/src/Eventium/ReadModel.hs +++ b/src/Eventium/ReadModel.hs @@ -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) => diff --git a/src/Eventium/Store/Class.hs b/src/Eventium/Store/Class.hs --- a/src/Eventium/Store/Class.hs +++ b/src/Eventium/Store/Class.hs @@ -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 diff --git a/src/Eventium/Store/Types.hs b/src/Eventium/Store/Types.hs --- a/src/Eventium/Store/Types.hs +++ b/src/Eventium/Store/Types.hs @@ -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 diff --git a/tests/Eventium/CommandDispatcherSpec.hs b/tests/Eventium/CommandDispatcherSpec.hs --- a/tests/Eventium/CommandDispatcherSpec.hs +++ b/tests/Eventium/CommandDispatcherSpec.hs @@ -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