diff --git a/cqrs.cabal b/cqrs.cabal --- a/cqrs.cabal +++ b/cqrs.cabal @@ -1,5 +1,5 @@ Name: cqrs -Version: 0.1.0 +Version: 0.2.0 Synopsis: Command-Query Responsibility Segregation Description: Haskell implementation of the CQRS architectural pattern. An SQLite3-based backend is included. @@ -13,11 +13,11 @@ Library Build-Depends: base == 4.* - , binary >= 0.5 && < 0.6 , bytestring >= 0.9.0.1 , containers >= 0.4 , 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 ExistentialQuantification @@ -31,6 +31,7 @@ ghc-options: -Wall hs-source-dirs: src Exposed-modules: Data.CQRS + Data.CQRS.Aggregate Data.CQRS.AggregateRef Data.CQRS.Event Data.CQRS.EventStore diff --git a/src/Data/CQRS.hs b/src/Data/CQRS.hs --- a/src/Data/CQRS.hs +++ b/src/Data/CQRS.hs @@ -1,12 +1,14 @@ {-| CQRS module. This module exports all the data types and functions needed for typical CQRS applications. -} module Data.CQRS - ( module Data.CQRS.AggregateRef + ( module Data.CQRS.Aggregate + , module Data.CQRS.AggregateRef , module Data.CQRS.Event , module Data.CQRS.GUID , module Data.CQRS.Transaction ) where +import Data.CQRS.Aggregate import Data.CQRS.AggregateRef import Data.CQRS.Event import Data.CQRS.GUID diff --git a/src/Data/CQRS/Aggregate.hs b/src/Data/CQRS/Aggregate.hs new file mode 100644 --- /dev/null +++ b/src/Data/CQRS/Aggregate.hs @@ -0,0 +1,18 @@ +-- | Aggregate type class. +module Data.CQRS.Aggregate + ( Aggregate(..) + ) where + +import Data.ByteString (ByteString) + +-- | Type class for aggregates. +class Aggregate a where + -- | Encode aggregate into a ByteString representation. The + -- representation should contain some metadata (a UUID for example) + -- which can be used to check reliably whether the encoded + -- representation is valid upon decoding. This can be used if the + -- actual aggregate structure changes. + encodeAggregate :: a -> ByteString + -- | Decode ByteString to aggregate state. If decoding is not + -- possible should return 'Nothing'. + decodeAggregate :: ByteString -> Maybe a diff --git a/src/Data/CQRS/Event.hs b/src/Data/CQRS/Event.hs --- a/src/Data/CQRS/Event.hs +++ b/src/Data/CQRS/Event.hs @@ -4,7 +4,15 @@ ( Event(..) ) where +import Data.ByteString (ByteString) + -- | Event class for applying events to aggregates. class Event e a | e -> a where -- | Apply an event to the aggregate and return the updated aggregate. applyEvent :: e -> a -> a + -- | Encode event into a strict ByteString. + encodeEvent :: e -> ByteString + -- | Decode event from a strict ByteString. To avoid corrupt + -- event stores, any events that have ever been stored {b must} + -- be decodable for all time. + decodeEvent :: ByteString -> e 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 @@ -3,9 +3,8 @@ ( EventStore(..) ) where -import Data.CQRS.Event +import Data.ByteString (ByteString) import Data.CQRS.GUID -import Data.Binary -- | Event stores are the backend used for reading and storing all the -- information about recorded events. @@ -15,12 +14,25 @@ -- into the event store, starting at the provided version number. -- If the version number does not match the expected value, a -- failure occurs. - storeEvents :: (Event e a, Binary e) => GUID a -> Int -> [e] -> IO (), + storeEvents :: forall a. GUID a -> Int -> [ByteString] -> IO (), -- | Retrieve the sequence of events associated with the aggregate - -- identified by the given GUID. The events are returned in increasing - -- order of version number. The version number of the last event is returned - -- as well. - retrieveEvents :: (Event e a, Binary e) => GUID a -> IO (Int,[e]), + -- 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. diff --git a/src/Data/CQRS/EventStore/Sqlite3.hs b/src/Data/CQRS/EventStore/Sqlite3.hs --- a/src/Data/CQRS/EventStore/Sqlite3.hs +++ b/src/Data/CQRS/EventStore/Sqlite3.hs @@ -6,24 +6,39 @@ import Control.Exception (catch, bracket, onException, SomeException) import Control.Monad (when, forM_) -import Data.Binary (encode, decode, Binary) -import qualified Data.ByteString.Char8 as B8 -import qualified Data.ByteString.Lazy.Char8 as BSL +import Data.ByteString (ByteString) import Data.CQRS.EventStore -import Data.CQRS.Event (Event) -import Data.CQRS.GUID +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 , PRIMARY KEY (guid, version) );" +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 = ?;" +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 ) VALUES (?, ?, ?);" +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 );" @@ -34,6 +49,15 @@ 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;" [] @@ -43,9 +67,6 @@ rollbackTransaction :: Database -> IO () rollbackTransaction database = execSql database "ROLLBACK TRANSACTION;" [] -toSQLBlob :: (Binary a) => a -> SQLData -toSQLBlob a = SQLBlob $ B8.concat $ BSL.toChunks $ encode a - withSqlStatement :: Database -> String -> [SQLData] -> (Statement -> IO a) -> IO a withSqlStatement database sql parameters action = bracket (SQL.prepare database sql) SQL.finalize $ \statement -> do @@ -58,24 +79,24 @@ _ <- SQL.step stmt return () -querySql :: Database -> String -> [SQLData] -> ([SQLData] -> IO a) -> IO [a] -querySql database sql parameters reader = +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 [ ] + go statement = loop a0 where loop acc = do res <- SQL.step statement case res of - Done -> return $ reverse acc + Done -> return acc Row -> do cols <- SQL.columns statement - a <- reader cols - loop (a:acc) + acc' <- reader cols acc + loop acc' -badQueryResult :: GUID a -> [SQLData] -> IO b -badQueryResult guid columns = - fail $ concat ["Invalid query result for ", show guid, ": ", show columns] +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 = @@ -83,13 +104,13 @@ , ", saw ", show cv, ")" ] -storeEvents_ :: Database -> (Event e a, Binary e) => GUID a -> Int -> [e] -> IO () +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 [toSQLBlob guid] $ \columns -> + versions <- querySql database getCurrentVersionSql [toSQLData guid] [] $ \columns acc -> case columns of - [ SQLInteger v ] -> return v - _ -> badQueryResult guid columns + [ SQLInteger v ] -> return (v:acc) + _ -> badQueryResult [show guid] columns let curVer = maximum (0 : versions) -- Sanity check current version number. @@ -98,32 +119,59 @@ -- Update de-normalized version number. execSql database updateCurrentVersionSql - [ toSQLBlob guid - , SQLInteger $ fromIntegral $ originatingVersion + length events + [ toSQLData guid + , toSQLData $ originatingVersion + length events ] -- Store the supplied events. forM_ (zip [1 + originatingVersion..] events) $ \(v,e) -> do execSql database insertEventSql - [ toSQLBlob guid - , SQLInteger $ fromIntegral v - , toSQLBlob e + [ toSQLData guid + , toSQLData v + , toSQLData e ] -retrieveEvents_ :: (Event e a, Binary e) => Database -> GUID a -> IO (Int,[e]) -retrieveEvents_ database guid = do +retrieveEvents_ :: Database -> GUID a -> Int -> IO (Int,[ByteString]) +retrieveEvents_ database guid v0 = do -- Find events. - results <- querySql database selectEventsSql [toSQLBlob guid] $ \columns -> do + results <- fmap reverse $ querySql database selectEventsSql [toSQLData guid, toSQLData v0] [] $ \columns acc -> do case columns of [SQLInteger version, SQLBlob eventData] -> - return (version, decode $ BSL.fromChunks [eventData]) + return $ (version, eventData) : acc _ -> - badQueryResult guid columns + badQueryResult [show guid] columns -- Find the max version number. - let maxVersion = maximum $ (:) 0 $ map fst results + 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 @@ -148,9 +196,13 @@ -- 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,22 +1,45 @@ -- | Globally Unique IDentifiers. module Data.CQRS.GUID ( GUID - , mkGUID + , fromByteString + , newGUID + , nil + , toByteString ) where -import Data.Binary (Binary(..)) -import Data.ByteString (ByteString) -import Data.Typeable (Typeable) +import Data.ByteString (ByteString) +import qualified Data.ByteString as B +import Data.ByteString.Char8 () +import Data.Typeable (Typeable) +import Data.Word (Word8) +import System.Random (randomRIO) -- | A GUID for values of type 'a'. newtype GUID a = GUID ByteString deriving (Show, Typeable, Eq) --- | Binary instance. -instance forall a . Binary (GUID a) where - get = fmap GUID get - put (GUID g) = put g +-- | The "nil" GUID. This is used for the root +-- aggregate root which all other aggregate roots +-- can be reached from. +nil :: GUID a +nil = GUID $ B.pack $ replicate 32 0 --- | Make a GUID. -mkGUID :: ByteString -> GUID a -mkGUID s = GUID s +-- | Get a random byte. +randomWord8 :: IO Word8 +randomWord8 = fmap fromInteger $ randomRIO (0,255) + +-- | Create a new random GUID. TODO: We should perhaps +-- be using a cryptographically secury random generator +-- for this. +newGUID :: IO (GUID a) +newGUID = do + 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 -- TODO: Should we sanity check? 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 @@ -1,13 +1,16 @@ module Data.CQRS.Internal.AggregateRef ( AggregateRef , arGUID + , arSnapshotVersion , arStartVersion + , getCurrentVersion , mkAggregateRef , publishEvent , readEvents , readValue ) where +import Control.Monad (liftM) import Control.Monad.IO.Class (MonadIO, liftIO) import Data.CQRS.GUID (GUID) import Data.CQRS.Event (Event(..)) @@ -20,15 +23,16 @@ , arEvents :: IORef [e] , arGUID :: GUID a , arStartVersion :: Int + , arSnapshotVersion :: Int } deriving (Typeable) -- | Make aggregate -mkAggregateRef :: (MonadIO m) => a -> GUID a -> Int -> m (AggregateRef a e) -mkAggregateRef a guid originatingVersion = do +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 [] - return $ AggregateRef a' e' guid originatingVersion + return $ AggregateRef a' e' guid originatingVersion snapshotVersion -- | Publish event to aggregate. publishEvent :: (MonadIO m, Event e a) => AggregateRef a e -> e -> m () @@ -45,3 +49,9 @@ -- | Read aggregate state. readValue :: (MonadIO m) => AggregateRef a e -> m a readValue = liftIO . readIORef . arValue + +-- | 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 + return $ nevs + (arStartVersion a) 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 @@ -1,16 +1,17 @@ -- | Run transactions against event stores. module Data.CQRS.Transaction ( TransactionT - , runTransactionT - , publishEvent , getAggregateRoot + , publishEvent + , retrieveEvents + , runTransactionT ) where -import Control.Monad (forM_) -import Control.Monad.IO.Class (MonadIO) +import Control.Monad (forM_, when) +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.Binary (Binary) +import Data.CQRS.Aggregate (Aggregate(..)) import Data.CQRS.Internal.AggregateRef (AggregateRef, mkAggregateRef) import qualified Data.CQRS.Internal.AggregateRef as AR import Data.CQRS.Event (Event(..)) @@ -22,26 +23,26 @@ import Data.Typeable (Typeable, cast) -- | Transaction monad transformer. -newtype TransactionT m a = TransactionT (Transaction m a) +newtype TransactionT e m a = TransactionT (Transaction e m a) deriving (Functor, Monad) -instance MonadTrans TransactionT where +instance MonadTrans (TransactionT e) where lift m = TransactionT $ lift m -- Existential wrapper for AggregateRef. -data BoxedAggregateRef = - forall a e . (Typeable a, Typeable e, Event e a, Default a, Binary e) => BoxedAggregateRef (AggregateRef a e) +data BoxedAggregateRef e = + forall a . (Typeable a, Typeable e, Event e a, Default a, Aggregate a) => BoxedAggregateRef (AggregateRef a e) -- | Transaction monad itself. -type Transaction = StateT TransactionState +type Transaction e = StateT (TransactionState e) -data TransactionState = +data TransactionState e = TransactionState { eventStore :: EventStore - , aggregateRefsToCommit :: [BoxedAggregateRef] + , aggregateRefsToCommit :: [BoxedAggregateRef e] } -- | Run transaction against an event store. -runTransactionT :: EventStore -> TransactionT IO a -> IO a +runTransactionT :: (Typeable a, Typeable e, Event e a, Default a, Aggregate a) => EventStore -> TransactionT e IO c -> IO c runTransactionT eventStore_ (TransactionT transaction) = do withTransaction eventStore_ $ do -- Run the computation. @@ -49,13 +50,20 @@ -- Write out all the accumulated aggregates forM_ (aggregateRefsToCommit s) $ \(BoxedAggregateRef a) -> do es <- AR.readEvents a - ES.storeEvents eventStore_ (AR.arGUID a) (AR.arStartVersion a) es + ES.storeEvents eventStore_ (AR.arGUID a) (AR.arStartVersion a) + (map encodeEvent es) + -- If we've advanced N events past the last snapshot, we + -- create a new snapshot. + v <- AR.getCurrentVersion a + when (v - AR.arSnapshotVersion a > 10) $ do + av <- AR.readValue a + ES.writeSnapshot eventStore_ (AR.arGUID a) (v, encodeAggregate av) -- Return the value. return r where s0 = TransactionState eventStore_ [ ] -- | Get an aggregate ref by GUID. -getById :: forall a e . (Typeable a, Typeable e, Event e a, Default a, Binary e) => GUID a -> TransactionT IO (AggregateRef a e) +getById :: forall a e . (Typeable a, Typeable e, Event e a, Default a, Aggregate a) => GUID a -> 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 @@ -73,27 +81,62 @@ Nothing -> do getByIdFromEventStore guid +-- 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, Event e a, Default a, Aggregate a) => GUID a -> Transaction e IO (Int,a) +getLatestSnapshot guid = do + es <- fmap eventStore get + r <- liftIO $ ES.getLatestSnapshot es guid + case r of + Just (v',a') -> do + case decodeAggregate a' :: Maybe a of + Just a'' -> return (v', a'') + Nothing -> return (0, def) + Nothing -> do + return (0, def) + -- Retrieve aggregate from event store. -getByIdFromEventStore :: forall a e . (Typeable a, Typeable e, Event e a, Default a, Binary e) => GUID a -> Transaction IO (AggregateRef a e) +getByIdFromEventStore :: forall a e . (Typeable a, Typeable e, Event e a, Default a, Aggregate a) => GUID a -> Transaction e IO (AggregateRef a e) getByIdFromEventStore guid = do es <- fmap eventStore get - (latestVersion :: Int, events :: [e]) <- - lift $ (ES.retrieveEvents es $ guid :: IO (Int,[e])) + -- Get latest snapshot (if any). + (v0,a0) <- getLatestSnapshot guid + -- Get events. + (latestVersion, events) <- lift $ ES.retrieveEvents es guid v0 -- Build the aggregate state from all the events. - let a = foldr applyEvent def events + let a = foldr applyEvent a0 $ map (\e -> decodeEvent e :: e) events -- Make the aggregate itself - (a' :: AggregateRef a e) <- lift $ mkAggregateRef a guid latestVersion + (a' :: AggregateRef a e) <- lift $ mkAggregateRef a guid latestVersion v0 -- Add to set of aggregates to commit later. modify $ \s -> s { aggregateRefsToCommit = (BoxedAggregateRef a' : aggregateRefsToCommit s) } -- Return the aggregate. return a' -- | Publish event for an aggregate root. -publishEvent :: (MonadIO m, Event e a) => AggregateRef a e -> e -> TransactionT m () +publishEvent :: (MonadIO m, Event e a) => 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 a . (Event e a) => 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 = decodeEvent ed + liftIO $ h e + -- | Get aggregate root. -getAggregateRoot :: (Default a, Binary e, Event e a, Typeable a, Typeable e) => GUID a -> TransactionT IO (AggregateRef a e, a) +getAggregateRoot :: (Default a, Event e a, Typeable a, Typeable e, Aggregate a) => GUID a -> TransactionT e IO (AggregateRef a e, a) getAggregateRoot guid = do aggregateRef <- getById guid aggregate <- lift $ AR.readValue aggregateRef