diff --git a/cqrs.cabal b/cqrs.cabal
--- a/cqrs.cabal
+++ b/cqrs.cabal
@@ -1,46 +1,69 @@
 Name:                cqrs
-Version:             0.8.0
+Version:             0.9.0
 Synopsis:            Command-Query Responsibility Segregation
 Description:         Haskell implementation of the CQRS architectural pattern.
 License:             MIT
 License-file:        LICENSE
 Category:            Data
-Cabal-version:       >=1.6.0.1
+Cabal-version:       >=1.10
 Build-type:          Simple
 Author:              Bardur Arantsson
 Maintainer:          Bardur Arantsson <bardur@scientician.net>
 
 Library
-  Build-Depends: base == 4.*
-               , base16-bytestring >= 0.1.1.3 && < 0.2
-               , bytestring >= 0.9.0.1
-               , cereal >= 0.3.3 && < 0.4
-               , conduit >= 0.1 && <0.2
-               , containers >= 0.4
-               , data-default >= 0.3 && < 0.4
-               , random >= 1.0 && < 1.1
-               , safecopy >= 0.6 && < 0.7
-               , transformers >= 0.2.2 && < 0.3
-  Extensions:          DeriveDataTypeable
+  Build-Depends:       base == 4.*
+                     , bytestring >= 0.9.0.1
+                     , conduit >= 1.0 && < 2
+                     , containers >= 0.5
+                     , cqrs-types >= 0.9.0 && < 0.10
+                     , deepseq >= 1.3 && < 2
+                     , random >= 1.0 && < 1.1
+                     , pool-conduit >= 0.1 && < 0.2
+                     , SafeSemaphore >= 0.9 && < 1.0
+                     , stm >= 2.4 && < 3
+                     , transformers >= 0.2.2 && < 0.4
+  Default-language:    Haskell2010
+  Default-Extensions:  DeriveDataTypeable
                        ExistentialQuantification
-                       FunctionalDependencies
                        GeneralizedNewtypeDeriving
                        MultiParamTypeClasses
-                       OverloadedStrings
-                       Rank2Types
+                       PackageImports
                        ScopedTypeVariables
   ghc-options:         -Wall
   hs-source-dirs:      src
-  Exposed-modules:     Data.CQRS
-                       Data.CQRS.Aggregate
-                       Data.CQRS.AggregateRef
-                       Data.CQRS.Eventable
-                       Data.CQRS.EventStore
-                       Data.CQRS.EventStore.Backend
+  Exposed-modules:     Data.CQRS.Command
+                       Data.CQRS.EventStore.Backend.Memory
                        Data.CQRS.GUID
-                       Data.CQRS.PersistedEvent
-                       Data.CQRS.Serialize
-                       Data.CQRS.Transaction
+                       Data.CQRS.Repository
+                       Data.CQRS.Query
+                       Data.CQRS.Serializable
   Other-modules:       Data.CQRS.Internal.AggregateRef
+                       Data.CQRS.Internal.EventBus
                        Data.CQRS.Internal.EventStore
-                       Data.CQRS.Internal.PersistedEvent
+                       Data.CQRS.Internal.Repository
+                       Data.CQRS.Internal.UnitOfWork
+
+Test-Suite cqrs-tests
+  Type:                exitcode-stdio-1.0
+  Hs-source-dirs:      src-test
+  Main-is:             Main.hs
+  Build-depends:       base == 4.*
+                     , async >= 2.0.1 && < 3
+                     , bytestring >= 0.9.0.1
+                     , conduit >= 0.5 && < 0.6
+                     , pool-conduit >= 0.1 && < 0.2
+                     , stm >= 2.4 && < 3
+                     , transformers >= 0.2.2 && < 0.4
+                     -- Self-dependency
+                     , cqrs
+                     , cqrs-test >= 0.9 && < 0.10
+                     -- Test framework
+                     , hspec >= 1.3 && < 2.0
+                     , HUnit >= 1.2 && < 2.0
+  Ghc-options:         -Wall
+  Default-language:    Haskell2010
+  Default-extensions:  DeriveDataTypeable
+                       MultiParamTypeClasses
+                       OverloadedStrings
+                       ScopedTypeVariables
+                       TemplateHaskell
diff --git a/src-test/Main.hs b/src-test/Main.hs
new file mode 100644
--- /dev/null
+++ b/src-test/Main.hs
@@ -0,0 +1,11 @@
+import Test.Hspec
+import Data.CQRS.TransactionTest
+import Data.CQRS.RepositoryTest
+import Data.CQRS.EventStore.Backend.MemoryTest
+
+main :: IO ()
+main = do
+  hspec $ do
+    memoryTest
+    transactionTest
+    repositoryTest
diff --git a/src/Data/CQRS.hs b/src/Data/CQRS.hs
deleted file mode 100644
--- a/src/Data/CQRS.hs
+++ /dev/null
@@ -1,17 +0,0 @@
-{-| CQRS module. This module exports all the data types
-and functions needed for typical CQRS applications. -}
-module Data.CQRS
-        ( 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
-
-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/Aggregate.hs b/src/Data/CQRS/Aggregate.hs
deleted file mode 100644
--- a/src/Data/CQRS/Aggregate.hs
+++ /dev/null
@@ -1,18 +0,0 @@
--- | 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/AggregateRef.hs b/src/Data/CQRS/AggregateRef.hs
deleted file mode 100644
--- a/src/Data/CQRS/AggregateRef.hs
+++ /dev/null
@@ -1,6 +0,0 @@
-{-| Aggregate root references. -}
-module Data.CQRS.AggregateRef
-       ( AggregateRef
-       ) where
-
-import Data.CQRS.Internal.AggregateRef
diff --git a/src/Data/CQRS/Command.hs b/src/Data/CQRS/Command.hs
new file mode 100644
--- /dev/null
+++ b/src/Data/CQRS/Command.hs
@@ -0,0 +1,32 @@
+{-| Module to import for the "Command" side of the application. -}
+module Data.CQRS.Command
+    ( -- Aggregate:
+      Aggregate
+      -- AggregateRef:
+    , AggregateRef
+      -- Eventable:
+    , Eventable(..)
+      -- EventStoreBackend:
+    , EventStoreBackend
+      -- GUID:
+    , GUID
+    , newGUID
+      -- Repository:
+    , Repository
+      -- UnitOfWork:
+    , UnitOfWorkT
+    , createAggregate
+    , createOrLoadAggregate
+    , findAggregate
+    , loadAggregate
+    , publishEvent
+    , runUnitOfWorkT
+    ) where
+
+import Data.CQRS.Aggregate (Aggregate)
+import Data.CQRS.Internal.AggregateRef (AggregateRef)
+import Data.CQRS.Internal.UnitOfWork
+import Data.CQRS.Internal.Repository (Repository)
+import Data.CQRS.Eventable (Eventable(..))
+import Data.CQRS.EventStore.Backend (EventStoreBackend)
+import Data.CQRS.GUID (GUID, newGUID)
diff --git a/src/Data/CQRS/EventStore.hs b/src/Data/CQRS/EventStore.hs
deleted file mode 100644
--- a/src/Data/CQRS/EventStore.hs
+++ /dev/null
@@ -1,8 +0,0 @@
--- | Event store functions.
-module Data.CQRS.EventStore
-       ( EventStore
-       , enumerateEventStore
-       , withEventStore
-       ) where
-
-import Data.CQRS.Internal.EventStore
diff --git a/src/Data/CQRS/EventStore/Backend.hs b/src/Data/CQRS/EventStore/Backend.hs
deleted file mode 100644
--- a/src/Data/CQRS/EventStore/Backend.hs
+++ /dev/null
@@ -1,59 +0,0 @@
--- | 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(..)
-       , RawEvent
-       , RawSnapshot(..)
-       ) where
-
-import Data.ByteString (ByteString)
-import Data.Conduit (Source)
-import Data.CQRS.GUID
-import Data.CQRS.PersistedEvent (PersistedEvent)
-
--- | Raw event type. The data associated with an event is not
--- translated in any way.
-type RawEvent = PersistedEvent ByteString
-
--- | Raw snapshot.
-data RawSnapshot =
-  RawSnapshot { rsVersion :: Int
-              , rsSnapshotData :: ByteString
-              }
-
--- | 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 :: GUID -> Int -> [RawEvent] -> 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.
-    esbRetrieveEvents :: GUID -> Int -> Source IO RawEvent,
-    -- | Enumerate all events later than given logical timestamp.
-    esbEnumerateAllEvents :: Int -> Source IO RawEvent,
-    -- | 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 :: GUID -> RawSnapshot -> 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 :: GUID -> IO (Maybe RawSnapshot),
-    -- | 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/Memory.hs b/src/Data/CQRS/EventStore/Backend/Memory.hs
new file mode 100644
--- /dev/null
+++ b/src/Data/CQRS/EventStore/Backend/Memory.hs
@@ -0,0 +1,73 @@
+-- | Memory-based event store backend. Used primarily
+-- for testing.
+module Data.CQRS.EventStore.Backend.Memory
+    ( MemoryEventStoreBackend
+    , createBackendPool
+    ) where
+
+-- External imports
+import           Control.Concurrent.STM (atomically)
+import           Control.Concurrent.STM.TVar (TVar, newTVar, readTVar, writeTVar)
+import           Control.Concurrent.MSem as MSem
+import           Control.Monad (unless, liftM)
+import           Control.Monad.IO.Class (liftIO)
+import qualified Data.Conduit.List as CL
+import           Data.Conduit.Pool (Pool, createPool)
+import qualified Data.Foldable as F
+import           Data.Map.Strict (Map)
+import qualified Data.Map.Strict as M
+import           Data.Sequence (Seq, (><))
+import qualified Data.Sequence as S
+
+-- Internal imports
+import Data.CQRS.GUID
+import Data.CQRS.EventStore.Backend
+
+-- | Memory-based event store backend.
+data MemoryEventStoreBackend = MESB
+    (TVar (Map GUID (Seq RawEvent)))
+    (MSem Int)
+    (TVar (Map GUID RawSnapshot))
+
+-- | Pool of memory event store backends.
+createBackendPool :: Int -> IO (Pool MemoryEventStoreBackend)
+createBackendPool n = do
+  -- New "backing store" for every pool.
+  events <- atomically $ newTVar M.empty
+  snapshots <- atomically $ newTVar M.empty
+  lock <- MSem.new 1
+  -- Create the pool
+  createPool (return $ MESB events lock snapshots) (\_ -> return ()) 1 1 n
+
+-- Instance
+instance EventStoreBackend MemoryEventStoreBackend where
+
+    esbStoreEvents (MESB eventsTVar _ _) g v0 newEvents =
+      atomically $ do
+        events <- readTVar eventsTVar
+        let eventsForAggregate = M.findWithDefault S.empty g events
+        let expectedV0 = S.length eventsForAggregate
+        unless (expectedV0 == v0) $ fail "Mismatched version numbers"
+        writeTVar eventsTVar $ M.insert g (eventsForAggregate >< S.fromList newEvents) events
+
+    esbRetrieveEvents (MESB eventsTVar _ _) g v0 = do
+        e <- liftIO $ atomically $ do
+          events <- readTVar eventsTVar
+          return $ F.toList $ M.findWithDefault S.empty g events
+        CL.sourceList $ drop v0 e
+
+    esbEnumerateAllEvents (MESB eventsTVar _ _) = do
+      events <- liftIO $ atomically $ readTVar eventsTVar
+      CL.sourceList $ concatMap F.toList $ M.elems $ events
+
+    esbWriteSnapshot (MESB _ _ v) g s =
+        atomically $ do
+          snapshots <- readTVar v
+          writeTVar v $ M.insert g s $ snapshots
+
+    esbGetLatestSnapshot (MESB _ _ v) g =
+        atomically $ do
+          liftM (M.lookup g) $ readTVar v
+
+    esbWithTransaction (MESB _ lock _) io =
+        MSem.with lock io
diff --git a/src/Data/CQRS/Eventable.hs b/src/Data/CQRS/Eventable.hs
deleted file mode 100644
--- a/src/Data/CQRS/Eventable.hs
+++ /dev/null
@@ -1,9 +0,0 @@
--- | Eventable type class.
-module Data.CQRS.Eventable
-       ( Eventable(..)
-       ) where
-
--- | Type class for applying events to aggregates.
-class Eventable a e | a -> e where
-  -- | Apply an event to the aggregate and return the updated aggregate.
-  applyEvent :: e -> a -> a
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,62 +1,5 @@
--- | Globally Unique IDentifiers.
 module Data.CQRS.GUID
-       ( GUID
-       , fromByteString
-       , hexDecode
-       , hexEncode
-       , newGUID
-       , toByteString
-       ) where
-
-import           Control.Applicative ((<$>))
-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.SafeCopy (SafeCopy(..), contain, safePut, safeGet)
-import           Data.Typeable (Typeable)
-import           Data.Word (Word8)
-import           System.Random (randomRIO)
-
--- | A Globally Unique IDentifier.
-newtype GUID = GUID ByteString
-              deriving (Show, Typeable, Eq, Ord)
-
--- | Default GUID value.
-instance Default GUID where
-    def = GUID $ B.pack $ replicate 32 0
-
--- | Get a random byte.
-randomWord8 :: IO Word8
-randomWord8 = fmap fromInteger $ randomRIO (0,255)
-
--- | Create a new random GUID.
-newGUID :: IO GUID
-newGUID = do
-  uuid <- sequence $ replicate 32 randomWord8
-  return $ GUID $ B.pack uuid
-
--- | Serialize instance.
-instance SafeCopy GUID where
-    putCopy = contain . safePut . toByteString
-    getCopy = contain $ fromByteString <$> safeGet
-
--- | 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
-
--- | Convert from ByteString.
-fromByteString :: ByteString -> GUID
-fromByteString = GUID
+    ( module G
+    ) where
 
--- | Convert to ByteString.
-toByteString :: GUID -> ByteString
-toByteString (GUID g) = g
+import "cqrs-types" Data.CQRS.GUID as G
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,49 +10,52 @@
        , readValue
        ) where
 
+import           Control.DeepSeq (NFData, ($!!))
 import           Control.Monad (liftM)
 import           Control.Monad.IO.Class (MonadIO, liftIO)
 import           Data.CQRS.Eventable (Eventable(..))
 import           Data.CQRS.GUID (GUID)
 import           Data.CQRS.PersistedEvent (PersistedEvent(..))
 import           Data.Foldable (toList)
-import           Data.IORef (IORef, modifyIORef, newIORef, readIORef)
+import           Data.IORef (IORef, modifyIORef', newIORef, readIORef, writeIORef)
 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
+  AggregateRef { arValue :: IORef (Maybe a)
                , arEvents :: IORef (Seq (PersistedEvent e))
-               , arGUID :: GUID
-               , arStartVersion :: Int
-               , arSnapshotVersion :: Int
+               , arGUID :: !GUID
+               , arStartVersion :: !Int
+               , arSnapshotVersion :: !Int
                }
   deriving (Typeable)
 
 -- | Make aggregate
-mkAggregateRef :: (MonadIO m) => a -> GUID -> Int -> Int -> m (AggregateRef a e)
+mkAggregateRef :: (MonadIO m) => Maybe a -> GUID -> Int -> Int -> m (AggregateRef a e)
 mkAggregateRef a guid originatingVersion snapshotVersion = do
   a' <- liftIO $ newIORef a
   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 -> Int -> m ()
-publishEvent aggregateRef event gv = liftIO $ do
+publishEvent :: (MonadIO m, Eventable a e, NFData e, NFData a) => AggregateRef a e -> e -> m ()
+publishEvent aggregateRef event = liftIO $ do
   -- Apply event to aggregate state.
-  modifyIORef (arValue aggregateRef) $ applyEvent event
+  a <- readIORef $ arValue aggregateRef
+  let a' = applyEvent a $!! event
+  writeIORef (arValue aggregateRef) $!! a'
   -- Add event to aggregate.
-  modifyIORef (arEvents aggregateRef) $ \events ->
-    events |> (PersistedEvent (arGUID aggregateRef) event (arStartVersion aggregateRef + 1 + S.length events) gv)
+  modifyIORef' (arEvents aggregateRef) $ \events -> do
+    (events |>) $!! (PersistedEvent (arGUID aggregateRef) event (arStartVersion aggregateRef + 1 + S.length events))
 
 -- | Read aggregate events.
 readEvents :: (MonadIO m) => AggregateRef a e -> m [PersistedEvent e]
 readEvents = liftM toList . liftIO . readIORef . arEvents
 
 -- | Read aggregate state.
-readValue :: (MonadIO m) => AggregateRef a e -> m a
+readValue :: (MonadIO m) => AggregateRef a e -> m (Maybe a)
 readValue = liftIO . readIORef . arValue
 
 -- | Get the current version of the aggregate in aggregate ref.
diff --git a/src/Data/CQRS/Internal/EventBus.hs b/src/Data/CQRS/Internal/EventBus.hs
new file mode 100644
--- /dev/null
+++ b/src/Data/CQRS/Internal/EventBus.hs
@@ -0,0 +1,40 @@
+module Data.CQRS.Internal.EventBus
+    ( EventBus(..)
+    , Subscription(..)
+    , newEventBus
+    ) where
+
+-- External imports
+import           Control.Concurrent.STM (atomically)
+import qualified Control.Concurrent.STM.TChan as C
+import           Control.DeepSeq (NFData, ($!!))
+import           Control.Monad.IO.Class (MonadIO, liftIO)
+
+-- Internal imports
+import           Data.CQRS.PersistedEvent
+
+-- | Event bus that broadcasts events of a certain type.
+data EventBus e = EventBus
+    { publishEventsToBus :: [PersistedEvent e] -> IO ()
+    , subscribeToEventBus :: IO (Subscription e)
+    , unsubscribeFromEventBus :: Subscription e -> IO ()
+    }
+
+-- | A subscription to an event bus.
+newtype Subscription e = Subscription
+    { readEventFromSubscription :: IO [PersistedEvent e]
+    }
+
+-- | Create a new event bus.
+newEventBus :: (MonadIO m, Show e, NFData e) => m (EventBus e)
+newEventBus = do
+  c <- liftIO $ atomically $ C.newBroadcastTChan
+  let publish events = do
+        atomically $ C.writeTChan c $!! events
+      subscribe = do
+        c' <- atomically $ C.dupTChan c
+        return $ Subscription (atomically $ C.readTChan c')
+      unsubscribe _ = do
+        -- TODO: Should we set a flag to detect post-unsubscribe use?
+        return () -- Needs no cleanup.
+  return $ EventBus publish subscribe unsubscribe
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
@@ -1,54 +1,37 @@
 -- | Event store functions.
 module Data.CQRS.Internal.EventStore
-       ( EventStore
+       ( EventStore(..)
        , enumerateEventStore
-       , getLatestSnapshot
-       , getLatestVersion
        , retrieveEvents
        , storeEvents
-       , withEventStore
-       , withTransaction
-       , writeSnapshot
        ) where
 
-import           Control.Exception (bracket)
-import           Control.Monad (liftM)
-import           Data.CQRS.EventStore.Backend (EventStoreBackend(..), RawSnapshot)
-import           Data.CQRS.GUID
-import           Data.CQRS.Internal.PersistedEvent (PersistedEvent(..), mapPersistedEvent)
-import           Data.CQRS.Serialize (decode', encode)
-import           Data.Conduit (Source, ($=))
+-- External imports
+import           Data.ByteString (ByteString)
+import           Data.Conduit (ResourceT, Source, ($=))
 import qualified Data.Conduit.List as CL
-import           Data.SafeCopy (SafeCopy)
+-- Internal imports
+import           Data.CQRS.EventStore.Backend
+import           Data.CQRS.GUID
+import           Data.CQRS.Serializable
+import           Data.CQRS.PersistedEvent (PersistedEvent(..))
 
 -- 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 e . (SafeCopy e) => EventStore e -> Int -> Source IO (PersistedEvent e)
-enumerateEventStore es minVersion =
-  esbEnumerateAllEvents (esBackend es) minVersion $= CL.map (mapPersistedEvent decode')
-
-withTransaction :: EventStore e -> IO a -> IO a
-withTransaction (EventStore esb) = esbWithTransaction esb
-
-storeEvents :: SafeCopy e => EventStore e -> GUID -> Int -> [PersistedEvent e] -> IO ()
-storeEvents (EventStore esb) guid v0 evs = esbStoreEvents esb guid v0 $ (map $ mapPersistedEvent $ encode) evs
+data EventStore e b = EventStore { esBackend :: b }
 
-retrieveEvents :: SafeCopy e => EventStore e -> GUID -> Int -> Source IO (PersistedEvent e)
-retrieveEvents (EventStore esb) guid v0 = fmap (mapPersistedEvent decode') $ esbRetrieveEvents esb guid v0
+enumerateEventStore :: forall e b . (Serializable e, EventStoreBackend b) => EventStore e b -> Source (ResourceT IO) [PersistedEvent e]
+enumerateEventStore es =
+  esbEnumerateAllEvents (esBackend es) $= CL.map (\re -> [fmap deserialize' re])
 
-writeSnapshot ::   EventStore e -> GUID -> RawSnapshot -> IO ()
-writeSnapshot (EventStore esb) = esbWriteSnapshot esb
+storeEvents :: (Serializable e, EventStoreBackend b) => EventStore e b -> GUID -> Int -> [PersistedEvent e] -> IO ()
+storeEvents (EventStore esb) guid v0 evs = do
+  esbStoreEvents esb guid v0 $ (map $ fmap $ serialize) evs
 
-getLatestSnapshot :: EventStore e -> GUID -> IO (Maybe RawSnapshot)
-getLatestSnapshot (EventStore esb) = esbGetLatestSnapshot esb
+retrieveEvents :: (Serializable e, EventStoreBackend b) => EventStore e b -> GUID -> Int -> Source (ResourceT IO) (PersistedEvent e)
+retrieveEvents (EventStore esb) guid v0 = (esbRetrieveEvents esb guid v0) $= (CL.map (fmap deserialize'))
 
-getLatestVersion :: EventStore e -> IO Int
-getLatestVersion (EventStore esb) = esbGetLatestVersion esb
+deserialize' :: Serializable e => ByteString -> e
+deserialize' s =
+    case deserialize s of
+      Nothing -> error "Could not deserialize event -- event deserialization MUST NOT fail"
+      Just e -> e
diff --git a/src/Data/CQRS/Internal/PersistedEvent.hs b/src/Data/CQRS/Internal/PersistedEvent.hs
deleted file mode 100644
--- a/src/Data/CQRS/Internal/PersistedEvent.hs
+++ /dev/null
@@ -1,18 +0,0 @@
-module Data.CQRS.Internal.PersistedEvent
-       ( PersistedEvent(..)
-       , mapPersistedEvent
-       ) where
-
-import Data.CQRS.GUID (GUID)
-
--- | Persisted Event.
-data PersistedEvent e =
-  PersistedEvent { peAggregateGUID :: !GUID -- ^ GUID of the aggregate.
-                 , peEvent :: e             -- ^ Event.
-                 , peSequenceNumber :: !Int -- ^ Sequence number within the aggregate.
-                 , peGlobalVer :: !Int      -- ^ Global sequence number.
-                 }
-
--- Transform the event data.
-mapPersistedEvent :: (e -> e') -> PersistedEvent e -> PersistedEvent e'
-mapPersistedEvent f (PersistedEvent g e v gv) = PersistedEvent g (f e) v gv
diff --git a/src/Data/CQRS/Internal/Repository.hs b/src/Data/CQRS/Internal/Repository.hs
new file mode 100644
--- /dev/null
+++ b/src/Data/CQRS/Internal/Repository.hs
@@ -0,0 +1,84 @@
+module Data.CQRS.Internal.Repository
+    ( Repository(..)
+    , Settings(..)
+    , defaultSettings
+    , enumerateEventStore
+    , enumerateAndStreamEvents
+    , newRepository
+    , withEventStoreBackend
+    ) where
+
+-- External imports
+import           Control.DeepSeq (NFData)
+import           Control.Monad (forever)
+import           Control.Monad.Trans.Class (lift)
+import           Data.Conduit (ResourceT, Source, addCleanup, bracketP, yield)
+import           Data.Conduit.Pool (Pool, withResource, takeResource, mrReuse, mrValue)
+
+-- Internal imports
+import           Data.CQRS.Internal.EventBus
+import qualified Data.CQRS.Internal.EventStore as ES
+import           Data.CQRS.Internal.EventStore (EventStore(..))
+import           Data.CQRS.EventStore.Backend
+import           Data.CQRS.PersistedEvent
+import           Data.CQRS.Serializable
+
+-- | Repository settings
+data Settings = Settings
+    { settingsSnapshotFrequency :: Maybe Int
+    }
+
+-- | Default repository settings.
+defaultSettings :: Settings
+defaultSettings = Settings
+    { settingsSnapshotFrequency = Nothing
+    }
+
+-- | Repository consisting of an event store and an event bus.
+data Repository e b = Repository
+    { repositoryEventStoreBackendPool :: Pool b  -- ^ Event store backend pool
+    , repositoryEventBus :: EventBus e           -- ^ Event bus to use
+    , repositorySettings :: Settings             -- ^ Repository settings
+    }
+
+-- | Create a repository from a pool of event store backends.
+newRepository :: (EventStoreBackend b, Show e, NFData e) => Settings -> Pool b -> IO (Repository e b)
+newRepository settings eventStoreBackendPool = do
+  eventBus <- newEventBus
+  return $ Repository { repositoryEventStoreBackendPool = eventStoreBackendPool
+                      , repositoryEventBus = eventBus
+                      , repositorySettings = settings
+                      }
+
+-- | Use event store backed from repository to perform an action.
+withEventStoreBackend :: (EventStoreBackend b) => Repository e b -> (b -> IO a) -> IO a
+withEventStoreBackend (Repository p _ _) action =
+    withResource p (\eventStoreBackend -> action eventStoreBackend)
+
+-- | Enumerate all events which satisfy certain criteria from event
+-- store associated with repository.
+enumerateEventStore :: (Serializable e, EventStoreBackend b) => Repository e b -> Source (ResourceT IO) [PersistedEvent e]
+enumerateEventStore (Repository p _ _) = do
+  eventStoreBackend <- lift $ takeResource p
+  addCleanup (mrReuse eventStoreBackend) $ ES.enumerateEventStore (EventStore $ mrValue eventStoreBackend)
+
+-- State for the streaming thread.
+data StreamingState e = Buffering [PersistedEvent e]
+                      | EnumerationComplete [PersistedEvent e]
+                      | Streaming
+                        deriving (Show)
+
+-- | Enumerate all events which satisfy criteria and stream
+-- all new events from repository. All events which arrive
+-- while enumerating are buffered until enumeration completes.
+enumerateAndStreamEvents :: (Show e, Serializable e, EventStoreBackend b) => Repository e b -> Source (ResourceT IO) [PersistedEvent e]
+enumerateAndStreamEvents repository@(Repository _ eventBus _) = do
+  bracketP (subscribeToEventBus eventBus)
+           (unsubscribeFromEventBus eventBus)
+           (\subscription -> do
+              enumerateEventStore repository
+              subscriptionSource subscription)
+  where
+    subscriptionSource subscription = forever $ do
+      e <- lift $ lift $ readEventFromSubscription subscription
+      yield e
diff --git a/src/Data/CQRS/Internal/UnitOfWork.hs b/src/Data/CQRS/Internal/UnitOfWork.hs
new file mode 100644
--- /dev/null
+++ b/src/Data/CQRS/Internal/UnitOfWork.hs
@@ -0,0 +1,183 @@
+-- | Run commands against repositories.
+module Data.CQRS.Internal.UnitOfWork
+       ( UnitOfWorkT
+       , createAggregate
+       , createOrLoadAggregate
+       , findAggregate
+       , loadAggregate
+       , publishEvent
+       , runUnitOfWorkT
+       ) where
+
+-- External imports
+import           Control.DeepSeq (NFData)
+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.Conduit (($$), runResourceT)
+import qualified Data.Conduit.List as CL
+import           Data.Foldable (forM_)
+import           Data.List (find)
+import           Data.Typeable (Typeable, cast)
+
+-- Internal imports
+import           Data.CQRS.Aggregate
+import           Data.CQRS.Eventable (Eventable(..))
+import           Data.CQRS.Internal.EventBus
+import           Data.CQRS.EventStore.Backend (EventStoreBackend(..), RawSnapshot(..))
+import           Data.CQRS.GUID (GUID)
+import           Data.CQRS.Internal.AggregateRef (AggregateRef, mkAggregateRef)
+import qualified Data.CQRS.Internal.AggregateRef as AR
+import           Data.CQRS.Internal.EventStore (EventStore(..))
+import qualified Data.CQRS.Internal.EventStore as ES
+import           Data.CQRS.Internal.Repository
+import           Data.CQRS.PersistedEvent (PersistedEvent(..))
+import           Data.CQRS.Serializable
+
+-- | UnitOfWork monad transformer.
+newtype UnitOfWorkT e m a = UnitOfWorkT (UnitOfWorkM e m a)
+    deriving (Functor, Monad)
+
+instance MonadTrans (UnitOfWorkT e) where
+  lift m = UnitOfWorkT $ lift m
+
+-- Existential wrapper for AggregateRef.
+data BoxedAggregateRef e =
+  forall a . (Typeable a, Typeable e, Serializable e, Aggregate a, Eventable a e) => BoxedAggregateRef (AggregateRef a e)
+
+-- Existential wrapper for event store.
+data BoxedEventStore e =
+    forall b . (EventStoreBackend b) => BoxedEventStore (EventStore e b)
+
+-- Existential wrapper for event store backend
+data BoxedEventStoreBackend =
+    forall b . (EventStoreBackend b) => BoxedEventStoreBackend b
+
+-- | UnitOfWork monad.
+type UnitOfWorkM e = StateT (UnitOfWork e)
+
+data UnitOfWork e =
+    UnitOfWork { txnEventStore :: BoxedEventStore e
+               , txnEventStoreBackend :: BoxedEventStoreBackend
+               , aggregateRefsToCommit :: [BoxedAggregateRef e]
+               }
+
+-- | Run transaction against an event store.
+runUnitOfWorkT :: forall b c e . (Typeable e, Serializable e, EventStoreBackend b) => Repository e b -> UnitOfWorkT e IO c -> IO c
+runUnitOfWorkT repository (UnitOfWorkT transaction) = do
+  (r, writtenEvents) <- withEventStoreBackend repository $ \eventStoreBackend -> do
+    esbWithTransaction eventStoreBackend $ do
+      -- Run the computation.
+      let eventStore = EventStore eventStoreBackend
+      (r,s) <- runStateT transaction $ UnitOfWork (BoxedEventStore eventStore) (BoxedEventStoreBackend eventStoreBackend) []
+      -- Write out all the aggregates.
+      writtenEvents <- forM (aggregateRefsToCommit s) $ \(BoxedAggregateRef a) -> do
+        -- 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.
+        forM_ (settingsSnapshotFrequency $ repositorySettings repository) $ \f -> do
+          v <- AR.getCurrentVersion a
+          when (v - AR.arSnapshotVersion a > f) $ do
+            mav <- AR.readValue a
+            case mav of
+              Just av ->
+                  esbWriteSnapshot eventStoreBackend (AR.arGUID a) $ RawSnapshot v $ serialize av
+              Nothing ->
+                  return ()
+        -- Return the written events for accumulator
+        return $ evs
+      -- Return the value.
+      return (r, concat writtenEvents)
+  -- Publish
+  publishEventsToBus (repositoryEventBus repository) writtenEvents
+  -- Return command return value
+  return r
+
+-- Get an aggregate ref by GUID.
+getById :: forall a e . (Typeable a, Typeable e, Serializable e, Aggregate a, Eventable a e) => GUID -> UnitOfWorkT e IO (AggregateRef a e)
+getById guid = UnitOfWorkT $ 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) -> AR.arGUID a == guid) aggregateRefs of
+    Just (BoxedAggregateRef a) ->
+      case cast a of
+        Just (a' :: AggregateRef a e) -> return a'
+        Nothing ->
+          -- This cast could only really fail if there are duplicate GUIDs for
+          -- different types of aggregates/events.
+          fail $ concat ["Duplicate GUID ", show guid, "!" ]
+    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, Serializable e, Aggregate a) => GUID -> UnitOfWorkM e IO (Int, Maybe a)
+getLatestSnapshot guid = do
+  (BoxedEventStoreBackend eventStoreBackend) <- fmap txnEventStoreBackend get
+  r <- liftIO $ esbGetLatestSnapshot eventStoreBackend guid
+  case r of
+    Just (RawSnapshot v a) -> do
+      case deserialize a :: Maybe a of
+        Just a' -> return (v, Just a')
+        Nothing -> return (0, Nothing)
+    Nothing -> do
+      return (0, Nothing)
+
+-- Retrieve aggregate from event store.
+getByIdFromEventStore :: forall a e . (Typeable a, Typeable e, Serializable e, Aggregate a, Eventable a e) => GUID -> UnitOfWorkM e IO (AggregateRef a e)
+getByIdFromEventStore guid = do
+  (BoxedEventStore es) <- fmap txnEventStore get
+  -- Get latest snapshot (if any).
+  (v0,ma0) <- getLatestSnapshot guid
+  -- Get events.
+  events <- lift $ runResourceT $ (ES.retrieveEvents es guid v0 $$ CL.consume)
+  let latestVersion = maximum $ (:) v0 (map peSequenceNumber events)
+  -- Build the aggregate state from all the events.
+  let a = foldl (\a0 e -> applyEvent a0 $ peEvent e) ma0 events
+  -- Make the aggregate itself
+  (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, Serializable e, Typeable a, Typeable e, Aggregate a, Eventable a e, NFData a, NFData e) => AggregateRef a e -> e -> UnitOfWorkT e m ()
+publishEvent aggregateRef event = UnitOfWorkT $ do
+  lift $ AR.publishEvent aggregateRef event
+
+-- | Find aggregate root.
+findAggregate :: (Serializable e, Typeable a, Typeable e, Aggregate a, Eventable a e) => GUID -> UnitOfWorkT e IO (Maybe (AggregateRef a e, a))
+findAggregate guid = do
+  aggregateRef <- getById guid
+  aggregate <- lift $ AR.readValue aggregateRef
+  case aggregate of
+    Nothing -> return Nothing
+    Just a -> return $ Just (aggregateRef, a)
+
+-- | Load aggregate root. The aggregate root must exist.
+loadAggregate :: (Serializable e, Typeable a, Typeable e, Aggregate a, Eventable a e) => GUID -> UnitOfWorkT e IO (AggregateRef a e, a)
+loadAggregate guid = do
+  mAggregate <- findAggregate guid
+  case mAggregate of
+    Nothing -> fail $ show $ "Aggregate with GUID " ++ show guid ++ " does not exist"
+    Just a -> return a
+
+-- | Add aggregate root. The aggregate root will be created upon
+-- transaction commit.
+createAggregate :: (Serializable e, Typeable a, Typeable e, Aggregate a, Eventable a e) => GUID -> UnitOfWorkT e IO (AggregateRef a e)
+createAggregate guid = do
+  aggregateRef <- getById guid
+  aggregate <- lift $ AR.readValue aggregateRef
+  case aggregate of
+    Nothing -> do
+      return aggregateRef
+    Just _ -> fail $ show $ "Aggregate with GUID " ++ show guid ++ " already exists"
+
+-- | Create or load aggregate. The aggregate root will be created (if necessary)
+-- upon transaction commit.
+createOrLoadAggregate :: (Serializable e, Typeable a, Typeable e, Aggregate a, Eventable a e) => GUID -> UnitOfWorkT e IO (AggregateRef a e)
+createOrLoadAggregate guid = getById guid
diff --git a/src/Data/CQRS/PersistedEvent.hs b/src/Data/CQRS/PersistedEvent.hs
deleted file mode 100644
--- a/src/Data/CQRS/PersistedEvent.hs
+++ /dev/null
@@ -1,5 +0,0 @@
-module Data.CQRS.PersistedEvent
-       ( PersistedEvent(..)
-       ) where
-
-import Data.CQRS.Internal.PersistedEvent
diff --git a/src/Data/CQRS/Query.hs b/src/Data/CQRS/Query.hs
new file mode 100644
--- /dev/null
+++ b/src/Data/CQRS/Query.hs
@@ -0,0 +1,19 @@
+module Data.CQRS.Query
+    ( -- cqrs-types:
+      PersistedEvent(..)
+      -- EventStoreBackend:
+    , EventStoreBackend
+      -- GUID:
+    , GUID
+      -- Repository:
+    , Repository
+    , enumerateAndStreamEvents
+    , enumerateEventStore
+    ) where
+
+import "cqrs-types" Data.CQRS.PersistedEvent
+import Data.CQRS.GUID (GUID)
+import Data.CQRS.Internal.Repository (Repository,
+                                      enumerateEventStore,
+                                      enumerateAndStreamEvents)
+import Data.CQRS.EventStore.Backend (EventStoreBackend)
diff --git a/src/Data/CQRS/Repository.hs b/src/Data/CQRS/Repository.hs
new file mode 100644
--- /dev/null
+++ b/src/Data/CQRS/Repository.hs
@@ -0,0 +1,11 @@
+module Data.CQRS.Repository
+    ( -- * Repository
+      Repository
+    , newRepository
+      -- * Settings
+    , Settings
+    , settingsSnapshotFrequency
+    , defaultSettings
+    ) where
+
+import Data.CQRS.Internal.Repository
diff --git a/src/Data/CQRS/Serializable.hs b/src/Data/CQRS/Serializable.hs
new file mode 100644
--- /dev/null
+++ b/src/Data/CQRS/Serializable.hs
@@ -0,0 +1,5 @@
+module Data.CQRS.Serializable
+    ( module G
+    ) where
+
+import "cqrs-types" Data.CQRS.Serializable as G
diff --git a/src/Data/CQRS/Serialize.hs b/src/Data/CQRS/Serialize.hs
deleted file mode 100644
--- a/src/Data/CQRS/Serialize.hs
+++ /dev/null
@@ -1,16 +0,0 @@
-module Data.CQRS.Serialize
-       ( decode'
-       , encode
-       ) where
-
-import Data.ByteString (ByteString)
-import Data.SafeCopy (SafeCopy, safeGet, safePut)
-import Data.Serialize (runGet, runPut)
-
--- | Decode a serialized value irrefutably.
-decode' :: SafeCopy e => ByteString -> e
-decode' = either error id . runGet safeGet
-
--- | Encode a value.
-encode :: SafeCopy e => e -> ByteString
-encode = runPut . safePut
diff --git a/src/Data/CQRS/Transaction.hs b/src/Data/CQRS/Transaction.hs
deleted file mode 100644
--- a/src/Data/CQRS/Transaction.hs
+++ /dev/null
@@ -1,134 +0,0 @@
--- | Run transactions against event stores.
-module Data.CQRS.Transaction
-       ( TransactionT
-       , getAggregateRoot
-       , publishEvent
-       , runTransactionT
-       ) where
-
-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, gets, modify, runStateT)
-import           Data.Conduit (($$), runResourceT)
-import qualified Data.Conduit.List as CL
-import           Data.CQRS.Aggregate (Aggregate(..))
-import           Data.CQRS.Eventable (Eventable(..))
-import           Data.CQRS.EventStore (EventStore)
-import           Data.CQRS.EventStore.Backend (RawSnapshot(..))
-import           Data.CQRS.GUID (GUID)
-import           Data.CQRS.Internal.AggregateRef (AggregateRef, mkAggregateRef)
-import qualified Data.CQRS.Internal.AggregateRef as AR
-import qualified Data.CQRS.Internal.EventStore as ES
-import           Data.CQRS.PersistedEvent (PersistedEvent(..))
-import           Data.Default (Default(..))
-import           Data.List (find)
-import           Data.SafeCopy (SafeCopy)
-import           Data.Typeable (Typeable, cast)
-
--- | Transaction monad transformer.
-newtype TransactionT e m a = TransactionT (Transaction e m a)
-                           deriving (Functor, Monad)
-
-instance MonadTrans (TransactionT e) where
-  lift m = TransactionT $ lift m
-
--- Existential wrapper for AggregateRef.
-data BoxedAggregateRef e =
-  forall a . (Typeable a, Typeable e, SafeCopy e, Default a, Aggregate a, Eventable a e) => BoxedAggregateRef (AggregateRef a e)
-
--- | Transaction monad itself.
-type Transaction e = StateT (TransactionState e)
-
-data TransactionState e =
-  TransactionState { eventStore :: EventStore e
-                   , aggregateRefsToCommit :: [BoxedAggregateRef e]
-                   , tsCurrentVersion :: Int
-                   }
-
--- | Run transaction against an event store.
-runTransactionT :: (Typeable e, SafeCopy e) => EventStore e -> TransactionT e IO c -> IO c
-runTransactionT eventStore_ (TransactionT transaction) = 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 gv0
-    -- Write out all the aggregates.
-    forM_ (aggregateRefsToCommit s) $ \(BoxedAggregateRef a) -> do
-      -- 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
-      when (v - AR.arSnapshotVersion a > 10) $ do
-        av <- AR.readValue a
-        ES.writeSnapshot eventStore_ (AR.arGUID a) $ RawSnapshot 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, SafeCopy 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) -> AR.arGUID a == guid) aggregateRefs of
-    Just (BoxedAggregateRef a) ->
-      case cast a of
-        Just (a' :: AggregateRef a e) -> return a'
-        Nothing ->
-          -- This cast could only really fail if there are duplicate GUIDs for
-          -- different types of aggregates/events.
-          fail $ concat ["Duplicate GUID ", show guid, "!" ]
-    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, SafeCopy e, Default a, Aggregate a) => GUID -> Transaction e IO (Int,a)
-getLatestSnapshot guid = do
-  es <- fmap eventStore get
-  r <- liftIO $ ES.getLatestSnapshot es guid
-  case r of
-    Just (RawSnapshot 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, SafeCopy 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).
-  (v0,a0) <- getLatestSnapshot guid
-  -- Get events.
-  events <- lift $ runResourceT $ ES.retrieveEvents es guid v0 $$ CL.consume
-  let latestVersion = maximum $ (:) v0 (map peSequenceNumber events)
-  -- Build the aggregate state from all the events.
-  let a = foldr (applyEvent . peEvent) a0 events
-  -- Make the aggregate itself
-  (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, SafeCopy 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, SafeCopy 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
-  return (aggregateRef, aggregate)