cqrs-postgresql 0.9.0 → 0.10.0

raw patch · 16 files changed

+915/−375 lines, 16 filesdep +cqrs-coredep +cqrs-testkitdep +deepseqdep −conduitdep −cqrs-testdep −cqrs-typesdep ~basedep ~hspecdep ~postgresql-libpqsetup-changedPVP ok

version bump matches the API change (PVP)

Dependencies added: cqrs-core, cqrs-testkit, deepseq, enclosed-exceptions, io-streams, pg-harness-client, random, resource-pool, uuid-types

Dependencies removed: conduit, cqrs-test, cqrs-types, old-locale, pool-conduit

Dependency ranges changed: base, hspec, postgresql-libpq, text, time, transformers

API changes (from Hackage documentation)

Files

• LICENSE +1/−1
• Setup.hs +2/−0
• Setup.lhs +0/−4
• cqrs-postgresql.cabal +32/−25
• src-test/Data/CQRS/PostgreSQL/Internal/MigrationSpec.hs +88/−0
• src-test/Data/CQRS/PostgreSQL/Internal/UtilsSpec.hs +39/−0
• src-test/Main.hs +49/−9
• src/Data/CQRS/EventStore/Backend/PostgreSQL.hs +0/−175
• src/Data/CQRS/EventStore/Backend/PostgreSQLUtils.hs +0/−161
• src/Data/CQRS/PostgreSQL.hs +9/−0
• src/Data/CQRS/PostgreSQL/Internal/ArchiveStore.hs +159/−0
• src/Data/CQRS/PostgreSQL/Internal/EventStore.hs +112/−0
• src/Data/CQRS/PostgreSQL/Internal/Migration.hs +64/−0
• src/Data/CQRS/PostgreSQL/Internal/SnapshotStore.hs +79/−0
• src/Data/CQRS/PostgreSQL/Internal/Utils.hs +236/−0
• src/Data/CQRS/PostgreSQL/Migrations.hs +45/−0

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

@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain

@@ -1,4 +0,0 @@-#! /usr/bin/env runhaskell--> import Distribution.Simple-> main = defaultMain

@@ -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 <bardur@scientician.net>  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

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

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

@@ -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)+                        }

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

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

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

@@ -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 <archiveSize> 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 <current> 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+             }+

@@ -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+             }

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

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

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

@@ -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 \+        \)"