packages feed

cqrs-core (empty) → 0.10.0

raw patch · 18 files changed

+688/−0 lines, 18 filesdep +basedep +bytestringdep +containerssetup-changed

Dependencies added: base, bytestring, containers, deepseq, io-streams, transformers, uuid-types

Files

+ LICENSE view
@@ -0,0 +1,19 @@+Copyright (c) 2011-2015 Bardur Arantsson++Permission is hereby granted, free of charge, to any person obtaining a copy+of this software and associated documentation files (the "Software"), to deal+in the Software without restriction, including without limitation the rights+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell+copies of the Software, and to permit persons to whom the Software is+furnished to do so, subject to the following conditions:++The above copyright notice and this permission notice shall be included in+all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN+THE SOFTWARE.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ cqrs-core.cabal view
@@ -0,0 +1,38 @@+Name:                cqrs-core+Version:             0.10.0+Synopsis:            Command-Query Responsibility Segregation+Description:         Haskell implementation of the CQRS architectural pattern.+License:             MIT+License-file:        LICENSE+Category:            Data+Cabal-version:       >=1.10+Build-type:          Simple+Author:              Bardur Arantsson+Maintainer:          Bardur Arantsson <bardur@scientician.net>++Library+  Build-Depends:       base >= 4.8 && < 5+                     , bytestring >= 0.9.0.1+                     , containers >= 0.5 && < 1+                     , deepseq >= 1.4 && < 2+                     , io-streams >= 1.2 && < 2+                     , transformers >= 0.4.1 && < 0.5+                     , uuid-types >= 1.0 && < 1.1+  Default-language:    Haskell2010+  ghc-options:         -Wall+  hs-source-dirs:      src+  Exposed-modules:     Data.CQRS.Command+                       Data.CQRS.Repository+                       Data.CQRS.Query+                       Data.CQRS.SnapshotStore+                       Data.CQRS.Internal.Aggregate+                       Data.CQRS.Internal.Repository+                       Data.CQRS.Types.AggregateAction+                       Data.CQRS.Types.ArchiveMetadata+                       Data.CQRS.Types.ArchiveRef+                       Data.CQRS.Types.ArchiveStore+                       Data.CQRS.Types.EventStore+                       Data.CQRS.Types.PersistedEvent+                       Data.CQRS.Types.Snapshot+                       Data.CQRS.Types.SnapshotStore+                       Data.CQRS.Types.StoreError
+ src/Data/CQRS/Command.hs view
@@ -0,0 +1,194 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-| Module to import for the "Command" side of the application. -}+module Data.CQRS.Command+    ( -- Re-exports for user convenience:+      AggregateAction+    , Repository+      -- Exports:+    , CommandT+    , createAggregate+    , execCommandT+    , freshUUID+    , publishEvent+    , readAggregate+    , runCommandT+    , UnitOfWorkT+    , updateAggregate+    ) where++import           Control.DeepSeq (NFData)+import           Control.Monad (forM, join, liftM, void, when)+import           Control.Monad.IO.Class (MonadIO, liftIO)+import           Control.Monad.Trans.Class (MonadTrans(..), lift)+import           Control.Monad.Trans.State.Strict (StateT, runStateT, get, modify')+import           Control.Monad.Trans.Reader (ReaderT, runReaderT, ask)+import           Data.CQRS.Internal.Aggregate (Aggregate)+import qualified Data.CQRS.Internal.Aggregate as A+import           Data.CQRS.Internal.Repository+import           Data.CQRS.Types.AggregateAction (AggregateAction)+import           Data.CQRS.Types.EventStore (EventStore(..))+import           Data.CQRS.Types.PersistedEvent (PersistedEvent(..))+import           Data.CQRS.Types.Snapshot (Snapshot(..))+import           Data.CQRS.Types.SnapshotStore+import           Data.Foldable (forM_)+import           Data.Maybe (fromJust)+import           Data.UUID.Types (UUID)+import qualified System.IO.Streams.Combinators as SC++-- | Command monad transformer.+newtype CommandT a e m b = CommandT { unCommandT :: CommandM a e m b }+    deriving (Functor, Applicative, Monad)++instance MonadTrans (CommandT a e) where+  lift m = CommandT $ lift m++instance MonadIO m => MonadIO (CommandT a e m) where+    liftIO m = CommandT $ liftIO m++-- | Command monad. This is just a reader to provide ambient access to the Repository.+type CommandM a e = ReaderT (Command a e)++data Command a e =+    Command { commandRepository :: Repository a e+            }++-- | Unit of work monad transformer.+newtype UnitOfWorkT a e m b = UnitOfWorkT { unUnitOfWorkT :: UnitOfWorkM a e m b }+  deriving (Functor, Applicative, Monad)++instance MonadTrans (UnitOfWorkT a e) where+    lift m = UnitOfWorkT $ lift m++instance MonadIO m => MonadIO (UnitOfWorkT a e m) where+    liftIO m = UnitOfWorkT $ liftIO m++-- | Unit of work monad.+type UnitOfWorkM a e = StateT (UnitOfWork a e)++data UnitOfWork a e =+   UnitOfWork { uowAggregate :: Aggregate a e+              }++-- | Run a command against a repository.+runCommandT :: (MonadIO m) => Repository a e -> CommandT a e m b -> m b+runCommandT repository (CommandT command) = runReaderT command $ Command repository++-- | Run a command against a repository, ignoring the result.+execCommandT :: (MonadIO m) => Repository a e -> CommandT a e m b -> m ()+execCommandT repository = void . runCommandT repository++-- | Create a fresh UUID from the repository.+freshUUID :: (MonadIO m) => CommandT a e m UUID+freshUUID = CommandT $ do+  repository <- liftM commandRepository ask+  liftIO $ repositoryUUIDSupply repository++-- Write out all the changes for a given aggregate.+writeChanges :: (MonadIO m) => UUID -> Aggregate a e -> CommandT a e m ()+writeChanges aggregateId aggregate = CommandT $ do+  repository <- liftM commandRepository ask+  let snapshotStore = repositorySnapshotStore repository+  let eventStore = repositoryEventStore repository+  let publishEvents = repositoryPublishEvents repository+  -- Convert all the accumulated events to PersistedEvent+  versionedEvents <- forM (A.versionedEvents aggregate) $ \(v, e) -> do+    i <- liftIO $ repositoryUUIDSupply repository+    return $ PersistedEvent e v i+  -- We only care if new events were generated+  when (length versionedEvents > 0) $ do+    -- Commit events to event store.+    liftIO $ (esStoreEvents eventStore) aggregateId versionedEvents+    -- Publish events written so far.+    liftIO $ publishEvents (aggregateId, versionedEvents)+    -- Write out the snapshot (if applicable).+    forM_ (settingsSnapshotFrequency $ repositorySettings repository) $ \snapshotFrequency -> do+      forM_ (snapshotForAggregate snapshotFrequency aggregate) $ \snapshot -> do+        liftIO $ ssWriteSnapshot snapshotStore aggregateId snapshot+  where+    snapshotForAggregate maxDelta aggregate = join $ (flip fmap) (A.aggregateSnapshot aggregate) $ \(v, a) ->+      -- If we've advanced N events past the last snapshot, we create a+      -- new snapshot.+      let sv = A.aggregateSnapshotVersion aggregate in+        if (v - sv > maxDelta) then+          Just $ Snapshot v a+        else+          Nothing++-- Get the aggregateAction from the repository.+getAggregateAction :: (Monad m) => CommandT a e m (AggregateAction a e)+getAggregateAction = CommandT $ liftM (repositoryAggregateAction . commandRepository) ask++-- | Create a new aggregate using the supplied unit of work. Throws a+-- 'Data.CQRS.Types.VersionConflict' exception if there is already an+-- aggregate with the given aggregate ID. __NOTE__: The exception may+-- be thrown at the __END__ of the unit of work, and so any operations+-- in the unit of work that are lifted into the nested monad may be+-- performed regardless. (This is due to optimistic concurrency+-- control.)+createAggregate :: (MonadIO m, Monad m) => UUID -> (UnitOfWorkT a e (CommandT a e m) (Maybe a) -> UnitOfWorkT a e (CommandT a e m) b) -> CommandT a e m b+createAggregate aggregateId unitOfWork = do+  -- We use an "empty" aggregate state as the starting point+  -- here. We'll automatically conflict when trying to save if there+  -- is a conflict or if the aggregate already exists.  We cannot+  -- check this preemptively since aggregates may be created+  -- concurrently.+  (r, s) <- do+    aggregateAction <- getAggregateAction+    runStateT run $ UnitOfWork $ A.emptyAggregate aggregateAction+  -- Write out any changes.+  writeChanges aggregateId $ uowAggregate s+  -- Return the result of the computation+  CommandT $ return r+  where+    run = do+      let getter = liftM (A.aggregateValue . uowAggregate) get+      unUnitOfWorkT $ unitOfWork $ UnitOfWorkT getter++-- | Update aggregate with the supplied unit of work. The unit of work+-- is given a function to get the current value of the aggregate.+-- Returns the value returned by the unit of work, or 'Nothing' if+-- there was no aggregate with the given ID. Throws a+-- 'Data.CQRS.Types.VersionConflict' exception if a version conflict+-- occurs during commit. __NOTE__: The exception may be thrown at the+-- END of the unit of work, and so any operations in the unit of work+-- that are lifted into the nested monad may be performed+-- regardless. (This is due to optimistic concurrency control.)+updateAggregate :: (MonadIO m) => UUID -> (UnitOfWorkT a e (CommandT a e m) a -> UnitOfWorkT a e (CommandT a e m) b) -> CommandT a e m (Maybe b)+updateAggregate aggregateId unitOfWork = CommandT $ do+  aggregate <- getByIdFromEventStore aggregateId+  unCommandT $ case A.aggregateValue $ aggregate of+    Nothing -> do+      return Nothing+    Just _ -> do+      (r, s) <- runStateT run $ UnitOfWork aggregate+      writeChanges aggregateId $ uowAggregate s+      return $ Just r+  where+    run = do+      let getter = liftM (fromJust . A.aggregateValue . uowAggregate) get+      unUnitOfWorkT $ unitOfWork $ UnitOfWorkT getter++-- | Read value of an aggregate if it exists. If any update needs to+-- be performed on the aggregate, use of the 'getter' function (see+-- 'createAggregate' and 'updateAggregate') should be preferred.+readAggregate :: (MonadIO m) => UUID -> CommandT a e m (Maybe a)+readAggregate = (flip updateAggregate) id++-- Retrieve aggregate from event store.+getByIdFromEventStore :: (MonadIO m) => UUID -> CommandM a e m (Aggregate a e)+getByIdFromEventStore aggregateId = do+  r <- liftM commandRepository ask+  let es = repositoryEventStore r+  let ss = repositorySnapshotStore r+  let aa = repositoryAggregateAction r+  -- Start from a snapshot (if any).+  a' <- liftM (A.applySnapshot $ A.emptyAggregate aa) $ liftIO $ ssReadSnapshot ss $ aggregateId+  -- Apply any subsequent events.+  a'' <- liftIO $ esRetrieveEvents es aggregateId (A.aggregateVersion0 a') (SC.fold A.applyEvent a')+  -- Return the aggregate ref.+  return a''++-- | Publish event for the current aggregate.+publishEvent :: (MonadIO m, NFData a, NFData e) => e -> UnitOfWorkT a e (CommandT a e m) ()+publishEvent event = UnitOfWorkT $ do+  modify' (\s -> s { uowAggregate = A.publishEvent (uowAggregate s) event })
+ src/Data/CQRS/Internal/Aggregate.hs view
@@ -0,0 +1,75 @@+module Data.CQRS.Internal.Aggregate+       ( Aggregate+       , aggregateSnapshotVersion+       , aggregateSnapshot+       , aggregateValue+       , aggregateVersion0+       , applyEvent+       , applySnapshot+       , emptyAggregate+       , versionedEvents+       , publishEvent+       ) where++import           Control.DeepSeq (NFData, ($!!))+import           Data.CQRS.Types.AggregateAction (AggregateAction)+import           Data.CQRS.Types.PersistedEvent (PersistedEvent(..))+import           Data.CQRS.Types.Snapshot (Snapshot(..))+import qualified Data.Foldable as F+import           Data.Sequence (Seq, (|>))+import qualified Data.Sequence as S+import           Data.Typeable (Typeable)++-- Aggregate's in-memory state.+data Aggregate a e = Aggregate+    { aggregateAction :: AggregateAction a e+    , aggregateValue :: Maybe a+    , aggregateEvents :: Seq e+    , aggregateVersion0 :: {-# UNPACK #-} !Int+    , aggregateSnapshotVersion :: {-# UNPACK #-} !Int+    } deriving (Typeable)++-- Make "empty" aggregate for applying snapshots and events to.+emptyAggregate :: AggregateAction a e -> Aggregate a e+emptyAggregate aggregateAction' =+    Aggregate { aggregateAction = aggregateAction'+                   , aggregateValue = Nothing+                   , aggregateEvents = S.empty+                   , aggregateVersion0 = -1+                   , aggregateSnapshotVersion = -1+                   }++-- Apply snapshot to aggregate.+applySnapshot :: Aggregate a e -> Maybe (Snapshot a) -> Aggregate a e+applySnapshot a0 Nothing = a0+applySnapshot a0 (Just s) =+  a0 { aggregateValue = Just $ sSnapshot s+     , aggregateVersion0 = sVersion s+     , aggregateSnapshotVersion = sVersion s }++-- Apply event from event store to aggregate. The event will not be published.+applyEvent :: Aggregate a e -> PersistedEvent e -> Aggregate a e+applyEvent a pe =+    a { aggregateValue = Just $ (aggregateAction a) (aggregateValue a) $ peEvent pe+      , aggregateVersion0 = max (peSequenceNumber pe) (aggregateVersion0 a)+      }++-- Publish event to aggregate.+publishEvent :: (NFData e, NFData a) => Aggregate a e -> e -> Aggregate a e+publishEvent a e =+    a { aggregateValue = Just $!! (aggregateAction a) (aggregateValue a) e+      , aggregateEvents = (|>) (aggregateEvents a) $!! e+      }++-- Return events with attached version numbers.+versionedEvents :: Aggregate a e -> [(Int, e)]+versionedEvents a = zip [v0+1 ..] evs+  where+    evs = F.toList $ aggregateEvents a+    v0 = aggregateVersion0 a++-- Return a snapshot of the aggregate state.+aggregateSnapshot :: Aggregate a e -> Maybe (Int, a)+aggregateSnapshot a = fmap (\av -> (v, av)) (aggregateValue a)+  where+    v = S.length (aggregateEvents a) + aggregateVersion0 a
+ src/Data/CQRS/Internal/Repository.hs view
@@ -0,0 +1,54 @@+module Data.CQRS.Internal.Repository+    ( Repository(..)+    , Settings(..)+    , setSnapshotFrequency+    , defaultSettings+    , newRepository+    ) where++import           Control.DeepSeq (NFData)+import           Control.Monad (void)+import           Data.CQRS.Types.EventStore (EventStore(..))+import           Data.CQRS.Types.PersistedEvent (PersistedEvent)+import           Data.CQRS.Types.SnapshotStore+import           Data.CQRS.Types.AggregateAction (AggregateAction)+import           Data.UUID.Types (UUID)++-- | Repository settings+data Settings = Settings+    { settingsSnapshotFrequency :: Maybe Int+    }++-- | Default repository settings:+--+--     * There are no snapshot by default+--+defaultSettings :: Settings+defaultSettings = Settings+    { settingsSnapshotFrequency = Nothing+    }++-- | Set the snapshot frequency. 0 or negative means+-- that no snapshots will be written. Any existing snapshots+-- will still be used.+setSnapshotFrequency :: Int -> Settings -> Settings+setSnapshotFrequency n s = s { settingsSnapshotFrequency = n' }+  where n' | n <= 0    = Nothing+           | otherwise = Just n++-- | Repository consisting of an event store and an event bus.+data Repository a e = Repository+    { repositoryAggregateAction :: AggregateAction a e+    , repositoryEventStore :: EventStore e+    , repositorySnapshotStore :: SnapshotStore a+    , repositoryPublishEvents :: (UUID, [PersistedEvent e]) -> IO ()+    , repositoryUUIDSupply :: IO UUID+    , repositorySettings :: Settings+    }++-- | Create a repository from a pool of event store backends.+newRepository :: (Show e, NFData e) => Settings -> AggregateAction a e -> EventStore e -> SnapshotStore a -> ((UUID, [PersistedEvent e]) -> IO r) -> IO UUID -> (Repository a e)+newRepository settings aggregateAction eventStore snapshotStore publishEvents uuidSupply = do+  Repository aggregateAction eventStore snapshotStore publishEvents' uuidSupply settings+  where+    publishEvents' inputStream = void $ publishEvents inputStream
+ src/Data/CQRS/Query.hs view
@@ -0,0 +1,7 @@+module Data.CQRS.Query+    ( PersistedEvent(..)+    , Repository+    ) where++import Data.CQRS.Types.PersistedEvent+import Data.CQRS.Internal.Repository
+ src/Data/CQRS/Repository.hs view
@@ -0,0 +1,11 @@+module Data.CQRS.Repository+    ( -- * Repository+      Repository+    , newRepository+      -- * Settings+    , Settings+    , setSnapshotFrequency+    , defaultSettings+    ) where++import Data.CQRS.Internal.Repository
+ src/Data/CQRS/SnapshotStore.hs view
@@ -0,0 +1,6 @@+module Data.CQRS.SnapshotStore+    ( nullSnapshotStore+    , SnapshotStore+    ) where++import Data.CQRS.Types.SnapshotStore
+ src/Data/CQRS/Types/AggregateAction.hs view
@@ -0,0 +1,10 @@+-- | Aggregate action type definition.+module Data.CQRS.Types.AggregateAction+       ( AggregateAction+       ) where++-- | An aggregate action is just a function for applying an event to+-- an aggregate. Aggregates that have not been created yet will be+-- passed in as @Nothing@ and aggregates which are being updated+-- will be passed in as @Just x@.+type AggregateAction a e = (Maybe a -> e -> a)
+ src/Data/CQRS/Types/ArchiveMetadata.hs view
@@ -0,0 +1,14 @@+module Data.CQRS.Types.ArchiveMetadata+    ( ArchiveMetadata(..)+    ) where++import Data.CQRS.Types.ArchiveRef (ArchiveRef)+import Data.Typeable (Typeable)+import Data.UUID.Types (UUID)++-- | Archive metadata.+data ArchiveMetadata = ArchiveMetadata+    { amArchiveId :: UUID+    , amPreviousArchiveId :: Maybe UUID+    , amNextArchiveId :: ArchiveRef+    } deriving (Typeable, Show, Eq)
+ src/Data/CQRS/Types/ArchiveRef.hs view
@@ -0,0 +1,10 @@+module Data.CQRS.Types.ArchiveRef+       ( ArchiveRef(..)+       ) where++import Data.Typeable (Typeable)+import Data.UUID.Types (UUID)++-- | Archive reference.+data ArchiveRef = CurrentArchive | NamedArchive UUID+  deriving (Typeable, Show, Eq)
+ src/Data/CQRS/Types/ArchiveStore.hs view
@@ -0,0 +1,111 @@+{-# LANGUAGE RankNTypes, ScopedTypeVariables #-}+module Data.CQRS.Types.ArchiveStore+       ( ArchiveRef+       , ArchiveMetadata(..)+       , ArchiveStore(..)+       , StoreError(..)+       , applyIso+       , enumerateAllEvents+       , rotateArchives+       ) where++import           Control.Monad ((>=>), when, void)+import           Data.CQRS.Types.ArchiveRef+import           Data.CQRS.Types.ArchiveMetadata+import           Data.CQRS.Types.PersistedEvent+import           Data.CQRS.Types.StoreError+import           Data.UUID.Types (UUID)+import           System.IO.Streams (InputStream)+import qualified System.IO.Streams.Combinators as SC++-- | ArchiveStore for events of type e.+data ArchiveStore e = ArchiveStore {+      -- | 'esGetUnarchivedEventCount` returns the number of currently+      -- unarchived events. Because other processes/threads may be+      -- accessing the event store concurrently, the returned value+      -- may only be treated as an estimate.  The caller is+      -- guaranteed, however, that -- absent any other process/thread+      -- calling 'esArchiveEvents' -- the value will be monotonically+      -- increasing.+      asGetUnarchivedEventCount :: IO Int+    ,+      -- | 'esArchiveEvents n' archives up to 'n' events. This+      -- function does nothing if 'n' is less than or equal to 0.  If+      -- there are fewer than 'n' unarchived events, then all those+      -- events will be archived. Returns the UUID of the newly+      -- created archive, if any.+      asArchiveEvents :: Int -> IO (Maybe UUID)+    ,+      -- | Retrieve the metadata of the latest archive. Returns+      -- Returns `Nothing` if there are no archives.+      asReadLatestArchiveMetadata :: IO (Maybe ArchiveMetadata)+    ,+      -- | Retrieve archive metadata for a specified archive.+      asReadArchiveMetadata :: UUID -> IO (Maybe ArchiveMetadata)+    ,+      -- | Read all the events in an archive. There's no guarantee on+      -- the ordering of the returned events except that the events+      -- for any specific aggregate root are returned in order of+      -- incresing version number.+      asReadArchive :: forall a . ArchiveRef -> (InputStream (UUID, PersistedEvent e) -> IO a) -> IO a+    }++-- | Transform an implementation of 'ArchiveStore a' to an+-- implementation of 'ArchiveStore b' via an isomorphism. This can be+-- used to add serialization/deserialization to event stores which do+-- not support storing anything other than binary data.+applyIso :: forall e' e . (e' -> e, e -> e') -> ArchiveStore e -> ArchiveStore e'+applyIso (_, g) (ArchiveStore getUnarchivedEventCount archiveEvents readLatestArchiveMetadata' readArchiveMetadata' readArchive') =+    ArchiveStore getUnarchivedEventCount archiveEvents  readLatestArchiveMetadata' readArchiveMetadata' readArchive+  where+    readArchive :: ArchiveRef -> (InputStream (UUID, PersistedEvent e') -> IO a) -> IO a+    readArchive archiveRef p = readArchive' archiveRef $ SC.map (\(aggregateId, e) -> (aggregateId, fmap g e)) >=> p++-- | Enumerate all events in all archives. __This should ONLY be used+-- for debugging purposes.__+enumerateAllEvents :: ArchiveStore e -> (InputStream (UUID, PersistedEvent e) -> IO ()) -> IO ()+enumerateAllEvents (ArchiveStore _ _ readLatestArchiveMetadata readArchiveMetadata readArchive) p =+  findFirstArchive >>= rollForward+  where+    readArchiveMetadata' archiveId =+        readArchiveMetadata archiveId >>= \x -> case x of+          Just archiveMetadata ->+            return archiveMetadata+          Nothing ->+            error $ "Archive inconsistency; archive reference chain contains non-existent archive " ++ show archiveId++    rollForward archiveRef = do+      readArchive archiveRef p+      case archiveRef of+        CurrentArchive         -> return ()+        NamedArchive archiveId -> readArchiveMetadata' archiveId >>= rollForward . amNextArchiveId++    findFirstArchive =+      readLatestArchiveMetadata >>= \x -> case x of+        Nothing              -> return CurrentArchive+        Just archiveMetadata -> search archiveMetadata+      where+        search archiveMetadata =+          case amPreviousArchiveId archiveMetadata of+            Nothing                -> return $ NamedArchive $ amArchiveId archiveMetadata+            Just previousArchiveId -> readArchiveMetadata' previousArchiveId >>= search++-- | Perform event archival until the current number of unarchived+-- events goes below the given 'archiveSize'. Does nothing if the+-- given 'archiveSize' is not a positive number.+rotateArchives :: ArchiveStore e -> Int -> IO ()+rotateArchives eventStore archiveSize =+    if archiveSize > 0 then+        loop+      else+        return ()+  where+    loop = do+      -- Count number of unarchived events+      count <- asGetUnarchivedEventCount eventStore+      -- Do we have enough to fill an archive?+      when (count >= archiveSize) $ do+        -- Archive an archive's worth of events.+        void $ asArchiveEvents eventStore $ archiveSize+        -- Try again to see if there's more we can archive.+        loop
+ src/Data/CQRS/Types/EventStore.hs view
@@ -0,0 +1,49 @@+{-# LANGUAGE RankNTypes, ScopedTypeVariables #-}+module Data.CQRS.Types.EventStore+       ( EventStore(..)+       , StoreError(..)+       , applyIso+       ) where++import           Control.Monad ((>=>))+import           Data.CQRS.Types.PersistedEvent+import           Data.CQRS.Types.StoreError+import           Data.UUID.Types (UUID)+import           System.IO.Streams (InputStream)+import qualified System.IO.Streams.Combinators as SC++-- | EventStore for events of type e.+data EventStore e = EventStore {+      -- | Store new events for an aggregate. May throw 'StoreError' exception+      -- if there's a problem storing the events. Guarantees atomicity, i.e.+      -- either all the events are stored, or none of them are (in the case of+      -- errors or conflicts).+      esStoreEvents :: UUID -> [PersistedEvent e] -> IO ()+    ,+      -- | Process sequence of events associated with the aggregate+      -- identified by the given UUID. Only events at or after the+      -- given version number are supplied by the input stream. The+      -- events are supplied in increasing order of version number.+      esRetrieveEvents :: forall a . UUID -> Int -> (InputStream (PersistedEvent e) -> IO a) -> IO a+    ,+      -- | Read all events from the event store. Events will be+      -- returned in order of increasing version number, grouped by+      -- aggregate UUID. __This function should ONLY be used for+      -- debugging purposes.__+      esRetrieveAllEvents :: forall a . (InputStream (UUID, PersistedEvent e) -> IO a) -> IO a+    }++-- | Transform an implementation of 'EventStore a' to an+-- implementation of 'EventStore b' via an isomorphism. This can be+-- used to add serialization/deserialization to event stores which do+-- not support storing anything other than binary data.+applyIso :: forall e' e . (e' -> e, e -> e') -> EventStore e -> EventStore e'+applyIso (f, g) (EventStore storeEvents' retrieveEvents' retrieveAllEvents') =+    EventStore storeEvents retrieveEvents retrieveAllEvents+  where+    storeEvents :: UUID -> [PersistedEvent e'] -> IO ()+    storeEvents aggregateId = storeEvents' aggregateId . map (fmap f)+    retrieveEvents :: forall a . UUID -> Int -> (InputStream (PersistedEvent e') -> IO a) -> IO a+    retrieveEvents aggregateId v0 p = retrieveEvents' aggregateId v0 $ SC.map (fmap g) >=> p+    retrieveAllEvents :: forall a. (InputStream (UUID, PersistedEvent e') -> IO a) -> IO a+    retrieveAllEvents p = retrieveAllEvents' $ SC.map (\(aggregateId, e) -> (aggregateId, fmap g e)) >=> p
+ src/Data/CQRS/Types/PersistedEvent.hs view
@@ -0,0 +1,18 @@+{-# LANGUAGE DeriveFunctor, DeriveGeneric #-}+module Data.CQRS.Types.PersistedEvent+       ( PersistedEvent(..)+       ) where++import Control.DeepSeq (NFData(..))+import Data.UUID.Types (UUID)+import GHC.Generics (Generic)++-- | Persisted Event.+data PersistedEvent e =+  PersistedEvent { peEvent :: !e            -- ^ Event.+                 , peSequenceNumber :: !Int -- ^ Sequence number within the aggregate.+                 , peEventId :: !UUID       -- ^ UUID for the event itself.+                 }+  deriving (Show, Eq, Functor, Generic)++instance NFData e => NFData (PersistedEvent e)
+ src/Data/CQRS/Types/Snapshot.hs view
@@ -0,0 +1,9 @@+module Data.CQRS.Types.Snapshot+       ( Snapshot(..)+       ) where++-- | Snapshot of a value at some particular version.+data Snapshot a =+  Snapshot { sVersion :: {-# UNPACK #-} !Int+           , sSnapshot :: !a+           } deriving (Eq, Ord, Show)
+ src/Data/CQRS/Types/SnapshotStore.hs view
@@ -0,0 +1,48 @@+module Data.CQRS.Types.SnapshotStore+    ( SnapshotStore(..)+    , applyPrism+    , nullSnapshotStore+    ) where++import Data.CQRS.Types.Snapshot (Snapshot(..))+import Data.UUID.Types (UUID)++-- | A snapshot store is used for storing snapshots.+data SnapshotStore a = SnapshotStore+    {+      -- | Write out a snapshot. Snapshot version numbers are NOT+      -- checked for validity (e.g. whether they are greater than the+      -- existing snapshot version numbers).+      ssWriteSnapshot :: UUID -> Snapshot a -> IO ()+    ,+      -- | Read latest snapshot of an aggregate identified by UUID. A+      -- snapshot store is permitted to return 'Nothing' in all cases.+      ssReadSnapshot :: UUID -> IO (Maybe (Snapshot a))+    }++-- | Transform an implementation of 'SnapshotStore a' to an+-- implementation of 'SnapshotStore b' via a prism. This is typically+-- used to add serialization/deserialization to snapshot stores which+-- do not support storing anything other than binary data. Note that+-- the prism is allowed to return 'Nothing' to indicate that data could+-- not be transformed. This can be used to avoid the need for complicated+-- versioning of snapshot data by using simple version tags/hashes to+-- determine compatibility with stored snapshots.+applyPrism :: (a -> b, b -> Maybe a) -> SnapshotStore b -> SnapshotStore a+applyPrism (f, g) (SnapshotStore writeSnapshot' readSnapshot') =+    SnapshotStore writeSnapshot readSnapshot+    where+      writeSnapshot aggregateId (Snapshot v a) = do+          writeSnapshot' aggregateId $ Snapshot v $ f a+      readSnapshot aggregateId = do+          fmap (f' =<<) $ (readSnapshot' aggregateId)+          where+            f' (Snapshot v a) = g a >>= Just . Snapshot v++-- | The "null" snapshot store, i.e. a snapshot store which never+-- actually stores any snapshots.+nullSnapshotStore :: SnapshotStore a+nullSnapshotStore =+    SnapshotStore { ssWriteSnapshot = \_ _ -> return ()+                  , ssReadSnapshot = \_ -> return Nothing+                  }
+ src/Data/CQRS/Types/StoreError.hs view
@@ -0,0 +1,13 @@+module Data.CQRS.Types.StoreError+    ( StoreError(..)+    ) where++import Control.Exception (Exception)+import Data.Typeable (Typeable)+import Data.UUID.Types (UUID)++-- | Errors that can happen during 'esStoreEvents'.+data StoreError = VersionConflict UUID+  deriving (Typeable, Show, Eq)++instance Exception StoreError