diff --git a/cqrs.cabal b/cqrs.cabal --- a/cqrs.cabal +++ b/cqrs.cabal @@ -1,8 +1,7 @@ Name: cqrs -Version: 0.5.0 +Version: 0.6.0 Synopsis: Command-Query Responsibility Segregation Description: Haskell implementation of the CQRS architectural pattern. - An SQLite3-based backend is included. License: MIT License-file: LICENSE Category: Data @@ -13,12 +12,12 @@ Library Build-Depends: base == 4.* + , base16-bytestring >= 0.1.1.3 && < 0.2 , 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 , transformers >= 0.2.2 && < 0.3 Extensions: DeriveDataTypeable @@ -37,9 +36,8 @@ Data.CQRS.Eventable Data.CQRS.EventStore Data.CQRS.EventStore.Backend - Data.CQRS.EventStore.Backend.Sqlite3 Data.CQRS.GUID + Data.CQRS.Serialize Data.CQRS.Transaction - Other-modules: Data.CQRS.Serialize - Data.CQRS.Internal.AggregateRef + Other-modules: Data.CQRS.Internal.AggregateRef Data.CQRS.Internal.EventStore diff --git a/src/Data/CQRS/EventStore/Backend.hs b/src/Data/CQRS/EventStore/Backend.hs --- a/src/Data/CQRS/EventStore/Backend.hs +++ b/src/Data/CQRS/EventStore/Backend.hs @@ -17,25 +17,25 @@ -- 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 (), + esbStoreEvents :: GUID -> 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]), + esbRetrieveEvents :: GUID -> 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, + esbEnumerateAllEvents :: forall a. Int -> Enumerator (Int, (GUID, Int, ByteString)) IO a, -- | 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 (), + esbWriteSnapshot :: GUID -> (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)), + esbGetLatestSnapshot :: GUID -> IO (Maybe (Int,ByteString)), -- | Get the latest version global number. esbGetLatestVersion :: IO Int, -- | Run transaction against the event store. The transaction is diff --git a/src/Data/CQRS/EventStore/Backend/Sqlite3.hs b/src/Data/CQRS/EventStore/Backend/Sqlite3.hs deleted file mode 100644 --- a/src/Data/CQRS/EventStore/Backend/Sqlite3.hs +++ /dev/null @@ -1,229 +0,0 @@ -{-| 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/GUID.hs b/src/Data/CQRS/GUID.hs --- a/src/Data/CQRS/GUID.hs +++ b/src/Data/CQRS/GUID.hs @@ -1,12 +1,15 @@ -- | Globally Unique IDentifiers. module Data.CQRS.GUID ( GUID + , hexDecode + , hexEncode , newGUID ) where import Control.Monad (liftM) import Data.ByteString (ByteString) import qualified Data.ByteString as B +import qualified Data.ByteString.Base16 as B16 import Data.ByteString.Char8 () import Data.Default (Default(..)) import Data.Serialize (Serialize(..)) @@ -14,12 +17,12 @@ import Data.Word (Word8) import System.Random (randomRIO) --- | A GUID for values of type 'a'. -newtype GUID a = GUID ByteString - deriving (Show, Typeable, Eq) +-- | A Globally Unique IDentifier. +newtype GUID = GUID ByteString + deriving (Show, Typeable, Eq, Ord) -- | Default GUID value. -instance Default (GUID a) where +instance Default GUID where def = GUID $ B.pack $ replicate 32 0 -- | Get a random byte. @@ -27,12 +30,23 @@ randomWord8 = fmap fromInteger $ randomRIO (0,255) -- | Create a new random GUID. -newGUID :: IO (GUID a) +newGUID :: IO GUID newGUID = do uuid <- sequence $ replicate 32 randomWord8 return $ GUID $ B.pack uuid -- | Serialize instance. -instance Serialize (GUID a) where +instance Serialize GUID where put (GUID s) = put s get = liftM GUID get + +-- | Hex encode a GUID. +hexEncode :: GUID -> ByteString +hexEncode (GUID s) = B16.encode s + +-- | Decode a GUID from hex representation. +hexDecode :: ByteString -> Maybe GUID +hexDecode s = + case B16.decode s of + (a,b) | B.length b == 0 -> Just $ GUID a + _ -> Nothing 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 @@ -24,14 +24,14 @@ data AggregateRef a e = AggregateRef { arValue :: IORef a , arEvents :: IORef (Seq (e, Int)) - , arGUID :: GUID a + , arGUID :: GUID , arStartVersion :: Int , arSnapshotVersion :: Int } deriving (Typeable) -- | Make aggregate -mkAggregateRef :: (MonadIO m) => a -> GUID a -> Int -> Int -> m (AggregateRef a e) +mkAggregateRef :: (MonadIO m) => a -> GUID -> Int -> Int -> m (AggregateRef a e) mkAggregateRef a guid originatingVersion snapshotVersion = do a' <- liftIO $ newIORef a e' <- liftIO $ newIORef S.empty diff --git a/src/Data/CQRS/Internal/EventStore.hs b/src/Data/CQRS/Internal/EventStore.hs --- a/src/Data/CQRS/Internal/EventStore.hs +++ b/src/Data/CQRS/Internal/EventStore.hs @@ -42,27 +42,27 @@ -- | 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 :: forall a e . (Serialize e) => EventStore e -> Int -> Enumerator (Int, (GUID, Int, e)) IO a 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 :: Serialize e => EventStore e -> GUID -> 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 :: Serialize e => EventStore e -> GUID -> 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 :: forall a e. Serialize e => EventStore e -> Int -> Enumerator (Int, (GUID, Int, e)) IO a enumerateAllEvents (EventStore esb) v0 = esbEnumerateAllEvents esb v0 $= EL.map (map2nd $ map3rd decode') -writeSnapshot :: EventStore e -> GUID a -> (Int, ByteString) -> IO () +writeSnapshot :: EventStore e -> GUID -> (Int, ByteString) -> IO () writeSnapshot (EventStore esb) = esbWriteSnapshot esb -getLatestSnapshot :: EventStore e -> GUID a -> IO (Maybe (Int, ByteString)) +getLatestSnapshot :: EventStore e -> GUID -> IO (Maybe (Int, ByteString)) getLatestSnapshot (EventStore esb) = esbGetLatestSnapshot esb getLatestVersion :: EventStore e -> IO Int 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 @@ -66,14 +66,11 @@ where s0 = TransactionState eventStore_ [ ] -- | Get an aggregate ref by GUID. -getById :: forall a e . (Typeable a, Typeable e, Serialize e, Default a, Aggregate a, Eventable a e) => GUID a -> TransactionT e IO (AggregateRef a e) +getById :: forall a e . (Typeable a, Typeable e, Serialize e, Default a, Aggregate a, Eventable a e) => GUID -> TransactionT e IO (AggregateRef a e) getById guid = TransactionT $ do -- Check through list to see if we've given out a reference to the aggregate before. aggregateRefs <- fmap aggregateRefsToCommit get - case find (\(BoxedAggregateRef a) -> - case cast $ AR.arGUID a of - Just (guid' :: GUID a) -> guid == guid' - Nothing -> False) aggregateRefs of + case find (\(BoxedAggregateRef a) -> AR.arGUID a == guid) aggregateRefs of Just (BoxedAggregateRef a) -> case cast a of Just (a' :: AggregateRef a e) -> return a' @@ -86,7 +83,7 @@ -- Get the latest snapshot from database, filling in a default -- if a) no snapshot exists, or b) snapshot state was not decodable. -getLatestSnapshot :: forall a e . (Typeable a, Typeable e, Serialize e, Default a, Aggregate a) => GUID a -> Transaction e IO (Int,a) +getLatestSnapshot :: forall a e . (Typeable a, Typeable e, Serialize e, Default a, Aggregate a) => GUID -> Transaction e IO (Int,a) getLatestSnapshot guid = do es <- fmap eventStore get r <- liftIO $ ES.getLatestSnapshot es guid @@ -99,7 +96,7 @@ return (0, def) -- Retrieve aggregate from event store. -getByIdFromEventStore :: forall a e . (Typeable a, Typeable e, Serialize e, Default a, Aggregate a, Eventable a e) => GUID a -> Transaction e IO (AggregateRef a e) +getByIdFromEventStore :: forall a e . (Typeable a, Typeable e, Serialize e, Default a, Aggregate a, Eventable a e) => GUID -> Transaction e IO (AggregateRef a e) getByIdFromEventStore guid = do es <- fmap eventStore get -- Get latest snapshot (if any). @@ -125,7 +122,7 @@ 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) +getAggregateRoot :: (Default a, Serialize e, Typeable a, Typeable e, Aggregate a, Eventable a e) => GUID -> TransactionT e IO (AggregateRef a e, a) getAggregateRoot guid = do aggregateRef <- getById guid aggregate <- lift $ AR.readValue aggregateRef