cqrs 0.5.0 → 0.6.0

raw patch · 7 files changed

+42/−262 lines, 7 filesdep +base16-bytestringdep −direct-sqlitePVP ok

version bump matches the API change (PVP)

Dependencies added: base16-bytestring

Dependencies removed: direct-sqlite

API changes (from Hackage documentation)

Files

• cqrs.cabal +4/−6
• src/Data/CQRS/EventStore/Backend.hs +5/−5
• src/Data/CQRS/EventStore/Backend/Sqlite3.hs +0/−229
• src/Data/CQRS/GUID.hs +20/−6
• src/Data/CQRS/Internal/AggregateRef.hs +2/−2
• src/Data/CQRS/Internal/EventStore.hs +6/−6
• src/Data/CQRS/Transaction.hs +5/−8

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

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

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

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

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

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

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