diff --git a/cqrs-example.cabal b/cqrs-example.cabal --- a/cqrs-example.cabal +++ b/cqrs-example.cabal @@ -1,31 +1,67 @@ Name: cqrs-example -Version: 0.5.0.1 +Version: 0.6.0 Synopsis: Example for cqrs package +Description: Example for cqrs package License: MIT License-file: LICENSE Author: Bardur Arantsson Maintainer: spam@scientician.net Category: Web Build-type: Simple -Cabal-version: >=1.6.0.1 +Cabal-version: >=1.10 Executable cqrs-example Main-is: Main.hs + Ghc-options: -Wall Build-depends: base == 4.* + , aeson >= 0.3.2.12 && <0.4 , bytestring >= 0.9 && < 0.10 , cereal >= 0.3 && < 0.4 , containers >= 0.4 && < 0.5 - , cqrs >= 0.5 && < 0.6 + , cqrs >= 0.6 && < 0.7 + , cqrs-sqlite3 >= 0.6 && < 0.7 , data-default >= 0.3 && < 0.4 + , direct-sqlite >= 1.1 && < 1.2 , enumerator >= 0.4.15 && < 0.5 + , happstack-lite >= 6.0 && < 7.0 , text >= 0.11 && < 0.12 + , time >= 1.2 && < 1.3 , transformers >= 0.2.2 && < 0.3 - Ghc-options: -Wall - Extensions: DeriveDataTypeable - MultiParamTypeClasses - OverloadedStrings - ScopedTypeVariables + Default-extensions: DeriveDataTypeable + MultiParamTypeClasses + OverloadedStrings + ScopedTypeVariables Hs-source-dirs: src Other-modules: Aggregates Events Query + Default-language: Haskell2010 + +Test-Suite cqrs-example-tests + type: exitcode-stdio-1.0 + hs-source-dirs: src + main-is: Test.hs + build-depends: base == 4.* + , bytestring >= 0.9 && < 0.10 + , cereal >= 0.3 && < 0.4 + , containers >= 0.4 && < 0.5 + , cqrs >= 0.5 && < 0.6 + , data-default >= 0.3 && < 0.4 + , direct-sqlite >= 1.1 && < 1.2 + , enumerator >= 0.4.15 && < 0.5 + , text >= 0.11 && < 0.12 + , time >= 1.2 && < 1.3 + , transformers >= 0.2.2 && < 0.3 + -- Test framework: + , HUnit >= 1.2 && < 2.0 + , test-framework >= 0.4.0 + , test-framework-hunit >= 0.2.0 + , test-framework-quickcheck >= 0.2.0 + , QuickCheck >= 1.1 && < 2 + default-language: Haskell2010 + ghc-options: -threaded -O0 + default-extensions: DeriveDataTypeable + MultiParamTypeClasses + OverloadedStrings + ScopedTypeVariables + diff --git a/src/Aggregates.hs b/src/Aggregates.hs --- a/src/Aggregates.hs +++ b/src/Aggregates.hs @@ -3,39 +3,54 @@ , ProjectId , Task(..) , TaskId + , User(..) + , UserId + , WorkUnit(..) ) where import Control.Monad (liftM) import Data.CQRS (Aggregate(..), GUID) import Data.Default (Default(..)) +import Data.Map (Map) import Data.Serialize (Serialize(..), decode, encode) import Data.Text (Text) import Data.Text.Encoding (decodeUtf8, encodeUtf8) +import Data.Time.Calendar(Day(..), toGregorian, fromGregorian) import Data.Typeable (Typeable) import Data.Word (Word8) +-- Instances for Haskell built-ins. +instance Serialize Day where + put = put . toGregorian + get = do + (a,b,c) <- get + return $ fromGregorian a b c + -- Projects. -type ProjectId = GUID Project +type ProjectId = GUID data Project = DefaultProject | ActiveProject { projectName :: Text + , projectCustomer :: Text } - deriving (Typeable) + deriving (Typeable, Eq, Show) instance Serialize Project where put DefaultProject = do put (0 :: Word8) - put (ActiveProject pn) = do + put (ActiveProject pn pc) = do put (1 :: Word8) put $ encodeUtf8 pn + put $ encodeUtf8 pc get = do i :: Word8 <- get case i of 0 -> return DefaultProject 1 -> do pn <- liftM decodeUtf8 get - return $ ActiveProject pn + pc <- liftM decodeUtf8 get + return $ ActiveProject pn pc _ -> fail $ "Cannot decode project state" ++ show i @@ -51,21 +66,25 @@ -- Tasks. -type TaskId = GUID Task +type TaskId = GUID data Task = NewTask | ActiveTask { taskProjectId :: ProjectId , taskShortDescription :: Text + , taskLongDescription :: Text + , taskWorkUnits :: Map Day WorkUnit } - deriving (Typeable) + deriving (Typeable, Eq, Show) instance Serialize Task where put NewTask = do put (0 :: Word8) - put (ActiveTask tpid tsd) = do + put (ActiveTask tpid tsd tld twu) = do put (1 :: Word8) put tpid put $ encodeUtf8 tsd + put $ encodeUtf8 tld + put twu get = do i :: Word8 <- get case i of @@ -73,7 +92,9 @@ 1 -> do tpid <- get tsd <- liftM decodeUtf8 get - return $ ActiveTask tpid tsd + tld <- liftM decodeUtf8 get + twu <- get + return $ ActiveTask tpid tsd tld twu _ -> fail $ "Cannot decode task state: " ++ show i @@ -87,3 +108,66 @@ instance Default Task where def = NewTask + +-- Users. + +type UserId = GUID + +data User = UncreatedUser + | ActiveUser { auUserName :: Text + , auLastName :: Text + , auFirstName :: Text + } + deriving (Typeable, Eq, Show) + +instance Serialize User where + put UncreatedUser = do + put $ (0 :: Word8) + put (ActiveUser auun auln aufn) = do + put $ (1 :: Word8) + put $ encodeUtf8 auun + put $ encodeUtf8 auln + put $ encodeUtf8 aufn + get = do + i :: Word8 <- get + case i of + 0 -> return UncreatedUser + 1 -> do + auun <- liftM decodeUtf8 get + auln <- liftM decodeUtf8 get + aufn <- liftM decodeUtf8 get + return $ ActiveUser auun auln aufn + _ -> + fail $ "Cannot decode user state: " ++ show i + +instance Aggregate User where + encodeAggregate = encode + decodeAggregate s = + case decode s of + Left e -> error e + Right a -> a + +instance Default User where + def = UncreatedUser + + +-- Work Units (NOT aggregates!) +data WorkUnit = WorkUnit { workUnitId :: GUID + , workUnitComment :: Text + , workUnitDuration :: Integer + , workUnitUser :: UserId + } + deriving (Eq, Show) + +instance Serialize WorkUnit where + put (WorkUnit wuid wuc wud wuu) = do + put $ wuid + put $ encodeUtf8 wuc + put wud + put wuu + get = do + wuid <- get + wuc <- liftM decodeUtf8 get + wud <- get + wuu <- get + return $ WorkUnit wuid wuc wud wuu diff --git a/src/Events.hs b/src/Events.hs --- a/src/Events.hs +++ b/src/Events.hs @@ -4,21 +4,26 @@ import Aggregates import Control.Monad (liftM) +import Data.CQRS (GUID) import Data.Serialize (Serialize(..)) import Data.Text (Text) import Data.Text.Encoding (decodeUtf8, encodeUtf8) +import Data.Time.Calendar (Day) import Data.Typeable (Typeable) import Data.Word (Word8) -data Event = ProjectCreated Text +data Event = ProjectCreated Text Text | ProjectRenamed Text | TaskAdded ProjectId Text + | RecordedWorkUnit GUID Day Integer Text UserId + | UserCreated Text Text Text deriving (Typeable, Show) instance Serialize Event where - put (ProjectCreated name) = do + put (ProjectCreated name shortDesc) = do put (0 :: Word8) put $ encodeUtf8 name + put $ encodeUtf8 shortDesc put (ProjectRenamed name) = do put (1 :: Word8) put $ encodeUtf8 name @@ -26,12 +31,25 @@ put (2 :: Word8) put $ pid put $ encodeUtf8 tsd + put (RecordedWorkUnit wuId wuDay wuDuration wuComment wuUserId) = do + put (3 :: Word8) + put $ wuId + put $ wuDay + put $ wuDuration + put $ encodeUtf8 wuComment + put wuUserId + put (UserCreated ucUserName ucFirst ucLast) = do + put (4 :: Word8) + put $ encodeUtf8 ucUserName + put $ encodeUtf8 ucFirst + put $ encodeUtf8 ucLast get = do i :: Word8 <- get case i of 0 -> do n <- liftM decodeUtf8 get - return $ ProjectCreated n + sd <- liftM decodeUtf8 get + return $ ProjectCreated n sd 1 -> do n <- liftM decodeUtf8 get return $ ProjectRenamed n @@ -39,6 +57,18 @@ pid <- get tsd <- liftM decodeUtf8 get return $ TaskAdded pid tsd + 3 -> do + wuId <- get + wuDay <- get + wuDuration <- get + wuComment <- liftM decodeUtf8 get + wuUserId <- get + return $ RecordedWorkUnit wuId wuDay wuDuration wuComment wuUserId + 4 -> do + ucUserName <- liftM decodeUtf8 get + ucFirst <- liftM decodeUtf8 get + ucLast <- liftM decodeUtf8 get + return $ UserCreated ucUserName ucFirst ucLast _ -> do fail $ "Unrecognized event type " ++ show i diff --git a/src/Main.hs b/src/Main.hs --- a/src/Main.hs +++ b/src/Main.hs @@ -1,58 +1,124 @@ -import Control.Concurrent (forkIO, threadDelay) -import Control.Monad.Trans.Class (lift) -import Data.CQRS -import Data.CQRS.EventStore.Backend.Sqlite3 -import Data.IORef (newIORef, readIORef) -import Aggregates -import Events -import Instances () -import Query +import Control.Concurrent (forkIO, threadDelay) +import Control.Monad (liftM) +import Control.Monad.Trans.Class (lift) +import Data.Aeson (ToJSON, encode) +import Data.CQRS +import Data.CQRS.GUID (hexEncode) +import Data.CQRS.EventStore.Backend.Sqlite3 +import qualified Data.Map as M +import qualified Data.Text as T +import qualified Data.Text.Encoding as TE +import Data.Time (fromGregorian) +import Database.SQLite3 (Database) +import Aggregates +import Events +import Instances () +import Query +import Happstack.Lite sqliteFile :: String sqliteFile = "example.db" -eventSourcingThread :: EventStore Event -> IO () -eventSourcingThread eventStore = do - qsRef <- newIORef emptyQueryState - loop qsRef +eventSourcingThread :: Database -> EventStore Event -> IO () +eventSourcingThread queryDb eventStore = do + loop where - loop qsRef = do + loop = do putStrLn "Sourcing events..." - updateQueryStateRef qsRef eventStore - qs <- readIORef qsRef - putStrLn $ " => qs: " ++ show qs + updateQueryStateRef queryDb eventStore -- Wait 1 second threadDelay 1000000 -- Go again. - loop qsRef + loop +-- Serve JSON ok response. +okJson :: ToJSON a => a -> ServerPart Response +okJson a = ok $ toResponseBS "text/json" $ encode a +-- Serve JSON project list. +projectsJson :: Database -> ServerPart Response +projectsJson queryDb = do + p <- liftM (map f) $ lift $ qProjectList queryDb + okJson $ p + where + f (g,pn,psd) = + M.fromList [ (T.pack "id" , TE.decodeUtf8 $ hexEncode g) + , (T.pack "name" , pn) + , (T.pack "short_desc", psd) + ] + +-- Serve JSON task list. +tasksJson :: Database -> ServerPart Response +tasksJson queryDb = do + p <- liftM (map f) $ lift $ qTaskList queryDb + okJson $ p + where + f (g,tsd) = + M.fromList [ (T.pack "id" , TE.decodeUtf8 $ hexEncode g) + , (T.pack "short_description", tsd) + ] + +-- Serve JSON data. +jsonPart :: Database -> ServerPart Response +jsonPart queryDb = + msum + [ dir "projects" $ projectsJson queryDb + , dir "tasks" $ tasksJson queryDb + , dir "time-sheet" $ okJson ([ ] :: [String]) + ] + +-- Serve. +myApp :: Database -> ServerPart Response +myApp queryDb = msum + [ dir "js" $ serveDirectory EnableBrowsing [ ] "static/js" + , dir "css" $ serveDirectory EnableBrowsing [ ] "static/css" + , dir "json" $ jsonPart queryDb + , serveDirectory EnableBrowsing [ "index.html" ] "static" + ] + +-- test1 :: IO () test1 = do + let queryDbFile = "query.db" -- Start sourcing events. _ <- forkIO $ do - withEventStore (openSqliteEventStore sqliteFile) eventSourcingThread + queryDb <- initializeQueryDatabase queryDbFile + withEventStore (openSqliteEventStore sqliteFile) $ eventSourcingThread queryDb + -- Web serving thread. + _ <- forkIO $ do + queryDb <- initializeQueryDatabase queryDbFile + serve Nothing $ myApp queryDb + -- Do a few things withEventStore (openSqliteEventStore sqliteFile) $ \eventStore -> do runTransactionT eventStore $ do -- Create new project. - projectId :: GUID Project <- lift $ newGUID - (projectRef, _) <- getAggregateRoot projectId - publishEvent projectRef $ ProjectCreated "my project" + projectId <- lift $ newGUID + (projectRef, _ :: Project) <- getAggregateRoot projectId + publishEvent projectRef $ ProjectCreated "my project" "short desc of my project" -- Add a couple of tasks - taskId1 :: GUID Task <- lift $ newGUID - (taskRef1, _) <- getAggregateRoot taskId1 + taskId1 <- lift $ newGUID + (taskRef1, _ :: Task) <- getAggregateRoot taskId1 publishEvent taskRef1 $ TaskAdded projectId "tweak knob" - taskId2 :: GUID Task <- lift $ newGUID - (taskRef2, _) <- getAggregateRoot taskId2 + taskId2 <- lift $ newGUID + (taskRef2, _ :: Task) <- getAggregateRoot taskId2 publishEvent taskRef2 $ TaskAdded projectId "pull lever" -- Rename project. publishEvent projectRef $ ProjectRenamed "old project" + + -- Create user. + userId1 <- lift $ newGUID + (userRef1, _ :: User) <- getAggregateRoot userId1 + publishEvent userRef1 $ UserCreated "bardur" "Bardur" "Arantsson" + + -- Record time for task #2. + workUnitId1 <- lift $ newGUID + publishEvent taskRef2 $ RecordedWorkUnit workUnitId1 (fromGregorian 2011 12 01) 3 "found the lever to pull" userId1 main :: IO () diff --git a/src/Query.hs b/src/Query.hs --- a/src/Query.hs +++ b/src/Query.hs @@ -1,48 +1,123 @@ -module Query ( emptyQueryState +module Query ( initializeQueryDatabase , updateQueryStateRef - , QueryState + , qProjectList + , qTaskList ) where -import Control.Monad (forM_) +import Control.Monad (forM_, liftM) import Data.ByteString (ByteString) import Data.CQRS (enumerateEventStore, GUID, EventStore) -import Data.Enumerator (run_, (>>==)) +import Data.CQRS.EventStore.Backend.Sqlite3Utils (withTransaction, execSql, enumQueryResult) +import Data.Enumerator (run_, (>>==), ($=)) import qualified Data.Enumerator.List as EL -import Data.IORef (IORef, readIORef, atomicModifyIORef) -import Data.Map (Map) -import qualified Data.Map as M -import Data.Serialize (encode) +import Data.Serialize (Serialize, decode, encode) import Data.Text (Text) +import qualified Data.Text as T +import qualified Database.SQLite3 as SQL +import Database.SQLite3 (Database, SQLData(..)) import Events (Event(..)) --- Update query state with latest events. -updateQueryStateRef :: IORef QueryState -> EventStore Event -> IO () -updateQueryStateRef queryStateRef eventStore = do - lastVersion <- fmap ((+) 1 . qsLastVersion) $ readIORef queryStateRef - evs <- run_ (EL.consume >>== enumerateEventStore eventStore lastVersion) - forM_ evs $ \(gv,(g,v,e)) -> do - putStrLn $ show $ "ev(" ++ show gv ++ "): " ++ show e - atomicModifyIORef queryStateRef $ \s -> (updateState s gv g v e, ()) - return () +-- TODO: Export Data.CQRS.Serialize decode' instead? +decode' :: Serialize a => ByteString -> a +decode' = either error id . decode --- Update state per event. -updateState :: QueryState -> Int -> GUID a -> Int -> Event -> QueryState -updateState queryState gv guid _ (ProjectCreated pn) = - queryState { qsLastVersion = gv - , qsProjectNamesById = M.insert (encode guid) pn $ qsProjectNamesById queryState - } -updateState queryState gv guid v (ProjectRenamed pn) = - updateState queryState gv guid v (ProjectCreated pn) -updateState queryState gv _ _ (TaskAdded _ _) = - queryState { qsLastVersion = gv } +-- Initialize the query state. +initializeQueryDatabase :: String -> IO Database +initializeQueryDatabase databaseFileName = do + -- Open database. + database <- SQL.open databaseFileName + -- Create prerequisite tables if necessary. + execSql database "CREATE TABLE IF NOT EXISTS gversion ( gversion INTEGER )" [] + execSql database "CREATE TABLE IF NOT EXISTS projects ( guid BLOB , name TEXT , short_desc TEXT )" [] + execSql database "CREATE TABLE IF NOT EXISTS tasks ( guid BLOB , project_guid BLOB , short_desc TEXT , long_desc TEXT )" [] + execSql database "CREATE TABLE IF NOT EXISTS users ( user_id BLOB , user_name TEXT , first_name TEXT , last_name TEXT )" [] + execSql database "CREATE TABLE IF NOT EXISTS work_units ( task_guid BLOB , wu_guid BLOB , wu_date TEXT , wu_duration INTEGER , wu_comment TEXT , user_id BLOB )" [] + -- Return the open database. + return database --- In-memory data structure for querying. We should really use --- a persistent store for this, but this is just an example. -data QueryState = QueryState { qsLastVersion :: Int - , qsProjectNamesById :: Map ByteString Text - } - deriving (Show) +-- React to event. +reactToEvent :: Database -> (Int, (GUID, Int, Event)) -> IO () +reactToEvent database (gv, (g, _, e)) = do + putStrLn $ show $ "gv(" ++ show gv ++ "): event: " ++ show e + react e + where + react (ProjectCreated pn psd) = do + execSql database "INSERT INTO projects ( guid , name , short_desc ) VALUES ( ? , ? , ? );" [ SQLBlob $ encode g , SQLText $ T.unpack pn , SQLText $ T.unpack psd ] --- Create "empty" data structure for querying. -emptyQueryState :: QueryState -emptyQueryState = QueryState 0 M.empty + react (ProjectRenamed pn) = do + execSql database "UPDATE projects SET name = ? WHERE guid = ?;" [ SQLText $ T.unpack pn , SQLBlob $ encode g ] + + react (TaskAdded pid tsd) = do + execSql database "INSERT INTO tasks ( guid , project_guid , short_desc ) VALUES (?, ?, ?);" [ SQLBlob $ encode g, SQLBlob $ encode pid , SQLText $ T.unpack tsd ] + + react (RecordedWorkUnit wuId wuDay wuDuration wuComment wuUserId) = do + execSql database "INSERT INTO work_units ( task_guid , wu_guid , wu_date , wu_duration , wu_comment , user_id ) VALUES ( ? , ? , ? , ? , ? , ? );" + [ SQLBlob $ encode g + , SQLBlob $ encode wuId + , SQLText $ show wuDay + , SQLInteger $ fromIntegral wuDuration + , SQLText $ T.unpack wuComment + , SQLBlob $ encode wuUserId + ] + + react (UserCreated ucUserName ucFirstName ucLastName) = do + execSql database "INSERT INTO users ( user_id , user_name , first_name , last_name ) VALUES ( ? , ? , ? , ? );" + [ SQLBlob $ encode g + , SQLText $ T.unpack ucUserName + , SQLText $ T.unpack ucFirstName + , SQLText $ T.unpack ucLastName + ] + +-- Get the global version number. +getStartVersion :: Database -> IO Int +getStartVersion queryDatabase = do + lastVersion <- + liftM head $ + run_ $ EL.consume >>== + (enumQueryResult queryDatabase "SELECT COALESCE(MAX(gversion), 1) FROM gversion" [] $= + (EL.map (\columns -> case columns of + [ SQLInteger i ] -> fromIntegral i + _ -> error "Invalid columns in SQL query result"))) + return $ lastVersion + +updateStartVersion :: Database -> Int -> IO () +updateStartVersion queryDatabase newStartVersion = do + execSql queryDatabase "DELETE FROM gversion;" [] + execSql queryDatabase "INSERT INTO gversion (gversion) VALUES (?);" [ SQLInteger $ fromIntegral $ newStartVersion ] + +-- Update query state with latest events. +updateQueryStateRef :: Database -> EventStore Event -> IO () +updateQueryStateRef queryDatabase eventStore = + withTransaction queryDatabase $ do + -- Find the version number to start at. + startVersion <- getStartVersion queryDatabase + -- Enumerate all the events form the event store since last run. + evs <- run_ (EL.consume >>== enumerateEventStore eventStore startVersion) + -- React to all the events. + forM_ evs $ reactToEvent queryDatabase + -- Update the global version number. + updateStartVersion queryDatabase $ startVersion + length evs + -- Done. + return () + +-- Query project list +qProjectList :: Database -> IO [(GUID, Text, Text)] +qProjectList queryDatabase = do + run_ $ EL.consume >>== + (enumQueryResult queryDatabase "SELECT guid, name, short_desc FROM projects" [] $= + (EL.map (\columns -> case columns of + [ SQLBlob guid, SQLText name, SQLText short_desc ] -> + (decode' guid , T.pack name, T.pack short_desc) + _ -> + error "Invalid columns in SQL query result"))) + +-- Query task list +qTaskList :: Database -> IO [(GUID, Text)] +qTaskList queryDatabase = do + run_ $ EL.consume >>== + (enumQueryResult queryDatabase "SELECT guid, short_desc FROM tasks;" [] $= + (EL.map (\columns -> case columns of + [ SQLBlob guid, SQLText short_desc ] -> + (decode' guid, T.pack short_desc) + _ -> + error "Invalid columns in SQL query result")))