packages feed

domaindriven-core (empty) → 0.5.0

raw patch · 13 files changed

+1183/−0 lines, 13 filesdep +aesondep +asyncdep +basesetup-changed

Dependencies added: aeson, async, base, bytestring, containers, deepseq, domaindriven-core, exceptions, generic-lens, hspec, http-types, microlens, mtl, postgresql-simple, random, streamly, template-haskell, time, transformers, unliftio, unliftio-pool, unordered-containers, uuid, vector

Files

+ ChangeLog.md view
@@ -0,0 +1,5 @@+# Changelog for domaindriven++## 0.5.0++First release published on hackage.
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright Tommy Engström (c) 2019++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++    * Redistributions of source code must retain the above copyright+      notice, this list of conditions and the following disclaimer.++    * Redistributions in binary form must reproduce the above+      copyright notice, this list of conditions and the following+      disclaimer in the documentation and/or other materials provided+      with the distribution.++    * Neither the name of Tommy Engström nor the names of other+      contributors may be used to endorse or promote products derived+      from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ README.md view
@@ -0,0 +1,3 @@+# domaindriven-core++This library contains the core model and persistance handlers for domaindriven.
+ Setup.hs view
@@ -0,0 +1,3 @@+import Distribution.Simple++main = defaultMain
+ domaindriven-core.cabal view
@@ -0,0 +1,151 @@+cabal-version: 1.12++-- This file has been generated from package.yaml by hpack version 0.35.1.+--+-- see: https://github.com/sol/hpack++name:           domaindriven-core+version:        0.5.0+synopsis:       Batteries included event sourcing and CQRS+description:    Please see the README on GitHub at <https://github.com/tommyengstrom/domaindriven/tree/master/domaindriven-core#readme>+category:       Web+homepage:       https://github.com/tommyengstrom/domaindriven#readme+bug-reports:    https://github.com/tommyengstrom/domaindriven/issues+author:         Tommy Engström+maintainer:     tommy@tommyengstrom.com+copyright:      2023 Tommy Engström+license:        BSD3+license-file:   LICENSE+build-type:     Simple+extra-source-files:+    README.md+    ChangeLog.md++source-repository head+  type: git+  location: https://github.com/tommyengstrom/domaindriven++library+  exposed-modules:+      DomainDriven.Persistance.Class+      DomainDriven.Persistance.ForgetfulInMemory+      DomainDriven.Persistance.Postgres+      DomainDriven.Persistance.Postgres.Internal+      DomainDriven.Persistance.Postgres.Migration+      DomainDriven.Persistance.Postgres.Types+  other-modules:+      Paths_domaindriven_core+  hs-source-dirs:+      src+  default-extensions:+      Arrows+      ConstraintKinds+      DataKinds+      DeriveAnyClass+      DeriveFunctor+      DeriveGeneric+      DeriveTraversable+      DerivingStrategies+      DuplicateRecordFields+      FlexibleContexts+      FlexibleInstances+      FunctionalDependencies+      GADTs+      GeneralizedNewtypeDeriving+      ImportQualifiedPost+      LambdaCase+      MultiParamTypeClasses+      MultiWayIf+      NamedFieldPuns+      NoImplicitPrelude+      OverloadedLabels+      OverloadedStrings+      PolyKinds+      RankNTypes+      ScopedTypeVariables+      StandaloneDeriving+      TupleSections+      TypeApplications+      TypeFamilyDependencies+      TypeOperators+      TypeSynonymInstances+      ViewPatterns+  ghc-options: -Wall -Wcompat -Widentities -Wincomplete-record-updates -Wincomplete-uni-patterns -Wpartial-fields -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-patterns+  build-depends:+      aeson >=2.0.3 && <2.2+    , async >=2.2.4 && <2.3+    , base >=4.7 && <5+    , bytestring >=0.11.3 && <0.12+    , containers >=0.6.5.1 && <0.7+    , deepseq >=1.4.6.1 && <1.5+    , exceptions >=0.10.4 && <0.11+    , generic-lens >=2.2.1.0 && <2.3+    , http-types >=0.12.3 && <0.13+    , microlens >=0.4.12.0 && <0.5+    , mtl >=2.2.2 && <2.3+    , postgresql-simple >=0.6.4 && <0.7+    , random >=1.2.1.1 && <1.3+    , streamly >=0.8.1.1 && <0.9+    , template-haskell >=2.18.0.0 && <2.19+    , time >=1.11.1 && <1.12+    , transformers >=0.5.6.2 && <0.6+    , unliftio >=0.2.0.1 && <0.3+    , unliftio-pool >=0.2.2.0 && <0.3+    , unordered-containers >=0.2.19.1 && <0.3+    , uuid >=1.3.15 && <1.4+    , vector >=0.12.3.1 && <0.14+  default-language: Haskell2010++test-suite domaindriven-core-test+  type: exitcode-stdio-1.0+  main-is: Spec.hs+  other-modules:+      DomainDriven.Persistance.PostgresIORefStateSpec+      Paths_domaindriven_core+  hs-source-dirs:+      test+  default-extensions:+      Arrows+      ConstraintKinds+      DataKinds+      DeriveAnyClass+      DeriveFunctor+      DeriveGeneric+      DeriveTraversable+      DerivingStrategies+      DuplicateRecordFields+      FlexibleContexts+      FlexibleInstances+      FunctionalDependencies+      GADTs+      GeneralizedNewtypeDeriving+      ImportQualifiedPost+      LambdaCase+      MultiParamTypeClasses+      MultiWayIf+      NamedFieldPuns+      NoImplicitPrelude+      OverloadedLabels+      OverloadedStrings+      PolyKinds+      RankNTypes+      ScopedTypeVariables+      StandaloneDeriving+      TupleSections+      TypeApplications+      TypeFamilyDependencies+      TypeOperators+      TypeSynonymInstances+      ViewPatterns+  ghc-options: -Wall -Wcompat -Widentities -Wincomplete-record-updates -Wincomplete-uni-patterns -Wpartial-fields -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-patterns -threaded -rtsopts -with-rtsopts=-N -Wall -Wunused-packages+  build-depends:+      aeson >=2.0.3 && <2.2+    , base >=4.7 && <5+    , domaindriven-core >=0.5.0 && <0.6+    , hspec >=2.9.7 && <2.10+    , postgresql-simple >=0.6.4 && <0.7+    , streamly >=0.8.1.1 && <0.9+    , time >=1.11.1 && <1.12+    , unliftio-pool >=0.2.2.0 && <0.3+    , uuid >=1.3.15 && <1.4+  default-language: Haskell2010
+ src/DomainDriven/Persistance/Class.hs view
@@ -0,0 +1,61 @@+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE TypeFamilyDependencies #-}+{-# LANGUAGE TypeSynonymInstances #-}+{-# LANGUAGE UndecidableInstances #-}++module DomainDriven.Persistance.Class where++import Control.DeepSeq (NFData)+import Control.Monad.Reader+import Data.Aeson+import Data.Kind+import Data.Time+import Data.UUID (UUID)+import GHC.Generics (Generic)+import Streamly.Prelude (SerialT)+import System.Random+import UnliftIO+import Prelude++class ReadModel p where+    type Model p :: Type+    type Event p :: Type+    applyEvent :: p -> Model p -> Stored (Event p) -> Model p+    getModel :: p -> IO (Model p)+    getEventList :: p -> IO [Stored (Event p)]+    getEventStream :: p -> SerialT IO (Stored (Event p))++class ReadModel p => WriteModel p where+    transactionalUpdate+        :: forall m a+         . MonadUnliftIO m+        => p+        -> (Model p -> m (Model p -> a, [Event p]))+        -> m a++-- | Wrapper for stored data+-- This ensures all events have a unique ID and a timestamp, without having to deal with+-- that when implementing the model.+data Stored a = Stored+    { storedEvent :: a+    , storedTimestamp :: UTCTime+    , storedUUID :: UUID+    }+    deriving+        ( Show+        , Eq+        , Ord+        , Generic+        , FromJSON+        , ToJSON+        , Functor+        , Foldable+        , Traversable+        , NFData+        )++mkId :: MonadIO m => m UUID+mkId = liftIO randomIO++toStored :: MonadIO m => e -> m (Stored e)+toStored e = Stored e <$> liftIO getCurrentTime <*> mkId
+ src/DomainDriven/Persistance/ForgetfulInMemory.hs view
@@ -0,0 +1,51 @@+module DomainDriven.Persistance.ForgetfulInMemory where++import Data.List (foldl')+import DomainDriven.Persistance.Class+import GHC.Generics (Generic)+import qualified Streamly.Prelude as S+import UnliftIO+import Prelude++createForgetful+    :: MonadIO m+    => (model -> Stored event -> model)+    -> model+    -- ^ initial model+    -> m (ForgetfulInMemory model event)+createForgetful appEvent m0 = do+    state <- newIORef m0+    evs <- newIORef []+    lock <- newQSem 1+    pure $ ForgetfulInMemory state appEvent m0 evs lock++-- | STM state without event persistance+data ForgetfulInMemory model event = ForgetfulInMemory+    { stateRef :: IORef model+    , apply :: model -> Stored event -> model+    , seed :: model+    , events :: IORef [Stored event]+    , lock :: QSem+    }+    deriving (Generic)++instance ReadModel (ForgetfulInMemory model e) where+    type Model (ForgetfulInMemory model e) = model+    type Event (ForgetfulInMemory model e) = e+    applyEvent ff = apply ff+    getModel ff = readIORef $ stateRef ff+    getEventList ff = readIORef $ events ff+    getEventStream ff = do+        l <- liftIO $ getEventList ff+        S.fromList l++instance WriteModel (ForgetfulInMemory model e) where+    transactionalUpdate ff evalCmd =+        bracket_ (waitQSem $ lock ff) (signalQSem $ lock ff) $ do+            model <- readIORef $ stateRef ff+            (returnFun, evs) <- evalCmd model+            storedEvs <- traverse toStored evs+            let newModel = foldl' (apply ff) model storedEvs+            modifyIORef (events ff) (<> storedEvs)+            writeIORef (stateRef ff) newModel+            pure $ returnFun newModel
+ src/DomainDriven/Persistance/Postgres.hs view
@@ -0,0 +1,20 @@+module DomainDriven.Persistance.Postgres+    ( module X+    )+where++import DomainDriven.Persistance.Postgres.Internal as X+    ( PostgresEvent (..)+    , postgresWriteModel+    , postgresWriteModelNoMigration+    , simplePool+    , simplePool'+    )+import DomainDriven.Persistance.Postgres.Types as X+    ( ChunkSize+    , EventMigration+    , EventTable (..)+    , EventTableBaseName+    , EventTableName+    , PreviousEventTableName+    )
+ src/DomainDriven/Persistance/Postgres/Internal.hs view
@@ -0,0 +1,465 @@+-- | Postgres events with state as an IORef+module DomainDriven.Persistance.Postgres.Internal where++import Control.Monad+import Control.Monad.Catch+import Control.Monad.IO.Class+import Data.Aeson+import Data.Foldable+import Data.Generics.Product+import Data.IORef+import Data.Int+import Data.Sequence (Seq (..))+import qualified Data.Sequence as Seq+import Data.String+import Database.PostgreSQL.Simple as PG+import qualified Database.PostgreSQL.Simple.Cursor as Cursor+import DomainDriven.Persistance.Class+import DomainDriven.Persistance.Postgres.Types+import GHC.Generics (Generic)+import Lens.Micro+    ( to+    , (^.)+    )+import qualified Streamly.Data.Unfold as Unfold+import qualified Streamly.Prelude as S+import UnliftIO (MonadUnliftIO (..))+import UnliftIO.Pool+    ( LocalPool+    , Pool+    , createPool+    , destroyResource+    , putResource+    , takeResource+    , withResource+    )+import Prelude++data PostgresEvent model event = PostgresEvent+    { connectionPool :: Pool Connection+    , eventTableName :: EventTableName+    , modelIORef :: IORef (NumberedModel model)+    , app :: model -> Stored event -> model+    , seed :: model+    , chunkSize :: ChunkSize+    -- ^ Number of events read from postgres per batch+    }+    deriving (Generic)++data PostgresEventTrans model event = PostgresEventTrans+    { transaction :: OngoingTransaction+    , eventTableName :: EventTableName+    , modelIORef :: IORef (NumberedModel model)+    , app :: model -> Stored event -> model+    , seed :: model+    , chunkSize :: ChunkSize+    -- ^ Number of events read from postgres per batch+    }+    deriving (Generic)++instance (FromJSON e) => ReadModel (PostgresEvent m e) where+    type Model (PostgresEvent m e) = m+    type Event (PostgresEvent m e) = e+    applyEvent pg = pg ^. field @"app"+    getModel pg = withIOTrans pg getModel'++    getEventList pg = withResource (connectionPool pg) $ \conn ->+        fmap fst <$> queryEvents conn (pg ^. field @"eventTableName")++    getEventStream pg = withStreamReadTransaction pg getEventStream'++getEventTableName :: EventTable -> EventTableName+getEventTableName = go 0+  where+    go :: Int -> EventTable -> String+    go i = \case+        MigrateUsing _ u -> go (i + 1) u+        InitialVersion n -> n <> "_v" <> show (i + 1)++-- | Create the table required for storing state and events, if they do not yet exist.+createEventTable+    :: (FromJSON e, WriteModel (PostgresEventTrans m e))+    => PostgresEventTrans m e+    -> IO ()+createEventTable pgt = do+    void (getModel pgt)+        `catch` ( const @_ @SqlError $ do+                    let etName = pgt ^. field @"eventTableName"+                    _ <-+                        createEventTable'+                            (pgt ^. field @"transaction" . to connection)+                            etName+                    void $ refreshModel pgt+                )++createEventTable' :: Connection -> EventTableName -> IO Int64+createEventTable' conn eventTable =+    execute_ conn $+        "create table if not exists \""+            <> fromString eventTable+            <> "\" \+               \( id uuid primary key\+               \, event_number bigint not null generated always as identity\+               \, timestamp timestamptz not null default now()\+               \, event jsonb not null\+               \);"++retireTable :: Connection -> EventTableName -> IO ()+retireTable conn tableName = do+    createRetireFunction conn+    void $+        execute_ conn $+            "create trigger retired before insert on \""+                <> fromString tableName+                <> "\" execute procedure retired_table()"++createRetireFunction :: Connection -> IO ()+createRetireFunction conn =+    void+        . execute_ conn+        $ "create or replace function retired_table() returns trigger as \+          \$$ begin raise exception 'Event table has been retired.'; end; $$ \+          \language plpgsql;"++simplePool' :: MonadUnliftIO m => PG.ConnectInfo -> m (Pool Connection)+simplePool' connInfo =+    createPool (liftIO $ PG.connect connInfo) (liftIO . PG.close) 1 5 5++simplePool :: MonadUnliftIO m => IO Connection -> m (Pool Connection)+simplePool getConn =+    createPool (liftIO getConn) (liftIO . PG.close) 1 5 5++-- | Setup the persistance model and verify that the tables exist.+postgresWriteModelNoMigration+    :: (FromJSON e, WriteModel (PostgresEventTrans m e))+    => Pool Connection+    -> EventTableName+    -> (m -> Stored e -> m)+    -> m+    -> IO (PostgresEvent m e)+postgresWriteModelNoMigration pool eventTable app' seed' = do+    pg <- createPostgresPersistance pool eventTable app' seed'+    withIOTrans pg createEventTable+    pure pg++-- | Setup the persistance model and verify that the tables exist.+postgresWriteModel+    :: Pool Connection+    -> EventTable+    -> (m -> Stored e -> m)+    -> m+    -> IO (PostgresEvent m e)+postgresWriteModel pool eventTable app' seed' = do+    pg <- createPostgresPersistance pool (getEventTableName eventTable) app' seed'+    withIOTrans pg $ \pgt -> runMigrations (pgt ^. field @"transaction") eventTable+    pure pg++newtype Exists = Exists+    { exists :: Bool+    }+    deriving (Show, Eq, Generic)+    deriving anyclass (FromRow)++runMigrations :: OngoingTransaction -> EventTable -> IO ()+runMigrations trans et = do+    tableExistQuery <-+        query+            conn+            "select exists (select * from information_schema.tables where table_schema='public' and table_name=?)"+            (Only $ getEventTableName et)++    case (et, tableExistQuery) of+        (InitialVersion _, [Only True]) -> pure ()+        (MigrateUsing{}, [Only True]) -> pure ()+        (InitialVersion _, [Only False]) -> createTable+        (MigrateUsing mig prevEt, [Only False]) -> do+            runMigrations trans prevEt+            createTable+            mig (getEventTableName prevEt) (getEventTableName et) conn+            retireTable conn (getEventTableName prevEt)+        (_, r) -> fail $ "Unexpected table query result: " <> show r+  where+    conn :: Connection+    conn = connection trans++    createTable :: IO ()+    createTable = do+        let tableName = getEventTableName et+        void $ createEventTable' conn tableName++createPostgresPersistance+    :: forall event model+     . Pool Connection+    -> EventTableName+    -> (model -> Stored event -> model)+    -> model+    -> IO (PostgresEvent model event)+createPostgresPersistance pool eventTable app' seed' = do+    ref <- newIORef $ NumberedModel seed' 0+    pure $+        PostgresEvent+            { connectionPool = pool+            , eventTableName = eventTable+            , modelIORef = ref+            , app = app'+            , seed = seed'+            , chunkSize = 50+            }++queryEvents+    :: (FromJSON a)+    => Connection+    -> EventTableName+    -> IO [(Stored a, EventNumber)]+queryEvents conn eventTable = do+    traverse fromEventRow =<< query_ conn q+  where+    q :: PG.Query+    q =+        "select id, event_number,timestamp,event from \""+            <> fromString eventTable+            <> "\" order by event_number"++queryEventsAfter+    :: (FromJSON a)+    => Connection+    -> EventTableName+    -> EventNumber+    -> IO [(Stored a, EventNumber)]+queryEventsAfter conn eventTable (EventNumber lastEvent) =+    traverse fromEventRow+        =<< query_+            conn+            ( "select id, event_number,timestamp,event from \""+                <> fromString eventTable+                <> "\" where event_number > "+                <> fromString (show lastEvent)+                <> " order by event_number"+            )++newtype EventQuery = EventQuery {getPgQuery :: PG.Query}+    deriving (Show, Generic)++mkEventsAfterQuery :: EventTableName -> EventNumber -> EventQuery+mkEventsAfterQuery eventTable (EventNumber lastEvent) =+    EventQuery $+        "select id, event_number,timestamp,event from \""+            <> fromString eventTable+            <> "\" where event_number > "+            <> fromString (show lastEvent)+            <> " order by event_number"++mkEventQuery :: EventTableName -> EventQuery+mkEventQuery eventTable =+    EventQuery $+        "select id, event_number,timestamp,event from \""+            <> fromString eventTable+            <> "\" order by event_number"++headMay :: [a] -> Maybe a+headMay = \case+    a : _ -> Just a+    [] -> Nothing++queryHasEventsAfter :: Connection -> EventTableName -> EventNumber -> IO Bool+queryHasEventsAfter conn eventTable (EventNumber lastEvent) =+    maybe True fromOnly . headMay <$> query_ conn q+  where+    q :: PG.Query+    q =+        "select count(*) > 0 from \""+            <> fromString eventTable+            <> "\" where event_number > "+            <> fromString (show lastEvent)++writeEvents+    :: forall a+     . (ToJSON a)+    => Connection+    -> EventTableName+    -> [Stored a]+    -> IO EventNumber+writeEvents conn eventTable storedEvents = do+    _ <-+        executeMany+            conn+            ( "insert into \""+                <> fromString eventTable+                <> "\" (id, timestamp, event) \+                   \values (?, ?, ?)"+            )+            ( fmap+                (\x -> (storedUUID x, storedTimestamp x, encode $ storedEvent x))+                storedEvents+            )+    foldl' max 0 . fmap fromOnly+        <$> query_+            conn+            ("select coalesce(max(event_number),1) from \"" <> fromString eventTable <> "\"")++getEventStream'+    :: FromJSON event => PostgresEventTrans model event -> S.SerialT IO (Stored event)+getEventStream' pgt =+    S.map fst $+        mkEventStream+            (pgt ^. field @"chunkSize")+            (pgt ^. field @"transaction" . field @"connection")+            (pgt ^. field @"eventTableName" . to mkEventQuery)++-- | A transaction that is always rolled back at the end.+-- This is useful when using cursors as they can only be used inside a transaction.+withStreamReadTransaction+    :: forall t m a model event+     . (S.IsStream t, S.MonadAsync m, MonadCatch m)+    => PostgresEvent model event+    -> (PostgresEventTrans model event -> t m a)+    -> t m a+withStreamReadTransaction pg = S.bracket startTrans rollbackTrans+  where+    startTrans :: m (PostgresEventTrans model event)+    startTrans = liftIO $ do+        (conn, localPool) <- takeResource (connectionPool pg)+        PG.begin conn+        pure $+            PostgresEventTrans+                { transaction = OngoingTransaction conn localPool+                , eventTableName = pg ^. field @"eventTableName"+                , modelIORef = pg ^. field @"modelIORef"+                , app = pg ^. field @"app"+                , seed = pg ^. field @"seed"+                , chunkSize = pg ^. field @"chunkSize"+                }++    rollbackTrans :: PostgresEventTrans model event -> m ()+    rollbackTrans pgt = liftIO $ do+        -- Nothing changes. We just need the transaction to be able to stream events.+        let OngoingTransaction conn localPool = pgt ^. field' @"transaction"++            giveBackConn :: IO ()+            giveBackConn = do+                PG.rollback conn+                putResource localPool conn+        giveBackConn `catchAll` \_ ->+            destroyResource (connectionPool pg) localPool conn++withIOTrans+    :: forall a model event+     . PostgresEvent model event+    -> (PostgresEventTrans model event -> IO a)+    -> IO a+withIOTrans pg f = do+    transactionCompleted <- newIORef False+    (conn, localPool) <- takeResource (connectionPool pg)+    bracket (prepareTransaction conn localPool) (cleanup transactionCompleted) $ \pgt -> do+        a <- f pgt+        writeIORef transactionCompleted True+        pure a+  where+    cleanup :: IORef Bool -> PostgresEventTrans model event -> IO ()+    cleanup transactionCompleted pgt = do+        let OngoingTransaction conn localPool = pgt ^. field' @"transaction"++            giveBackConn :: IO ()+            giveBackConn = do+                readIORef transactionCompleted >>= \case+                    True -> PG.commit conn+                    False -> PG.rollback conn+                putResource localPool conn+        giveBackConn `catchAll` \_ ->+            destroyResource (connectionPool pg) localPool conn++    prepareTransaction :: Connection -> LocalPool Connection -> IO (PostgresEventTrans model event)+    prepareTransaction conn localPool = do+        PG.begin conn+        pure $+            PostgresEventTrans+                { transaction = OngoingTransaction conn localPool+                , eventTableName = pg ^. field @"eventTableName"+                , modelIORef = pg ^. field @"modelIORef"+                , app = pg ^. field @"app"+                , seed = pg ^. field @"seed"+                , chunkSize = pg ^. field @"chunkSize"+                }++mkEventStream+    :: FromJSON event+    => ChunkSize+    -> Connection+    -> EventQuery+    -> S.SerialT IO (Stored event, EventNumber)+mkEventStream chunkSize conn q = do+    let step :: Cursor.Cursor -> IO (Maybe (Seq EventRowOut, Cursor.Cursor))+        step cursor = do+            r <- Cursor.foldForward cursor chunkSize (\a r -> pure (a :|> r)) Seq.Empty+            case r of+                Left Seq.Empty -> pure Nothing+                Left a -> pure $ Just (a, cursor)+                Right a -> pure $ Just (a, cursor)++    cursor <- liftIO $ Cursor.declareCursor conn (getPgQuery q)+    S.mapM fromEventRow $+        S.unfoldMany Unfold.fromList . fmap toList $+            S.unfoldrM+                step+                cursor++getModel' :: forall e m. (FromJSON e) => PostgresEventTrans m e -> IO m+getModel' pgt = do+    NumberedModel model lastEventNo <- readIORef (pgt ^. field @"modelIORef")+    hasNewEvents <-+        queryHasEventsAfter+            (pgt ^. field @"transaction" . to connection)+            (pgt ^. field @"eventTableName")+            lastEventNo+    if hasNewEvents then fst <$> refreshModel pgt else pure model++refreshModel+    :: forall m e+     . (FromJSON e)+    => PostgresEventTrans m e+    -> IO (m, EventNumber)+refreshModel pg = do+    -- refresh doesn't write any events but changes the state and thus needs a lock+    exclusiveLock (pg ^. field @"transaction") (pg ^. field @"eventTableName")+    NumberedModel model lastEventNo <- readIORef (pg ^. field @"modelIORef")+    let eventStream :: S.SerialT IO (Stored e, EventNumber)+        eventStream =+            mkEventStream+                (pg ^. field @"chunkSize")+                (pg ^. field @"transaction" . field @"connection")+                (mkEventsAfterQuery (pg ^. field @"eventTableName") lastEventNo)++        applyModel :: NumberedModel m -> (Stored e, EventNumber) -> NumberedModel m+        applyModel (NumberedModel m _) (ev, evNumber) =+            NumberedModel ((pg ^. field @"app") m ev) evNumber++    NumberedModel newModel lastNewEventNo <-+        S.foldl'+            applyModel+            (NumberedModel model lastEventNo)+            eventStream++    _ <- writeIORef (pg ^. field @"modelIORef") $ NumberedModel newModel lastNewEventNo+    pure (newModel, lastNewEventNo)++exclusiveLock :: OngoingTransaction -> EventTableName -> IO ()+exclusiveLock (OngoingTransaction conn _) etName =+    void $ execute_ conn ("lock \"" <> fromString etName <> "\" in exclusive mode")++instance (ToJSON e, FromJSON e) => WriteModel (PostgresEvent m e) where+    transactionalUpdate pg cmd = withRunInIO $ \runInIO -> do+        withIOTrans pg $ \pgt -> do+            let eventTable = pg ^. field @"eventTableName"+            exclusiveLock (pgt ^. field @"transaction") eventTable+            m <- getModel' pgt+            (returnFun, evs) <- runInIO $ cmd m+            NumberedModel m' _ <- readIORef (pg ^. field @"modelIORef")+            storedEvs <- traverse toStored evs+            let newM = foldl' (pg ^. field @"app") m' storedEvs+            lastEventNo <-+                writeEvents+                    (pgt ^. field @"transaction" . to connection)+                    eventTable+                    storedEvs+            _ <- writeIORef (pg ^. field @"modelIORef") $ NumberedModel newM lastEventNo+            pure $ returnFun newM
+ src/DomainDriven/Persistance/Postgres/Migration.hs view
@@ -0,0 +1,86 @@+module DomainDriven.Persistance.Postgres.Migration where++import Data.Aeson+import Data.Int+import Data.String+import Database.PostgreSQL.Simple as PG+import DomainDriven.Persistance.Class+import DomainDriven.Persistance.Postgres.Internal+    ( createEventTable'+    , mkEventQuery+    , mkEventStream+    )+import DomainDriven.Persistance.Postgres.Types+import qualified Streamly.Data.Unfold as Unfold+import qualified Streamly.Prelude as S+import UnliftIO (liftIO)+import Prelude++migrateValue1to1+    :: Connection -> PreviousEventTableName -> EventTableName -> (Value -> Value) -> IO ()+migrateValue1to1 conn prevTName tName f = migrate1to1 conn prevTName tName (fmap f)++migrate1to1+    :: forall a b+     . (FromJSON a, ToJSON b)+    => Connection+    -> PreviousEventTableName+    -> EventTableName+    -> (Stored a -> Stored b)+    -> IO ()+migrate1to1 conn prevTName tName f = migrate1toMany conn prevTName tName (pure . f)++migrate1toMany+    :: forall a b+     . (FromJSON a, ToJSON b)+    => Connection+    -> PreviousEventTableName+    -> EventTableName+    -> (Stored a -> [Stored b])+    -> IO ()+migrate1toMany conn prevTName tName f = do+    _ <- createEventTable' conn tName+    S.mapM_ (liftIO . writeIt)+        . S.unfoldMany Unfold.fromList+        $ S.map (f . fst)+        $ mkEventStream 1 conn (mkEventQuery prevTName)+  where+    writeIt :: Stored b -> IO Int64+    writeIt event =+        PG.executeMany+            conn+            ( "insert into \""+                <> fromString tName+                <> "\" (id, timestamp, event) \+                   \values (?, ?, ?)"+            )+            (fmap (\x -> (storedUUID x, storedTimestamp x, encode $ storedEvent x)) [event])++migrate1toManyWithState+    :: forall a b state+     . (FromJSON a, ToJSON b)+    => Connection+    -> PreviousEventTableName+    -> EventTableName+    -> (state -> Stored a -> (state, [Stored b]))+    -> state+    -> IO ()+migrate1toManyWithState conn prevTName tName f initialState = do+    _ <- createEventTable' conn tName+    S.mapM_ (liftIO . writeIt)+        . S.unfoldMany Unfold.fromList+        . S.map snd+        $ S.scanl' (\b -> f (fst b)) (initialState, [])+        $ S.map fst+        $ mkEventStream 1 conn (mkEventQuery prevTName)+  where+    writeIt :: Stored b -> IO Int64+    writeIt event =+        PG.executeMany+            conn+            ( "insert into \""+                <> fromString tName+                <> "\" (id, timestamp, event) \+                   \values (?, ?, ?)"+            )+            (fmap (\x -> (storedUUID x, storedTimestamp x, encode $ storedEvent x)) [event])
+ src/DomainDriven/Persistance/Postgres/Types.hs view
@@ -0,0 +1,78 @@+module DomainDriven.Persistance.Postgres.Types where++import Control.Monad.Catch+import Data.Aeson+import Data.Int+import Data.Time+import Data.Typeable+import Data.UUID (UUID)+import Database.PostgreSQL.Simple (Connection)+import qualified Database.PostgreSQL.Simple as PG+import qualified Database.PostgreSQL.Simple.FromField as FF+import DomainDriven.Persistance.Class+import GHC.Generics (Generic)+import UnliftIO.Pool (LocalPool)+import Prelude++data PersistanceError+    = EncodingError String+    | ValueError String+    deriving (Show, Eq, Typeable, Exception)++type EventTableBaseName = String+type EventVersion = Int+type EventTableName = String+type PreviousEventTableName = String+type ChunkSize = Int++type EventMigration = PreviousEventTableName -> EventTableName -> Connection -> IO ()++data EventTable+    = MigrateUsing EventMigration EventTable+    | InitialVersion EventTableBaseName++newtype EventNumber = EventNumber {unEventNumber :: Int64}+    deriving (Show, Generic)+    deriving newtype (Eq, Ord, Num)++instance FF.FromField EventNumber where+    fromField f bs = EventNumber <$> FF.fromField f bs++data NumberedModel m = NumberedModel+    { model :: !m+    , eventNumber :: !EventNumber+    }+    deriving (Show, Generic)++data NumberedEvent e = NumberedEvent+    { event :: !(Stored e)+    , eventNumber :: !EventNumber+    }+    deriving (Show, Generic)++data OngoingTransaction = OngoingTransaction+    { connection :: Connection+    , localPool :: LocalPool Connection+    }+    deriving (Generic)++data EventRowOut = EventRowOut+    { key :: UUID+    , commitNumber :: EventNumber+    , timestamp :: UTCTime+    , event :: Value+    }+    deriving (Show, Eq, Generic, PG.FromRow)++fromEventRow :: (FromJSON e, MonadThrow m) => EventRowOut -> m (Stored e, EventNumber)+fromEventRow (EventRowOut evKey no ts ev) = case fromJSON ev of+    Success a -> pure (Stored a ts evKey, no)+    Error err ->+        throwM+            . EncodingError+            $ "Failed to parse event "+                <> show evKey+                <> ": "+                <> err+                <> "\nWhen trying to parse:\n"+                <> show ev
+ test/DomainDriven/Persistance/PostgresIORefStateSpec.hs view
@@ -0,0 +1,229 @@+module DomainDriven.Persistance.PostgresIORefStateSpec where++import Control.Exception (SomeException)+import Control.Monad+import Data.Aeson (FromJSON, ToJSON, Value)+import Data.String (fromString)+import Data.Time+import Data.Traversable+import Data.UUID (nil)+import qualified Data.UUID.V4 as V4+import Database.PostgreSQL.Simple+import DomainDriven.Persistance.Class+import DomainDriven.Persistance.Postgres+import DomainDriven.Persistance.Postgres.Internal+    ( getEventTableName+    , queryEvents+    , writeEvents+    )+import DomainDriven.Persistance.Postgres.Migration+import GHC.Generics (Generic)+import qualified Streamly.Prelude as S+import Test.Hspec+import UnliftIO.Pool+import Prelude++-- import           Text.Pretty.Simple++eventTable :: EventTable+eventTable =+    MigrateUsing (\_ _ _ -> pure ()) . MigrateUsing (\_ _ _ -> pure ()) $+        InitialVersion+            "test_events"++eventTable2 :: EventTable+eventTable2 = MigrateUsing mig eventTable+  where+    mig :: PreviousEventTableName -> EventTableName -> Connection -> IO ()+    mig prevName name conn = migrate1to1 @Value conn prevName name id++spec :: Spec+spec = do+    aroundAll setupPersistance streamingSpec+    aroundAll setupPersistance $ do+        writeEventsSpec+        queryEventsSpec+        migrationSpec -- make sure migrationSpec is run last!++type TestModel = Int++data TestEvent+    = AddOne+    | SubtractOne+    deriving (Show, Eq, Generic, FromJSON, ToJSON)++applyTestEvent :: TestModel -> Stored TestEvent -> TestModel+applyTestEvent m ev = case storedEvent ev of+    AddOne -> m + 1+    SubtractOne -> m - 1++setupPersistance+    :: ((PostgresEvent TestModel TestEvent, Pool Connection) -> IO ())+    -> IO ()+setupPersistance test = do+    dropEventTables =<< mkTestConn+    pool <- createPool (mkTestConn) close 1 5 1+    p <- postgresWriteModel pool eventTable applyTestEvent 0+    test (p{chunkSize = 2}, pool)++mkTestConn :: IO Connection+mkTestConn =+    connect $+        ConnectInfo+            { connectHost = "localhost"+            , connectPort = 5432+            , connectUser = "postgres"+            , connectPassword = "postgres"+            , connectDatabase = "domaindriven"+            }++dropEventTables :: Connection -> IO ()+dropEventTables conn = void . for (tableNames eventTable2) $ \n ->+    void $ execute_ conn ("drop table if exists \"" <> fromString n <> "\"")++tableNames :: EventTable -> [EventTableName]+tableNames et = case et of+    MigrateUsing _ next -> getEventTableName et : tableNames next+    InitialVersion{} -> [getEventTableName et]++withPool :: (Pool Connection -> IO a) -> IO a+withPool f = do+    pool <- createPool (mkTestConn) close 1 5 1+    r <- f pool+    destroyAllResources pool+    pure r++writeEventsSpec :: SpecWith (PostgresEvent TestModel TestEvent, Pool Connection)+writeEventsSpec = describe "queryEvents" $ do+    let ev1 :: Stored TestEvent+        ev1 =+            Stored+                { storedEvent = AddOne+                , storedTimestamp = UTCTime (fromGregorian 2020 10 15) 0+                , storedUUID = nil+                }++    it "Can write event to database" $ \(_p, pool) -> withResource pool $ \conn -> do+        i <- writeEvents conn (getEventTableName eventTable) [ev1]+        i `shouldBe` 1++    it "Writing the same event again fails" $ \(_p, pool) -> withResource pool $ \conn -> do+        writeEvents conn (getEventTableName eventTable) [ev1]+            `shouldThrow` (== FatalError)+                . sqlExecStatus++    it "Writing multiple events at once works" $ \(p, pool) -> do+        let evs =+                [ AddOne+                , SubtractOne+                ]+        storedEvs <-+            traverse+                (\e -> Stored e (UTCTime (fromGregorian 2020 10 15) 10) <$> mkId)+                evs+        _ <- withResource pool $ \conn ->+            writeEvents conn (getEventTableName eventTable) storedEvs+        evs' <- getEventList p+        drop (length evs' - 2) (fmap storedEvent evs') `shouldBe` evs++streamingSpec :: SpecWith (PostgresEvent TestModel TestEvent, Pool Connection)+streamingSpec = describe "steaming" $ do+    it "getEventList and getEventStream yields the same result" $ \(p, pool) -> do+        storedEvs <- for ([1 .. 10] :: [Int]) $ \i -> do+            enKey <- mkId+            pure $ Stored AddOne (UTCTime (fromGregorian 2020 10 15) (fromIntegral i)) enKey+        _ <- withResource pool $ \conn ->+            writeEvents conn (getEventTableName eventTable) storedEvs+        evList <- getEventList p+        evStream <- S.toList $ getEventStream p+        -- pPrint evList+        evList `shouldSatisfy` (== 10) . length -- must be at least two to verify order+        fmap storedEvent evStream `shouldBe` fmap storedEvent evList+        evStream `shouldBe` evList++queryEventsSpec :: SpecWith (PostgresEvent TestModel TestEvent, Pool Connection)+queryEventsSpec = describe "queryEvents" $ do+    it "Can query events" $ \(_p, pool) -> withResource pool $ \conn -> do+        evs <- queryEvents @TestEvent conn (getEventTableName eventTable)+        evs `shouldSatisfy` (>= 1) . length+    it "Events come out in the right order" $ \(_p, pool) -> withResource pool $ \conn -> do+        -- write few more events before+        --+        _ <- do+            id1 <- mkId+            let ev1 = SubtractOne+            _ <-+                writeEvents+                    conn+                    (getEventTableName eventTable)+                    [Stored ev1 (UTCTime (fromGregorian 2020 10 20) 1) id1]++            id2 <- mkId+            let ev2 = AddOne+            writeEvents+                conn+                (getEventTableName eventTable)+                [Stored ev2 (UTCTime (fromGregorian 2020 10 18) 1) id2]++        evs <- queryEvents @TestEvent conn (getEventTableName eventTable)+        evs `shouldSatisfy` (> 1) . length+        let event_numbers = fmap snd evs+        event_numbers `shouldSatisfy` (\n -> and $ zipWith (>) (drop 1 n) n)++migrationSpec :: SpecWith (PostgresEvent TestModel TestEvent, Pool Connection)+migrationSpec = describe "migrate1to1" $ do+    it "Keeps all events when using `id` to update" $ \(_p, pool) -> do+        evs <- withResource pool $ \conn ->+            queryEvents @TestEvent conn (getEventTableName eventTable)+        evs `shouldSatisfy` (>= 1) . length++        _ <- postgresWriteModel pool eventTable2 applyTestEvent 0+        evs' <- withResource pool $ \conn ->+            queryEvents @TestEvent conn (getEventTableName eventTable2)++        fmap fst evs' `shouldBe` fmap fst evs+    it "Can no longer write new events to old table after migration" $ \(_p, pool) -> do+        uuid <- V4.nextRandom+        let ev =+                Stored+                    AddOne+                    (UTCTime (fromGregorian 2020 10 15) 0)+                    uuid+        withResource pool (\conn -> writeEvents conn (getEventTableName eventTable) [ev])+            `shouldThrow` (== FatalError)+                . sqlExecStatus+    it "But can write to the new table" $ \(_p, pool) -> do+        uuid <- V4.nextRandom+        let ev =+                Stored+                    AddOne+                    (UTCTime (fromGregorian 2020 10 15) 0)+                    uuid++        void . withResource pool $ \conn -> writeEvents conn (getEventTableName eventTable2) [ev]++    it "Broken migration throws and rollbacks transaction" $ \(_, pool) -> do+        let eventTable3 :: EventTable+            eventTable3 = MigrateUsing undefined eventTable2++        print $ tableNames eventTable3++        postgresWriteModel pool eventTable3 applyTestEvent 0+            `shouldThrow` const @_ @SomeException True+        conn <- mkTestConn++        case tableNames eventTable3 of+            failedMig : prevMig : _ -> do+                [Only prevExists] <-+                    query_ @(Only Bool) conn $+                        "select exists(select * from pg_tables where tablename='"+                            <> fromString prevMig+                            <> "')"+                [Only brokenExists] <-+                    query_ @(Only Bool) conn $+                        "select exists(select * from pg_tables where tablename='"+                            <> fromString failedMig+                            <> "')"+                prevExists `shouldBe` True+                brokenExists `shouldBe` False+            _ -> fail "Unexpectedly lacking table versions!"
+ test/Spec.hs view
@@ -0,0 +1,1 @@+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}