cqrs-example-0.6.0: src/Query.hs

module Query ( initializeQueryDatabase
             , updateQueryStateRef
             , qProjectList
             , qTaskList
             ) where

import           Control.Monad (forM_, liftM)
import           Data.ByteString (ByteString)
import           Data.CQRS (enumerateEventStore, GUID, EventStore)
import           Data.CQRS.EventStore.Backend.Sqlite3Utils (withTransaction, execSql, enumQueryResult)
import           Data.Enumerator (run_, (>>==), ($=))
import qualified Data.Enumerator.List as EL
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(..))

-- TODO: Export Data.CQRS.Serialize decode' instead?
decode' :: Serialize a => ByteString -> a
decode' = either error id . decode

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

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

    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")))