diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,19 @@ +Copyright (c) 2011 Bardur Arantsson + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in +all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +THE SOFTWARE. diff --git a/Setup.lhs b/Setup.lhs new file mode 100644 --- /dev/null +++ b/Setup.lhs @@ -0,0 +1,4 @@ +#! /usr/bin/env runhaskell + +> import Distribution.Simple +> main = defaultMain diff --git a/cqrs-sqlite3.cabal b/cqrs-sqlite3.cabal new file mode 100644 --- /dev/null +++ b/cqrs-sqlite3.cabal @@ -0,0 +1,24 @@ +Name: cqrs-sqlite3 +Version: 0.6.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 +Build-type: Simple +Author: Bardur Arantsson +Maintainer: Bardur Arantsson + +Library + build-depends: base == 4.* + , bytestring >= 0.9.0.1 + , cereal >= 0.3.3 && < 0.4 + , cqrs >= 0.6.0 && < 0.7 + , enumerator >= 0.4.15 && < 0.5 + , 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 diff --git a/src/Data/CQRS/EventStore/Backend/Sqlite3.hs b/src/Data/CQRS/EventStore/Backend/Sqlite3.hs new file mode 100644 --- /dev/null +++ b/src/Data/CQRS/EventStore/Backend/Sqlite3.hs @@ -0,0 +1,173 @@ +{-| Implementation of an SQLite3-based event store. -} +module Data.CQRS.EventStore.Backend.Sqlite3 + ( openSqliteEventStore + ) where + +import Control.Monad (when, forM_, liftM) +import Data.ByteString (ByteString) +import Data.CQRS.EventStore.Backend (EventStoreBackend(..)) +import Data.CQRS.EventStore.Backend.Sqlite3Utils (withTransaction, execSql, enumQueryResult) +import Data.CQRS.GUID (GUID) +import Data.CQRS.Serialize (decode') +import Data.Enumerator (Enumerator, (>>==), ($=), run_) +import qualified Data.Enumerator.List as EL +import Data.Serialize (encode) +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 + +instance ToSQLData Int where + toSQLData = SQLInteger . fromIntegral + +instance ToSQLData ByteString where + 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) );" + +selectEventsSql :: String +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;" + +insertEventSql :: String +insertEventSql = "INSERT INTO events ( guid, version, ev_data, gversion ) VALUES (?, ?, ?, ?);" + +createAggregateVersionsSql :: String +createAggregateVersionsSql = "CREATE TABLE IF NOT EXISTS versions ( guid BLOB PRIMARY KEY , version INTEGER );" + +getCurrentVersionSql :: String +getCurrentVersionSql = "SELECT version FROM versions WHERE guid = ?;" + +getLatestVersionSql :: String +getLatestVersionSql = "SELECT COALESCE(MAX(gversion), 0) FROM events;" + +updateCurrentVersionSql :: String +updateCurrentVersionSql = "INSERT OR REPLACE INTO versions ( guid, version ) VALUES (?,?);" + +createSnapshotSql :: String +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] + +versionConflict :: (Show a, Show b) => a -> b -> IO c +versionConflict ov cv = + fail $ concat [ "Version conflict detected (expected ", show ov + , ", saw ", show cv, ")" + ] + +storeEvents :: Database -> GUID -> Int -> [(ByteString,Int)] -> IO () +storeEvents database guid originatingVersion events = do + -- Column unpacking. + let unpackColumns [ SQLInteger v ] = v + unpackColumns columns = error $ badQueryResultMsg [show guid] columns + -- Get the current version number of the aggregate. + curVer <- run_ $ EL.fold (\x -> max x . unpackColumns) 0 >>== + (enumQueryResult database getCurrentVersionSql [toSQLData guid]) + + -- Sanity check current version number. + when (fromIntegral curVer /= originatingVersion) $ + versionConflict originatingVersion curVer + + -- Update de-normalized version number. + execSql database updateCurrentVersionSql + [ toSQLData guid + , toSQLData $ originatingVersion + length events + ] + + -- Store the supplied events. + forM_ (zip [1 + originatingVersion..] events) $ \(v,(e,gv)) -> do + execSql database insertEventSql + [ toSQLData guid + , toSQLData v + , toSQLData e + , toSQLData gv + ] + +retrieveEvents :: Database -> GUID -> Int -> IO (Int,[ByteString]) +retrieveEvents database guid v0 = do + -- Unpack the columns into tuples. + let unpackColumns [SQLInteger version, SQLBlob eventData] = (version, eventData) + unpackColumns columns = error $ badQueryResultMsg [show guid, show v0] columns + -- Find events with version numbers. + results <- + run_ $ EL.consume >>== + (enumQueryResult database selectEventsSql [toSQLData guid, toSQLData v0] $= + (EL.map unpackColumns)) + -- Find the max version number. + let maxVersion = maximum $ (:) (fromIntegral v0) $ map fst results + return (fromIntegral maxVersion, map snd results) + +enumerateAllEvents :: Database -> Int -> Enumerator (Int,(GUID, Int, ByteString)) IO a +enumerateAllEvents database minVersion = do + enumQueryResult database enumerateAllEventsSql [toSQLData minVersion] $= EL.map + (\columns -> do + case columns of + [ SQLInteger gv, SQLBlob g, SQLInteger v, SQLBlob ed ] -> + ( fromIntegral gv, (decode' g, fromIntegral v, ed) ) + _ -> + error $ badQueryResultMsg [show minVersion] columns) + +writeSnapshot :: Database -> GUID -> (Int, ByteString) -> IO () +writeSnapshot database guid (v,a) = do + execSql database writeSnapshotSql + [ toSQLData guid + , toSQLData a + , toSQLData v + ] + +getLatestSnapshot :: Database -> GUID -> IO (Maybe (Int, ByteString)) +getLatestSnapshot database guid = do + -- Unpack columns from result. + let unpackColumns :: [SQLData] -> Maybe (Int,ByteString) + unpackColumns [SQLBlob a, SQLInteger v] = Just (fromIntegral v, a) + unpackColumns columns = error $ badQueryResultMsg [show guid] columns + -- Run the query. + run_ $ EL.fold const Nothing >>== + (enumQueryResult database selectSnapshotSql [toSQLData guid] $= (EL.map unpackColumns)) + +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 $ run_ $ EL.consume >>== (enumQueryResult database getLatestVersionSql [] $= (EL.map unpackColumns)) + +-- | 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 + } diff --git a/src/Data/CQRS/EventStore/Backend/Sqlite3Utils.hs b/src/Data/CQRS/EventStore/Backend/Sqlite3Utils.hs new file mode 100644 --- /dev/null +++ b/src/Data/CQRS/EventStore/Backend/Sqlite3Utils.hs @@ -0,0 +1,72 @@ +{-| Implementation of an SQLite3-based event store. -} +module Data.CQRS.EventStore.Backend.Sqlite3Utils + ( enumQueryResult + , execSql + , withTransaction + ) where + +import Control.Exception (catch, bracket, finally, onException, SomeException) +import Data.Enumerator (Enumerator, Iteratee(..), tryIO, continue, (>>==), Stream(..)) +import Data.Enumerator.Internal (checkContinue0) +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;" [] + +commitTransaction :: Database -> IO () +commitTransaction database = execSql database "COMMIT TRANSACTION;" [] + +rollbackTransaction :: Database -> IO () +rollbackTransaction database = execSql database "ROLLBACK TRANSACTION;" [] + +withSqlStatement :: Database -> String -> [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 sql parameters = + withSqlStatement database sql parameters $ \stmt -> do + _ <- SQL.step stmt + return () + +enumQueryResults :: Statement -> Enumerator [SQLData] IO b +enumQueryResults stmt = checkContinue0 $ \loop k -> + do + nextResult <- tryIO $ SQL.step stmt + case nextResult of + Done -> continue k + Row -> do + cols <- tryIO $ SQL.columns stmt + k (Chunks [cols]) >>== loop + +-- | Perform a query and enumerate the results. Each result is a list +-- of returned columns. +enumQueryResult :: Database -> String -> [SQLData] -> Enumerator [SQLData] IO b +enumQueryResult database sql parameters step = do + stmt <- tryIO $ SQL.prepare database sql + Iteratee $ finally + (do + SQL.bind stmt parameters + runIteratee $ enumQueryResults stmt step) + (SQL.finalize stmt) + +-- | Execute an IO action with an active transaction. +withTransaction :: Database -> IO a -> IO a +withTransaction database action = do + beginTransaction database + onException runAction tryRollback + where + runAction = do + r <- action + commitTransaction database + return r + + tryRollback = + -- Try rollback while discarding exception; we want to preserve + -- original exception. + catch (rollbackTransaction database) (\(_::SomeException) -> return ())