diff --git a/eventful-core.cabal b/eventful-core.cabal --- a/eventful-core.cabal +++ b/eventful-core.cabal @@ -3,7 +3,7 @@ -- see: https://github.com/sol/hpack name: eventful-core -version: 0.1.3 +version: 0.2.0 synopsis: Core module for eventful description: Core module for eventful category: Database,Eventsourcing @@ -80,7 +80,6 @@ , HUnit other-modules: Eventful.SerializerSpec - HLint Eventful Eventful.Aggregate Eventful.EventBus @@ -95,28 +94,4 @@ Eventful.TH.Projection Eventful.TH.SumTypeSerializer Eventful.UUID - default-language: Haskell2010 - -test-suite style - type: exitcode-stdio-1.0 - main-is: HLint.hs - hs-source-dirs: - tests - ghc-options: -Wall - build-depends: - base >= 4.9 && < 5 - , aeson - , containers - , contravariant - , http-api-data - , path-pieces - , sum-type-boilerplate - , template-haskell - , text - , transformers - , uuid - , hlint - other-modules: - Eventful.SerializerSpec - Spec default-language: Haskell2010 diff --git a/src/Eventful/Aggregate.hs b/src/Eventful/Aggregate.hs --- a/src/Eventful/Aggregate.hs +++ b/src/Eventful/Aggregate.hs @@ -44,15 +44,16 @@ -- saves the events back to the store as well. commandStoredAggregate :: (Monad m) - => EventStore serialized m - -> Aggregate state serialized command + => EventStoreWriter m event + -> VersionedEventStoreReader m event + -> Aggregate state event command -> UUID -> command - -> m [serialized] -commandStoredAggregate store (Aggregate handler proj) uuid command = do - StreamProjection{..} <- getLatestProjection store (streamProjection proj uuid) + -> m [event] +commandStoredAggregate writer reader (Aggregate handler proj) uuid command = do + StreamProjection{..} <- getLatestStreamProjection reader (versionedStreamProjection uuid proj) let events = handler streamProjectionState command - mError <- storeEvents store (ExactVersion streamProjectionVersion) uuid events + mError <- storeEvents writer (ExactVersion streamProjectionPosition) uuid events case mError of (Just err) -> error $ "TODO: Create aggregate restart logic. " ++ show err Nothing -> return events diff --git a/src/Eventful/EventBus.hs b/src/Eventful/EventBus.hs --- a/src/Eventful/EventBus.hs +++ b/src/Eventful/EventBus.hs @@ -12,31 +12,25 @@ -- is also useful for integration testing along with an in-memory event store. synchronousEventBusWrapper :: (Monad m) - => EventStore serialized m - -> [EventStore serialized m -> UUID -> serialized -> m ()] - -> EventStore serialized m -synchronousEventBusWrapper store handlers = - let + => EventStoreWriter m event + -> [EventStoreWriter m event -> UUID -> event -> m ()] + -> EventStoreWriter m event +synchronousEventBusWrapper writer handlers = wrappedStore + where -- NB: We need to use recursive let bindings so we can put wrappedStore -- inside the event handlers handlers' = map ($ wrappedStore) handlers - wrappedStore = - EventStore - { getLatestVersion = getLatestVersion store - , getEvents = getEvents store - , storeEvents = storeAndPublishEvents store handlers' - } - in wrappedStore + wrappedStore = EventStoreWriter $ storeAndPublishEvents writer handlers' -- | Stores events in the store and them publishes them to the event handlers. -- This is used in the 'synchronousEventBusWrapper'. storeAndPublishEvents :: (Monad m) - => EventStore serialized m - -> [UUID -> serialized -> m ()] + => EventStoreWriter m event + -> [UUID -> event -> m ()] -> ExpectedVersion -> UUID - -> [serialized] + -> [event] -> m (Maybe EventWriteError) storeAndPublishEvents store handlers expectedVersion uuid events = do result <- storeEvents store expectedVersion uuid events diff --git a/src/Eventful/ProcessManager.hs b/src/Eventful/ProcessManager.hs --- a/src/Eventful/ProcessManager.hs +++ b/src/Eventful/ProcessManager.hs @@ -21,9 +21,9 @@ -- appropriate Aggregates or Projections in other streams. data ProcessManager state event command = ProcessManager - { processManagerProjection :: Projection state (ProjectionEvent event) + { processManagerProjection :: Projection state (VersionedStreamEvent event) , processManagerPendingCommands :: state -> [ProcessManagerCommand event command] - , processManagerPendingEvents :: state -> [ProjectionEvent event] + , processManagerPendingEvents :: state -> [StreamEvent UUID () event] } -- | This is a @command@ along with the UUID of the target 'Aggregate', and @@ -46,11 +46,12 @@ applyProcessManagerCommandsAndEvents :: (Monad m) => ProcessManager state event command - -> EventStore event m + -> EventStoreWriter m event + -> VersionedEventStoreReader m event -> state -> m () -applyProcessManagerCommandsAndEvents ProcessManager{..} store state = do +applyProcessManagerCommandsAndEvents ProcessManager{..} writer reader state = do forM_ (processManagerPendingCommands state) $ \(ProcessManagerCommand aggregateId aggregate command) -> - void $ commandStoredAggregate store aggregate aggregateId command - forM_ (processManagerPendingEvents state) $ \(ProjectionEvent projectionId event) -> - storeEvents store AnyVersion projectionId [event] + void $ commandStoredAggregate writer reader aggregate aggregateId command + forM_ (processManagerPendingEvents state) $ \(StreamEvent projectionId () event) -> + storeEvents writer AnyVersion projectionId [event] diff --git a/src/Eventful/Projection.hs b/src/Eventful/Projection.hs --- a/src/Eventful/Projection.hs +++ b/src/Eventful/Projection.hs @@ -5,16 +5,15 @@ , latestProjection , allProjections , StreamProjection (..) + , VersionedStreamProjection + , GlobalStreamProjection , streamProjection - , getLatestProjection - , GloballyOrderedProjection (..) - , globallyOrderedProjection - , globallyOrderedProjectionEventHandler - , getLatestGlobalProjection + , versionedStreamProjection + , globalStreamProjection + , getLatestStreamProjection , serializedProjection , projectionMapMaybe - ) - where + ) where import Data.Foldable (foldl') import Data.Functor.Contravariant @@ -56,91 +55,67 @@ allProjections (Projection seed handler) = scanl' handler seed -- | A 'StreamProjection' is a 'Projection' that has been constructed from --- events from a particular event stream. This is mostly useful so we can --- associate an 'EventVersion' with some state. -data StreamProjection state event +-- events from a particular event stream. This is useful when we want to cache +-- the resulting state and also keep track of what part of the stream the state +-- is caught up to. +data StreamProjection key position state event = StreamProjection - { streamProjectionProjection :: Projection state event - , streamProjectionUuid :: !UUID - , streamProjectionVersion :: EventVersion + { streamProjectionKey :: !key + , streamProjectionPosition :: !position + , streamProjectionProjection :: !(Projection state event) , streamProjectionState :: !state } --- | Initialize a 'StreamProjection' with a 'Projection'. -streamProjection - :: Projection state event - -> UUID - -> StreamProjection state event -streamProjection projection@Projection{..} uuid = - StreamProjection projection uuid (-1) projectionSeed +type VersionedStreamProjection = StreamProjection UUID EventVersion +type GlobalStreamProjection state event = StreamProjection () SequenceNumber state (VersionedStreamEvent event) --- | Gets the latest projection from a store by using 'getEvents' and then --- applying the events using the Projection's event handler. -getLatestProjection - :: (Monad m) - => EventStore event m - -> StreamProjection state event - -> m (StreamProjection state event) -getLatestProjection store projection@StreamProjection{..} = do - events <- getEvents store streamProjectionUuid (eventsStartingAt $ streamProjectionVersion + 1) - let - latestVersion = newEventVersion events - latestState = foldl' (projectionEventHandler streamProjectionProjection) streamProjectionState $ storedEventEvent <$> events - return $ - projection - { streamProjectionVersion = latestVersion - , streamProjectionState = latestState - } - where - newEventVersion [] = streamProjectionVersion - newEventVersion es = maximum $ storedEventVersion <$> es +-- | Initialize a 'StreamProjection' with a 'Projection', key, and order key. +streamProjection + :: key + -> position + -> Projection state event + -> StreamProjection key position state event +streamProjection key position projection@Projection{..} = + StreamProjection key position projection projectionSeed --- | This is a combination of a 'Projection' and the latest projection state --- with respect to some 'SequenceNumber'. This is useful for in-memory read --- models, and for querying the latest state starting from some previous state --- at a lower 'SequenceNumber'. -data GloballyOrderedProjection state serialized - = GloballyOrderedProjection - { globallyOrderedProjectionProjection :: !(Projection state (GloballyOrderedEvent serialized)) - , globallyOrderedProjectionSequenceNumber :: !SequenceNumber - , globallyOrderedProjectionState :: !state - } +-- | Initialize a 'VersionedStreamProjection'. +versionedStreamProjection + :: UUID + -> Projection state event + -> VersionedStreamProjection state event +versionedStreamProjection uuid = streamProjection uuid (-1) --- | Initialize a 'GloballyOrderedProjection' at 'SequenceNumber' 0 and with --- the projection's seed value. -globallyOrderedProjection - :: Projection state (GloballyOrderedEvent serialized) - -> GloballyOrderedProjection state serialized -globallyOrderedProjection projection@Projection{..} = - GloballyOrderedProjection projection 0 projectionSeed +-- | Initialize a 'GlobalStreamProjection'. +globalStreamProjection + :: Projection state (VersionedStreamEvent event) + -> GlobalStreamProjection state event +globalStreamProjection = streamProjection () 0 --- | This applies an event to a 'GloballyOrderedProjection'. NOTE: There is no --- guarantee that the 'SequenceNumber' for the event is the previous --- 'SequenceNumber' plus one (in fact, that isn't even a guarantee that some --- stores can provide). This function will update the --- 'GloballyOrderedProjetion' to use the sequence number of the event. -globallyOrderedProjectionEventHandler - :: GloballyOrderedProjection state serialized - -> GloballyOrderedEvent serialized - -> GloballyOrderedProjection state serialized -globallyOrderedProjectionEventHandler GloballyOrderedProjection{..} event@GloballyOrderedEvent{..} = +-- | Apply an event to the 'StreamProjection'. NOTE: There is no guarantee that +-- the order key for the event is greater than the current order key in the +-- 'StreamProjection'. This function simple will update the 'StreamProjection' +-- to use the order key of the event. +streamProjectionEventHandler + :: StreamProjection key position state event + -> StreamEvent eventKey position event + -> StreamProjection key position state event +streamProjectionEventHandler StreamProjection{..} event = let - Projection{..} = globallyOrderedProjectionProjection - seqNum = globallyOrderedEventSequenceNumber - state' = projectionEventHandler globallyOrderedProjectionState event - in GloballyOrderedProjection globallyOrderedProjectionProjection seqNum state' + Projection{..} = streamProjectionProjection + position' = streamEventPosition event + state' = projectionEventHandler streamProjectionState (streamEventEvent event) + in StreamProjection streamProjectionKey position' streamProjectionProjection state' --- | Gets globally ordered events from the event store and builds a --- 'Projection' based on 'ProjectionEvent'. Optionally accepts the current --- projection state as an argument. -getLatestGlobalProjection - :: (Monad m) - => GloballyOrderedEventStore serialized m - -> GloballyOrderedProjection state serialized - -> m (GloballyOrderedProjection state serialized) -getLatestGlobalProjection store globalProjection@GloballyOrderedProjection{..} = do - events <- getSequencedEvents store (eventsStartingAt $ globallyOrderedProjectionSequenceNumber + 1) - return $ foldl' globallyOrderedProjectionEventHandler globalProjection events +-- | Gets the latest projection from a store by querying events from the latest +-- order key and then applying the events using the Projection's event handler. +getLatestStreamProjection + :: (Monad m, Num position) + => EventStoreReader key position m (StreamEvent key position event) + -> StreamProjection key position state event + -> m (StreamProjection key position state event) +getLatestStreamProjection (EventStoreReader getEvents') projection@StreamProjection{..} = do + events <- getEvents' (eventsStartingAt streamProjectionKey $ streamProjectionPosition + 1) + return $ foldl' streamProjectionEventHandler projection events -- | Use a 'Serializer' to wrap a 'Projection' with event type @event@ so it -- uses the @serialized@ type. diff --git a/src/Eventful/ProjectionCache/Types.hs b/src/Eventful/ProjectionCache/Types.hs --- a/src/Eventful/ProjectionCache/Types.hs +++ b/src/Eventful/ProjectionCache/Types.hs @@ -3,11 +3,11 @@ module Eventful.ProjectionCache.Types ( ProjectionCache (..) - , StreamProjectionCache - , GloballyOrderedProjectionCache + , VersionedProjectionCache + , GlobalStreamProjectionCache , runProjectionCacheUsing , serializedProjectionCache - , getLatestProjectionWithCache + , getLatestVersionedProjectionWithCache , getLatestGlobalProjectionWithCache , updateProjectionCache , updateGlobalProjectionCache @@ -27,34 +27,34 @@ -- helper functions in this module to interpret the stored values using a -- 'Projection'. -- --- The @key@ and @orderKey@ type parameters are polymorphic so we can abstract +-- 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 orderKey serialized m +data ProjectionCache key position serialized m = ProjectionCache - { storeProjectionSnapshot :: key -> orderKey -> serialized -> m () - -- ^ Stores the state for a projection at a given @key@ and @orderKey@. + { storeProjectionSnapshot :: key -> position -> serialized -> m () + -- ^ 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 -- actually derived from the events in the stream. Consider using -- 'updateProjectionCache'. - , loadProjectionSnapshot :: key -> m (Maybe (orderKey, serialized)) + , loadProjectionSnapshot :: key -> m (Maybe (position, serialized)) -- ^ Loads the latest projection state from the cache. } -- | Type synonym for a 'ProjectionCache' used on individual event streams. -type StreamProjectionCache serialized m = ProjectionCache UUID EventVersion serialized m +type VersionedProjectionCache serialized m = ProjectionCache UUID EventVersion serialized m -- | Type synonym for a 'ProjectionCache' that is used in conjunction with a --- 'GloballyOrderedEventStore'. -type GloballyOrderedProjectionCache key serialized m = ProjectionCache key SequenceNumber serialized m +-- 'GlobalStreamEventStore'. +type GlobalStreamProjectionCache key serialized m = ProjectionCache key SequenceNumber serialized m -- | Changes the monad a 'ProjectionCache' runs in. This is useful to run the -- cache in another 'Monad' while forgetting the original 'Monad'. runProjectionCacheUsing :: (Monad m, Monad mstore) => (forall a. mstore a -> m a) - -> ProjectionCache key orderKey serialized mstore - -> ProjectionCache key orderKey serialized m + -> ProjectionCache key position serialized mstore + -> ProjectionCache key position serialized m runProjectionCacheUsing runCache ProjectionCache{..} = ProjectionCache { storeProjectionSnapshot = \uuid version state -> runCache $ storeProjectionSnapshot uuid version state @@ -67,8 +67,8 @@ serializedProjectionCache :: (Monad m) => Serializer state serialized - -> ProjectionCache key orderKey serialized m - -> ProjectionCache key orderKey state m + -> ProjectionCache key position serialized m + -> ProjectionCache key position state m serializedProjectionCache Serializer{..} ProjectionCache{..} = ProjectionCache storeProjectionSnapshot' loadProjectionSnapshot' where @@ -77,71 +77,68 @@ mState <- loadProjectionSnapshot uuid return $ mState >>= traverse deserialize --- | Like 'getLatestProjection', but uses a 'ProjectionCache' if it contains +-- | Like 'getLatestVersionedProjection', but uses a 'ProjectionCache' if it contains -- more recent state. -getLatestProjectionWithCache +getLatestVersionedProjectionWithCache :: (Monad m) - => EventStore event m - -> StreamProjectionCache state m - -> StreamProjection state event - -> m (StreamProjection state event) -getLatestProjectionWithCache store cache originalProj = do - mLatestState <- loadProjectionSnapshot cache (streamProjectionUuid originalProj) - let - mkProjection' (version, state) = - if version > streamProjectionVersion originalProj - then - originalProj - { streamProjectionVersion = version - , streamProjectionState = state - } - else originalProj - projection' = maybe originalProj mkProjection' mLatestState - getLatestProjection store projection' + => VersionedEventStoreReader m event + -> VersionedProjectionCache state m + -> VersionedStreamProjection state event + -> m (VersionedStreamProjection state event) +getLatestVersionedProjectionWithCache store cache projection = + getLatestProjectionWithCache' cache projection (streamProjectionKey projection) >>= getLatestStreamProjection store -- | Like 'getLatestGlobalProjection', but uses a 'ProjectionCache' if it -- contains more recent state. getLatestGlobalProjectionWithCache :: (Monad m) - => GloballyOrderedEventStore event m - -> GloballyOrderedProjectionCache key state m - -> GloballyOrderedProjection state event + => GlobalEventStoreReader m event + -> GlobalStreamProjectionCache key state m + -> GlobalStreamProjection state event -> key - -> m (GloballyOrderedProjection state event) -getLatestGlobalProjectionWithCache store cache originalProj key = do + -> m (GlobalStreamProjection state event) +getLatestGlobalProjectionWithCache store cache projection key = + getLatestProjectionWithCache' cache projection key >>= getLatestStreamProjection store + +getLatestProjectionWithCache' + :: (Monad m, Ord position) + => ProjectionCache key position state m + -> StreamProjection projKey position state event + -> key + -> m (StreamProjection projKey position state event) +getLatestProjectionWithCache' cache projection key = do mLatestState <- loadProjectionSnapshot cache key let - mkProjection' (seqNum, state) = - if seqNum > globallyOrderedProjectionSequenceNumber originalProj + mkProjection' (position, state) = + if position > streamProjectionPosition projection then - originalProj - { globallyOrderedProjectionSequenceNumber = seqNum - , globallyOrderedProjectionState = state + projection + { streamProjectionPosition = position + , streamProjectionState = state } - else originalProj - projection' = maybe originalProj mkProjection' mLatestState - getLatestGlobalProjection store projection' + else projection + return $ maybe projection mkProjection' mLatestState -- | Loads the latest projection state from the cache/store and stores this -- value back into the projection cache. updateProjectionCache :: (Monad m) - => EventStore event m - -> StreamProjectionCache state m - -> StreamProjection state event + => VersionedEventStoreReader m event + -> VersionedProjectionCache state m + -> VersionedStreamProjection state event -> m () -updateProjectionCache store cache projection = do - StreamProjection{..} <- getLatestProjectionWithCache store cache projection - storeProjectionSnapshot cache streamProjectionUuid streamProjectionVersion streamProjectionState +updateProjectionCache reader cache projection = do + StreamProjection{..} <- getLatestVersionedProjectionWithCache reader cache projection + storeProjectionSnapshot cache streamProjectionKey streamProjectionPosition streamProjectionState --- | Analog of 'updateProjectionCache' for a 'GloballyOrderedProjectionCache'. +-- | Analog of 'updateProjectionCache' for a 'GlobalStreamProjectionCache'. updateGlobalProjectionCache :: (Monad m) - => GloballyOrderedEventStore event m - -> GloballyOrderedProjectionCache key state m - -> GloballyOrderedProjection state event + => GlobalEventStoreReader m event + -> GlobalStreamProjectionCache key state m + -> GlobalStreamProjection state event -> key -> m () -updateGlobalProjectionCache store cache projection key = do - GloballyOrderedProjection{..} <- getLatestGlobalProjectionWithCache store cache projection key - storeProjectionSnapshot cache key globallyOrderedProjectionSequenceNumber globallyOrderedProjectionState +updateGlobalProjectionCache reader cache projection key = do + StreamProjection{..} <- getLatestGlobalProjectionWithCache reader cache projection key + storeProjectionSnapshot cache key streamProjectionPosition streamProjectionState diff --git a/src/Eventful/ReadModel/Class.hs b/src/Eventful/ReadModel/Class.hs --- a/src/Eventful/ReadModel/Class.hs +++ b/src/Eventful/ReadModel/Class.hs @@ -16,7 +16,7 @@ ReadModel { readModelModel :: model , readModelLatestAppliedSequence :: model -> m SequenceNumber - , readModelHandleEvents :: model -> [GloballyOrderedEvent serialized] -> m () + , readModelHandleEvents :: model -> [GlobalStreamEvent serialized] -> m () } type PollingPeriodSeconds = Double @@ -24,14 +24,14 @@ runPollingReadModel :: (MonadIO m, Monad mstore) => ReadModel model serialized m - -> GloballyOrderedEventStore serialized mstore + -> GlobalEventStoreReader mstore serialized -> (forall a. mstore a -> m a) -> PollingPeriodSeconds -> m () -runPollingReadModel ReadModel{..} getGloballyOrderedEvents runStore waitSeconds = forever $ do +runPollingReadModel ReadModel{..} globalReader runStore waitSeconds = forever $ do -- Get new events starting from latest applied sequence number latestSeq <- readModelLatestAppliedSequence readModelModel - newEvents <- runStore $ getSequencedEvents getGloballyOrderedEvents (eventsStartingAt $ latestSeq + 1) + newEvents <- runStore $ getEvents globalReader (eventsStartingAt () $ latestSeq + 1) -- Handle the new events readModelHandleEvents readModelModel newEvents diff --git a/src/Eventful/Serializer.hs b/src/Eventful/Serializer.hs --- a/src/Eventful/Serializer.hs +++ b/src/Eventful/Serializer.hs @@ -11,6 +11,7 @@ , composeSerializers -- * Common serializers , idSerializer + , traverseSerializer , jsonSerializer , jsonTextSerializer , dynamicSerializer @@ -25,6 +26,7 @@ import Data.Maybe (fromMaybe) import qualified Data.Text.Lazy as TL import qualified Data.Text.Lazy.Encoding as TLE +import Data.Typeable (typeOf) import GHC.Generics -- | A 'Serializer' describes the injective conversion between types @a@ and @@ -66,6 +68,18 @@ -- serializer but you don't need to actually change types. idSerializer :: Serializer a a idSerializer = simpleSerializer id Just + +-- | Uses 'Traversable' to wrap a 'Serializer'. +traverseSerializer + :: (Traversable t) + => Serializer a b + -> Serializer (t a) (t b) +traverseSerializer Serializer{..} = + Serializer serialize' deserialize' deserializeEither' + where + serialize' = fmap serialize + deserialize' = traverse deserialize + deserializeEither' = traverse deserializeEither -- | A 'Serializer' for aeson 'Value's. jsonSerializer :: (ToJSON a, FromJSON a) => Serializer a Value diff --git a/src/Eventful/Store/Class.hs b/src/Eventful/Store/Class.hs --- a/src/Eventful/Store/Class.hs +++ b/src/Eventful/Store/Class.hs @@ -7,24 +7,24 @@ module Eventful.Store.Class ( -- * EventStore - EventStore (..) - , GloballyOrderedEventStore (..) + EventStoreReader (..) + , EventStoreWriter (..) + , VersionedEventStoreReader + , GlobalEventStoreReader + , StreamEvent (..) + , VersionedStreamEvent + , GlobalStreamEvent , ExpectedVersion (..) , EventWriteError (..) - , runEventStoreUsing - , runGloballyOrderedEventStoreUsing + , runEventStoreReaderUsing + , runEventStoreWriterUsing , module Eventful.Store.Queries -- * Serialization - , serializedEventStore - , serializedGloballyOrderedEventStore + , serializedEventStoreReader + , serializedVersionedEventStoreReader + , serializedGlobalEventStoreReader + , serializedEventStoreWriter -- * Utility types - , ProjectionEvent (..) - , StoredEvent (..) - , storedEventToProjectionEvent - , GloballyOrderedEvent (..) - , globallyOrderedEventToStoredEvent - , globallyOrderedEventToProjectionEvent - , storedEventToGloballyOrderedEvent , EventVersion (..) , SequenceNumber (..) -- * Utility functions @@ -32,6 +32,7 @@ ) where import Data.Aeson +import Data.Functor.Contravariant import Data.Maybe (mapMaybe) import Web.HttpApiData import Web.PathPieces @@ -40,29 +41,36 @@ import Eventful.Store.Queries import Eventful.UUID --- | The 'EventStore' is the core type of eventful. A store operates in some --- monad @m@ and stores events by serializing them to the type @serialized@. -data EventStore serialized m - = EventStore - { getLatestVersion :: UUID -> m EventVersion - -- ^ Gets the latest 'EventVersion' for a given 'Projection'. - , getEvents :: UUID -> EventStoreQueryRange EventVersion -> m [StoredEvent serialized] - -- ^ Retrieves all the events for a given 'Projection' using that - -- projection's UUID. If an event version is provided then all events with - -- a version greater than or equal to that version are returned. - , storeEvents :: ExpectedVersion -> UUID -> [serialized] -> m (Maybe EventWriteError) - -- ^ Stores the events for a given 'Projection' using that projection's - -- UUID. - } +-- | An 'EventStoreReader' is a function to query a stream from an event store. +-- It operates in some monad @m@ and returns events of type @event@ from a +-- stream at @key@ ordered by @position@. +newtype EventStoreReader key position m event = EventStoreReader { getEvents :: QueryRange key position -> m [event] } --- | Gets all the events ordered starting with a given 'SequenceNumber', and --- ordered by 'SequenceNumber'. This is used when replaying all the events in a --- store. -newtype GloballyOrderedEventStore serialized m = - GloballyOrderedEventStore - { getSequencedEvents :: EventStoreQueryRange SequenceNumber -> m [GloballyOrderedEvent serialized] - } +instance (Functor m) => Functor (EventStoreReader key position m) where + fmap f (EventStoreReader reader) = EventStoreReader $ fmap (fmap f) <$> reader +type VersionedEventStoreReader m event = EventStoreReader UUID EventVersion m (VersionedStreamEvent event) +type GlobalEventStoreReader m event = EventStoreReader () SequenceNumber m (GlobalStreamEvent event) + +-- | An 'EventStoreWriter' is a function to write some events of type @event@ +-- to an event store in some monad @m@. +newtype EventStoreWriter m event = EventStoreWriter { storeEvents :: ExpectedVersion -> UUID -> [event] -> m (Maybe EventWriteError) } + +instance Contravariant (EventStoreWriter m) where + contramap f (EventStoreWriter writer) = EventStoreWriter $ \vers uuid -> writer vers uuid . fmap f + +-- | An event along with the @key@ for the event stream it is from and its +-- @position@ in that event stream. +data StreamEvent key position event + = StreamEvent + { streamEventKey :: !key + , streamEventPosition :: !position + , streamEventEvent :: !event + } deriving (Show, Eq, Functor, Foldable, Traversable) + +type VersionedStreamEvent event = StreamEvent UUID EventVersion event +type GlobalStreamEvent event = StreamEvent () SequenceNumber (VersionedStreamEvent event) + -- | ExpectedVersion is used to assert the event stream is at a certain version -- number. This is used when multiple writers are concurrently writing to the -- event store. If the expected version is incorrect, then storing fails. @@ -87,8 +95,8 @@ transactionalExpectedWriteHelper :: (Monad m) => (UUID -> m EventVersion) - -> (UUID -> [serialized] -> m ()) - -> ExpectedVersion -> UUID -> [serialized] -> m (Maybe EventWriteError) + -> (UUID -> [event] -> m ()) + -> ExpectedVersion -> UUID -> [event] -> m (Maybe EventWriteError) transactionalExpectedWriteHelper getLatestVersion' storeEvents' expected = go expected getLatestVersion' storeEvents' where @@ -101,8 +109,8 @@ :: (Monad m) => Maybe (EventVersion -> Bool) -> (UUID -> m EventVersion) - -> (UUID -> [serialized] -> m ()) - -> UUID -> [serialized] -> m (Maybe EventWriteError) + -> (UUID -> [event] -> m ()) + -> UUID -> [event] -> m (Maybe EventWriteError) transactionalExpectedWriteHelper' Nothing _ storeEvents' uuid events = storeEvents' uuid events >> return Nothing transactionalExpectedWriteHelper' (Just f) getLatestVersion' storeEvents' uuid events = do @@ -111,133 +119,62 @@ then storeEvents' uuid events >> return Nothing else return $ Just $ EventStreamNotAtExpectedVersion latestVersion --- | Changes the monad an 'EventStore' runs in. This is useful to run event --- stores in another 'Monad' while forgetting the original 'Monad'. -runEventStoreUsing +-- | Changes the monad an 'EventStoreReader' runs in. This is useful to run +-- event stores in another 'Monad' while forgetting the original 'Monad'. +runEventStoreReaderUsing :: (Monad m, Monad mstore) => (forall a. mstore a -> m a) - -> EventStore serialized mstore - -> EventStore serialized m -runEventStoreUsing runStore EventStore{..} = - EventStore - { getLatestVersion = runStore . getLatestVersion - , getEvents = \uuid range -> runStore $ getEvents uuid range - , storeEvents = \vers uuid events -> runStore $ storeEvents vers uuid events - } + -> EventStoreReader key position mstore event + -> EventStoreReader key position m event +runEventStoreReaderUsing runStore (EventStoreReader f) = EventStoreReader (runStore . f) --- | Analog of 'runEventStoreUsing' for a 'GloballyOrderedEventStore'. -runGloballyOrderedEventStoreUsing +-- | Analog of 'runEventStoreReaderUsing' for a 'EventStoreWriter'. +runEventStoreWriterUsing :: (Monad m, Monad mstore) => (forall a. mstore a -> m a) - -> GloballyOrderedEventStore serialized mstore - -> GloballyOrderedEventStore serialized m -runGloballyOrderedEventStoreUsing runStore GloballyOrderedEventStore{..} = - GloballyOrderedEventStore - { getSequencedEvents = runStore . getSequencedEvents - } + -> EventStoreWriter mstore event + -> EventStoreWriter m event +runEventStoreWriterUsing runStore (EventStoreWriter f) = + EventStoreWriter $ \vers uuid events -> runStore $ f vers uuid events --- | Wraps an 'EventStore' and transparently serializes/deserializes events for --- you. Note that in this implementation deserialization errors when using --- 'getEvents' are simply ignored (the event is not returned). -serializedEventStore +-- | Wraps an 'EventStoreReader' and transparently serializes/deserializes +-- events for you. Note that in this implementation deserialization errors are +-- simply ignored (the event is not returned). +serializedEventStoreReader :: (Monad m) => Serializer event serialized - -> EventStore serialized m - -> EventStore event m -serializedEventStore Serializer{..} store = - EventStore - (getLatestVersion store) - getEvents' - storeEvents' - where - getEvents' uuid mVersion = mapMaybe (traverse deserialize) <$> getEvents store uuid mVersion - storeEvents' expectedVersion uuid events = storeEvents store expectedVersion uuid (serialize <$> events) + -> EventStoreReader key position m serialized + -> EventStoreReader key position m event +serializedEventStoreReader Serializer{..} (EventStoreReader reader) = + EventStoreReader $ fmap (mapMaybe deserialize) . reader --- | Like 'serializedEventStore' except for 'GloballyOrderedEventStore'. -serializedGloballyOrderedEventStore +-- | Convenience wrapper around 'serializedEventStoreReader' for +-- 'VersionedEventStoreReader'. +serializedVersionedEventStoreReader :: (Monad m) => Serializer event serialized - -> GloballyOrderedEventStore serialized m - -> GloballyOrderedEventStore event m -serializedGloballyOrderedEventStore Serializer{..} store = - GloballyOrderedEventStore getSequencedEvents' - where - getSequencedEvents' sequenceNumber = - mapMaybe (traverse deserialize) <$> getSequencedEvents store sequenceNumber - --- | A 'ProjectionEvent' is an event that is associated with a 'Projection' via --- the projection's 'UUID'. -data ProjectionEvent event - = ProjectionEvent - { projectionEventProjectionId :: !UUID - -- ^ The UUID of the 'Projection' that the event belongs to. - , projectionEventEvent :: !event - -- ^ The actual event type. Note that this can be a serialized event or the - -- actual Haskell event type. - } deriving (Show, Eq, Functor, Foldable, Traversable) - --- | A 'StoredEvent' is an event with associated storage metadata. -data StoredEvent event - = StoredEvent - { storedEventProjectionId :: !UUID - -- ^ The UUID of the 'Projection' that the event belongs to. - , storedEventVersion :: !EventVersion - -- ^ The version of the Projection corresponding to this event. - , storedEventEvent :: !event - -- ^ The actual event type. Note that this can be a serialized event or the - -- actual Haskell event type. - } deriving (Show, Eq, Functor, Foldable, Traversable) - -storedEventToProjectionEvent :: StoredEvent event -> ProjectionEvent event -storedEventToProjectionEvent StoredEvent{..} = - ProjectionEvent - { projectionEventProjectionId = storedEventProjectionId - , projectionEventEvent = storedEventEvent - } - --- | A 'GloballyOrderedEvent' is like a 'StoredEvent' but has a global --- 'SequenceNumber'. -data GloballyOrderedEvent event - = GloballyOrderedEvent - { globallyOrderedEventProjectionId :: !UUID - -- ^ The UUID of the 'Projection' that the event belongs to. - , globallyOrderedEventVersion :: !EventVersion - -- ^ The version of the Projection corresponding to this event. - , globallyOrderedEventSequenceNumber :: !SequenceNumber - -- ^ The global sequence number of this event. - , globallyOrderedEventEvent :: !event - -- ^ The actual event type. Note that this can be a serialized event or the - -- actual Haskell event type. - } deriving (Show, Eq, Functor, Foldable, Traversable) - --- | Extract the 'StoredEvent' from a 'GloballyOrderedEvent' -globallyOrderedEventToStoredEvent :: GloballyOrderedEvent event -> StoredEvent event -globallyOrderedEventToStoredEvent GloballyOrderedEvent{..} = - StoredEvent - { storedEventProjectionId = globallyOrderedEventProjectionId - , storedEventVersion = globallyOrderedEventVersion - , storedEventEvent = globallyOrderedEventEvent - } + -> VersionedEventStoreReader m serialized + -> VersionedEventStoreReader m event +serializedVersionedEventStoreReader serializer = serializedEventStoreReader (traverseSerializer serializer) --- | Extract the 'ProjectionEvent' from a 'GloballyOrderedEvent' -globallyOrderedEventToProjectionEvent :: GloballyOrderedEvent event -> ProjectionEvent event -globallyOrderedEventToProjectionEvent GloballyOrderedEvent{..} = - ProjectionEvent - { projectionEventProjectionId = globallyOrderedEventProjectionId - , projectionEventEvent = globallyOrderedEventEvent - } +-- | Convenience wrapper around 'serializedEventStoreReader' for +-- 'GlobalEventStoreReader'. +serializedGlobalEventStoreReader + :: (Monad m) + => Serializer event serialized + -> GlobalEventStoreReader m serialized + -> GlobalEventStoreReader m event +serializedGlobalEventStoreReader serializer = serializedEventStoreReader (traverseSerializer (traverseSerializer serializer)) --- | Convert a 'StoredEvent' to a 'GloballyOrderedEvent' by adding the --- 'SequenceNumber'. This is mainly used by event stores to create globally --- ordered events. -storedEventToGloballyOrderedEvent :: SequenceNumber -> StoredEvent event -> GloballyOrderedEvent event -storedEventToGloballyOrderedEvent sequenceNumber StoredEvent{..} = - GloballyOrderedEvent - { globallyOrderedEventProjectionId = storedEventProjectionId - , globallyOrderedEventVersion = storedEventVersion - , globallyOrderedEventSequenceNumber = sequenceNumber - , globallyOrderedEventEvent = storedEventEvent - } +-- | Like 'serializedEventStoreReader' but for an 'EventStoreWriter'. Note that +-- 'EventStoreWriter' is an instance of 'Contravariant', so you can just use +-- @contramap serialize@ instead of this function. +serializedEventStoreWriter + :: (Monad m) + => Serializer event serialized + -> EventStoreWriter m serialized + -> EventStoreWriter m event +serializedEventStoreWriter Serializer{..} = contramap serialize -- | 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/src/Eventful/Store/Queries.hs b/src/Eventful/Store/Queries.hs --- a/src/Eventful/Store/Queries.hs +++ b/src/Eventful/Store/Queries.hs @@ -1,9 +1,10 @@ {-# LANGUAGE DeriveFunctor #-} +{-# LANGUAGE TypeFamilies #-} module Eventful.Store.Queries - ( EventStoreQueryRange (..) - , EventStoreQueryStart (..) - , EventStoreQueryLimit (..) + ( QueryRange (..) + , QueryStart (..) + , QueryLimit (..) , allEvents , eventsUntil , eventsStartingAt @@ -11,38 +12,39 @@ , eventsStartingAtTakeLimit ) where --- | This type defines how to query an event stream. It defines both where to --- start and where to stop in the stream. -data EventStoreQueryRange a - = EventStoreQueryRange - { eventStoreQueryRangeStart :: EventStoreQueryStart a - , eventStoreQueryRangeLimit :: EventStoreQueryLimit a - } deriving (Functor) +-- | This type defines how to query an event stream. It defines the stream key +-- and the start/stop points for the query. +data QueryRange key position + = QueryRange + { queryRangeKey :: key + , queryRangeStart :: QueryStart position + , queryRangeLimit :: QueryLimit position + } deriving (Show, Eq) -- | This type defines where an event store query starts. -data EventStoreQueryStart a +data QueryStart position = StartFromBeginning - | StartQueryAt a + | StartQueryAt position deriving (Show, Eq, Functor) -- | This type is used to limit the results of a query from an event store. -data EventStoreQueryLimit a +data QueryLimit position = NoQueryLimit | MaxNumberOfEvents Int - | StopQueryAt a + | StopQueryAt position deriving (Show, Eq, Functor) -allEvents :: EventStoreQueryRange a -allEvents = EventStoreQueryRange StartFromBeginning NoQueryLimit +allEvents :: key -> QueryRange key position +allEvents key = QueryRange key StartFromBeginning NoQueryLimit -eventsUntil :: a -> EventStoreQueryRange a -eventsUntil end = EventStoreQueryRange StartFromBeginning (StopQueryAt end) +eventsUntil :: key -> position -> QueryRange key position +eventsUntil key end = QueryRange key StartFromBeginning (StopQueryAt end) -eventsStartingAt :: a -> EventStoreQueryRange a -eventsStartingAt start = EventStoreQueryRange (StartQueryAt start) NoQueryLimit +eventsStartingAt :: key -> position -> QueryRange key position +eventsStartingAt key start = QueryRange key (StartQueryAt start) NoQueryLimit -eventsStartingAtUntil :: a -> a -> EventStoreQueryRange a -eventsStartingAtUntil start end = EventStoreQueryRange (StartQueryAt start) (StopQueryAt end) +eventsStartingAtUntil :: key -> position -> position -> QueryRange key position +eventsStartingAtUntil key start end = QueryRange key (StartQueryAt start) (StopQueryAt end) -eventsStartingAtTakeLimit :: a -> Int -> EventStoreQueryRange a -eventsStartingAtTakeLimit start maxNum = EventStoreQueryRange (StartQueryAt start) (MaxNumberOfEvents maxNum) +eventsStartingAtTakeLimit :: key -> position -> Int -> QueryRange key position +eventsStartingAtTakeLimit key start maxNum = QueryRange key (StartQueryAt start) (MaxNumberOfEvents maxNum) diff --git a/tests/Eventful/SerializerSpec.hs b/tests/Eventful/SerializerSpec.hs --- a/tests/Eventful/SerializerSpec.hs +++ b/tests/Eventful/SerializerSpec.hs @@ -4,6 +4,7 @@ module Eventful.SerializerSpec (spec) where import Data.Dynamic +import Data.Typeable (typeOf) import GHC.Generics import Test.Hspec diff --git a/tests/HLint.hs b/tests/HLint.hs deleted file mode 100644 --- a/tests/HLint.hs +++ /dev/null @@ -1,21 +0,0 @@ -module Main (main) where - -import Language.Haskell.HLint (hlint) -import System.Exit (exitFailure, exitSuccess) - -import Prelude (String, IO, null) - -arguments :: [String] -arguments = - [ "src" - , "tests" - , "-i=Redundant do" - , "-i=Unused LANGUAGE pragma" -- This fails on DeriveGeneric - , "-i=Use newtype instead of data" - , "-i=Eta reduce" - ] - -main :: IO () -main = do - hints <- hlint arguments - if null hints then exitSuccess else exitFailure