cqrs-sqlite3 0.8.0 → 0.9.0

raw patch · 4 files changed

+145/−111 lines, 4 filesdep +cqrs-sqlite3dep +cqrs-testdep +cqrs-typesdep −cerealdep −cqrsdep ~conduitdep ~direct-sqlitePVP ok

version bump matches the API change (PVP)

Dependencies added: cqrs-sqlite3, cqrs-test, cqrs-types, hspec, pool-conduit, text, transformers

Dependencies removed: cereal, cqrs

Dependency ranges changed: conduit, direct-sqlite

API changes (from Hackage documentation)

Files

• cqrs-sqlite3.cabal +40/−13
• src-test/Main.hs +17/−0
• src/Data/CQRS/EventStore/Backend/Sqlite3.hs +66/−70
• src/Data/CQRS/EventStore/Backend/Sqlite3Utils.hs +22/−28

@@ -1,24 +1,51 @@ Name:                cqrs-sqlite3-Version:             0.8.0+Version:             0.9.0 Synopsis:            SQLite3 backend for the cqrs package. Description:         SQLite3 backend for the cqrs package. License:             MIT License-file:        LICENSE Category:            Data-Cabal-version:       >=1.6.0.1+Cabal-version:       >=1.10 Build-type:          Simple Author:              Bardur Arantsson Maintainer:          Bardur Arantsson <bardur@scientician.net>  Library-  build-depends:   base == 4.*-                 , bytestring >= 0.9.0.1-                 , cereal >= 0.3.3 && < 0.4-                 , conduit >= 0.1 && < 0.2-                 , cqrs >= 0.8.0 && < 0.9-                 , direct-sqlite >= 1.1 && < 1.2-  extensions:      ScopedTypeVariables-  ghc-options:     -Wall-  hs-source-dirs:  src-  exposed-modules: Data.CQRS.EventStore.Backend.Sqlite3-                   Data.CQRS.EventStore.Backend.Sqlite3Utils+  build-depends:      base == 4.*+                    , bytestring >= 0.9.0.1+                    , conduit >= 1.0 && < 2+                    , cqrs-types >= 0.9.0 && < 0.10.0+                    , pool-conduit >= 0.1 && < 0.2+                    , direct-sqlite >= 2.2 && < 2.3+                    , text >= 0.11 && < 0.12+                    , transformers >= 0.2 && <0.4+  Default-Language:   Haskell2010+  Default-Extensions: OverloadedStrings+                      ScopedTypeVariables+                      TypeFamilies+  ghc-options:        -Wall+  hs-source-dirs:     src+  exposed-modules:    Data.CQRS.EventStore.Backend.Sqlite3+                      Data.CQRS.EventStore.Backend.Sqlite3Utils++Test-Suite cqrs-sqlite3-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-test >= 0.9 && < 0.10.0+                     , pool-conduit >= 0.1 && < 0.2+                     , direct-sqlite >= 2.2 && < 2.3+                     , text >= 0.11 && < 0.12+                     , transformers >= 0.2.2 && < 0.4+                     -- Self-dependency+                     , cqrs-sqlite3+                     -- Test framework+                     , hspec >= 1.3 && < 2.0+  Ghc-options:         -Wall+  Default-language:    Haskell2010+  Default-extensions:  OverloadedStrings+                       ScopedTypeVariables+                       TypeFamilies

@@ -0,0 +1,17 @@+module Main where++import Data.CQRS.Test.EventStore.BackendTest (testBackend)+import Data.CQRS.EventStore.Backend.Sqlite3+import Test.Hspec (hspec, Spec)++sqlite3BackendTest :: Spec+sqlite3BackendTest = do+  testBackend mkPool setup+  where+    mkPool = createBackendPool 1 ":memory:"+    setup = return ()++main :: IO ()+main = do+  hspec $ do+    sqlite3BackendTest

@@ -1,27 +1,29 @@ {-| Implementation of an SQLite3-based event store. -} module Data.CQRS.EventStore.Backend.Sqlite3-       ( openSqliteEventStore+       ( SQLite3EventStoreBackend+       , createBackendPool        ) where -import           Control.Monad (when, forM_, liftM)+import           Control.Monad (when, forM_) import           Data.ByteString (ByteString)-import           Data.Conduit (Source, ($=), ($$), runResourceT)+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.EventStore.Backend.Sqlite3Utils (withTransaction, execSql, sourceQuery) import           Data.CQRS.GUID (GUID)+import qualified Data.CQRS.GUID as G import           Data.CQRS.PersistedEvent (PersistedEvent(..))-import           Data.CQRS.Serialize (decode', encode)+import           Data.Text (Text) import qualified Database.SQLite3 as SQL import           Database.SQLite3 (Database, SQLData(..))-import           Prelude hiding (catch)  -- Convenience class for converting values to SQLData. class ToSQLData a where   toSQLData :: a -> SQLData  instance ToSQLData GUID where-  toSQLData = SQLBlob . encode+  toSQLData = SQLBlob . G.toByteString  instance ToSQLData Int where   toSQLData = SQLInteger . fromIntegral@@ -30,39 +32,30 @@   toSQLData = SQLBlob  -- SQL-createEventsSql :: String-createEventsSql = "CREATE TABLE IF NOT EXISTS events ( guid BLOB , ev_data BLOB , version INTEGER , gversion INTEGER, PRIMARY KEY (guid, version) );"+createEventsSql :: Text+createEventsSql = "CREATE TABLE IF NOT EXISTS events ( guid BLOB , ev_data BLOB , version INTEGER , PRIMARY KEY (guid, version) );" -selectEventsSql :: String-selectEventsSql = "SELECT version, gversion, ev_data FROM events WHERE guid = ? AND version > ? ORDER BY version ASC;"+selectEventsSql :: Text+selectEventsSql = "SELECT version, ev_data FROM events WHERE guid = ? AND version >= ? ORDER BY version ASC;" -enumerateAllEventsSql :: String-enumerateAllEventsSql = "SELECT gversion, guid, version, ev_data FROM events WHERE gversion >= ? ORDER BY gversion ASC;"+enumerateAllEventsSql :: Text+enumerateAllEventsSql = "SELECT guid, version, ev_data FROM events ORDER BY version ASC;" -insertEventSql :: String-insertEventSql = "INSERT INTO events ( guid, version, ev_data, gversion ) VALUES (?, ?, ?, ?);"+insertEventSql :: Text+insertEventSql = "INSERT INTO events ( guid, version, ev_data ) VALUES (?, ?, ?);" -createAggregateVersionsSql :: String+createAggregateVersionsSql :: Text createAggregateVersionsSql = "CREATE TABLE IF NOT EXISTS versions ( guid BLOB PRIMARY KEY , version INTEGER );" -getCurrentVersionSql :: String+getCurrentVersionSql :: Text getCurrentVersionSql = "SELECT version FROM versions WHERE guid = ?;" -getLatestVersionSql :: String-getLatestVersionSql = "SELECT COALESCE(MAX(gversion), 0) FROM events;"--updateCurrentVersionSql :: String+updateCurrentVersionSql :: Text updateCurrentVersionSql = "INSERT OR REPLACE INTO versions ( guid, version ) VALUES (?,?);" -createSnapshotSql :: String+createSnapshotSql :: Text createSnapshotSql = "CREATE TABLE IF NOT EXISTS snapshots ( guid BLOB PRIMARY KEY , data BLOB , version INTEGER );" -writeSnapshotSql :: String-writeSnapshotSql = "INSERT OR REPLACE INTO snapshots ( guid , data, version ) VALUES ( ?, ?, ? );"--selectSnapshotSql :: String-selectSnapshotSql = "SELECT data, version FROM snapshots WHERE guid = ?;"- badQueryResultMsg :: [String] -> [SQLData] -> String badQueryResultMsg params columns = concat ["Invalid query result shape. Params: ", show params, ". Result columns: ", show columns] @@ -96,30 +89,25 @@       [ toSQLData guid       , toSQLData $ peSequenceNumber e       , toSQLData $ peEvent e-      , toSQLData $ peGlobalVer e       ] -retrieveEvents :: Database -> GUID -> Int -> Source IO RawEvent+retrieveEvents :: Database -> GUID -> Int -> Source (ResourceT IO) RawEvent retrieveEvents database guid v0 = do   -- Unpack the columns into tuples.-  let unpackColumns [SQLInteger version, SQLInteger gversion, SQLBlob eventData] = PersistedEvent guid eventData (fromIntegral version) (fromIntegral gversion)+  let unpackColumns [SQLInteger version, SQLBlob eventData] = PersistedEvent guid eventData (fromIntegral version)       unpackColumns columns = error $ badQueryResultMsg [show guid, show v0] columns   -- Find events with version numbers.-  fmap unpackColumns $ sourceQuery database selectEventsSql [toSQLData guid, toSQLData v0]+  sourceQuery database selectEventsSql [toSQLData guid, toSQLData v0] $= CL.map unpackColumns -enumerateAllEvents :: Database -> Int -> Source IO RawEvent-enumerateAllEvents database minVersion = do-  sourceQuery database enumerateAllEventsSql [toSQLData minVersion] $= CL.map-    (\columns -> do-        case columns of-          [ SQLInteger gv, SQLBlob g, SQLInteger v, SQLBlob ed ] ->-            PersistedEvent (decode' g) ed (fromIntegral v) (fromIntegral gv)-          _ ->-            error $ badQueryResultMsg [show minVersion] columns)+enumerateAllEvents :: Database -> Source (ResourceT IO) RawEvent+enumerateAllEvents database = do+  let unpackColumns [ SQLBlob g, SQLInteger v, SQLBlob ed ] = PersistedEvent (G.fromByteString g) ed (fromIntegral v)+      unpackColumns columns = error $ badQueryResultMsg [] columns+  sourceQuery database enumerateAllEventsSql [] $= CL.map unpackColumns  writeSnapshot :: Database -> GUID -> RawSnapshot -> IO () writeSnapshot database guid (RawSnapshot v d) = do-  execSql database writeSnapshotSql+  execSql database "INSERT OR REPLACE INTO snapshots ( guid , data, version ) VALUES ( ?, ?, ? );"     [ toSQLData guid     , toSQLData d     , toSQLData v@@ -131,34 +119,42 @@   let unpackColumns [SQLBlob d, SQLInteger v] = Just $ RawSnapshot (fromIntegral v) d       unpackColumns columns                   = error $ badQueryResultMsg [show guid] columns   -- Run the query.-  runResourceT $ (sourceQuery database selectSnapshotSql [toSQLData guid] $= (CL.map unpackColumns)) $$ CL.fold const Nothing+  runResourceT $ (sourceQuery database selectSnapshotSql [toSQLData guid] $= (CL.map unpackColumns)) $$ CL.fold (\_ x -> x) Nothing -getLatestVersion :: Database -> IO Int-getLatestVersion database = do-  -- Unpack columns from result.-  let unpackColumns :: [SQLData] -> Int-      unpackColumns [SQLInteger v] = fromIntegral v-      unpackColumns columns        = error $ badQueryResultMsg [] columns-  -- Run the query.-  liftM head $ runResourceT $ (sourceQuery database getLatestVersionSql [] $= (CL.map unpackColumns)) $$ CL.consume+  where+    selectSnapshotSql = "SELECT data, version FROM snapshots WHERE guid = ?;" --- | Open an SQLite3-based event store using the named SQLite database file.--- The database file is created if it does not exist.-openSqliteEventStore :: String -> IO EventStoreBackend-openSqliteEventStore databaseFileName = do-  -- Create the database.-  database <- SQL.open databaseFileName-  -- Set up tables.-  execSql database createEventsSql []-  execSql database createAggregateVersionsSql []-  execSql database createSnapshotSql []-  -- Return event store.-  return $ EventStoreBackend { esbStoreEvents = storeEvents database-                             , esbRetrieveEvents = retrieveEvents database-                             , esbEnumerateAllEvents = enumerateAllEvents database-                             , esbWriteSnapshot = writeSnapshot database-                             , esbGetLatestSnapshot = getLatestSnapshot database-                             , esbGetLatestVersion = getLatestVersion database-                             , esbWithTransaction = withTransaction database-                             , esbCloseEventStoreBackend = SQL.close database-                             }+-- | SQLite3 event store backend type alias.+newtype SQLite3EventStoreBackend = ESB Database++-- | Instance of EventStoreBackend for SQLite3EventStoreBackend+instance EventStoreBackend SQLite3EventStoreBackend where+    esbStoreEvents (ESB database) =+        storeEvents database+    esbRetrieveEvents (ESB database) =+        retrieveEvents database+    esbEnumerateAllEvents (ESB database) =+        enumerateAllEvents database+    esbWriteSnapshot (ESB database) =+        writeSnapshot database+    esbGetLatestSnapshot (ESB database) =+        getLatestSnapshot database+    esbWithTransaction (ESB database) =+        withTransaction database++-- | Create a pool of SQLite3-based event store backends.+createBackendPool :: Int -> Text -> IO (Pool SQLite3EventStoreBackend)+createBackendPool n databaseFileName = do+  createPool open close 1 1 n+  where+    open = do+      -- Create the database.+      database <- SQL.open databaseFileName+      -- Set up tables if necessary.+      execSql database createEventsSql []+      execSql database createAggregateVersionsSql []+      execSql database createSnapshotSql []+      -- Return event store.+      return $ ESB database+    close (ESB db) = do+      SQL.close db

@@ -6,13 +6,11 @@        ) where  import           Control.Exception (catch, bracket, onException, SomeException)-import           Control.Monad (liftM, when)-import           Data.Conduit (Source)-import qualified Data.Conduit as C-import           Data.IORef (newIORef, readIORef, writeIORef)+import           Control.Monad.IO.Class (liftIO)+import           Data.Conduit (ResourceT, Source, bracketP, yield)+import           Data.Text (Text) import qualified Database.SQLite3 as SQL import           Database.SQLite3 (Database, Statement, SQLData(..), StepResult(..))-import           Prelude hiding (catch)  beginTransaction :: Database -> IO () beginTransaction database = execSql database "BEGIN TRANSACTION;" []@@ -23,42 +21,38 @@ rollbackTransaction :: Database -> IO () rollbackTransaction database = execSql database "ROLLBACK TRANSACTION;" [] -withSqlStatement :: Database -> String -> [SQLData] -> (Statement -> IO a) -> IO a+withSqlStatement :: Database -> Text -> [SQLData] -> (Statement -> IO a) -> IO a withSqlStatement database sql parameters action =   bracket (SQL.prepare database sql) SQL.finalize $ \statement -> do     SQL.bind statement parameters     action statement  -- | Execute an SQL statement for which no result is expected.-execSql :: Database -> String -> [SQLData] -> IO ()+execSql :: Database -> Text -> [SQLData] -> IO () execSql database sql parameters =   withSqlStatement database sql parameters $ \stmt -> do     _ <- SQL.step stmt     return () -data State = Unbound-           | Bound-             deriving (Eq)+data State = Unbound | Bound+           deriving (Eq) -sourceQuery :: Database -> String -> [SQLData] -> Source IO [SQLData]+sourceQuery :: Database -> Text -> [SQLData] -> Source (ResourceT IO) [SQLData] sourceQuery database sql parameters =-  C.sourceIO-  (do-      stateRef <- newIORef Unbound-      stmt <- SQL.prepare database sql-      return (stateRef, stmt))-  (\(_,stmt) -> SQL.finalize stmt)-  (\(stateRef,stmt) -> do-      -- Bind parameters if necessary.-      state <- readIORef stateRef-      when (state == Unbound) $ do-        SQL.bind stmt parameters-        writeIORef stateRef $ Bound-      -- Fetch results.-      nextResult <- SQL.step stmt-      case nextResult of-        Done -> return C.Closed-        Row -> liftM C.Open $ SQL.columns stmt)+    bracketP (SQL.prepare database sql)+             (SQL.finalize)+             run+    where+      run stmt = do+        liftIO $ SQL.bind stmt parameters+        loop stmt+      loop stmt = do+        nextResult <- liftIO $ SQL.step stmt+        case nextResult of+          Done -> return ()+          Row -> do+            (liftIO $ SQL.columns stmt) >>= yield+            loop stmt  -- | Execute an IO action with an active transaction. withTransaction :: Database -> IO a -> IO a