eventium-postgresql 0.1.0 → 0.2.1
raw patch · 7 files changed
+239/−148 lines, 7 filesdep +eventium-testkitdep −eventium-test-helpersdep ~eventium-coredep ~eventium-sql-commondep ~persistent
Dependencies added: eventium-testkit
Dependencies removed: eventium-test-helpers
Dependency ranges changed: eventium-core, eventium-sql-common, persistent
Files
- CHANGELOG.md +17/−0
- README.md +41/−127
- eventium-postgresql.cabal +31/−11
- src/Eventium/ProjectionCache/Postgresql.hs +41/−0
- src/Eventium/Store/Postgresql.hs +12/−2
- tests/Eventium/ProjectionCache/PostgresqlSpec.hs +92/−0
- tests/Eventium/Store/PostgresqlSpec.hs +5/−8
CHANGELOG.md view
@@ -0,0 +1,17 @@+# eventium-postgresql Changelog++## 0.2.1 (Unreleased)++- `postgresqlCheckpointStore` -- PostgreSQL-backed `CheckpointStore` for `SequenceNumber` tracking.+- Re-exports `CheckpointName` from `eventium-sql-common`.++## 0.2.0 (Unreleased)++- Updated for eventium-core 0.2.0 API changes.+- Single `events` table with metadata columns (event_type, correlation_id, causation_id, created_at).+- Uses `LOCK IN EXCLUSIVE MODE` for monotonic global sequence ordering.+- Enabled `NoFieldSelectors`, `DuplicateRecordFields`, `OverloadedRecordDot` default extensions.++## 0.1.0++Initial release.
README.md view
@@ -1,42 +1,27 @@ # Eventium PostgreSQL -PostgreSQL-based event store implementation for production event sourcing systems.+PostgreSQL event store backend for Eventium. ## Overview -`eventium-postgresql` provides a robust, production-ready event store implementation backed by PostgreSQL. It leverages PostgreSQL's ACID guarantees, indexing capabilities, and reliability for persistent event storage with high performance and data integrity.--## Features--- ✅ **ACID Transactions** - Full consistency guarantees-- ✅ **Optimistic Concurrency** - Prevents lost updates with version checks-- ✅ **Efficient Indexing** - Fast event retrieval by stream and position-- ✅ **Global Event Ordering** - Sequence numbers for time-ordered queries-- ✅ **Type-Safe Access** - Uses Persistent library for database operations-- ✅ **Production Ready** - Battle-tested PostgreSQL backend-- ✅ **Multi-Process Support** - Concurrent access from multiple applications--## Database Schema--The implementation creates two main tables:-- **Events Table** - Stores events with aggregate keys, versions, and payloads-- **Global Events Table** - Maintains global ordering with sequence numbers+`eventium-postgresql` provides a production-grade event store backed by PostgreSQL.+It uses the `persistent` library for type-safe database access and guarantees+monotonically increasing global sequence numbers via `LOCK IN EXCLUSIVE MODE`. -Indexes ensure fast lookups by:-- Stream key + version-- Global sequence number-- Event types (for projections)+## API -## Installation+```haskell+postgresqlEventStoreWriter+ :: (MonadIO m)+ => SqlEventStoreConfig entity serialized+ -> VersionedEventStoreWriter (SqlPersistT m) serialized+``` -Add to your `package.yaml`:+Readers come from `eventium-sql-common` (re-exported): -```yaml-dependencies:- - eventium-core- - eventium-sql-common- - eventium-postgresql- - persistent-postgresql # PostgreSQL driver+```haskell+sqlEventStoreReader :: SqlEventStoreConfig entity serialized -> VersionedEventStoreReader (SqlPersistT m) serialized+sqlGlobalEventStoreReader :: SqlEventStoreConfig entity serialized -> GlobalEventStoreReader (SqlPersistT m) serialized ``` ## Usage@@ -46,120 +31,49 @@ import Database.Persist.Postgresql main :: IO ()-main = do- let connStr = "host=localhost dbname=eventstore user=postgres"- +main = runStdoutLoggingT $ withPostgresqlPool connStr 10 $ \pool -> do- -- Initialize schema+ -- Use defaultSqlEventStoreConfig for the standard schema+ let writer = postgresqlEventStoreWriter defaultSqlEventStoreConfig+ reader = sqlEventStoreReader defaultSqlEventStoreConfig flip runSqlPool pool $ do- runMigration migrateAll- - -- Create event store- let store = makePostgresqlEventStore pool- - -- Use with command handlers- result <- applyCommandHandler - (eventStoreWriter store)- (eventStoreReader store)- commandHandler- aggregateId- command-```--## Configuration--### Connection String--```haskell--- Basic connection-"host=localhost port=5432 dbname=mydb user=myuser password=mypass"---- With connection pool-withPostgresqlPool connectionString poolSize $ \pool -> ...-```--### Connection Pooling--Recommended settings for production:-```haskell--- Pool size based on concurrent requests-poolSize = numCores * 2 + effectiveSpindleCount---- Example: 10 connections for typical web app-withPostgresqlPool connStr 10 $ \pool -> ...+ runMigration migrateSqlEvent+ -- writer and reader are ready to use ``` ## Setup -### Start PostgreSQL with Docker- ```bash-# Using docker-compose (provided in project root)-docker-compose up -d postgres--# Or manually-docker run -d \- --name eventium-postgres \- -e POSTGRES_PASSWORD=postgres \- -e POSTGRES_DB=eventstore \- -p 5432:5432 \- postgres:15-```--### Run Migrations+# Start PostgreSQL with docker-compose (from project root)+docker compose up -d -```haskell-runSqlPool (runMigration migrateAll) pool+# Default connection settings (matching docker-compose.yaml):+# POSTGRES_HOST=127.0.0.1 POSTGRES_PORT=5432+# POSTGRES_USER=postgres POSTGRES_PASSWORD=password+# POSTGRES_DBNAME=eventium_test ``` -## Performance--PostgreSQL provides excellent performance characteristics:-- **Writes**: ~1000-5000 events/sec (single connection)-- **Reads**: ~10000-50000 events/sec (with proper indexing)-- **Scalability**: Read replicas for query scaling--See `postgres-event-store-bench/` for benchmarking scripts.--## Best Practices--1. **Use Connection Pooling** - Essential for web applications-2. **Index Strategy** - Default indexes cover common queries-3. **Backup Strategy** - Regular PostgreSQL backups-4. **Monitoring** - Watch connection pool usage and query performance-5. **Read Replicas** - Scale read models with PostgreSQL replication--## Production Considerations--- **High Availability** - Use PostgreSQL replication-- **Backup & Recovery** - Point-in-time recovery with WAL archiving-- **Monitoring** - Track event growth and query performance-- **Connection Limits** - Configure max_connections appropriately+## Concurrency -## Example: Complete Setup+PostgreSQL uses `LOCK table_name IN EXCLUSIVE MODE` during writes. This ensures+that auto-increment IDs are assigned in commit order, so concurrent readers+always see a gapless, monotonically increasing global sequence. -```haskell-import Eventium.Store.Postgresql-import Control.Monad.Logger (runStdoutLoggingT)+## Installation -setupEventStore :: IO ()-setupEventStore = runStdoutLoggingT $ do- let connStr = "host=localhost dbname=eventstore"- - withPostgresqlPool connStr 10 $ \pool -> do- -- Run migrations- flip runSqlPool pool $ runMigration migrateAll- - -- Event store is ready to use- liftIO $ putStrLn "Event store initialized"+```yaml+dependencies:+ - eventium-core+ - eventium-postgresql+ - persistent-postgresql ``` ## Documentation -- [Main README](../README.md) - Project overview-- [SQL Common](../eventium-sql-common/) - Shared SQL utilities-- [Design Documentation](../DESIGN.md) - Architecture details+- [Main README](../README.md)+- [SQL Common](../eventium-sql-common/)+- [Design](../DESIGN.md) ## License -MIT - see [LICENSE.md](LICENSE.md)+MIT -- see [LICENSE.md](LICENSE.md)
eventium-postgresql.cabal view
@@ -1,11 +1,11 @@ cabal-version: 1.12 --- This file has been generated from package.yaml by hpack version 0.37.0.+-- This file has been generated from package.yaml by hpack version 0.38.3. -- -- see: https://github.com/sol/hpack name: eventium-postgresql-version: 0.1.0+version: 0.2.1 synopsis: Postgres implementations for eventium description: Eventium-postgresql provides a PostgreSQL-based event store implementation for the Eventium event sourcing framework. It uses the Persistent library for type-safe database access and provides@@ -27,36 +27,54 @@ type: git location: https://github.com/aleks-sidorenko/eventium +flag ci+ description: Enable -Werror for CI builds+ manual: True+ default: False+ library exposed-modules:+ Eventium.ProjectionCache.Postgresql Eventium.Store.Postgresql other-modules: Paths_eventium_postgresql hs-source-dirs: src- ghc-options: -Wall+ default-extensions:+ NoFieldSelectors+ DuplicateRecordFields+ OverloadedRecordDot+ ghc-options: -Wall -Wcompat -Widentities -Wincomplete-record-updates -Wredundant-constraints -Wno-type-defaults -Wno-incomplete-uni-patterns build-depends: aeson >=1.5 && <2.3 , base >=4.9 && <5 , bytestring >=0.10 && <0.13- , eventium-core- , eventium-sql-common+ , eventium-core >=0.2.0 && <0.3.0+ , eventium-sql-common >=0.2.0 && <0.3.0 , mtl >=2.2 && <2.4- , persistent >=2.13 && <2.15+ , persistent >=2.14 && <2.18 , text >=1.2 && <2.2 default-language: Haskell2010+ if flag(ci)+ ghc-options: -Werror test-suite spec type: exitcode-stdio-1.0 main-is: Spec.hs other-modules:+ Eventium.ProjectionCache.PostgresqlSpec Eventium.Store.PostgresqlSpec+ Eventium.ProjectionCache.Postgresql Eventium.Store.Postgresql Paths_eventium_postgresql hs-source-dirs: tests src- ghc-options: -Wall+ default-extensions:+ NoFieldSelectors+ DuplicateRecordFields+ OverloadedRecordDot+ ghc-options: -Wall -Wcompat -Widentities -Wincomplete-record-updates -Wredundant-constraints -Wno-type-defaults -Wno-incomplete-uni-patterns build-tool-depends: hspec-discover:hspec-discover build-depends:@@ -64,13 +82,15 @@ , aeson >=1.5 && <2.3 , base >=4.9 && <5 , bytestring >=0.10 && <0.13- , eventium-core- , eventium-sql-common- , eventium-test-helpers+ , eventium-core >=0.2.0 && <0.3.0+ , eventium-sql-common >=0.2.0 && <0.3.0+ , eventium-testkit , hspec , mtl >=2.2 && <2.4- , persistent >=2.13 && <2.15+ , persistent >=2.14 && <2.18 , persistent-postgresql , text >=1.2 && <2.2 , utf8-string default-language: Haskell2010+ if flag(ci)+ ghc-options: -Werror
+ src/Eventium/ProjectionCache/Postgresql.hs view
@@ -0,0 +1,41 @@+module Eventium.ProjectionCache.Postgresql+ ( postgresqlVersionedProjectionCache,+ postgresqlGlobalProjectionCache,+ postgresqlCheckpointStore,+ CheckpointName (..),+ migrateProjectionSnapshot,+ )+where++import Control.Monad.IO.Class (MonadIO)+import Database.Persist.Sql (SqlPersistT)+import Eventium.EventSubscription (CheckpointStore)+import Eventium.ProjectionCache.Sql (CheckpointName (..), ProjectionName, migrateProjectionSnapshot, sqlCheckpointStore, sqlGlobalProjectionCache, sqlVersionedProjectionCache)+import Eventium.ProjectionCache.Types (ProjectionCache)+import Eventium.Store.Class (EventVersion, SequenceNumber)+import Eventium.Store.Sql.JSONString (JSONString)+import Eventium.UUID (UUID)++-- | PostgreSQL-backed 'ProjectionCache' for per-entity snapshots.+-- Alias for 'sqlVersionedProjectionCache'.+postgresqlVersionedProjectionCache ::+ (MonadIO m) =>+ ProjectionName ->+ ProjectionCache UUID EventVersion JSONString (SqlPersistT m)+postgresqlVersionedProjectionCache = sqlVersionedProjectionCache++-- | PostgreSQL-backed 'ProjectionCache' for global blob snapshots.+-- Alias for 'sqlGlobalProjectionCache'.+postgresqlGlobalProjectionCache ::+ (MonadIO m) =>+ ProjectionName ->+ ProjectionCache () SequenceNumber JSONString (SqlPersistT m)+postgresqlGlobalProjectionCache = sqlGlobalProjectionCache++-- | PostgreSQL-backed 'CheckpointStore' for tracking subscription position.+-- Alias for 'sqlCheckpointStore'.+postgresqlCheckpointStore ::+ (MonadIO m) =>+ CheckpointName ->+ CheckpointStore (SqlPersistT m) SequenceNumber+postgresqlCheckpointStore = sqlCheckpointStore
src/Eventium/Store/Postgresql.hs view
@@ -5,16 +5,15 @@ -- | Defines an Postgresql event store. module Eventium.Store.Postgresql ( postgresqlEventStoreWriter,+ postgresqlEventStoreWriterTagged, module Eventium.Store.Class, module Eventium.Store.Sql, ) where import Control.Monad.Reader-import Data.Monoid ((<>)) import Data.Text (Text) import Database.Persist-import Database.Persist.Names (EntityNameDB (..), FieldNameDB (..)) import Database.Persist.Sql import Eventium.Store.Class import Eventium.Store.Sql@@ -33,6 +32,17 @@ maxPostgresVersionSql :: FieldNameDB -> FieldNameDB -> FieldNameDB -> Text maxPostgresVersionSql (FieldNameDB tableName) (FieldNameDB uuidFieldName) (FieldNameDB versionFieldName) = "SELECT COALESCE(MAX(" <> versionFieldName <> "), -1) FROM " <> tableName <> " WHERE " <> uuidFieldName <> " = ?"++-- | Like 'postgresqlEventStoreWriter' but accepts 'TaggedEvent's,+-- preserving the metadata attached to each event.+postgresqlEventStoreWriterTagged ::+ (MonadIO m, PersistEntity entity, PersistEntityBackend entity ~ SqlBackend, SafeToInsert entity) =>+ SqlEventStoreConfig entity serialized ->+ VersionedEventStoreWriter (SqlPersistT m) (TaggedEvent serialized)+postgresqlEventStoreWriterTagged config = EventStoreWriter $ transactionalExpectedWriteHelper getLatestVersion storeEvents'+ where+ getLatestVersion = sqlMaxEventVersion config maxPostgresVersionSql+ storeEvents' = sqlStoreEventsTagged config (Just tableLockFunc) maxPostgresVersionSql -- | We need to lock the events table or else our global sequence number might -- not be monotonically increasing over time from the point of view of a
+ tests/Eventium/ProjectionCache/PostgresqlSpec.hs view
@@ -0,0 +1,92 @@+{-# LANGUAGE OverloadedStrings #-}++module Eventium.ProjectionCache.PostgresqlSpec (spec) where++import Data.ByteString (ByteString)+import qualified Data.ByteString.UTF8 as UTF8+import Database.Persist.Postgresql+import Eventium.ProjectionCache.Cache (codecProjectionCache)+import Eventium.ProjectionCache.Postgresql+import Eventium.Store.Postgresql+import Eventium.Testkit+import System.Environment (lookupEnv)+import Test.Hspec++spec :: Spec+spec = do+ describe "PostgreSQL versioned projection cache" $ do+ versionedProjectionCacheSpec postgresVersionedProjectionCacheRunner++ describe "PostgreSQL global projection cache" $ do+ globalProjectionCacheSpec postgresGlobalProjectionCacheRunner++ describe "PostgreSQL checkpoint store" $ do+ checkpointStoreSpec postgresCheckpointStoreRunner++makePool :: IO ConnectionPool+makePool = do+ let makeConnString host port user pass db =+ "host="+ <> host+ <> " port="+ <> port+ <> " user="+ <> user+ <> " dbname="+ <> db+ <> " password="+ <> pass+ connString <-+ makeConnString+ <$> getEnvDef "POSTGRES_HOST" "127.0.0.1"+ <*> getEnvDef "POSTGRES_PORT" "5432"+ <*> getEnvDef "POSTGRES_USER" "postgres"+ <*> getEnvDef "POSTGRES_PASSWORD" "password"+ <*> getEnvDef "POSTGRES_DBNAME" "eventium_test"+ pool <- runNoLoggingT (createPostgresqlPool connString 1)+ flip runSqlPool pool $ do+ void $ runMigrationSilent migrateSqlEvent+ void $ runMigrationSilent migrateProjectionSnapshot+ truncateTables+ return pool++getEnvDef :: String -> ByteString -> IO ByteString+getEnvDef name def = maybe def UTF8.fromString <$> lookupEnv name++truncateTables :: (MonadIO m) => SqlPersistT m ()+truncateTables = do+ rawExecute "TRUNCATE TABLE events RESTART IDENTITY" []+ rawExecute "TRUNCATE TABLE projection_snapshots" []++postgresVersionedProjectionCacheRunner :: VersionedProjectionCacheRunner (SqlPersistT IO)+postgresVersionedProjectionCacheRunner = VersionedProjectionCacheRunner $ \action -> do+ pool <- makePool+ let writer =+ codecEventStoreWriter jsonStringCodec $+ postgresqlEventStoreWriter defaultSqlEventStoreConfig+ reader =+ codecVersionedEventStoreReader jsonStringCodec $+ sqlEventStoreReader defaultSqlEventStoreConfig+ cache =+ codecProjectionCache jsonStringCodec $+ postgresqlVersionedProjectionCache (ProjectionName "test_versioned")+ runSqlPool (action writer reader cache) pool++postgresGlobalProjectionCacheRunner :: GlobalProjectionCacheRunner (SqlPersistT IO)+postgresGlobalProjectionCacheRunner = GlobalProjectionCacheRunner $ \action -> do+ pool <- makePool+ let writer =+ codecEventStoreWriter jsonStringCodec $+ postgresqlEventStoreWriter defaultSqlEventStoreConfig+ globalReader =+ codecGlobalEventStoreReader jsonStringCodec $+ sqlGlobalEventStoreReader defaultSqlEventStoreConfig+ cache =+ codecProjectionCache jsonStringCodec $+ postgresqlGlobalProjectionCache (ProjectionName "test_global")+ runSqlPool (action writer globalReader cache) pool++postgresCheckpointStoreRunner :: CheckpointStoreRunner (SqlPersistT IO)+postgresCheckpointStoreRunner = CheckpointStoreRunner $ \action -> do+ pool <- makePool+ runSqlPool (action (postgresqlCheckpointStore (CheckpointName "test_checkpoint"))) pool
tests/Eventium/Store/PostgresqlSpec.hs view
@@ -2,14 +2,11 @@ module Eventium.Store.PostgresqlSpec (spec) where -import Control.Monad.Reader (ask) import Data.ByteString (ByteString) import qualified Data.ByteString.UTF8 as UTF8-import Data.Maybe (maybe)-import Data.Text (Text) import Database.Persist.Postgresql import Eventium.Store.Postgresql-import Eventium.TestHelpers+import Eventium.Testkit import System.Environment (lookupEnv) import Test.Hspec @@ -39,14 +36,14 @@ <> " password=" <> pass writer =- serializedEventStoreWriter jsonStringSerializer $+ codecEventStoreWriter jsonStringCodec $ postgresqlEventStoreWriter defaultSqlEventStoreConfig reader =- serializedVersionedEventStoreReader jsonStringSerializer $+ codecVersionedEventStoreReader jsonStringCodec $ sqlEventStoreReader defaultSqlEventStoreConfig connString <- makeConnString- <$> getEnvDef "POSTGRES_HOST" "localhost"+ <$> getEnvDef "POSTGRES_HOST" "127.0.0.1" <*> getEnvDef "POSTGRES_PORT" "5432" <*> getEnvDef "POSTGRES_USER" "postgres" <*> getEnvDef "POSTGRES_PASSWORD" "password"@@ -73,5 +70,5 @@ postgresStoreGlobalRunner :: GlobalStreamEventStoreRunner (SqlPersistT IO) postgresStoreGlobalRunner = GlobalStreamEventStoreRunner $ \action -> do (writer, _, pool) <- makeStore- let globalReader = serializedGlobalEventStoreReader jsonStringSerializer (sqlGlobalEventStoreReader defaultSqlEventStoreConfig)+ let globalReader = codecGlobalEventStoreReader jsonStringCodec (sqlGlobalEventStoreReader defaultSqlEventStoreConfig) runSqlPool (action writer globalReader) pool