diff --git a/LICENSE b/LICENSE --- a/LICENSE +++ b/LICENSE @@ -1,4 +1,4 @@ -Copyright (c) 2011 Bardur Arantsson +Copyright (c) 2011-2015 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 diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/Setup.lhs b/Setup.lhs deleted file mode 100644 --- a/Setup.lhs +++ /dev/null @@ -1,4 +0,0 @@ -#! /usr/bin/env runhaskell - -> import Distribution.Simple -> main = defaultMain diff --git a/cqrs-postgresql.cabal b/cqrs-postgresql.cabal --- a/cqrs-postgresql.cabal +++ b/cqrs-postgresql.cabal @@ -1,5 +1,5 @@ Name: cqrs-postgresql -Version: 0.9.0 +Version: 0.10.0 Synopsis: PostgreSQL backend for the cqrs package. Description: PostgreSQL backend for the cqrs package. License: MIT @@ -11,42 +11,49 @@ Maintainer: Bardur Arantsson Library - build-depends: base == 4.* + build-depends: base >= 4.8 && < 5 , 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 + , bytestring-lexing >= 0.4 && < 0.5 + , cqrs-core >= 0.10.0 && < 0.11.0 + , deepseq >= 1.4 && < 2 + , enclosed-exceptions >= 1.0.1 && < 2 + , io-streams >= 1.2 && < 2 + , resource-pool >= 0.2.1 && < 0.3 + , postgresql-libpq >= 0.9.0.1 && < 0.10 + , text >= 1.0 && < 2 + , time >= 1.5 && < 1.6 + , transformers >= 0.4.1 && < 0.5 + , uuid-types >= 1.0 && < 1.1 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 + exposed-modules: Data.CQRS.PostgreSQL + Data.CQRS.PostgreSQL.Migrations + Data.CQRS.PostgreSQL.Internal.ArchiveStore + Data.CQRS.PostgreSQL.Internal.EventStore + Data.CQRS.PostgreSQL.Internal.Migration + Data.CQRS.PostgreSQL.Internal.SnapshotStore + Data.CQRS.PostgreSQL.Internal.Utils -Test-Suite cqrs-postgresql-tests +Test-Suite 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 + , cqrs-core >= 0.10.0 && < 0.11.0 + , cqrs-testkit >= 0.10.0 && < 0.11.0 + , io-streams >= 1.2 && < 2 + , pg-harness-client >= 0.4.0 && < 0.5.0 + , postgresql-libpq >= 0.9.0.1 && < 0.10 + , random >= 1.1 && < 2 + , resource-pool >= 0.2.1 && < 0.3 + , uuid-types >= 1.0 && < 1.1 -- Self-dependency , cqrs-postgresql -- Test framework: - , hspec >= 1.3 && < 2.0 + , hspec >= 2.2.0 && < 3.0 default-language: Haskell2010 ghc-options: -Wall - default-extensions: MultiParamTypeClasses - OverloadedStrings + other-modules: Data.CQRS.PostgreSQL.Internal.UtilsSpec + Data.CQRS.PostgreSQL.Internal.MigrationSpec diff --git a/src-test/Data/CQRS/PostgreSQL/Internal/MigrationSpec.hs b/src-test/Data/CQRS/PostgreSQL/Internal/MigrationSpec.hs new file mode 100644 --- /dev/null +++ b/src-test/Data/CQRS/PostgreSQL/Internal/MigrationSpec.hs @@ -0,0 +1,88 @@ +{-# LANGUAGE OverloadedStrings #-} +module Data.CQRS.PostgreSQL.Internal.MigrationSpec + ( mkApplyMigrationsSpec + ) where + +import Control.Exception (bracket) +import Data.ByteString (ByteString) +import Data.CQRS.PostgreSQL.Internal.Utils (runQuery) +import Data.CQRS.PostgreSQL.Internal.Migration (applyMigrations, uuid) +import Data.Pool (Pool, withResource, destroyAllResources) +import Data.UUID.Types (UUID) +import Database.PostgreSQL.LibPQ (Connection) +import Test.Hspec + +-- Tests for Data.CQRS.PostgreSQL.Internal.Migration +mkApplyMigrationsSpec :: IO (Pool Connection) -> Spec +mkApplyMigrationsSpec mkConnectionPool = do + describe "applyMigrations" $ do + it "can apply a single migration" $ withConnectionPool $ \connectionPool -> do + -- Apply the migration + applyMigrations' connectionPool + [ (uuid0, createXSql) + ] + -- Do a query which would fail without the migration + assertValidQuery connectionPool selectFromX + + it "ignores migrations that have already been applied (single call)" $ withConnectionPool $ \connectionPool -> do + -- Apply the migrations + applyMigrations' connectionPool + [ (uuid0, createXSql) + , (uuid0, createXSql) -- Would fail if applied + ] + -- Do a query which would fail without at least one of the migrations being applied + assertValidQuery connectionPool selectFromX + + it "ignores migrations that have already been applied (multiple calls)" $ withConnectionPool $ \connectionPool -> do + -- Apply first migration + applyMigrations' connectionPool + [ (uuid0, createXSql) + ] + -- Apply second migration + applyMigrations' connectionPool + [ (uuid0, createXSql) + ] + -- Do a query which would fail without at least one of the migrations being applied + assertValidQuery connectionPool selectFromX + + it "throws an error if SQL is changed for a given change set ID" $ withConnectionPool $ \connectionPool -> do + -- Apply first migration + applyMigrations' connectionPool + [ (uuid0, createXSql) + ] + -- Apply second migration, which SHOULD fail because the SQL is different + applyMigrations' connectionPool + [ (uuid0, createXSqlBad) + ] `shouldThrow` anyException + + it "can apply multiple distinct migrations in a single call" $ withConnectionPool $ \connectionPool -> do + -- Apply both migrations + applyMigrations' connectionPool + [ (uuid0, createXSql) + , (uuid1, createYSql) + ] + -- Do a query which would fail without at least one of the migrations being applied + assertValidQuery connectionPool joinXY + + where + withConnectionPool spec = + bracket mkConnectionPool destroyAllResources spec + + assertValidQuery connectionPool sql = do + rows <- runQuery connectionPool sql [] + rows `shouldSatisfy` (\rs -> length rs >= 0) -- Don't care about size of result, just that query succeeded + + createXSql = "CREATE TABLE X (A INT)" + createYSql = "CREATE TABLE Y (B INT)" + createXSqlBad = "CREATE TABLE X (Y CHAR(1))" + + selectFromX = "SELECT * FROM X" + joinXY = "SELECT * FROM X, Y where X.A = Y.B" + + uuid0 = uuid "a328156d-9875-4471-8192-0c86959badb3" + uuid1 = uuid "00c6159c-c7f6-4cec-b63f-f70c1c4c7bb1" + +applyMigrations' :: Pool Connection -> [(UUID, ByteString)] -> IO () +applyMigrations' connectionPool migrations = do + withResource connectionPool $ \connection -> + applyMigrations connection migrations diff --git a/src-test/Data/CQRS/PostgreSQL/Internal/UtilsSpec.hs b/src-test/Data/CQRS/PostgreSQL/Internal/UtilsSpec.hs new file mode 100644 --- /dev/null +++ b/src-test/Data/CQRS/PostgreSQL/Internal/UtilsSpec.hs @@ -0,0 +1,39 @@ +{-# LANGUAGE OverloadedStrings #-} +module Data.CQRS.PostgreSQL.Internal.UtilsSpec + ( mkUtilsSpec + ) where + +import Control.Exception (bracket) +import Data.ByteString (isInfixOf) +import Data.CQRS.PostgreSQL.Internal.Utils (SqlValue(..), QueryError(..), runQuery) +import Data.Pool (Pool, destroyAllResources) +import Database.PostgreSQL.LibPQ (Connection) +import Test.Hspec + +-- Tests for Data.CQRS.PostgreSQL.Internal.Utils +mkUtilsSpec :: IO (Pool Connection) -> Spec +mkUtilsSpec mkConnectionPool = do + describe "single-result query" $ do + it "produces a result (non-parametric)" $ withConnectionPool $ \connectionPool -> do + x <- runQuery connectionPool "SELECT TRUE" [] + length x `shouldBe` 1 + it "produces a result (parametric)" $ withConnectionPool $ \connectionPool -> do + x <- runQuery connectionPool "SELECT TRUE WHERE $1" [SqlBool $ Just True] + length x `shouldBe` 1 + it "produces no results if unsatisfiable" $ withConnectionPool $ \connectionPool -> do + x <- runQuery connectionPool "SELECT TRUE WHERE FALSE" [] + length x `shouldBe` 0 + it "throws a QueryError when invalid SQL statement is executed" $ withConnectionPool $ \connectionPool -> do + runQuery connectionPool "MY BAD QUERY" [] `shouldThrow` (\e -> + case e of + QueryError (Just "42601") "PGRES_FATAL_ERROR" (Just msg) | "MY BAD QUERY" `isInfixOf` msg -> + True + _ -> + False) + describe "multi-result query" $ do + it "produces the correct number of rows" $ withConnectionPool $ \connectionPool -> do + x <- runQuery connectionPool "SELECT GENERATE_SERIES(1,5)" [] + length x `shouldBe` 5 + where + withConnectionPool test = + bracket mkConnectionPool destroyAllResources test diff --git a/src-test/Main.hs b/src-test/Main.hs --- a/src-test/Main.hs +++ b/src-test/Main.hs @@ -1,14 +1,54 @@ -import qualified Database.PostgreSQL.LibPQ as P -import Test.Hspec (hspec) +module Main ( main ) where --- Import test suites. -import Data.CQRS.EventStore.Backend.PostgreSQLUtilsTest -import Data.CQRS.EventStore.Backend.PostgreSQLTest +import qualified Database.PostgreSQL.LibPQ as P +import Data.CQRS.PostgreSQL (newEventStore, newSnapshotStore, newArchiveStore) +import Data.CQRS.PostgreSQL.Migrations (migrate) +import Data.CQRS.PostgreSQL.Internal.UtilsSpec (mkUtilsSpec) +import Data.CQRS.PostgreSQL.Internal.MigrationSpec (mkApplyMigrationsSpec) +import Data.CQRS.Test.TestKit (mkArchiveStoreSpec, mkEventStoreSpec, mkRepositorySpec, mkSnapshotStoreSpec, TestKitSettings(..)) +import Data.Pool (createPool, withResource, destroyAllResources) +import qualified Database.PostgreSQL.Harness.Client as H +import System.Random (randomIO) +import Test.Hspec main :: IO () main = do - let connString = "dbname=cqrstest" - conn <- P.connectdb connString + -- HSpec has no easy way to get "other" command line parameters, so + -- we'll just settle for a hardcoded value here. + let url = "http://localhost:8900" + -- Connection pool creation function. We use a fresh temporary + -- database for every connection pool. + let mkConnectionPool = do + connectionString <- fmap H.toConnectionString $ H.createTemporaryDatabase url + createPool (P.connectdb connectionString) P.finish 1 1 5 + -- Setup for TestKit + let testKitSettings = TestKitSettings + { tksMakeContext = \_ -> return () + , tksSetUp = do + connectionPool <- mkConnectionPool + withResource connectionPool migrate + return connectionPool + , tksTearDown = destroyAllResources + } + -- Run the tests hspec $ do - runPostgreSQLUtilsTest conn - runPostgreSQLTest conn connString + mkUtilsSpec mkConnectionPool + mkApplyMigrationsSpec mkConnectionPool + mkArchiveStoreSpec $ testKitSettings { + tksMakeContext = \c -> do + as <- newArchiveStore randomIO c + es <- newEventStore c + return (as, es) + } + mkSnapshotStoreSpec $ testKitSettings { + tksMakeContext = newSnapshotStore + } + mkEventStoreSpec $ testKitSettings { + tksMakeContext = newEventStore + } + mkRepositorySpec $ testKitSettings { + tksMakeContext = \c -> do + es <- newEventStore c + ss <- newSnapshotStore c + return (es, ss) + } diff --git a/src/Data/CQRS/EventStore/Backend/PostgreSQL.hs b/src/Data/CQRS/EventStore/Backend/PostgreSQL.hs deleted file mode 100644 --- a/src/Data/CQRS/EventStore/Backend/PostgreSQL.hs +++ /dev/null @@ -1,175 +0,0 @@ -{-| 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 deleted file mode 100644 --- a/src/Data/CQRS/EventStore/Backend/PostgreSQLUtils.hs +++ /dev/null @@ -1,161 +0,0 @@ -{-| 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) diff --git a/src/Data/CQRS/PostgreSQL.hs b/src/Data/CQRS/PostgreSQL.hs new file mode 100644 --- /dev/null +++ b/src/Data/CQRS/PostgreSQL.hs @@ -0,0 +1,9 @@ +module Data.CQRS.PostgreSQL + ( newArchiveStore + , newEventStore + , newSnapshotStore + ) where + +import Data.CQRS.PostgreSQL.Internal.ArchiveStore (newArchiveStore) +import Data.CQRS.PostgreSQL.Internal.EventStore (newEventStore) +import Data.CQRS.PostgreSQL.Internal.SnapshotStore (newSnapshotStore) diff --git a/src/Data/CQRS/PostgreSQL/Internal/ArchiveStore.hs b/src/Data/CQRS/PostgreSQL/Internal/ArchiveStore.hs new file mode 100644 --- /dev/null +++ b/src/Data/CQRS/PostgreSQL/Internal/ArchiveStore.hs @@ -0,0 +1,159 @@ +{-# LANGUAGE OverloadedStrings #-} +module Data.CQRS.PostgreSQL.Internal.ArchiveStore + ( newArchiveStore + ) where + +import Control.Monad ((>=>)) +import Data.ByteString (ByteString) +import Data.Pool (Pool, withResource) +import Data.CQRS.Types.ArchiveRef (ArchiveRef(..)) +import Data.CQRS.Types.ArchiveStore (ArchiveStore(..), ArchiveMetadata(..)) +import Data.CQRS.Types.PersistedEvent (PersistedEvent(..)) +import Data.CQRS.PostgreSQL.Internal.Utils (execSql, ioQuery, SqlValue(..), withTransaction, badQueryResultMsg) +import Data.UUID.Types (UUID) +import Database.PostgreSQL.LibPQ (Connection) +import System.IO.Streams (InputStream) +import qualified System.IO.Streams as Streams +import qualified System.IO.Streams.Combinators as SC + +unpackArchiveMetadata :: [SqlValue] -> ArchiveMetadata +unpackArchiveMetadata [ SqlUUID (Just archiveId) + , SqlUUID maybePreviousArchiveId + , SqlUUID maybeNextArchiveId ] = ArchiveMetadata + { amArchiveId = archiveId + , amPreviousArchiveId = maybePreviousArchiveId + , amNextArchiveId = maybe CurrentArchive NamedArchive maybeNextArchiveId + } +unpackArchiveMetadata columns = + error $ badQueryResultMsg [ ] columns + +getUnarchivedEventCount :: Pool Connection -> IO Int +getUnarchivedEventCount cp = withResource cp $ \c -> do + maybeCount <- withTransaction c $ do + ioQuery c sqlUnarchivedCount [ ] $ SC.map unpackCount >=> Streams.read + case maybeCount of + Nothing -> + error "Query error counting number of unarchived events" + Just count -> + return count + where + unpackCount [ SqlInt32 (Just count) ] = fromIntegral count + unpackCount [ SqlInt64 (Just count) ] = fromIntegral count + unpackCount columns = error $ badQueryResultMsg [ ] columns + + sqlUnarchivedCount = + "SELECT COUNT(*) \ + \ FROM event \ + \ WHERE archive_uuid IS NULL" + +archiveEvents :: Pool Connection -> IO UUID -> Int -> IO (Maybe UUID) +archiveEvents cp uuidSupply archiveSize = + if (archiveSize > 0) then + withResource cp $ \c -> do + withTransaction c $ do + archiveId <- uuidSupply + -- Create new archive + execSql c sqlInsertNewCurrentArchive [ SqlUUID $ Just archiveId ] + execSql c sqlUpdatePreviousCurrentArchive [ ] + -- Move events which don't have an archive into the new archive + execSql c sqlFillNextArchive [ SqlUUID $ Just archiveId, SqlInt32 $ Just $ fromIntegral archiveSize ] + -- Return its ID + return $ Just archiveId + else + return Nothing + + where + + sqlInsertNewCurrentArchive = + -- Note that we use a RIGHT OUTER join here to ensure that this also works in the + -- case where there are no pre-existing archives. + "INSERT INTO archive (archive_uuid, prev_archive_uuid, next_archive_uuid) \ + \ SELECT $1, A.archive_uuid, NULL \ + \ FROM archive A RIGHT OUTER JOIN (SELECT 1) AS B \ + \ ON A.next_archive_uuid IS NULL" + + sqlUpdatePreviousCurrentArchive = + -- Update the previous archive to point to the new + -- archive as its "next" archive. This is a bit more + -- complicated then just setting the next_archive_uuid for the + -- archive with the next_archive_uuid=NULL because we've just + -- inserted a new archive with next_archive_uuid=NULL so now we + -- have TWO archives with next_archive_uuid=NULL. The nested + -- select takes care of finding the proper row. + "UPDATE archive A0 \ + \ SET next_archive_uuid = (SELECT archive_uuid \ + \ FROM archive AS A1 \ + \ WHERE A1.prev_archive_uuid = A0.archive_uuid) \ + \ WHERE next_archive_uuid IS NULL" + + sqlFillNextArchive = + "UPDATE event \ + \ SET archive_uuid = $1 \ + \ WHERE (aggregate_uuid, seq_no) IN \ + \ ( SELECT aggregate_uuid, seq_no \ + \ FROM event \ + \ WHERE archive_uuid IS NULL \ + \ ORDER BY aggregate_uuid, seq_no \ + \ LIMIT $2)" + +readLatestArchiveMetadata :: Pool Connection -> IO (Maybe ArchiveMetadata) +readLatestArchiveMetadata cp = + withResource cp $ \c -> withTransaction c $ do + ioQuery c sqlReadLatestArchiveMetadata [ ] $ SC.map unpackArchiveMetadata >=> Streams.read + where + sqlReadLatestArchiveMetadata = + "SELECT archive_uuid, prev_archive_uuid, next_archive_uuid \ + \ FROM archive \ + \ WHERE next_archive_uuid IS NULL" + +readArchiveMetadata :: Pool Connection -> UUID -> IO (Maybe ArchiveMetadata) +readArchiveMetadata cp archiveId = + withResource cp $ \c -> withTransaction c $ do + ioQuery c sqlReadArchiveMetadata [ SqlUUID $ Just archiveId ] $ SC.map unpackArchiveMetadata >=> Streams.read + where + sqlReadArchiveMetadata = + "SELECT archive_uuid, prev_archive_uuid, next_archive_uuid \ + \ FROM archive \ + \ WHERE archive_uuid = $1" + +readArchive :: Pool Connection -> ArchiveRef -> (InputStream (UUID, PersistedEvent ByteString) -> IO a) -> IO a +readArchive cp archiveRef p = + withResource cp $ \c -> withTransaction c $ do + query c archiveRef $ SC.map unpack >=> p + where + -- We use two different queries rather than using "IS NOT DISTINCT + -- FROM" because the query optimization engine in some (all?) + -- PostgreSQL versions has a lot of trouble optimizing this and + -- will do table scans. + query c CurrentArchive = ioQuery c sqlReadArchiveCurrent [ ] + query c (NamedArchive archiveId) = ioQuery c sqlReadArchiveNamed [ SqlUUID $ Just archiveId ] + -- Unpack result columns + unpack [ SqlUUID (Just aggregateId) + , SqlUUID (Just eventId) + , SqlByteArray (Just eventData) + , SqlInt32 (Just sequenceNumber) + ] = (aggregateId, PersistedEvent eventData (fromIntegral sequenceNumber) eventId) + unpack columns = error $ badQueryResultMsg [show archiveRef] columns + -- SQL + sqlReadArchiveCurrent = + " SELECT aggregate_uuid, event_uuid, event_data, seq_no \ + \ FROM event \ + \ WHERE archive_uuid IS NULL \ + \ORDER BY aggregate_uuid, seq_no ASC" + sqlReadArchiveNamed = + " SELECT aggregate_uuid, event_uuid, event_data, seq_no \ + \ FROM event \ + \ WHERE archive_uuid = $1 \ + \ORDER BY aggregate_uuid, seq_no ASC" + +-- | Create an archive store backed by a PostgreSQL connection pool. +newArchiveStore :: IO UUID -> Pool Connection -> IO (ArchiveStore ByteString) +newArchiveStore uuidSupply connectionPool = do + return $ ArchiveStore + { asGetUnarchivedEventCount = getUnarchivedEventCount connectionPool + , asArchiveEvents = archiveEvents connectionPool uuidSupply + , asReadLatestArchiveMetadata = readLatestArchiveMetadata connectionPool + , asReadArchiveMetadata = readArchiveMetadata connectionPool + , asReadArchive = readArchive connectionPool + } + diff --git a/src/Data/CQRS/PostgreSQL/Internal/EventStore.hs b/src/Data/CQRS/PostgreSQL/Internal/EventStore.hs new file mode 100644 --- /dev/null +++ b/src/Data/CQRS/PostgreSQL/Internal/EventStore.hs @@ -0,0 +1,112 @@ +{-# LANGUAGE OverloadedStrings #-} +module Data.CQRS.PostgreSQL.Internal.EventStore + ( newEventStore + ) where + +import Control.Exception (throw, catchJust) +import Control.Monad (forM_, (>=>)) +import Data.ByteString (ByteString) +import Data.Pool (Pool, withResource) +import Data.CQRS.Types.EventStore (EventStore(..), StoreError(..)) +import Data.CQRS.Types.PersistedEvent (PersistedEvent(..)) +import Data.CQRS.PostgreSQL.Internal.Utils (execSql, ioQuery, SqlValue(..), isDuplicateKey, withTransaction, badQueryResultMsg) +import Data.Time.Clock.POSIX (getPOSIXTime) +import Data.UUID.Types (UUID) +import Database.PostgreSQL.LibPQ (Connection) +import System.IO.Streams (InputStream) +import qualified System.IO.Streams.Combinators as SC + +-- Store events for a given aggregate. We do not have a separate table +-- storing aggregate (UUID -> version) mappings, which would +-- ordinarily be required to avoid the potential for version "gaps" +-- caused by phantom reads (at any non-SERIALIZABLE isolation level). +-- The scenario would be this: Thread A loads an aggregate and +-- generates some events. Before A commits, but after it has inserted +-- the events into the "event" table, thread B comes along and loads +-- the same aggregate, sees A's events and begins to append events +-- (continuing from A's last sequence number). When A comes to commit +-- it fails for some reason, but B's commit succeeds (since it is pure +-- inserts there's no data dependency on A to prevent it from +-- committing). Thus we would end up with a gap in the version +-- numbers, not to mention that B may have depended semantically on +-- A's events. However, in PostgreSQL the initial read that B +-- performs cannot see A's events because READ COMMITTED doesn't +-- permit it, even if the events were inserts. +storeEvents :: Pool Connection -> UUID -> [PersistedEvent ByteString] -> IO () +storeEvents cp aggregateId es = do + withResource cp $ \c -> do + translateExceptions aggregateId $ do + withTransaction c $ do + forM_ es $ \e -> do + -- Add a timestamp for informational purposes. + timestamp <- fmap (\t -> round $ t * 1000) $ getPOSIXTime + -- Insert + execSql c sqlInsertEvent + [ SqlUUID $ Just aggregateId + , SqlUUID $ Just $ peEventId e + , SqlByteArray $ Just $ peEvent e + , SqlInt32 $ Just $ fromIntegral $ peSequenceNumber e + , SqlInt64 $ Just $ timestamp + ] + + where + -- Translate duplicate key exceptions into StoreError. + translateExceptions uuid action = + catchJust isDuplicateKey action $ \_ -> + throw $ VersionConflict uuid + -- SQL for event insertion + sqlInsertEvent = + "INSERT INTO event (\ + \ aggregate_uuid, \ + \ event_uuid, \ + \ event_data, \ + \ seq_no, \ + \ \"timestamp\", \ + \ archive_uuid \ + \) VALUES ($1, $2, $3, $4, $5, NULL)" + +retrieveEvents :: Pool Connection -> UUID -> Int -> (InputStream (PersistedEvent ByteString) -> IO a) -> IO a +retrieveEvents cp aggregateId v0 f = withResource cp $ \c -> withTransaction c $ do + ioQuery c sqlSelectEvent [ SqlUUID $ Just aggregateId + , SqlInt32 $ Just $ fromIntegral v0 + ] $ SC.map unpack >=> f + where + unpack [ SqlInt32 (Just sequenceNumber) + , SqlUUID (Just eventId) + , SqlByteArray (Just eventData) + ] = PersistedEvent eventData (fromIntegral sequenceNumber) eventId + unpack columns = error $ badQueryResultMsg [show aggregateId, show v0] columns + + sqlSelectEvent = + " SELECT seq_no, event_uuid, event_data \ + \ FROM event \ + \ WHERE aggregate_uuid = $1 \ + \ AND seq_no > $2 \ + \ORDER BY seq_no ASC" + +retrieveAllEvents :: Pool Connection -> (InputStream (UUID, PersistedEvent ByteString) -> IO a) -> IO a +retrieveAllEvents cp f = withResource cp $ \c -> withTransaction c $ do + ioQuery c sqlSelectAllEvents [ ] $ SC.map unpack >=> f + where + unpack [ SqlUUID (Just aggregateId) + , SqlInt32 (Just sequenceNumber) + , SqlUUID (Just eventId) + , SqlByteArray (Just eventData) + ] = (aggregateId, PersistedEvent eventData (fromIntegral sequenceNumber) eventId) + unpack columns = error $ badQueryResultMsg [] columns + + sqlSelectAllEvents = + " SELECT aggregate_uuid, seq_no, event_uuid, event_data \ + \ FROM event \ + \ORDER BY aggregate_uuid, seq_no ASC" + +-- | Create an event store backed by a PostgreSQL connection pool. +-- The database which the connections go to must have an appropriate +-- schema as defined by 'applyMigrations'. +newEventStore :: Pool Connection -> IO (EventStore ByteString) +newEventStore connectionPool = do + return $ EventStore + { esStoreEvents = storeEvents connectionPool + , esRetrieveEvents = retrieveEvents connectionPool + , esRetrieveAllEvents = retrieveAllEvents connectionPool + } diff --git a/src/Data/CQRS/PostgreSQL/Internal/Migration.hs b/src/Data/CQRS/PostgreSQL/Internal/Migration.hs new file mode 100644 --- /dev/null +++ b/src/Data/CQRS/PostgreSQL/Internal/Migration.hs @@ -0,0 +1,64 @@ +{-# LANGUAGE OverloadedStrings #-} +module Data.CQRS.PostgreSQL.Internal.Migration + ( applyMigrations + , uuid + ) where + +import Control.Monad (forM_, (>=>)) +import Data.ByteString (ByteString) +import Data.CQRS.PostgreSQL.Internal.Utils (badQueryResultMsg, execSql, ioQuery, withTransaction, SqlValue(..)) +import Data.UUID.Types (UUID) +import qualified Data.UUID.Types as U +import Data.Text.Encoding (decodeUtf8) +import Database.PostgreSQL.LibPQ (Connection) +import qualified System.IO.Streams as Streams +import qualified System.IO.Streams.Combinators as SC + +-- | Convert valid UUID string to a value. +uuid :: ByteString -> UUID +uuid s = + case U.fromASCIIBytes s of + Nothing -> error $ "Invalid UUID string: " ++ show s + Just u -> u + +-- | Apply all necessary migrations to event store database. +-- We use a special table to track which migrations have been +-- applied. +applyMigrations :: Connection -> [(UUID, ByteString)] -> IO () +applyMigrations c migrations = do + -- Must always create the change log if necessary + withTransaction c $ execSql c sqlCreateChangeSetTbl [ ] + -- Apply all the migrations. + forM_ migrations $ \(changeSetId, sql) -> do + let changeSetIdSql = SqlUUID $ Just changeSetId + let sqlText = decodeUtf8 sql + withTransaction c $ do + -- Check if change set has already been applied + existingChangeSet <- ioQuery c sqlFindChangeSet [ changeSetIdSql ] $ + SC.map (unpackChangeSet changeSetId) >=> Streams.read + case existingChangeSet of + Just (_, sqlText') | sqlText == sqlText' -> + return () -- Already applied, do nothing + Just _ -> + -- Applied, but SQL doesn't match. That's a huge problem, so we'll error out. + error $ "Migration error: Changeset SQL modified: UUID " ++ show changeSetId + Nothing -> do + execSql c sqlInsertChangeSet [ changeSetIdSql, SqlText $ Just sqlText ] + execSql c sql [ ] + + where + unpackChangeSet _ [ SqlUUID (Just changeSetId), SqlText (Just sqlText) ] = (changeSetId, sqlText) + unpackChangeSet changeSetId columns = error $ badQueryResultMsg [show changeSetId] columns + + -- Migrations support SQL: + sqlCreateChangeSetTbl = + "CREATE TABLE IF NOT EXISTS changeset ( \ + \ changeset_id UUID PRIMARY KEY, \ + \ sql TEXT NOT NULL \ + \)" + sqlFindChangeSet = + "SELECT changeset_id, sql FROM changeset \ + \ WHERE changeset_id = $1" + sqlInsertChangeSet = + "INSERT INTO changeset (changeset_id, sql) \ + \ VALUES ($1, $2)" diff --git a/src/Data/CQRS/PostgreSQL/Internal/SnapshotStore.hs b/src/Data/CQRS/PostgreSQL/Internal/SnapshotStore.hs new file mode 100644 --- /dev/null +++ b/src/Data/CQRS/PostgreSQL/Internal/SnapshotStore.hs @@ -0,0 +1,79 @@ +{-# LANGUAGE OverloadedStrings #-} +module Data.CQRS.PostgreSQL.Internal.SnapshotStore + ( newSnapshotStore + ) where + +import Control.Exception (catchJust) +import Data.ByteString (ByteString) +import Data.Pool (Pool, withResource) +import Data.CQRS.Types.Snapshot (Snapshot(..)) +import Data.CQRS.Types.SnapshotStore (SnapshotStore(..)) +import Data.CQRS.PostgreSQL.Internal.Utils (execSql, execSql', ioQuery, isDuplicateKey, withTransaction, badQueryResultMsg, SqlValue(..)) +import Data.UUID.Types (UUID) +import Database.PostgreSQL.LibPQ (Connection) +import qualified System.IO.Streams as Streams + +writeSnapshot :: Pool Connection -> UUID -> Snapshot ByteString -> IO () +writeSnapshot connectionPool aggregateId (Snapshot v d) = + withResource connectionPool $ \c -> do + -- We ignore duplicate key exceptions since snapshots aren't + -- important enough to merit aborting the user's command. We'll + -- probably have opportunities to write a snapshot for the + -- aggregate later anyway. + ignoreDuplicateKey $ + withTransaction c $ do + maybeUpdatedRows <- execSql' c updateSnapshotSql + [ SqlByteArray (Just d) + , SqlInt32 $ Just $ fromIntegral v + , SqlUUID $ Just aggregateId + ] + if (maybe 0 id maybeUpdatedRows) > 0 then + return () -- Update happened, done! + else + -- Not updated; this means that the snapshot row didn't exist, + -- so we'll try to INSERT it now. + execSql c insertSnapshotSql + [ SqlUUID $ Just aggregateId + , SqlByteArray (Just d) + , SqlInt32 $ Just $ fromIntegral v + ] + + where + updateSnapshotSql = + "UPDATE snapshot SET data=$1, version=$2 WHERE aggregate_uuid=$3" + insertSnapshotSql = + "INSERT INTO snapshot \ + \ (aggregate_uuid, data, version) \ + \ VALUES ($1, $2, $3)" + -- Run an action, ignoring duplicate key exceptions. + ignoreDuplicateKey action = + catchJust isDuplicateKey action $ \_ -> + return () -- Ignore + +readSnapshot :: Pool Connection -> UUID -> IO (Maybe (Snapshot ByteString)) +readSnapshot connectionPool aggregateId = do + withResource connectionPool $ \connection -> do + withTransaction connection $ 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 aggregateId] columns + -- Run the query. + r <- ioQuery connection selectSnapshotSql [SqlUUID $ Just aggregateId] $ \inputStream -> do + cs <- Streams.read inputStream + return $ fmap unpackColumns cs + case r of + Just (d,v) -> return $ Just $ Snapshot v d + Nothing -> return Nothing + + where + selectSnapshotSql = + "SELECT data, version FROM snapshot WHERE aggregate_uuid=$1;" + +-- | Create an snapshot store backed by a PostgreSQL connection pool. +newSnapshotStore :: Pool Connection -> IO (SnapshotStore ByteString) +newSnapshotStore connectionPool = do + return $ SnapshotStore + (writeSnapshot connectionPool) + (readSnapshot connectionPool) diff --git a/src/Data/CQRS/PostgreSQL/Internal/Utils.hs b/src/Data/CQRS/PostgreSQL/Internal/Utils.hs new file mode 100644 --- /dev/null +++ b/src/Data/CQRS/PostgreSQL/Internal/Utils.hs @@ -0,0 +1,236 @@ +{-# LANGUAGE BangPatterns, DeriveGeneric, OverloadedStrings #-} +module Data.CQRS.PostgreSQL.Internal.Utils + ( SqlValue(..) + , QueryError(..) + , badQueryResultMsg + , execSql + , execSql' + , ioQuery + , ioQuery' + , isDuplicateKey + , runQuery + , withTransaction + ) where + +import Control.DeepSeq (NFData(..), ($!!)) +import Control.Exception (Exception, throw) +import Control.Exception.Enclosed (catchAny) +import Control.Monad (forM) +import Control.Monad.IO.Class (liftIO) +import Control.Exception (SomeException, bracket) +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.Int (Int16, Int32, Int64) +import Data.Pool (Pool, withResource) +import Data.Text (Text) +import Data.Text.Encoding (decodeUtf8', encodeUtf8) +import Data.Typeable (Typeable) +import Data.UUID.Types (UUID) +import qualified Data.UUID.Types as U +import Database.PostgreSQL.LibPQ (Connection, Oid(..), Format(..), ExecStatus(..), Column(..), Row(..), FieldCode(..)) +import qualified Database.PostgreSQL.LibPQ as P +import GHC.Generics (Generic) +import System.IO.Streams (InputStream) +import qualified System.IO.Streams.Combinators as SC +import qualified System.IO.Streams.List as SL + +-- | 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) + | SqlUUID (Maybe UUID) + | Unmatched (Oid, Maybe ByteString) + deriving (Eq, Show) + +-- | Is the given query exception a duplicate key exception? +isDuplicateKey :: QueryError -> Maybe () +isDuplicateKey qe | qeSqlState qe == Just "23505" = Just () + | otherwise = Nothing + +-- | Execute an IO action with an active transaction. +withTransaction :: Connection -> IO a -> IO a +withTransaction connection action = do + begin + catchAny runAction tryRollback + where + runAction = do + r <- action + commit + return r + + tryRollback :: SomeException -> IO a + tryRollback e = + -- Try explicit rollback; we want to preserve original exception. + catchAny (rollback >> throw e) $ \_ -> + -- Rethrow original exception; resource pool will make sure the database + -- connection is properly destroyed (rather than being returned to the + -- pool). + throw e + + begin = execSql connection "START TRANSACTION;" [ ] + commit = execSql connection "COMMIT TRANSACTION;" [ ] + rollback = execSql 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 _ (SqlInt64 Nothing) = return Nothing +fromSqlValue _ (SqlInt64 (Just i)) = return $ Just (Oid 20, 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 _ (SqlUUID Nothing) = return Nothing +fromSqlValue _ (SqlUUID (Just u)) = return $ Just (Oid 2950, U.toASCIIBytes u, Text) +fromSqlValue _ _ = error "fromSqlValue: 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 2950 -> c (return . U.fromASCIIBytes) SqlUUID + + _ -> 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 -> error "toSqlValue: Conversion failed" + Just _ -> return $ construct mvalue' + +-- | Execute a query with no result. +execSql :: Connection -> ByteString -> [SqlValue] -> IO () +execSql connection sql parameters = + ioQuery connection sql parameters (\_ -> return ()) + +-- | Execute a query an return the number of updated rows (if available). +execSql' :: Connection -> ByteString -> [SqlValue] -> IO (Maybe Int) +execSql' connection sql parameters = + ioQuery' connection sql parameters (\n _ -> return n) + +-- | Error happened during query. +data QueryError = QueryError + { qeSqlState :: Maybe ByteString + , qeStatusMessage :: ByteString + , qeErrorMessage :: Maybe ByteString + } deriving (Show, Typeable, Generic) + +instance Exception QueryError + +instance NFData QueryError + +-- | Run a query and fold over the results. The action receives an +-- 'InputStream' over all the rows in the result. +ioQuery :: Connection -> ByteString -> [SqlValue] -> (InputStream [SqlValue] -> IO a) -> IO a +ioQuery connection sql parameters f = + ioQuery' connection sql parameters $ \_ is -> f is + +-- | Run a query and fold over the results. The action receives the number of rows affected +-- and an 'InputStream' over all the rows in the result. +ioQuery' :: Connection -> ByteString -> [SqlValue] -> (Maybe Int -> InputStream [SqlValue] -> IO a) -> IO a +ioQuery' connection sql parameters f = + bracket open done $ \r -> do + -- Check the status + status <- P.resultStatus r + case status of + CommandOk -> do + n <- affectedRows r + go r >>= f n + TuplesOk -> do + n <- affectedRows r + go r >>= f n + _ -> do + -- Extract error information. We need to be careful to + -- COPY the values here since freeing the result will + -- cause the "original" values to become garbage. + sqlState <- fmap (fmap B.copy) $ P.resultErrorField r DiagSqlstate + statusMessage <- fmap (fmap B.copy) P.resStatus status + errorMessage <- fmap (fmap B.copy) $ P.resultErrorMessage r + throw $!! QueryError + { qeSqlState = sqlState + , qeStatusMessage = statusMessage + , qeErrorMessage = errorMessage + } + + where + affectedRows r = do + !cmdTuples <- P.cmdTuples r + case cmdTuples of + Nothing -> return $ Nothing + Just !x -> return $! fmap fst $! readDecimal $! B.copy x + + 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 set; something is very wrong" + Just r -> return r + + go :: P.Result -> IO (InputStream [SqlValue]) + go r = do + Col nFields <- liftIO $ P.nfields r + Row nRows <- liftIO $ P.ntuples r + let columns = map P.toColumn [0..nFields-1] + let loop [] = return Nothing + loop (row:rows) = do + columnValues <- forM columns $ getSqlVal r row + return $ Just (columnValues, rows) + SC.unfoldM loop $ map P.toRow [0..nRows-1] + + getSqlVal r row c = do + mval <- P.getvalue' r row c + typ <- P.ftype r c + toSqlValue (typ,mval) + +-- Run a query and result a list of the rows in the result. +runQuery :: Pool Connection -> ByteString -> [SqlValue] -> IO [[SqlValue]] +runQuery connectionPool sql parameters = + withResource connectionPool $ \c -> do + ioQuery c sql parameters (\inputStream -> SL.toList inputStream) + +-- | Format a message indicating a bad query result due to the "shape". +badQueryResultMsg :: [String] -> [SqlValue] -> String +badQueryResultMsg params columns = concat ["Invalid query result shape. Params: ", show params, ". Result columns: ", show columns] diff --git a/src/Data/CQRS/PostgreSQL/Migrations.hs b/src/Data/CQRS/PostgreSQL/Migrations.hs new file mode 100644 --- /dev/null +++ b/src/Data/CQRS/PostgreSQL/Migrations.hs @@ -0,0 +1,45 @@ +{-# LANGUAGE OverloadedStrings #-} +module Data.CQRS.PostgreSQL.Migrations + ( migrate + ) where + +import Database.PostgreSQL.LibPQ (Connection) +import Data.CQRS.PostgreSQL.Internal.Migration (applyMigrations, uuid) + +-- | Apply all the necessary migrations to use any subset of the CQRS +-- functionality. +migrate :: Connection -> IO () +migrate connection = + applyMigrations connection + [ (uuid "d0bf19c5-6bca-4bb0-8dd8-8e1340ec7503", sqlCreateEventTbl) + , (uuid "1bbd4b4c-db0c-445c-90dd-19f7b725cac4", sqlCreateEventArchiveIdx) + , (uuid "f1c4cec1-4b8c-4693-a48b-1a6f02e66741", sqlCreateArchiveTable) + , (uuid "c7258b1c-bc3c-467a-bc1f-f63ad330651f", sqlCreateSnapshotTbl) + ] + where + sqlCreateEventTbl = + "CREATE TABLE event ( \ + \ aggregate_uuid UUID NOT NULL, \ + \ event_uuid UUID NOT NULL, \ + \ event_data BYTEA NOT NULL, \ + \ seq_no INTEGER NOT NULL, \ + \ \"timestamp\" BIGINT NOT NULL, \ + \ archive_uuid UUID, \ + \ PRIMARY KEY (aggregate_uuid, seq_no) \ + \)" + sqlCreateEventArchiveIdx = + "CREATE INDEX event_archive_idx ON event ( \ + \ archive_uuid \ + \)" + sqlCreateArchiveTable = + "CREATE TABLE archive ( \ + \ archive_uuid UUID NOT NULL, \ + \ prev_archive_uuid UUID DEFAULT NULL, \ + \ next_archive_uuid UUID DEFAULT NULL, \ + \ PRIMARY KEY (archive_uuid) \ + \)" + sqlCreateSnapshotTbl = + "CREATE TABLE snapshot ( \ + \ aggregate_uuid UUID PRIMARY KEY, \ + \ data BYTEA, version INTEGER \ + \)"