diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/Setup.lhs b/Setup.lhs new file mode 100644 --- /dev/null +++ b/Setup.lhs @@ -0,0 +1,4 @@ +#! /usr/bin/env runhaskell + +> import Distribution.Simple +> main = defaultMain diff --git a/cqrs-postgresql.cabal b/cqrs-postgresql.cabal new file mode 100644 --- /dev/null +++ b/cqrs-postgresql.cabal @@ -0,0 +1,52 @@ +Name: cqrs-postgresql +Version: 0.9.0 +Synopsis: PostgreSQL backend for the cqrs package. +Description: PostgreSQL backend for the cqrs package. +License: MIT +License-file: LICENSE +Category: Data +Cabal-version: >=1.10 +Build-type: Simple +Author: Bardur Arantsson +Maintainer: Bardur Arantsson + +Library + build-depends: base == 4.* + , bytestring >= 0.9.0.1 + , bytestring-lexing >= 0.4 && <0.5 + , conduit >= 1.0 && < 2 + , cqrs-types >= 0.9.0 && < 0.10.0 + , old-locale >= 1.0 && < 1.1 + , pool-conduit >= 0.1 && < 0.2 + , postgresql-libpq >= 0.8.2 && <0.9 + , text >= 0.11 && < 0.12 + , time >= 1.2 && < 1.5 + , transformers >= 0.2 && < 0.4 + default-language: Haskell2010 + default-extensions: MultiParamTypeClasses + OverloadedStrings + ScopedTypeVariables + ghc-options: -Wall + hs-source-dirs: src + exposed-modules: Data.CQRS.EventStore.Backend.PostgreSQL + Data.CQRS.EventStore.Backend.PostgreSQLUtils + +Test-Suite cqrs-postgresql-tests + type: exitcode-stdio-1.0 + hs-source-dirs: src-test + main-is: Main.hs + build-depends: base == 4.* + , bytestring >= 0.9.0.1 + , conduit >= 0.5 && < 0.6 + , cqrs-types >= 0.9.0 && < 0.10.0 + , cqrs-test >= 0.9 && < 0.10.0 + , pool-conduit >= 0.1 && < 0.2 + , postgresql-libpq >= 0.7 && < 0.8 + -- Self-dependency + , cqrs-postgresql + -- Test framework: + , hspec >= 1.3 && < 2.0 + default-language: Haskell2010 + ghc-options: -Wall + default-extensions: MultiParamTypeClasses + OverloadedStrings 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,14 @@ +import qualified Database.PostgreSQL.LibPQ as P +import Test.Hspec (hspec) + +-- Import test suites. +import Data.CQRS.EventStore.Backend.PostgreSQLUtilsTest +import Data.CQRS.EventStore.Backend.PostgreSQLTest + +main :: IO () +main = do + let connString = "dbname=cqrstest" + conn <- P.connectdb connString + hspec $ do + runPostgreSQLUtilsTest conn + runPostgreSQLTest conn connString diff --git a/src/Data/CQRS/EventStore/Backend/PostgreSQL.hs b/src/Data/CQRS/EventStore/Backend/PostgreSQL.hs new file mode 100644 --- /dev/null +++ b/src/Data/CQRS/EventStore/Backend/PostgreSQL.hs @@ -0,0 +1,175 @@ +{-| Implementation of a PostgreSQL-based backend pool. -} +module Data.CQRS.EventStore.Backend.PostgreSQL + ( createBackendPool + ) where + +import Control.Monad (when, forM_, void) +import Data.ByteString (ByteString) +import Data.Conduit (ResourceT, Source, ($=), ($$), runResourceT) +import qualified Data.Conduit.List as CL +import Data.Conduit.Pool (Pool, createPool) +import Data.CQRS.EventStore.Backend (EventStoreBackend(..), RawEvent, RawSnapshot(..)) +import Data.CQRS.GUID (GUID) +import qualified Data.CQRS.GUID as G +import Data.CQRS.PersistedEvent (PersistedEvent(..)) +import Database.PostgreSQL.LibPQ (Connection) +import qualified Database.PostgreSQL.LibPQ as P + +import Data.CQRS.EventStore.Backend.PostgreSQLUtils + +-- SQL +createEventsSql :: ByteString +createEventsSql = "CREATE TABLE IF NOT EXISTS events ( guid BYTEA , ev_data BYTEA , version INTEGER , PRIMARY KEY (guid, version) );" + +selectEventsSql :: ByteString +selectEventsSql = "SELECT version, ev_data FROM events WHERE guid = $1 AND version >= $2 ORDER BY version ASC;" + +enumerateAllEventsSql :: ByteString +enumerateAllEventsSql = "SELECT guid, version, ev_data FROM events ORDER BY version ASC;" + +insertEventSql :: ByteString +insertEventSql = "INSERT INTO events ( guid, version, ev_data ) VALUES ($1, $2, $3);" + +insertVersionSql :: ByteString +insertVersionSql = "INSERT INTO versions ( guid , version ) SELECT $1 , $2 WHERE $1 NOT IN ( SELECT guid FROM versions );" + +insertSnapshotSql :: ByteString +insertSnapshotSql = "INSERT INTO snapshots ( guid, data, version) SELECT $1 , $2, $3 WHERE $1 NOT IN ( SELECT guid FROM snapshots );" + +createAggregateVersionsSql :: ByteString +createAggregateVersionsSql = "CREATE TABLE IF NOT EXISTS versions ( guid BYTEA PRIMARY KEY , version INTEGER );" + +getCurrentVersionSql :: ByteString +getCurrentVersionSql = "SELECT version FROM versions WHERE guid = $1;" + +updateCurrentVersionSql :: ByteString +updateCurrentVersionSql = "UPDATE versions SET version = $1 WHERE guid = $2;" + +createSnapshotSql :: ByteString +createSnapshotSql = "CREATE TABLE IF NOT EXISTS snapshots ( guid BYTEA PRIMARY KEY , data BYTEA , version INTEGER );" + +writeSnapshotSql :: ByteString +writeSnapshotSql = "UPDATE snapshots SET data=$1, version=$2 WHERE guid=$3;" + +selectSnapshotSql :: ByteString +selectSnapshotSql = "SELECT data, version FROM snapshots WHERE guid = $1;" + +badQueryResultMsg :: [String] -> [SqlValue] -> String +badQueryResultMsg params columns = concat ["Invalid query result shape. Params: ", show params, ". Result columns: ", 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, ")" + ] + +sqlGuid :: GUID -> SqlValue +sqlGuid g = SqlByteArray (Just $ G.toByteString g) + +storeEvents :: Connection -> GUID -> Int -> [RawEvent] -> IO () +storeEvents c guid originatingVersion events = do + -- Column unpacking. + let unpackColumns [ SqlInt32 (Just v) ] = fromIntegral v + unpackColumns columns = error $ badQueryResultMsg [show guid] columns + -- Get the current version number of the aggregate. + curVer <- runResourceT $ (sourceQuery c getCurrentVersionSql [sqlGuid guid]) $$ CL.fold (\x -> max x . unpackColumns) 0 + + -- Sanity check current version number. + when (curVer /= originatingVersion) $ + versionConflict originatingVersion curVer + + -- Update de-normalized version number. + let version' = originatingVersion + length events + void $ run c updateCurrentVersionSql + [ SqlInt32 $ Just $ fromIntegral $ version' + , sqlGuid guid + ] + void $ run c insertVersionSql + [ sqlGuid guid + , SqlInt32 $ Just $ fromIntegral $ version' + ] + + -- Store the supplied events. + forM_ events $ \e -> do + void $ run c insertEventSql + [ sqlGuid guid + , SqlInt32 $ Just $ fromIntegral $ peSequenceNumber e + , SqlByteArray $ Just $ peEvent e + ] + + +retrieveEvents :: Connection -> GUID -> Int -> Source (ResourceT IO) RawEvent +retrieveEvents connection guid v0 = do + -- Unpack the columns into tuples. + let unpackColumns [ SqlInt32 (Just version) + , SqlByteArray (Just eventData) + ] = PersistedEvent guid eventData (fromIntegral version) + unpackColumns columns = error $ badQueryResultMsg [show guid, show v0] columns + -- Find events with version numbers. + sourceQuery connection selectEventsSql + [sqlGuid guid, SqlInt32 $ Just $ fromIntegral v0] $= CL.map unpackColumns + +enumerateAllEvents :: Connection -> Source (ResourceT IO) RawEvent +enumerateAllEvents connection = do + sourceQuery connection enumerateAllEventsSql [] $= CL.map + (\columns -> do + case columns of + [ SqlByteArray (Just g), SqlInt32 (Just v), SqlByteArray (Just ed)] -> + PersistedEvent (G.fromByteString g) ed (fromIntegral v) + _ -> + error $ badQueryResultMsg [] columns) + +writeSnapshot :: Connection -> GUID -> RawSnapshot -> IO () +writeSnapshot c guid (RawSnapshot v d) = do + void $ run c writeSnapshotSql + [ SqlByteArray (Just d) + , SqlInt32 $ Just $ fromIntegral v + , sqlGuid guid + ] + void $ run c insertSnapshotSql + [ sqlGuid guid + , SqlByteArray (Just d) + , SqlInt32 $ Just $ fromIntegral v + ] + +getLatestSnapshot :: Connection -> GUID -> IO (Maybe RawSnapshot) +getLatestSnapshot connection guid = do + -- Unpack columns from result. + let unpackColumns :: [SqlValue] -> (ByteString, Int) + unpackColumns [ SqlByteArray (Just d) + , SqlInt32 (Just v) ] = (d, fromIntegral v) + unpackColumns columns = error $ badQueryResultMsg [show guid] columns + -- Run the query. + r <- runResourceT $ (sourceQuery connection selectSnapshotSql [sqlGuid guid] $= (CL.map unpackColumns)) $$ CL.take 1 + case r of + ((d,v):_) -> return $ Just $ RawSnapshot v d + [] -> return Nothing + +-- | PostgreSQL backend +newtype PostgreSQLEventStoreBackend = ESB Connection + +-- | Instance of EventStoreBackend for PostgreSQLBackend. +instance EventStoreBackend PostgreSQLEventStoreBackend where + esbStoreEvents (ESB c) = storeEvents c + esbRetrieveEvents (ESB c) = retrieveEvents c + esbEnumerateAllEvents (ESB c) = enumerateAllEvents c + esbWriteSnapshot (ESB c) = writeSnapshot c + esbGetLatestSnapshot (ESB c) = getLatestSnapshot c + esbWithTransaction (ESB c) = withTransaction c + +-- | Create a pool of PostgreSQL-based event store backends. +createBackendPool :: Int -> ByteString -> IO (Pool PostgreSQLEventStoreBackend) +createBackendPool n connectionString = do + createPool open close 1 1 n + where + open = do + -- Connect + c <- P.connectdb connectionString + -- Set up tables if necessary. + void $ run c createEventsSql [] + void $ run c createAggregateVersionsSql [] + void $ run c createSnapshotSql [] + -- Return backend + return $ ESB c + close (ESB c) = do + P.finish c diff --git a/src/Data/CQRS/EventStore/Backend/PostgreSQLUtils.hs b/src/Data/CQRS/EventStore/Backend/PostgreSQLUtils.hs new file mode 100644 --- /dev/null +++ b/src/Data/CQRS/EventStore/Backend/PostgreSQLUtils.hs @@ -0,0 +1,161 @@ +{-| Implementation of a PostgreSQL-based event store. -} +module Data.CQRS.EventStore.Backend.PostgreSQLUtils + ( run + , sourceQuery + , withTransaction + , SqlValue(..) + ) where + +import Control.Monad (forM) +import Control.Monad.IO.Class (liftIO) +import Control.Exception (catch, onException, SomeException) +import qualified Data.ByteString.Char8 as B8 +import Data.ByteString (ByteString) +import qualified Data.ByteString as B +import Data.ByteString.Lex.Integral (readDecimal) +import Data.ByteString.Lex.Double (readDouble) +import Data.Conduit (ResourceT, Source, ($$), bracketP, yield) +import qualified Data.Conduit as C +import qualified Data.Conduit.List as CL +import Data.Int (Int16, Int32, Int64) +import Data.Text (Text) +import Data.Text.Encoding (decodeUtf8', encodeUtf8) +import Data.Time (Day) +import Data.Time.Format (parseTime) +import Database.PostgreSQL.LibPQ (Connection, Oid(..), Format(..), ExecStatus(..), Column(..), Row(..)) +import qualified Database.PostgreSQL.LibPQ as P +import System.Locale (defaultTimeLocale) + +-- | Known field types. +data SqlValue = SqlByteArray (Maybe ByteString) + | SqlBlankPaddedString (Maybe ByteString) + | SqlBool (Maybe Bool) + | SqlInt16 (Maybe Int16) + | SqlInt32 (Maybe Int32) + | SqlInt64 (Maybe Int64) + | SqlFloating (Maybe Double) + | SqlVarChar (Maybe Text) + | SqlText (Maybe Text) + | SqlDate (Maybe Day) + | Unmatched (Oid, Maybe ByteString) + deriving (Eq, Show) + +-- | Execute an IO action with an active transaction. +withTransaction :: Connection -> IO a -> IO a +withTransaction connection action = do + begin + onException runAction tryRollback + where + runAction = do + r <- action + commit + return r + + tryRollback = + -- Try rollback while discarding exception; we want to preserve + -- original exception. + catch rollback (\(_::SomeException) -> return ()) + + begin = run connection "BEGIN TRANSACTION;" [] + commit = run connection "COMMIT TRANSACTION;" [] + rollback = run connection "ROLLBACK TRANSACTION;" [] + +-- | Read a boolean. +readBoolean :: ByteString -> Maybe Bool +readBoolean "t" = Just True +readBoolean "f" = Just False +readBoolean _ = Nothing + +-- | Map an SqlValue to a parameter. +fromSqlValue :: Connection -> SqlValue -> IO (Maybe (Oid, ByteString, Format)) +fromSqlValue connection (SqlByteArray a) = do + case a of + Nothing -> return Nothing + Just a' -> do + x <- P.escapeByteaConn connection a' + case x of + Nothing -> error "Conversion failed" + Just x' -> return $ Just (Oid 17, x', Text) +fromSqlValue _ (SqlBool (Just True)) = return $ Just (Oid 16, "t", Text) +fromSqlValue _ (SqlBool (Just False)) = return $ Just (Oid 16, "f", Text) +fromSqlValue _ (SqlBool Nothing) = return Nothing +fromSqlValue _ (SqlInt32 Nothing) = return Nothing +fromSqlValue _ (SqlInt32 (Just i)) = return $ Just (Oid 23, B8.pack (show i), Text) +fromSqlValue _ (SqlVarChar Nothing) = return Nothing +fromSqlValue _ (SqlVarChar (Just t)) = return $ Just (Oid 1043, encodeUtf8 t, Binary) +fromSqlValue _ (SqlText Nothing) = return Nothing +fromSqlValue _ (SqlText (Just t)) = return $ Just (Oid 25, encodeUtf8 t, Text) +fromSqlValue _ (SqlDate Nothing) = return Nothing +fromSqlValue _ (SqlDate (Just d)) = return $ Just (Oid 1082, B8.pack (show d), Text) +fromSqlValue _ _ = error "Parameter conversion failed" + +-- | Map field to an SqlValue. +toSqlValue :: (Oid, Maybe ByteString) -> IO SqlValue +toSqlValue (oid, mvalue) = + case oid of + Oid 17 -> c P.unescapeBytea SqlByteArray + Oid 16 -> c (return . readBoolean) SqlBool + Oid 20 -> c (return . fmap fst . readDecimal) SqlInt64 + Oid 21 -> c (return . fmap fst . readDecimal) SqlInt16 + Oid 23 -> c (return . fmap fst . readDecimal) SqlInt32 + Oid 25 -> c (return . either (const Nothing) Just . decodeUtf8') SqlText + Oid 700 -> c (return . fmap fst . readDouble) SqlFloating + Oid 701 -> c (return . fmap fst . readDouble) SqlFloating + Oid 1042 -> c (return . Just) SqlBlankPaddedString + Oid 1043 -> c (return . either (const Nothing) Just . decodeUtf8') SqlVarChar + Oid 1082 -> c (return . parseTime defaultTimeLocale "%F" . B8.unpack) SqlDate + + _ -> return $ Unmatched (oid,mvalue) + where + c :: Monad m => (ByteString -> m (Maybe a)) -> (Maybe a -> SqlValue) -> m SqlValue + c convert construct = + case mvalue of + Nothing -> return $ construct Nothing + Just value -> do + mvalue' <- convert value + case mvalue' of + Nothing -> fail "Conversion failed" + Just _ -> return $ construct mvalue' + +-- | Execute a query with no result. +run :: Connection -> ByteString -> [SqlValue] -> IO () +run connection sql parameters = do + C.runResourceT (sourceQuery connection sql parameters $$ CL.sinkNull) + +-- | Source for traversing all the results of a PostgreSQL query. +sourceQuery :: Connection -> ByteString -> [SqlValue] -> Source (ResourceT IO) [SqlValue] +sourceQuery connection sql parameters = + bracketP open done go + where + done r = P.unsafeFreeResult r + open = do + parameters' <- forM parameters $ fromSqlValue connection + mr <- P.execParams connection sql parameters' Text + case mr of + Nothing -> error "No result" + Just r -> do + status <- P.resultStatus r + case status of + CommandOk -> return r + TuplesOk -> return r + _ -> do + statusMessage <- P.resStatus status + errorMessage <- fmap (maybe "(Unknown error)" id) $ P.resultErrorMessage r + error $ B8.unpack $ B.concat [statusMessage, ": ", errorMessage] + + go r = do + Col nFields <- liftIO $ P.nfields r + Row nRows <- liftIO $ P.ntuples r + let rows = map P.toRow [0..nRows-1] + let columns = map P.toColumn [0..nFields-1] + loop columns rows + where + loop _ [] = return () + loop columns (row:rows) = do + (forM columns $ (liftIO . getSqlVal r row)) >>= yield + loop columns rows + + getSqlVal r row c = do + mval <- P.getvalue' r row c + typ <- P.ftype r c + toSqlValue (typ,mval)