eventful-core 0.1.3 → 0.2.0

raw patch · 12 files changed

+273/−397 lines, 12 filesdep −hlintdep ~basePVP ok

version bump matches the API change (PVP)

Dependencies removed: hlint

Dependency ranges changed: base

API changes (from Hackage documentation)

Files

• eventful-core.cabal +1/−26
• src/Eventful/Aggregate.hs +7/−6
• src/Eventful/EventBus.hs +9/−15
• src/Eventful/ProcessManager.hs +8/−7
• src/Eventful/Projection.hs +57/−82
• src/Eventful/ProjectionCache/Types.hs +57/−60
• src/Eventful/ReadModel/Class.hs +4/−4
• src/Eventful/Serializer.hs +14/−0
• src/Eventful/Store/Class.hs +89/−152
• src/Eventful/Store/Queries.hs +26/−24
• tests/Eventful/SerializerSpec.hs +1/−0
• tests/HLint.hs +0/−21

@@ -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

@@ -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

@@ -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

@@ -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]

@@ -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.

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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)

@@ -4,6 +4,7 @@ module Eventful.SerializerSpec (spec) where  import Data.Dynamic+import Data.Typeable (typeOf) import GHC.Generics import Test.Hspec 

@@ -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