packages feed

cqrs (empty) → 0.1.0

raw patch · 11 files changed

+447/−0 lines, 11 filesdep +basedep +binarydep +bytestringsetup-changed

Dependencies added: base, binary, bytestring, containers, data-default, direct-sqlite, transformers

Files

+ LICENSE view
@@ -0,0 +1,19 @@+Copyright (c) 2011 Bardur Arantsson++Permission is hereby granted, free of charge, to any person obtaining a copy+of this software and associated documentation files (the "Software"), to deal+in the Software without restriction, including without limitation the rights+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell+copies of the Software, and to permit persons to whom the Software is+furnished to do so, subject to the following conditions:++The above copyright notice and this permission notice shall be included in+all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN+THE SOFTWARE.
+ Setup.lhs view
@@ -0,0 +1,4 @@+#! /usr/bin/env runhaskell++> import Distribution.Simple+> main = defaultMain
+ cqrs.cabal view
@@ -0,0 +1,40 @@+Name:                cqrs+Version:             0.1.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+Cabal-version:       >=1.6.0.1+Build-type:          Simple+Author:              Bardur Arantsson+Maintainer:          Bardur Arantsson <bardur@scientician.net>++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+               , transformers >= 0.2.2 && < 0.3+  Extensions:          DeriveDataTypeable+                       ExistentialQuantification+                       FunctionalDependencies+                       GeneralizedNewtypeDeriving+                       MultiParamTypeClasses+                       OverloadedStrings+                       Rank2Types+                       ScopedTypeVariables+                       TemplateHaskell+  ghc-options:         -Wall+  hs-source-dirs:      src+  Exposed-modules:     Data.CQRS+                       Data.CQRS.AggregateRef+                       Data.CQRS.Event+                       Data.CQRS.EventStore+                       Data.CQRS.EventStore.Sqlite3+                       Data.CQRS.GUID+                       Data.CQRS.Transaction+  Other-modules:       Data.CQRS.Internal.AggregateRef
+ src/Data/CQRS.hs view
@@ -0,0 +1,13 @@+{-| 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.Event+        , module Data.CQRS.GUID+        , module Data.CQRS.Transaction+        ) where++import Data.CQRS.AggregateRef+import Data.CQRS.Event+import Data.CQRS.GUID+import Data.CQRS.Transaction
+ src/Data/CQRS/AggregateRef.hs view
@@ -0,0 +1,6 @@+{-| Aggregate root references. -}+module Data.CQRS.AggregateRef+       ( AggregateRef+       ) where++import Data.CQRS.Internal.AggregateRef
+ src/Data/CQRS/Event.hs view
@@ -0,0 +1,10 @@+-- | Event type class. All events that can be applied to aggregate roots+-- are required have an instance for this class.+module Data.CQRS.Event+       ( Event(..)+       ) where++-- | 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
+ src/Data/CQRS/EventStore.hs view
@@ -0,0 +1,30 @@+-- | Event store data types.+module Data.CQRS.EventStore+       ( EventStore(..)+       ) where++import Data.CQRS.Event+import Data.CQRS.GUID+import Data.Binary++-- | Event stores are the backend used for reading and storing all the+-- information about recorded events.+data EventStore =+  EventStore {+    -- | Store a sequence of events for aggregate identified by GUID+    -- into the event store, starting at the provided version number.+    -- If the version number does not match the expected value, a+    -- failure occurs.+    storeEvents :: (Event e a, Binary e) => GUID a -> Int -> [e] -> 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]),+    -- | Run transaction against the event store. The transaction is+    -- expected to commit if the supplied IO action runs to completion+    -- (i.e. doesn't throw an exception) and to rollback otherwise.+    withTransaction :: forall a . IO a -> IO a,+    -- | Close the event store.+    closeEventStore :: IO ()+    }
+ src/Data/CQRS/EventStore/Sqlite3.hs view
@@ -0,0 +1,156 @@+{-| Implementation of an SQLite3-based event store. -}+module Data.CQRS.EventStore.Sqlite3+       ( openSqliteEventStore+       , closeEventStore -- Re-export for convenience+       ) where++import           Control.Exception (catch, bracket, onException, SomeException)+import           Control.Monad (when, forM_)+import           Data.Binary (encode, decode, Binary)+import qualified Data.ByteString.Char8 as B8+import qualified Data.ByteString.Lazy.Char8 as BSL+import           Data.CQRS.EventStore+import           Data.CQRS.Event (Event)+import           Data.CQRS.GUID+import qualified Database.SQLite3 as SQL+import           Database.SQLite3 (Database, Statement, SQLData(..), StepResult(..))+import           Prelude hiding (catch)++createEventsSql :: String+createEventsSql = "CREATE TABLE IF NOT EXISTS events ( guid BLOB , ev_data BLOB , version INTEGER , PRIMARY KEY (guid, version) );"++selectEventsSql :: String+selectEventsSql = "SELECT version, ev_data FROM events WHERE guid = ?;"++insertEventSql :: String+insertEventSql = "INSERT INTO events ( guid, version, ev_data ) 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 = ?;"++updateCurrentVersionSql :: String+updateCurrentVersionSql = "INSERT OR REPLACE INTO versions ( guid, version ) VALUES (?,?);"++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;" []++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+    SQL.bind statement parameters+    action statement++execSql :: Database -> String -> [SQLData] -> IO ()+execSql database sql parameters =+  withSqlStatement database sql parameters $ \stmt -> do+    _ <- SQL.step stmt+    return ()++querySql :: Database -> String -> [SQLData] -> ([SQLData] -> IO a) -> IO [a]+querySql database sql parameters reader =+  withSqlStatement database sql parameters go+  where+    go statement = loop [ ]+      where+        loop acc = do+          res <- SQL.step statement+          case res of+            Done -> return $ reverse acc+            Row -> do+              cols <- SQL.columns statement+              a <- reader cols+              loop (a:acc)++badQueryResult :: GUID a -> [SQLData] -> IO b+badQueryResult guid columns =+  fail $ concat ["Invalid query result for ", show guid, ": ", 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_ :: Database -> (Event e a, Binary e) => GUID a -> Int -> [e] -> IO ()+storeEvents_ database guid originatingVersion events = do+  -- Get the current version number of the aggregate.+  versions <- querySql database getCurrentVersionSql [toSQLBlob guid] $ \columns ->+    case columns of+      [ SQLInteger v ] -> return v+      _ -> badQueryResult guid columns+  let curVer = maximum (0 : versions)++  -- Sanity check current version number.+  when (fromIntegral curVer /= originatingVersion) $+    versionConflict originatingVersion curVer++  -- Update de-normalized version number.+  execSql database updateCurrentVersionSql+    [ toSQLBlob guid+    , SQLInteger $ fromIntegral $ 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+      ]++retrieveEvents_ :: (Event e a, Binary e) => Database -> GUID a -> IO (Int,[e])+retrieveEvents_ database guid = do+  -- Find events.+  results <- querySql database selectEventsSql [toSQLBlob guid] $ \columns -> do+    case columns of+      [SQLInteger version, SQLBlob eventData] ->+        return (version, decode $ BSL.fromChunks [eventData])+      _ ->+        badQueryResult guid columns++  -- Find the max version number.+  let maxVersion = maximum $ (:) 0 $ map fst results+  return (fromIntegral maxVersion, map snd results)++withTransaction_ :: forall a . Database -> IO a -> IO a+withTransaction_ database action = do+  beginTransaction database+  onException runAction tryRollback+  where+    runAction = do+      r <- action+      commitTransaction database+      return r++    tryRollback =+      -- Try rollback while discarding exception; we want to preserve+      -- original exception.+      catch (rollbackTransaction database) (\(_::SomeException) -> return ())++-- | Open an SQLite3-based event store using the named SQLite database file.+-- The database file is created if it does not exist.+openSqliteEventStore :: String -> IO EventStore+openSqliteEventStore databaseFileName = do+  -- Create the database.+  database <- SQL.open databaseFileName+  -- Set up tables.+  execSql database createEventsSql []+  execSql database createAggregateVersionsSql []+  -- Return event store.+  return $ EventStore { storeEvents = storeEvents_ database+                      , retrieveEvents = retrieveEvents_ database+                      , withTransaction = withTransaction_ database+                      , closeEventStore = SQL.close database+                      }
+ src/Data/CQRS/GUID.hs view
@@ -0,0 +1,22 @@+-- | Globally Unique IDentifiers.+module Data.CQRS.GUID+       ( GUID+       , mkGUID+       ) where++import Data.Binary (Binary(..))+import Data.ByteString (ByteString)+import Data.Typeable (Typeable)++-- | 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++-- | Make a GUID.+mkGUID :: ByteString -> GUID a+mkGUID s = GUID s
+ src/Data/CQRS/Internal/AggregateRef.hs view
@@ -0,0 +1,47 @@+module Data.CQRS.Internal.AggregateRef+       ( AggregateRef+       , arGUID+       , arStartVersion+       , mkAggregateRef+       , publishEvent+       , readEvents+       , readValue+       ) where++import Control.Monad.IO.Class (MonadIO, liftIO)+import Data.CQRS.GUID (GUID)+import Data.CQRS.Event (Event(..))+import Data.IORef (IORef, modifyIORef, newIORef, readIORef)+import Data.Typeable (Typeable)++-- | Aggregate root reference.+data AggregateRef a e =+  AggregateRef { arValue :: IORef a+               , arEvents :: IORef [e]+               , arGUID :: GUID a+               , arStartVersion :: Int+               }+  deriving (Typeable)++-- | Make aggregate+mkAggregateRef :: (MonadIO m) => a -> GUID a -> Int -> m (AggregateRef a e)+mkAggregateRef a guid originatingVersion = do+  a' <- liftIO $ newIORef a+  e' <- liftIO $ newIORef []+  return $ AggregateRef a' e' guid originatingVersion++-- | Publish event to aggregate.+publishEvent :: (MonadIO m, Event e a) => AggregateRef a e -> e -> m ()+publishEvent aggregateRef event = liftIO $ do+  -- Apply event to aggregate state.+  modifyIORef (arValue aggregateRef) $ applyEvent event+  -- Add event to aggregate.+  modifyIORef (arEvents aggregateRef) $ \es -> event:es++-- | Read aggregate events.+readEvents :: (MonadIO m) => AggregateRef a e -> m [e]+readEvents = liftIO . readIORef . arEvents++-- | Read aggregate state.+readValue :: (MonadIO m) => AggregateRef a e -> m a+readValue = liftIO . readIORef . arValue
+ src/Data/CQRS/Transaction.hs view
@@ -0,0 +1,100 @@+-- | Run transactions against event stores.+module Data.CQRS.Transaction+       ( TransactionT+       , runTransactionT+       , publishEvent+       , getAggregateRoot+       ) where++import           Control.Monad (forM_)+import           Control.Monad.IO.Class (MonadIO)+import           Control.Monad.Trans.Class (MonadTrans(..), lift)+import           Control.Monad.Trans.State (StateT, get, modify, runStateT)+import           Data.Binary (Binary)+import           Data.CQRS.Internal.AggregateRef (AggregateRef, mkAggregateRef)+import qualified Data.CQRS.Internal.AggregateRef as AR+import           Data.CQRS.Event (Event(..))+import           Data.CQRS.EventStore (EventStore, withTransaction)+import qualified Data.CQRS.EventStore as ES+import           Data.CQRS.GUID (GUID)+import           Data.Default (Default(..))+import           Data.List (find)+import           Data.Typeable (Typeable, cast)++-- | Transaction monad transformer.+newtype TransactionT m a = TransactionT (Transaction m a)+                           deriving (Functor, Monad)++instance MonadTrans TransactionT 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)++-- | Transaction monad itself.+type Transaction = StateT TransactionState++data TransactionState =+  TransactionState { eventStore :: EventStore+                   , aggregateRefsToCommit :: [BoxedAggregateRef]+                   }++-- | Run transaction against an event store.+runTransactionT :: EventStore -> TransactionT IO a -> IO a+runTransactionT eventStore_ (TransactionT transaction) = do+  withTransaction eventStore_ $ do+    -- Run the computation.+    (r,s) <- runStateT transaction s0+    -- 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+    -- 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 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+    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++-- 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 guid = do+  es <- fmap eventStore get+  (latestVersion :: Int, events :: [e]) <-+    lift $ (ES.retrieveEvents es $ guid :: IO (Int,[e]))+  -- Build the aggregate state from all the events.+  let a = foldr applyEvent def events+  -- Make the aggregate itself+  (a' :: AggregateRef a e) <- lift $ mkAggregateRef a guid latestVersion+  -- 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 aggregateRef event = lift $ AR.publishEvent aggregateRef event++-- | Get aggregate root.+getAggregateRoot :: (Default a, Binary e, Event e a, Typeable a, Typeable e) => GUID a -> TransactionT IO (AggregateRef a e, a)+getAggregateRoot guid = do+  aggregateRef <- getById guid+  aggregate <- lift $ AR.readValue aggregateRef+  return (aggregateRef, aggregate)