diff --git a/cqrs.cabal b/cqrs.cabal --- a/cqrs.cabal +++ b/cqrs.cabal @@ -1,5 +1,5 @@ Name: cqrs -Version: 0.4.0 +Version: 0.5.0 Synopsis: Command-Query Responsibility Segregation Description: Haskell implementation of the CQRS architectural pattern. An SQLite3-based backend is included. @@ -16,6 +16,7 @@ , bytestring >= 0.9.0.1 , cereal >= 0.3.3 && < 0.4 , containers >= 0.4 + , enumerator >= 0.4.15 && < 0.5 , data-default >= 0.3 && < 0.4 , direct-sqlite >= 1.1 && < 1.2 , random >= 1.0 && < 1.1 @@ -35,7 +36,10 @@ Data.CQRS.AggregateRef Data.CQRS.Eventable Data.CQRS.EventStore - Data.CQRS.EventStore.Sqlite3 + Data.CQRS.EventStore.Backend + Data.CQRS.EventStore.Backend.Sqlite3 Data.CQRS.GUID Data.CQRS.Transaction - Other-modules: Data.CQRS.Internal.AggregateRef + Other-modules: Data.CQRS.Serialize + Data.CQRS.Internal.AggregateRef + Data.CQRS.Internal.EventStore diff --git a/src/Data/CQRS.hs b/src/Data/CQRS.hs --- a/src/Data/CQRS.hs +++ b/src/Data/CQRS.hs @@ -4,6 +4,7 @@ ( module Data.CQRS.Aggregate , module Data.CQRS.AggregateRef , module Data.CQRS.Eventable + , module Data.CQRS.EventStore , module Data.CQRS.GUID , module Data.CQRS.Transaction ) where @@ -11,5 +12,6 @@ import Data.CQRS.Aggregate import Data.CQRS.AggregateRef import Data.CQRS.Eventable +import Data.CQRS.EventStore import Data.CQRS.GUID(GUID, newGUID) import Data.CQRS.Transaction diff --git a/src/Data/CQRS/EventStore.hs b/src/Data/CQRS/EventStore.hs --- a/src/Data/CQRS/EventStore.hs +++ b/src/Data/CQRS/EventStore.hs @@ -1,49 +1,8 @@ --- | Event store data types. +-- | Event store functions. module Data.CQRS.EventStore - ( EventStore(..) + ( EventStore + , enumerateEventStore , withEventStore ) where -import Control.Exception (bracket) -import Data.ByteString (ByteString) -import Data.CQRS.GUID - --- | Event stores are the backend used for reading and storing all the --- information about recorded events. -data EventStore = - EventStore { - -- | Store a sequence of events for aggregate identified by GUID - -- into the event store, starting at the provided version number. - -- If the version number does not match the expected value, a - -- failure occurs. - storeEvents :: forall a. GUID a -> Int -> [ByteString] -> IO (), - -- | Retrieve the sequence of events associated with the aggregate - -- identified by the given GUID. Only events at or after the given - -- version number are retrieved. The events are returned in - -- increasing order of version number. The version number of the - -- last event is returned as well. - retrieveEvents :: forall a. GUID a -> Int -> IO (Int,[ByteString]), - -- | Read events and pass them to the provided monadic action. - readAllEvents :: Int -> Int -> (ByteString -> IO ()) -> IO (), - -- | Write snapshot for aggregate identified by GUID and - -- the given version number. The version number is NOT checked - -- for validity. If the event store does not support snapshots - -- this function may do nothing. - writeSnapshot :: forall a. GUID a -> (Int,ByteString) -> IO (), - -- | Get latest snapshot of an aggregate identified by GUID. - -- Returns the version number of the snapshot in addition to the - -- data. An event store which does not support snapshots is - -- permitted to return 'Nothing' in all cases. - getLatestSnapshot :: forall a. GUID a -> IO (Maybe (Int,ByteString)), - -- | Run transaction against the event store. The transaction is - -- expected to commit if the supplied IO action runs to completion - -- (i.e. doesn't throw an exception) and to rollback otherwise. - withTransaction :: forall a . IO a -> IO a, - -- | Close the event store. - closeEventStore :: IO () - } - --- | Perform an IO action with an open event store. -withEventStore :: (IO EventStore) -> (EventStore -> IO a) -> IO a -withEventStore open action = bracket open closeEventStore action - +import Data.CQRS.Internal.EventStore diff --git a/src/Data/CQRS/EventStore/Backend.hs b/src/Data/CQRS/EventStore/Backend.hs new file mode 100644 --- /dev/null +++ b/src/Data/CQRS/EventStore/Backend.hs @@ -0,0 +1,47 @@ +-- | Event store backend. You only need to import this +-- module if you're planning on implementing a custom +-- event store backend. +module Data.CQRS.EventStore.Backend + ( EventStoreBackend(..) + ) where + +import Data.ByteString (ByteString) +import Data.CQRS.GUID +import Data.Enumerator (Enumerator) + +-- | Event stores are the backend used for reading and storing all the +-- information about recorded events. +data EventStoreBackend = + EventStoreBackend { + -- | Store a sequence of events for aggregate identified by GUID + -- into the event store, starting at the provided version number. + -- If the version number does not match the expected value, a + -- failure occurs. + esbStoreEvents :: forall a. GUID a -> Int -> [(ByteString,Int)] -> IO (), + -- | Retrieve the sequence of events associated with the aggregate + -- identified by the given GUID. Only events at or after the given + -- version number are retrieved. The events are returned in + -- increasing order of version number. The version number of the + -- last event is returned as well. + esbRetrieveEvents :: forall a. GUID a -> Int -> IO (Int,[ByteString]), + -- | Enumerate all events later than given logical timestamp. + esbEnumerateAllEvents :: forall a b. Int -> Enumerator (Int, (GUID a, Int, ByteString)) IO b, + -- | Write snapshot for aggregate identified by GUID and + -- the given version number. The version number is NOT checked + -- for validity. If the event store does not support snapshots + -- this function may do nothing. + esbWriteSnapshot :: forall a. GUID a -> (Int,ByteString) -> IO (), + -- | Get latest snapshot of an aggregate identified by GUID. + -- Returns the version number of the snapshot in addition to the + -- data. An event store which does not support snapshots is + -- permitted to return 'Nothing' in all cases. + esbGetLatestSnapshot :: forall a. GUID a -> IO (Maybe (Int,ByteString)), + -- | Get the latest version global number. + esbGetLatestVersion :: IO Int, + -- | Run transaction against the event store. The transaction is + -- expected to commit if the supplied IO action runs to completion + -- (i.e. doesn't throw an exception) and to rollback otherwise. + esbWithTransaction :: forall a . IO a -> IO a, + -- | Close the event store. + esbCloseEventStoreBackend :: IO () + } diff --git a/src/Data/CQRS/EventStore/Backend/Sqlite3.hs b/src/Data/CQRS/EventStore/Backend/Sqlite3.hs new file mode 100644 --- /dev/null +++ b/src/Data/CQRS/EventStore/Backend/Sqlite3.hs @@ -0,0 +1,229 @@ +{-| Implementation of an SQLite3-based event store. -} +module Data.CQRS.EventStore.Backend.Sqlite3 + ( openSqliteEventStore + ) where + +import Control.Exception (catch, bracket, finally, onException, SomeException) +import Control.Monad (when, forM_, liftM) +import Data.ByteString (ByteString) +import Data.CQRS.EventStore.Backend (EventStoreBackend(..)) +import Data.CQRS.GUID (GUID) +import Data.CQRS.Serialize (decode') +import Data.Enumerator (Enumerator, Iteratee(..), tryIO, continue, (>>==), ($=), run_, Stream(..)) +import qualified Data.Enumerator.List as EL +import Data.Enumerator.Internal (checkContinue0) +import Data.Serialize (encode) +import qualified Database.SQLite3 as SQL +import Database.SQLite3 (Database, Statement, SQLData(..), StepResult(..)) +import Prelude hiding (catch) + +-- Convenience class for converting values to SQLData. +class ToSQLData a where + toSQLData :: a -> SQLData + +instance ToSQLData (GUID a) where + toSQLData = SQLBlob . encode + +instance ToSQLData Int where + toSQLData = SQLInteger . fromIntegral + +instance ToSQLData ByteString where + toSQLData = SQLBlob + +-- SQL +createEventsSql :: String +createEventsSql = "CREATE TABLE IF NOT EXISTS events ( guid BLOB , ev_data BLOB , version INTEGER , gversion INTEGER, PRIMARY KEY (guid, version) );" + +selectEventsSql :: String +selectEventsSql = "SELECT version, ev_data FROM events WHERE guid = ? AND version > ? ORDER BY version ASC;" + +enumerateAllEventsSql :: String +enumerateAllEventsSql = "SELECT gversion, guid, version, ev_data FROM events WHERE gversion >= ? ORDER BY gversion ASC;" + +insertEventSql :: String +insertEventSql = "INSERT INTO events ( guid, version, ev_data, gversion ) VALUES (?, ?, ?, ?);" + +createAggregateVersionsSql :: String +createAggregateVersionsSql = "CREATE TABLE IF NOT EXISTS versions ( guid BLOB PRIMARY KEY , version INTEGER );" + +getCurrentVersionSql :: String +getCurrentVersionSql = "SELECT version FROM versions WHERE guid = ?;" + +getLatestVersionSql :: String +getLatestVersionSql = "SELECT COALESCE(MAX(gversion), 0) FROM events;" + +updateCurrentVersionSql :: String +updateCurrentVersionSql = "INSERT OR REPLACE INTO versions ( guid, version ) VALUES (?,?);" + +createSnapshotSql :: String +createSnapshotSql = "CREATE TABLE IF NOT EXISTS snapshots ( guid BLOB PRIMARY KEY , data BLOB , version INTEGER );" + +writeSnapshotSql :: String +writeSnapshotSql = "INSERT OR REPLACE INTO snapshots ( guid , data, version ) VALUES ( ?, ?, ? );" + +selectSnapshotSql :: String +selectSnapshotSql = "SELECT data, version FROM snapshots WHERE guid = ?;" + +beginTransaction :: Database -> IO () +beginTransaction database = execSql database "BEGIN TRANSACTION;" [] + +commitTransaction :: Database -> IO () +commitTransaction database = execSql database "COMMIT TRANSACTION;" [] + +rollbackTransaction :: Database -> IO () +rollbackTransaction database = execSql database "ROLLBACK TRANSACTION;" [] + +withSqlStatement :: Database -> String -> [SQLData] -> (Statement -> IO a) -> IO a +withSqlStatement database sql parameters action = + bracket (SQL.prepare database sql) SQL.finalize $ \statement -> do + SQL.bind statement parameters + action statement + +execSql :: Database -> String -> [SQLData] -> IO () +execSql database sql parameters = + withSqlStatement database sql parameters $ \stmt -> do + _ <- SQL.step stmt + return () + +enumQueryResults :: Statement -> Enumerator [SQLData] IO b +enumQueryResults stmt = checkContinue0 $ \loop k -> + do + nextResult <- tryIO $ SQL.step stmt + case nextResult of + Done -> continue k + Row -> do + cols <- tryIO $ SQL.columns stmt + k (Chunks [cols]) >>== loop + +enumQueryResult :: Database -> String -> [SQLData] -> Enumerator [SQLData] IO b +enumQueryResult database sql parameters step = do + stmt <- tryIO $ SQL.prepare database sql + Iteratee $ finally + (do + SQL.bind stmt parameters + runIteratee $ enumQueryResults stmt step) + (SQL.finalize stmt) + +badQueryResultMsg :: [String] -> [SQLData] -> String +badQueryResultMsg params columns = concat ["Invalid query result shape. Params: ", show params, ". Result columns: ", show columns] + +versionConflict :: (Show a, Show b) => a -> b -> IO c +versionConflict ov cv = + fail $ concat [ "Version conflict detected (expected ", show ov + , ", saw ", show cv, ")" + ] + +storeEvents :: forall a. Database -> GUID a -> Int -> [(ByteString,Int)] -> IO () +storeEvents database guid originatingVersion events = do + -- Column unpacking. + let unpackColumns [ SQLInteger v ] = v + unpackColumns columns = error $ badQueryResultMsg [show guid] columns + -- Get the current version number of the aggregate. + curVer <- run_ $ EL.fold (\x -> max x . unpackColumns) 0 >>== + (enumQueryResult database getCurrentVersionSql [toSQLData guid]) + + -- Sanity check current version number. + when (fromIntegral curVer /= originatingVersion) $ + versionConflict originatingVersion curVer + + -- Update de-normalized version number. + execSql database updateCurrentVersionSql + [ toSQLData guid + , toSQLData $ originatingVersion + length events + ] + + -- Store the supplied events. + forM_ (zip [1 + originatingVersion..] events) $ \(v,(e,gv)) -> do + execSql database insertEventSql + [ toSQLData guid + , toSQLData v + , toSQLData e + , toSQLData gv + ] + +retrieveEvents :: Database -> GUID a -> Int -> IO (Int,[ByteString]) +retrieveEvents database guid v0 = do + -- Unpack the columns into tuples. + let unpackColumns [SQLInteger version, SQLBlob eventData] = (version, eventData) + unpackColumns columns = error $ badQueryResultMsg [show guid, show v0] columns + -- Find events with version numbers. + results <- + run_ $ EL.consume >>== + (enumQueryResult database selectEventsSql [toSQLData guid, toSQLData v0] $= + (EL.map unpackColumns)) + -- Find the max version number. + let maxVersion = maximum $ (:) (fromIntegral v0) $ map fst results + return (fromIntegral maxVersion, map snd results) + +enumerateAllEvents :: forall a b. Database -> Int -> Enumerator (Int,(GUID a, Int, ByteString)) IO b +enumerateAllEvents database minVersion = do + enumQueryResult database enumerateAllEventsSql [toSQLData minVersion] $= EL.map + (\columns -> do + case columns of + [ SQLInteger gv, SQLBlob g, SQLInteger v, SQLBlob ed ] -> + ( fromIntegral gv, (decode' g, fromIntegral v, ed) ) + _ -> + error $ badQueryResultMsg [show minVersion] columns) + +writeSnapshot :: Database -> GUID a -> (Int, ByteString) -> IO () +writeSnapshot database guid (v,a) = do + execSql database writeSnapshotSql + [ toSQLData guid + , toSQLData a + , toSQLData v + ] + +getLatestSnapshot :: Database -> GUID a -> IO (Maybe (Int, ByteString)) +getLatestSnapshot database guid = do + -- Unpack columns from result. + let unpackColumns :: [SQLData] -> Maybe (Int,ByteString) + unpackColumns [SQLBlob a, SQLInteger v] = Just (fromIntegral v, a) + unpackColumns columns = error $ badQueryResultMsg [show guid] columns + -- Run the query. + run_ $ EL.fold const Nothing >>== + (enumQueryResult database selectSnapshotSql [toSQLData guid] $= (EL.map unpackColumns)) + +getLatestVersion :: Database -> IO Int +getLatestVersion database = do + -- Unpack columns from result. + let unpackColumns :: [SQLData] -> Int + unpackColumns [SQLInteger v] = fromIntegral v + unpackColumns columns = error $ badQueryResultMsg [] columns + -- Run the query. + liftM head $ run_ $ EL.consume >>== (enumQueryResult database getLatestVersionSql [] $= (EL.map unpackColumns)) + +withTransaction :: forall a . Database -> IO a -> IO a +withTransaction database action = do + beginTransaction database + onException runAction tryRollback + where + runAction = do + r <- action + commitTransaction database + return r + + tryRollback = + -- Try rollback while discarding exception; we want to preserve + -- original exception. + catch (rollbackTransaction database) (\(_::SomeException) -> return ()) + +-- | Open an SQLite3-based event store using the named SQLite database file. +-- The database file is created if it does not exist. +openSqliteEventStore :: String -> IO EventStoreBackend +openSqliteEventStore databaseFileName = do + -- Create the database. + database <- SQL.open databaseFileName + -- Set up tables. + execSql database createEventsSql [] + execSql database createAggregateVersionsSql [] + execSql database createSnapshotSql [] + -- Return event store. + return $ EventStoreBackend { esbStoreEvents = storeEvents database + , esbRetrieveEvents = retrieveEvents database + , esbEnumerateAllEvents = enumerateAllEvents database + , esbWriteSnapshot = writeSnapshot database + , esbGetLatestSnapshot = getLatestSnapshot database + , esbGetLatestVersion = getLatestVersion database + , esbWithTransaction = withTransaction database + , esbCloseEventStoreBackend = SQL.close database + } diff --git a/src/Data/CQRS/EventStore/Sqlite3.hs b/src/Data/CQRS/EventStore/Sqlite3.hs deleted file mode 100644 --- a/src/Data/CQRS/EventStore/Sqlite3.hs +++ /dev/null @@ -1,208 +0,0 @@ -{-| Implementation of an SQLite3-based event store. -} -module Data.CQRS.EventStore.Sqlite3 - ( openSqliteEventStore - , closeEventStore -- Re-export for convenience - ) where - -import Control.Exception (catch, bracket, onException, SomeException) -import Control.Monad (when, forM_) -import Data.ByteString (ByteString) -import Data.CQRS.EventStore -import Data.CQRS.GUID (GUID) -import qualified Data.CQRS.GUID as G -import qualified Database.SQLite3 as SQL -import Database.SQLite3 (Database, Statement, SQLData(..), StepResult(..)) -import Prelude hiding (catch) - --- Convenience class for converting values to SQLData. -class ToSQLData a where - toSQLData :: a -> SQLData - -instance ToSQLData (GUID a) where - toSQLData = SQLBlob . G.toByteString - -instance ToSQLData Int where - toSQLData = SQLInteger . fromIntegral - -instance ToSQLData ByteString where - toSQLData = SQLBlob - --- SQL -createEventsSql :: String -createEventsSql = "CREATE TABLE IF NOT EXISTS events ( guid BLOB , ev_data BLOB , version INTEGER , gversion INTEGER, PRIMARY KEY (guid, version) );" - -selectEventsSql :: String -selectEventsSql = "SELECT version, ev_data FROM events WHERE guid = ? AND version > ? ORDER BY version ASC;" - -selectAllEventsSql :: String -selectAllEventsSql = "SELECT version, ev_data FROM events WHERE gversion >= ? AND gversion < ? ORDER BY gversion ASC;" - -insertEventSql :: String -insertEventSql = "INSERT INTO events ( guid, version, ev_data, gversion ) VALUES (?, ?, ?, COALESCE((SELECT MAX(gversion) FROM events ), 0) + 1);" - -createAggregateVersionsSql :: String -createAggregateVersionsSql = "CREATE TABLE IF NOT EXISTS versions ( guid BLOB PRIMARY KEY , version INTEGER );" - -getCurrentVersionSql :: String -getCurrentVersionSql = "SELECT version FROM versions WHERE guid = ?;" - -updateCurrentVersionSql :: String -updateCurrentVersionSql = "INSERT OR REPLACE INTO versions ( guid, version ) VALUES (?,?);" - -createSnapshotSql :: String -createSnapshotSql = "CREATE TABLE IF NOT EXISTS snapshots ( guid BLOB PRIMARY KEY , data BLOB , version INTEGER );" - -writeSnapshotSql :: String -writeSnapshotSql = "INSERT OR REPLACE INTO snapshots ( guid , data, version ) VALUES ( ?, ?, ? );" - -selectSnapshotSql :: String -selectSnapshotSql = "SELECT data, version FROM snapshots WHERE guid = ?;" - -beginTransaction :: Database -> IO () -beginTransaction database = execSql database "BEGIN TRANSACTION;" [] - -commitTransaction :: Database -> IO () -commitTransaction database = execSql database "COMMIT TRANSACTION;" [] - -rollbackTransaction :: Database -> IO () -rollbackTransaction database = execSql database "ROLLBACK TRANSACTION;" [] - -withSqlStatement :: Database -> String -> [SQLData] -> (Statement -> IO a) -> IO a -withSqlStatement database sql parameters action = - bracket (SQL.prepare database sql) SQL.finalize $ \statement -> do - SQL.bind statement parameters - action statement - -execSql :: Database -> String -> [SQLData] -> IO () -execSql database sql parameters = - withSqlStatement database sql parameters $ \stmt -> do - _ <- SQL.step stmt - return () - -querySql :: Database -> String -> [SQLData] -> a -> ([SQLData] -> a -> IO a) -> IO a -querySql database sql parameters a0 reader = - withSqlStatement database sql parameters go - where - go statement = loop a0 - where - loop acc = do - res <- SQL.step statement - case res of - Done -> return acc - Row -> do - cols <- SQL.columns statement - acc' <- reader cols acc - loop acc' - -badQueryResult :: [String] -> [SQLData] -> IO b -badQueryResult params columns = - fail $ concat ["Invalid query result shape. Params: ", show params, ". Result columns: ", show columns] - -versionConflict :: (Show a, Show b) => a -> b -> IO c -versionConflict ov cv = - fail $ concat [ "Version conflict detected (expected ", show ov - , ", saw ", show cv, ")" - ] - -storeEvents_ :: forall a. Database -> GUID a -> Int -> [ByteString] -> IO () -storeEvents_ database guid originatingVersion events = do - -- Get the current version number of the aggregate. - versions <- querySql database getCurrentVersionSql [toSQLData guid] [] $ \columns acc -> - case columns of - [ SQLInteger v ] -> return (v:acc) - _ -> badQueryResult [show guid] columns - let curVer = maximum (0 : versions) - - -- Sanity check current version number. - when (fromIntegral curVer /= originatingVersion) $ - versionConflict originatingVersion curVer - - -- Update de-normalized version number. - execSql database updateCurrentVersionSql - [ toSQLData guid - , toSQLData $ originatingVersion + length events - ] - - -- Store the supplied events. - forM_ (zip [1 + originatingVersion..] events) $ \(v,e) -> do - execSql database insertEventSql - [ toSQLData guid - , toSQLData v - , toSQLData e - ] - -retrieveEvents_ :: Database -> GUID a -> Int -> IO (Int,[ByteString]) -retrieveEvents_ database guid v0 = do - -- Find events. - results <- fmap reverse $ querySql database selectEventsSql [toSQLData guid, toSQLData v0] [] $ \columns acc -> do - case columns of - [SQLInteger version, SQLBlob eventData] -> - return $ (version, eventData) : acc - _ -> - badQueryResult [show guid] columns - - -- Find the max version number. - let maxVersion = maximum $ (:) (fromIntegral v0) $ map fst results - return (fromIntegral maxVersion, map snd results) - -readAllEvents_ :: Database -> Int -> Int -> (ByteString -> IO ()) -> IO () -readAllEvents_ database minVersion maxVersion handler = do - querySql database selectAllEventsSql [toSQLData minVersion, toSQLData maxVersion] undefined $ \columns _ -> do - case columns of - [SQLInteger version, SQLBlob eventData] -> do - handler eventData - _ -> - badQueryResult [] columns - return () - -writeSnapshot_ :: Database -> GUID a -> (Int, ByteString) -> IO () -writeSnapshot_ database guid (v,a) = do - execSql database writeSnapshotSql - [ toSQLData guid - , toSQLData a - , toSQLData v - ] - -getLatestSnapshot_ :: Database -> GUID a -> IO (Maybe (Int, ByteString)) -getLatestSnapshot_ database guid = do - querySql database selectSnapshotSql [toSQLData guid] Nothing $ \columns _ -> do - case columns of - [SQLBlob aggData, SQLInteger version] -> - return $ Just (fromIntegral version, aggData) - _ -> - badQueryResult [show guid] columns - -withTransaction_ :: forall a . Database -> IO a -> IO a -withTransaction_ database action = do - beginTransaction database - onException runAction tryRollback - where - runAction = do - r <- action - commitTransaction database - return r - - tryRollback = - -- Try rollback while discarding exception; we want to preserve - -- original exception. - catch (rollbackTransaction database) (\(_::SomeException) -> return ()) - --- | Open an SQLite3-based event store using the named SQLite database file. --- The database file is created if it does not exist. -openSqliteEventStore :: String -> IO EventStore -openSqliteEventStore databaseFileName = do - -- Create the database. - database <- SQL.open databaseFileName - -- Set up tables. - execSql database createEventsSql [] - execSql database createAggregateVersionsSql [] - execSql database createSnapshotSql [] - -- Return event store. - return $ EventStore { storeEvents = storeEvents_ database - , retrieveEvents = retrieveEvents_ database - , readAllEvents = readAllEvents_ database - , writeSnapshot = writeSnapshot_ database - , getLatestSnapshot = getLatestSnapshot_ database - , withTransaction = withTransaction_ database - , closeEventStore = SQL.close database - } diff --git a/src/Data/CQRS/GUID.hs b/src/Data/CQRS/GUID.hs --- a/src/Data/CQRS/GUID.hs +++ b/src/Data/CQRS/GUID.hs @@ -1,9 +1,7 @@ -- | Globally Unique IDentifiers. module Data.CQRS.GUID ( GUID - , fromByteString , newGUID - , toByteString ) where import Control.Monad (liftM) @@ -34,15 +32,7 @@ uuid <- sequence $ replicate 32 randomWord8 return $ GUID $ B.pack uuid --- | Convert GUID to ByteString. -toByteString :: GUID a -> ByteString -toByteString (GUID s) = s - --- | Convert ByteString to GUID. -fromByteString :: ByteString -> GUID a -fromByteString s = GUID s - -- | Serialize instance. instance Serialize (GUID a) where - put = put . toByteString - get = liftM fromByteString get + put (GUID s) = put s + get = liftM GUID get diff --git a/src/Data/CQRS/Internal/AggregateRef.hs b/src/Data/CQRS/Internal/AggregateRef.hs --- a/src/Data/CQRS/Internal/AggregateRef.hs +++ b/src/Data/CQRS/Internal/AggregateRef.hs @@ -10,17 +10,20 @@ , readValue ) where -import Control.Monad (liftM) -import Control.Monad.IO.Class (MonadIO, liftIO) -import Data.CQRS.GUID (GUID) -import Data.CQRS.Eventable (Eventable(..)) -import Data.IORef (IORef, modifyIORef, newIORef, readIORef) -import Data.Typeable (Typeable) +import Control.Monad (liftM) +import Control.Monad.IO.Class (MonadIO, liftIO) +import Data.CQRS.GUID (GUID) +import Data.CQRS.Eventable (Eventable(..)) +import Data.Foldable (toList) +import Data.IORef (IORef, modifyIORef, newIORef, readIORef) +import Data.Sequence (Seq, (|>)) +import qualified Data.Sequence as S +import Data.Typeable (Typeable) -- | Aggregate root reference. data AggregateRef a e = AggregateRef { arValue :: IORef a - , arEvents :: IORef [e] + , arEvents :: IORef (Seq (e, Int)) , arGUID :: GUID a , arStartVersion :: Int , arSnapshotVersion :: Int @@ -31,20 +34,20 @@ mkAggregateRef :: (MonadIO m) => a -> GUID a -> Int -> Int -> m (AggregateRef a e) mkAggregateRef a guid originatingVersion snapshotVersion = do a' <- liftIO $ newIORef a - e' <- liftIO $ newIORef [] + e' <- liftIO $ newIORef S.empty return $ AggregateRef a' e' guid originatingVersion snapshotVersion -- | Publish event to aggregate. -publishEvent :: (MonadIO m, Eventable a e) => AggregateRef a e -> e -> m () -publishEvent aggregateRef event = liftIO $ do +publishEvent :: (MonadIO m, Eventable a e) => AggregateRef a e -> e -> Int -> m () +publishEvent aggregateRef event gv = liftIO $ do -- Apply event to aggregate state. modifyIORef (arValue aggregateRef) $ applyEvent event -- Add event to aggregate. - modifyIORef (arEvents aggregateRef) $ \es -> event:es + modifyIORef (arEvents aggregateRef) $ \events -> events |> (event,gv) -- | Read aggregate events. -readEvents :: (MonadIO m) => AggregateRef a e -> m [e] -readEvents = liftIO . readIORef . arEvents +readEvents :: (MonadIO m) => AggregateRef a e -> m [(e,Int)] +readEvents = liftM toList . liftIO . readIORef . arEvents -- | Read aggregate state. readValue :: (MonadIO m) => AggregateRef a e -> m a @@ -53,5 +56,5 @@ -- | Get the current version of the aggregate in aggregate ref. getCurrentVersion :: (MonadIO m) => AggregateRef a e -> m Int getCurrentVersion a = do - nevs <- liftM length $ liftIO $ readIORef $ arEvents a + nevs <- liftM S.length $ liftIO $ readIORef $ arEvents a return $ nevs + (arStartVersion a) diff --git a/src/Data/CQRS/Internal/EventStore.hs b/src/Data/CQRS/Internal/EventStore.hs new file mode 100644 --- /dev/null +++ b/src/Data/CQRS/Internal/EventStore.hs @@ -0,0 +1,69 @@ +-- | Event store functions. +module Data.CQRS.Internal.EventStore + ( EventStore + , enumerateAllEvents + , enumerateEventStore + , getLatestSnapshot + , getLatestVersion + , retrieveEvents + , storeEvents + , withEventStore + , withTransaction + , writeSnapshot + ) where + +import Control.Exception (bracket) +import Control.Monad (liftM) +import Data.ByteString (ByteString) +import Data.CQRS.EventStore.Backend (EventStoreBackend(..)) +import Data.CQRS.GUID +import Data.CQRS.Serialize (decode') +import Data.Enumerator (Enumerator, ($=)) +import qualified Data.Enumerator.List as EL +import Data.Serialize (Serialize, encode) + +-- Utilities. +map1st :: (a -> c) -> (a, b) -> (c, b) +map1st f (a,b) = (f a, b) + +map2nd :: (b -> c) -> (a,b) -> (a,c) +map2nd f (a,b) = (a, f b) + +map3rd :: (c -> d) -> (a,b,c) -> (a,b,d) +map3rd f (a,b,c) = (a,b,f c) + +-- Provide a type alias. +data EventStore e = EventStore { esBackend :: EventStoreBackend + } + +-- | Perform an IO action with an open event store. +withEventStore :: (IO EventStoreBackend) -> (EventStore e -> IO a) -> IO a +withEventStore open action = bracket (liftM EventStore open) (esbCloseEventStoreBackend . esBackend) action + +-- | Enumerate all the events from an event store that occur at or later +-- than a given logical timestamp. +enumerateEventStore :: forall a b e . (Serialize e) => EventStore e -> Int -> Enumerator (Int, (GUID a, Int, e)) IO b +enumerateEventStore es minVersion = + esbEnumerateAllEvents (esBackend es) minVersion $= EL.map (\(gv,(g,ev,ed)) -> (gv,(g,ev,decode' ed :: e))) + +withTransaction :: EventStore e -> IO a -> IO a +withTransaction (EventStore esb) = esbWithTransaction esb + +storeEvents :: Serialize e => EventStore e -> GUID a -> Int -> [(e,Int)] -> IO () +storeEvents (EventStore esb) guid v0 evs = esbStoreEvents esb guid v0 (map (map1st encode) evs) + +retrieveEvents :: Serialize e => EventStore e -> GUID a -> Int -> IO (Int,[e]) +retrieveEvents (EventStore esb) guid v0 = liftM (map2nd $ map decode') $ esbRetrieveEvents esb guid v0 + +enumerateAllEvents :: forall a b e. Serialize e => EventStore e -> Int -> Enumerator (Int, (GUID a, Int, e)) IO b +enumerateAllEvents (EventStore esb) v0 = + esbEnumerateAllEvents esb v0 $= EL.map (map2nd $ map3rd decode') + +writeSnapshot :: EventStore e -> GUID a -> (Int, ByteString) -> IO () +writeSnapshot (EventStore esb) = esbWriteSnapshot esb + +getLatestSnapshot :: EventStore e -> GUID a -> IO (Maybe (Int, ByteString)) +getLatestSnapshot (EventStore esb) = esbGetLatestSnapshot esb + +getLatestVersion :: EventStore e -> IO Int +getLatestVersion (EventStore esb) = esbGetLatestVersion esb diff --git a/src/Data/CQRS/Serialize.hs b/src/Data/CQRS/Serialize.hs new file mode 100644 --- /dev/null +++ b/src/Data/CQRS/Serialize.hs @@ -0,0 +1,10 @@ +module Data.CQRS.Serialize + ( decode' + ) where + +import Data.ByteString (ByteString) +import Data.Serialize (Serialize, decode) + +-- | Decode a serialized value "irrefutably". +decode' :: Serialize e => ByteString -> e +decode' = either error id . decode diff --git a/src/Data/CQRS/Transaction.hs b/src/Data/CQRS/Transaction.hs --- a/src/Data/CQRS/Transaction.hs +++ b/src/Data/CQRS/Transaction.hs @@ -3,25 +3,23 @@ ( TransactionT , getAggregateRoot , publishEvent - , retrieveEvents , runTransactionT ) where -import Control.Monad (forM_, when) +import Control.Monad (forM_, when, liftM) import Control.Monad.IO.Class (MonadIO, liftIO) import Control.Monad.Trans.Class (MonadTrans(..), lift) -import Control.Monad.Trans.State (StateT, get, modify, runStateT) -import Data.ByteString (ByteString) +import Control.Monad.Trans.State (StateT, get, gets, modify, runStateT) import Data.CQRS.Aggregate (Aggregate(..)) import Data.CQRS.Internal.AggregateRef (AggregateRef, mkAggregateRef) import qualified Data.CQRS.Internal.AggregateRef as AR import Data.CQRS.Eventable (Eventable(..)) -import Data.CQRS.EventStore (EventStore, withTransaction) -import qualified Data.CQRS.EventStore as ES +import Data.CQRS.EventStore (EventStore) +import qualified Data.CQRS.Internal.EventStore as ES import Data.CQRS.GUID (GUID) import Data.Default (Default(..)) import Data.List (find) -import Data.Serialize (Serialize, decode, encode) +import Data.Serialize (Serialize) import Data.Typeable (Typeable, cast) -- | Transaction monad transformer. @@ -39,25 +37,24 @@ type Transaction e = StateT (TransactionState e) data TransactionState e = - TransactionState { eventStore :: EventStore + TransactionState { eventStore :: EventStore e , aggregateRefsToCommit :: [BoxedAggregateRef e] + , tsCurrentVersion :: Int } --- | Decode a serialized value "irrefutably". -decode' :: Serialize e => ByteString -> e -decode' = either error id . decode - -- | Run transaction against an event store. -runTransactionT :: (Typeable e, Serialize e) => EventStore -> TransactionT e IO c -> IO c +runTransactionT :: (Typeable e, Serialize e) => EventStore e -> TransactionT e IO c -> IO c runTransactionT eventStore_ (TransactionT transaction) = do - withTransaction eventStore_ $ do + ES.withTransaction eventStore_ $ do + -- Get the latest global version number to use. + gv0 <- liftM ((+) 1) $ ES.getLatestVersion eventStore_ -- Run the computation. - (r,s) <- runStateT transaction s0 - -- Write out all the accumulated aggregates + (r,s) <- runStateT transaction $ s0 gv0 + -- Write out all the aggregates. forM_ (aggregateRefsToCommit s) $ \(BoxedAggregateRef a) -> do - es <- AR.readEvents a - ES.storeEvents eventStore_ (AR.arGUID a) (AR.arStartVersion a) - (map encode es) + -- Write out accumulated events. + evs <- AR.readEvents a + ES.storeEvents eventStore_ (AR.arGUID a) (AR.arStartVersion a) evs -- If we've advanced N events past the last snapshot, we -- create a new snapshot. v <- AR.getCurrentVersion a @@ -110,7 +107,7 @@ -- Get events. (latestVersion, events) <- lift $ ES.retrieveEvents es guid v0 -- Build the aggregate state from all the events. - let a = foldr applyEvent a0 $ map (\e -> decode' e :: e) events + let a = foldr applyEvent a0 events -- Make the aggregate itself (a' :: AggregateRef a e) <- lift $ mkAggregateRef a guid latestVersion v0 -- Add to set of aggregates to commit later. @@ -119,27 +116,13 @@ return a' -- | Publish event for an aggregate root. -publishEvent :: (MonadIO m, Serialize e, Eventable a e) => AggregateRef a e -> e -> TransactionT e m () -publishEvent aggregateRef event = lift $ AR.publishEvent aggregateRef event - --- | Retrieve a range of events, invoking a callback function for each event. --- For example, --- --- @ --- retrieveEvents 10 20 $ \\e -> do --- putStrLn $ \"Event: \" ++ show e --- @ --- --- would print all events between "logical time" 20 (inclusive) --- and "logical time" 30 (not inclusive). "Logical time" starts --- at 1. It is NOT an error to provide bounds for which there --- are no events. -retrieveEvents :: forall e . (Serialize e) => Int -> Int -> (e -> IO ()) -> TransactionT e IO () -retrieveEvents minVersion maxVersion h = TransactionT $ do - es <- fmap eventStore get - lift $ ES.readAllEvents es minVersion maxVersion $ \ed -> do - let e :: e = decode' ed - liftIO $ h e +publishEvent :: (MonadIO m, Serialize e, Typeable a, Typeable e, Aggregate a, Eventable a e, Default a) => AggregateRef a e -> e -> TransactionT e m () +publishEvent aggregateRef event = TransactionT $ do + -- Publish event to aggregate itself. + currentGlobalVersion <- gets tsCurrentVersion + lift $ AR.publishEvent aggregateRef event currentGlobalVersion + -- Increment global version number. + modify $ \s -> s { tsCurrentVersion = currentGlobalVersion + 1 } -- | Get aggregate root. getAggregateRoot :: (Default a, Serialize e, Typeable a, Typeable e, Aggregate a, Eventable a e) => GUID a -> TransactionT e IO (AggregateRef a e, a)