cqrs-postgresql (empty) → 0.9.0
raw patch · 6 files changed
+425/−0 lines, 6 filesdep +basedep +bytestringdep +bytestring-lexingsetup-changed
Dependencies added: base, bytestring, bytestring-lexing, conduit, cqrs-postgresql, cqrs-test, cqrs-types, hspec, old-locale, pool-conduit, postgresql-libpq, text, time, transformers
Files
- LICENSE +19/−0
- Setup.lhs +4/−0
- cqrs-postgresql.cabal +52/−0
- src-test/Main.hs +14/−0
- src/Data/CQRS/EventStore/Backend/PostgreSQL.hs +175/−0
- src/Data/CQRS/EventStore/Backend/PostgreSQLUtils.hs +161/−0
+ 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-postgresql.cabal view
@@ -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 <bardur@scientician.net>++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
+ src-test/Main.hs view
@@ -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
+ src/Data/CQRS/EventStore/Backend/PostgreSQL.hs view
@@ -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
+ src/Data/CQRS/EventStore/Backend/PostgreSQLUtils.hs view
@@ -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)