eventsourcing-postgresql (empty) → 0.9.0
raw patch · 12 files changed
+1683/−0 lines, 12 filesdep +basedep +bytestringdep +eventsourcingsetup-changed
Dependencies added: base, bytestring, eventsourcing, hashable, mtl, pipes, postgresql-simple, stm, unordered-containers
Files
- CHANGELOG.md +5/−0
- LICENSE +5/−0
- Setup.hs +2/−0
- eventsourcing-postgresql.cabal +47/−0
- src/Database/CQRS/PostgreSQL.hs +22/−0
- src/Database/CQRS/PostgreSQL/Internal.hs +49/−0
- src/Database/CQRS/PostgreSQL/Migration.hs +272/−0
- src/Database/CQRS/PostgreSQL/Projection.hs +335/−0
- src/Database/CQRS/PostgreSQL/SQLQuery.hs +41/−0
- src/Database/CQRS/PostgreSQL/Stream.hs +301/−0
- src/Database/CQRS/PostgreSQL/StreamFamily.hs +408/−0
- src/Database/CQRS/PostgreSQL/TrackingTable.hs +196/−0
+ CHANGELOG.md view
@@ -0,0 +1,5 @@+# Revision history for eventsourcing-postgresql++## 0.9.0 -- 2020-08-16++* First version.
+ LICENSE view
@@ -0,0 +1,5 @@+Copyright 2019-2020 Tom Feron <tho.feron@gmail.com>++Permission to use, copy, modify, and/or distribute this software for any purpose with or without fee is hereby granted, provided that the above copyright notice and this permission notice appear in all copies.++THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ eventsourcing-postgresql.cabal view
@@ -0,0 +1,47 @@+cabal-version: >=1.10+name: eventsourcing-postgresql+version: 0.9.0+synopsis: PostgreSQL adaptor for eventsourcing.+description: Adaptor to use PostgreSQL as a back-end with eventsourcing.+license: ISC+license-file: LICENSE+author: Tom Feron <tho.feron@gmail.com>+maintainer: Tom Feron <tho.feron@gmail.com>+build-type: Simple+extra-source-files: CHANGELOG.md+homepage: https://github.com/thoferon/eventsourcing+bug-reports: https://github.com/thoferon/eventsourcing/issues+category: Database++source-repository head+ type: git+ location: git://github.com/thoferon/eventsourcing.git+ subdir: eventsourcing-postgresql++library+ hs-source-dirs: src+ default-language: Haskell2010+ ghc-options: -Wall++ exposed-modules:+ Database.CQRS.PostgreSQL+ Database.CQRS.PostgreSQL.Migration+ Database.CQRS.PostgreSQL.Projection+ Database.CQRS.PostgreSQL.SQLQuery+ Database.CQRS.PostgreSQL.Stream+ Database.CQRS.PostgreSQL.StreamFamily+ Database.CQRS.PostgreSQL.TrackingTable++ other-modules:+ Database.CQRS.PostgreSQL.Internal++ build-depends:+ base >=4.12 && <5,+ bytestring >=0.10,+ eventsourcing,+ hashable,+ mtl,+ pipes,+ postgresql-simple >=0.6,+ stm,+ unordered-containers
+ src/Database/CQRS/PostgreSQL.hs view
@@ -0,0 +1,22 @@+module Database.CQRS.PostgreSQL+ ( -- * Stream+ Stream+ , makeStream+ , makeStream'++ -- * Stream family+ , StreamFamily+ , makeStreamFamily++ -- * Projection+ , Projection+ , executeSqlActions+ , executeCustomActions+ , fromTabularDataActions+ , createTrackingTable+ ) where++import Database.CQRS.PostgreSQL.Projection+import Database.CQRS.PostgreSQL.Stream+import Database.CQRS.PostgreSQL.StreamFamily+import Database.CQRS.PostgreSQL.TrackingTable
+ src/Database/CQRS/PostgreSQL/Internal.hs view
@@ -0,0 +1,49 @@+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeApplications #-}++module Database.CQRS.PostgreSQL.Internal where++import Control.Exception++import qualified Database.PostgreSQL.Simple as PG+import qualified Database.PostgreSQL.Simple.ToField as PG.To+import qualified Database.PostgreSQL.Simple.ToRow as PG.To++-- | Return all the 'Right' elements before the first 'Left' and the value of+-- the first 'Left'.+stopOnLeft :: [Either a b] -> ([b], Maybe a)+stopOnLeft = go id+ where+ go :: ([b] -> [b]) -> [Either a b] -> ([b], Maybe a)+ go f = \case+ [] -> (f [], Nothing)+ Left err : _ -> (f [], Just err)+ Right x : xs -> go (f . (x:)) xs++type SqlAction = (PG.Query, [PG.To.Action])++makeSqlAction :: PG.To.ToRow r => PG.Query -> r -> SqlAction+makeSqlAction query r = (query, PG.To.toRow r)++appendSqlActions :: [SqlAction] -> SqlAction+appendSqlActions = \case+ [] -> ("", [])+ action : actions -> foldl step action actions+ where+ step :: SqlAction -> SqlAction -> SqlAction+ step (q1,v1) (q2,v2) = (q1 <> ";" <> q2, v1 ++ v2)++handleError+ :: forall e e' a proxy. (Exception e, Show e)+ => proxy e -> (String -> e') -> Handler (Either e' a)+handleError _ f = Handler $ pure . Left . f . show @e++data SomeParams = forall r. PG.To.ToRow r => SomeParams r++instance PG.To.ToRow SomeParams where+ toRow (SomeParams x) = PG.To.toRow x
+ src/Database/CQRS/PostgreSQL/Migration.hs view
@@ -0,0 +1,272 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeApplications #-}++module Database.CQRS.PostgreSQL.Migration+ ( migrate+ ) where++import Control.Exception+import Control.Monad ((<=<), unless)+import Control.Monad.Trans (MonadIO(..), lift)+import Data.Hashable (Hashable)+import Data.List (foldl', intersperse)+import Data.Proxy (Proxy(..))+import Database.PostgreSQL.Simple ((:.)(..))+import Pipes ((>->))++import qualified Control.Monad.Except as Exc+import qualified Control.Monad.State.Strict as St+import qualified Data.HashMap.Strict as HM+import qualified Database.PostgreSQL.Simple as PG+import qualified Database.PostgreSQL.Simple.Types as PG+import qualified Database.PostgreSQL.Simple.FromField as PG.From+import qualified Database.PostgreSQL.Simple.FromRow as PG.From+import qualified Database.PostgreSQL.Simple.ToField as PG.To+import qualified Database.PostgreSQL.Simple.ToRow as PG.To+import qualified Pipes++import Database.CQRS.PostgreSQL.StreamFamily+import Database.CQRS.PostgreSQL.Internal++import qualified Database.CQRS as CQRS++-- | Migrate a stream family stored in a PostgreSQL database to the same+-- database. It is meant to run in parallel with the application using the+-- stream family without disturbing it.+--+-- An alternative use of this is to migrate a stream family to a new relation+-- without swapping the tables at the end. The old table stays in use by the+-- application and the new one can be read by an external system for instance.+--+-- If the new table already exists (and the initialisation query does not fail+-- in that case,) the migration will start over from the point it left off.+migrate+ :: forall streamId eventId metadata event transformedStreamFamily m.+ ( CQRS.WritableEvent+ (CQRS.EventType (CQRS.StreamType transformedStreamFamily))+ , CQRS.Stream m (CQRS.StreamType transformedStreamFamily)+ , CQRS.StreamFamily m transformedStreamFamily+ , Exc.MonadError CQRS.Error m+ , Hashable (CQRS.StreamIdentifier transformedStreamFamily)+ , MonadIO m+ , Ord (CQRS.EventIdentifier (CQRS.StreamType transformedStreamFamily))+ , Ord (CQRS.StreamIdentifier transformedStreamFamily)+ , PG.From.FromField (CQRS.StreamIdentifier transformedStreamFamily)+ , PG.From.FromField+ (CQRS.EventIdentifier (CQRS.StreamType transformedStreamFamily))+ , PG.From.FromRow+ (CQRS.EventMetadata (CQRS.StreamType transformedStreamFamily))+ , PG.From.FromField+ (CQRS.EncodingFormat+ (CQRS.EventType (CQRS.StreamType transformedStreamFamily)))+ , PG.To.ToField (CQRS.StreamIdentifier transformedStreamFamily)+ , PG.To.ToField+ (CQRS.EventIdentifier (CQRS.StreamType transformedStreamFamily))+ , PG.To.ToRow+ (CQRS.EventMetadata (CQRS.StreamType transformedStreamFamily))+ , PG.To.ToField+ (CQRS.EncodingFormat+ (CQRS.EventType (CQRS.StreamType transformedStreamFamily)))+ , Show+ (CQRS.EventIdentifier (CQRS.StreamType transformedStreamFamily))+ )+ => StreamFamily streamId eventId metadata event+ -> (StreamFamily streamId eventId metadata event -> transformedStreamFamily)+ -> PG.Query -- ^ Name of the new (temporary) relation.+ -> PG.Query -- ^ New stream identifier column name.+ -> PG.Query -- ^ New event identifier column name.+ -> [PG.Query] -- ^ New metadata column names.+ -> PG.Query -- ^ New event column name.+ -> (PG.Query -> PG.Query)+ -- ^ Initialisation query that creates the new relation. It is given the name+ -- of the current relation to migrate. It must be idempotent to be able to+ -- restart the migrator.+ -> (PG.Query -> PG.Query)+ -- ^ Query to lock the relation used by the application. It is given the name+ -- of the current relation. It must be idempotent to be able to restart the+ -- migrator.+ -> (PG.Query -> PG.Query)+ -- ^ Query to swap the relation used by the application. It is given the name+ -- of the current relation.+ -> m ()+migrate fam@StreamFamily { connectionPool, relation } transform+ tempRelation streamIdentifierColumn+ eventIdentifierColumn metadataColumns eventColumn+ initQuery lockQuery swapQuery = do++ let transformedStreamFamily = transform fam++ Exc.liftEither <=< liftIO . connectionPool $ \conn ->+ (const (Right ()) <$> PG.execute_ conn (initQuery relation))+ `catches` handlers++ newEvents <- CQRS.allNewEvents transformedStreamFamily++ flip St.evalStateT HM.empty $ do+ Pipes.runEffect . Pipes.for+ (Pipes.hoist lift (CQRS.latestEventIdentifiers tempStreamFamily))+ $ \(streamId, eventId) ->+ St.modify' $ HM.insert streamId eventId++ -- Phase 1: Process batches of events for all streams.+ -- FIXME: handle exceptions+ Pipes.runEffect $+ Pipes.hoist lift (CQRS.latestEventIdentifiers transformedStreamFamily)+ >-> migrateStream transformedStreamFamily++ -- Phase 2: Go through the new events that were created while phase 1 was+ -- still ongoing.+ processNewEvents newEvents++ -- Phase 3: Prevent writes to the original relation, go through+ -- notifications one last time and swap the relations.++ Exc.liftEither <=< liftIO . connectionPool $ \conn ->+ (const (Right ()) <$> PG.execute_ conn (lockQuery relation))+ `catches` handlers++ processNewEvents newEvents++ Exc.liftEither <=< liftIO . connectionPool $ \conn ->+ (const (Right ()) <$> PG.execute_ conn (swapQuery relation))+ `catches` handlers++ where+ migrateStream+ :: transformedStreamFamily+ -> Pipes.Consumer+ ( CQRS.StreamIdentifier transformedStreamFamily+ , CQRS.EventIdentifier (CQRS.StreamType transformedStreamFamily)+ )+ (St.StateT+ (HM.HashMap+ (CQRS.StreamIdentifier transformedStreamFamily)+ (CQRS.EventIdentifier (CQRS.StreamType transformedStreamFamily)))+ m)+ ()+ migrateStream transformedStreamFamily = do+ (streamId, eventId) <- Pipes.await+ stream <- lift . lift $ CQRS.getStream transformedStreamFamily streamId++ state <- St.get+ let bounds = case HM.lookup streamId state of+ Nothing -> CQRS.untilEvent eventId+ Just lastEventId ->+ CQRS.afterEvent lastEventId <> CQRS.untilEvent eventId++ lift . Pipes.runEffect . Pipes.for+ (Pipes.hoist lift (CQRS.streamEvents stream bounds))+ $ \batch -> do+ let (ewcs, mErr) = stopOnLeft batch+ params =+ map (\CQRS.EventWithContext{..} ->+ PG.Only streamId :. PG.Only identifier :. metadata+ :. PG.Only (CQRS.encodeEvent event))+ ewcs++ unless (null ewcs) $ do+ Exc.liftEither <=< liftIO . connectionPool $ \conn ->+ (const (Right ())+ <$> PG.execute conn insertQuery (PG.Only (PG.Values [] params)))+ `catches` handlers++ let lastEventId = CQRS.identifier . last $ ewcs+ St.modify' $ HM.insert streamId lastEventId++ case mErr of+ Nothing -> pure ()+ Just (errEventId, err) ->+ Exc.throwError $ CQRS.EventDecodingError (show errEventId) err++ processNewEvents+ :: Pipes.Producer+ [ ( CQRS.StreamIdentifier transformedStreamFamily+ , Either+ ( CQRS.EventIdentifier (CQRS.StreamType transformedStreamFamily)+ , String )+ (CQRS.EventWithContext'+ (CQRS.StreamType transformedStreamFamily))+ ) ]+ m ()+ -> St.StateT+ (HM.HashMap+ (CQRS.StreamIdentifier transformedStreamFamily)+ (CQRS.EventIdentifier (CQRS.StreamType transformedStreamFamily)))+ m ()+ processNewEvents newEvents =+ Pipes.runEffect . untilEmpty (Pipes.hoist lift newEvents) $ \batch -> do+ state <- St.get++ let (events, mErr) =+ stopOnLeft . map sequence $ batch+ params =+ map (\(streamId, CQRS.EventWithContext{..}) ->+ PG.Only streamId :. PG.Only identifier :. metadata+ :. PG.Only (CQRS.encodeEvent event))+ . filter (\(streamId, CQRS.EventWithContext{..}) ->+ maybe True (identifier >) . HM.lookup streamId $ state)+ $ events++ unless (null events) $+ Exc.liftEither <=< liftIO . connectionPool $ \conn ->+ (const (Right ())+ <$> PG.execute conn insertQuery (PG.Only (PG.Values [] params)))+ `catches` handlers++ St.put+ . foldl'+ (\s (streamId, CQRS.EventWithContext{..}) ->+ HM.insert streamId identifier s)+ state+ $ events++ case mErr of+ Nothing -> pure ()+ Just (errEventId, err) ->+ Exc.throwError $ CQRS.EventDecodingError (show errEventId) err++ tempStreamFamily+ :: StreamFamily+ (CQRS.StreamIdentifier transformedStreamFamily)+ (CQRS.EventIdentifier (CQRS.StreamType transformedStreamFamily))+ (CQRS.EventMetadata (CQRS.StreamType transformedStreamFamily))+ (CQRS.EventType (CQRS.StreamType transformedStreamFamily))+ tempStreamFamily = StreamFamily+ { relation = tempRelation+ , notificationChannel = "unused"+ , parseNotification = const $ Left "unused"+ , ..+ }++ untilEmpty+ :: Monad n+ => Pipes.Producer [b] n ()+ -> ([b] -> Pipes.Effect n ())+ -> Pipes.Effect n ()+ untilEmpty pipe f =+ let pipe' = do+ xs <- Pipes.await+ if null xs+ then pure ()+ else lift . Pipes.runEffect . f $ xs+ in+ pipe >-> pipe'++ insertQuery :: PG.Query+ insertQuery =+ "INSERT INTO " <> tempRelation+ <> " (" <> streamIdentifierColumn+ <> ", " <> eventIdentifierColumn+ <> ", " <> mconcat (intersperse "," metadataColumns)+ <> ", " <> eventColumn+ <> ") VALUES ?"++ handlers :: [Handler (Either CQRS.Error ())]+ handlers =+ [ handleError (Proxy @PG.FormatError) CQRS.MigrationError+ , handleError (Proxy @PG.SqlError) CQRS.MigrationError+ ]
+ src/Database/CQRS/PostgreSQL/Projection.hs view
@@ -0,0 +1,335 @@+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE KindSignatures #-}+{-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE TypeApplications #-}++module Database.CQRS.PostgreSQL.Projection+ ( Projection+ , executeSqlActions+ , executeCustomActions+ , fromTabularDataActions+ ) where++import Control.Exception+import Control.Monad ((<=<), forever, forM_)+import Control.Monad.Trans (MonadIO(..), lift)+import Data.List (intersperse)+import Data.Proxy (Proxy(..))+import Data.String (fromString)+import Database.PostgreSQL.Simple ((:.)(..))++import qualified Control.Monad.Except as Exc+import qualified Control.Monad.Identity as Id+import qualified Control.Monad.State.Strict as St+import qualified Data.Bifunctor as Bifunctor+import qualified Database.PostgreSQL.Simple as PG+import qualified Database.PostgreSQL.Simple.ToField as PG.To+import qualified Database.PostgreSQL.Simple.Types as PG+import qualified Pipes++import Database.CQRS.PostgreSQL.Internal+import Database.CQRS.PostgreSQL.TrackingTable++import qualified Database.CQRS as CQRS+import qualified Database.CQRS.TabularData as CQRS.Tab++type Projection event st = CQRS.Projection event st SqlAction++-- | Execute the SQL actions and update the tracking table in one transaction.+--+-- The custom actions are transformed into a list of SQL actions by the given+-- function. See 'fromTabularDataActions' for an example.+executeSqlActions+ :: forall streamId eventId action m st.+ ( Exc.MonadError CQRS.Error m+ , MonadIO m+ , PG.To.ToField eventId+ , PG.To.ToField streamId+ , PG.To.ToField st+ )+ => ([action] -> [SqlAction])+ -> (forall r. (PG.Connection -> IO r) -> IO r)+ -> TrackingTable streamId eventId st+ -> Pipes.Consumer (st, [action], streamId, eventId) m ()+executeSqlActions transform connectionPool trackingTable =+ forever $ do+ (st, actions, streamId, eventId) <- Pipes.await++ let sqlActions = transform actions+ (query, values) =+ appendSqlActions+ [ ("BEGIN", [])+ , appendSqlActions sqlActions+ , upsertTrackingTable+ trackingTable streamId eventId (Right st)+ , ("COMMIT", [])+ ]++ Exc.liftEither <=< liftIO . connectionPool $ \conn -> do+ eRes <-+ (Right <$> PG.execute conn query values)+ `catches`+ [ handleError (Proxy @PG.FormatError) id+ , handleError (Proxy @PG.SqlError) id+ ]++ case eRes of+ Left err -> do+ let (uquery, uvalues) =+ upsertTrackingTable+ trackingTable streamId eventId+ (Left err :: Either String st)+ (const (Right ()) <$> PG.execute conn uquery uvalues)+ `catches`+ [ handleError (Proxy @PG.FormatError) CQRS.ProjectionError+ , handleError (Proxy @PG.SqlError) CQRS.ProjectionError+ ]+ Right _ -> pure $ Right ()++-- | Execute custom actions by calling the runner function on each action in+-- turn and updating the tracking table accordingly.+executeCustomActions+ :: forall streamId eventId action m st.+ ( Exc.MonadError CQRS.Error m+ , MonadIO m+ , PG.To.ToField eventId+ , PG.To.ToField streamId+ , PG.To.ToField st+ )+ => (action -> m (Either String (m ())))+ -- ^ Run an action returning either an error or a rollback action.+ -- If any of the rollback actions fail, the others are not run.+ -- Rollback actions are run in reversed order.+ -> TrackingTable streamId eventId st+ -> Pipes.Consumer (st, [action], streamId, eventId) m ()+executeCustomActions runAction trackingTable =+ forever $ do+ (st, actions, streamId, eventId) <- Pipes.await++ (eRes, rollbackActions) <- lift . flip St.runStateT [] . Exc.runExceptT $+ forM_ actions $ \action -> do+ errOrRollback <- lift . lift . runAction $ action+ case errOrRollback of+ Left err -> Exc.throwError err+ Right rollbackAction -> St.modify' (rollbackAction :)++ lift $ case eRes of+ Left err -> do+ doUpsertTrackingTable trackingTable streamId eventId+ (Left err :: Either String st)+ sequence_ rollbackActions++ Right () ->+ doUpsertTrackingTable trackingTable streamId eventId (Right st)++fromTabularDataActions+ :: FromTabularDataAction cols+ => PG.Query -- ^ Relation name.+ -> [CQRS.Tab.TabularDataAction cols]+ -> [SqlAction]+fromTabularDataActions = map . fromTabularDataAction++class FromTabularDataAction cols where+ fromTabularDataAction+ :: PG.Query -> CQRS.Tab.TabularDataAction cols -> SqlAction++instance+ forall keyCols cols.+ ( CQRS.Tab.AllColumns+ PG.To.ToField (CQRS.Tab.Flatten ('CQRS.Tab.WithUniqueKey keyCols cols))+ , CQRS.Tab.AllColumns PG.To.ToField keyCols+ , CQRS.Tab.AllColumns PG.To.ToField cols+ , CQRS.Tab.MergeSplitTuple keyCols cols+ )+ => FromTabularDataAction ('CQRS.Tab.WithUniqueKey keyCols cols) where++ fromTabularDataAction relation = \case+ CQRS.Tab.Insert tuple ->+ makeInsertSqlAction @('CQRS.Tab.WithUniqueKey keyCols cols) relation tuple+ CQRS.Tab.Update updates conds ->+ makeUpdateSqlAction+ @('CQRS.Tab.WithUniqueKey keyCols cols)+ relation updates conds+ CQRS.Tab.Upsert tuple -> makeUpsertSqlAction @keyCols @cols relation tuple+ CQRS.Tab.Delete conds ->+ makeDeleteSqlAction+ @('CQRS.Tab.WithUniqueKey keyCols cols)+ relation conds++instance+ forall cols.+ CQRS.Tab.AllColumns PG.To.ToField cols+ => FromTabularDataAction ('CQRS.Tab.Flat cols) where++ fromTabularDataAction relation = \case+ CQRS.Tab.Insert tuple ->+ makeInsertSqlAction @('CQRS.Tab.Flat cols) relation tuple+ CQRS.Tab.Update updates conds ->+ makeUpdateSqlAction @('CQRS.Tab.Flat cols) relation updates conds+ CQRS.Tab.Delete conds ->+ makeDeleteSqlAction @('CQRS.Tab.Flat cols) relation conds++makeInsertSqlAction+ :: forall (cols :: CQRS.Tab.Columns).+ CQRS.Tab.AllColumns PG.To.ToField (CQRS.Tab.Flatten cols)+ => PG.Query -> CQRS.Tab.Tuple Id.Identity cols -> SqlAction+makeInsertSqlAction relation tuple =+ let (identifiers, values) =+ unzip+ . CQRS.Tab.toList @PG.To.ToField+ (\name (Id.Identity x) ->+ (fromString @PG.Identifier name, PG.To.toField x))+ $ tuple+ questionMarks =+ "(" <> mconcat (intersperse "," (map (const "?") values)) <> ")"+ query =+ "INSERT INTO " <> relation <> questionMarks+ <> " VALUES " <> questionMarks+ in+ makeSqlAction query (identifiers :. values)++makeUpdateSqlAction+ :: forall (cols :: CQRS.Tab.Columns).+ CQRS.Tab.AllColumns PG.To.ToField (CQRS.Tab.Flatten cols)+ => PG.Query+ -> CQRS.Tab.Tuple CQRS.Tab.Update cols+ -> CQRS.Tab.Tuple CQRS.Tab.Conditions cols+ -> SqlAction+makeUpdateSqlAction relation updates conds =+ let (updatesQuery, updatesValues) =+ Bifunctor.bimap (mconcat . intersperse ",") mconcat+ . unzip+ . CQRS.Tab.toList @PG.To.ToField+ (\name update -> case update of+ CQRS.Tab.NoUpdate -> ("", [])+ CQRS.Tab.Set x ->+ ( "? = ?"+ , [ PG.To.toField (fromString @PG.Identifier name)+ , PG.To.toField x+ ]+ )+ CQRS.Tab.Plus x ->+ ( "? = ? + ?"+ , [ PG.To.toField (fromString @PG.Identifier name)+ , PG.To.toField (fromString @PG.Identifier name)+ , PG.To.toField x+ ]+ )+ CQRS.Tab.Minus x ->+ ( "? = ? - ?"+ , [ PG.To.toField (fromString @PG.Identifier name)+ , PG.To.toField (fromString @PG.Identifier name)+ , PG.To.toField x+ ]+ )+ )+ $ updates++ (whereStmtQuery, whereStmtValues) = makeWhereStatement @cols conds++ values = updatesValues ++ whereStmtValues+ query =+ "UPDATE " <> relation <> " SET " <> updatesQuery <> whereStmtQuery+ in+ (query, values)++makeUpsertSqlAction+ :: forall keyCols cols.+ ( CQRS.Tab.AllColumns PG.To.ToField keyCols+ , CQRS.Tab.AllColumns PG.To.ToField cols+ , CQRS.Tab.MergeSplitTuple keyCols cols+ )+ => PG.Query+ -> CQRS.Tab.Tuple Id.Identity ('CQRS.Tab.WithUniqueKey keyCols cols)+ -> SqlAction+makeUpsertSqlAction relation tuple =+ let toSqlValues+ :: forall flatCols. CQRS.Tab.AllColumns PG.To.ToField flatCols+ => CQRS.Tab.FlatTuple Id.Identity flatCols+ -> [(PG.Identifier, PG.To.Action)]+ toSqlValues =+ CQRS.Tab.toList @PG.To.ToField+ (\name (Id.Identity x) ->+ (fromString @PG.Identifier name, PG.To.toField x))+ (keyTuple, otherTuple) = CQRS.Tab.splitTuple @keyCols @cols tuple+ keyPairs = toSqlValues keyTuple+ (keyIdentifiers, keyValues) = unzip keyPairs+ otherPairs = toSqlValues otherTuple+ (otherIdentifiers, otherValues) = unzip otherPairs++ mkQuestionMarks xs =+ mconcat (intersperse "," (map (const "?") xs))+ keyQuestionMarks = mkQuestionMarks keyValues+ rowQuestionMarks =+ "(" <> mkQuestionMarks (keyValues ++ otherValues) <> ")"++ mkValues =+ foldr (\(name, value) acc -> PG.To.toField name : value : acc) []+ updateSetValues = mkValues otherPairs+ updateWhereValues = mkValues keyPairs++ query =+ "INSERT INTO " <> relation <> rowQuestionMarks+ <> " VALUES " <> rowQuestionMarks+ <> " ON CONFLICT " <> keyQuestionMarks <> " DO UPDATE SET "+ <> mconcat+ (intersperse ", " (map (const "? = ?") otherIdentifiers))+ <> " WHERE "+ <> mconcat+ (intersperse " AND " (map (const "? = ?") keyIdentifiers))++ in+ makeSqlAction query $+ keyIdentifiers :. otherIdentifiers :. keyValues :. otherValues+ :. keyIdentifiers :. updateSetValues :. updateWhereValues++makeDeleteSqlAction+ :: forall (cols :: CQRS.Tab.Columns).+ CQRS.Tab.AllColumns PG.To.ToField (CQRS.Tab.Flatten cols)+ => PG.Query+ -> CQRS.Tab.Tuple CQRS.Tab.Conditions cols+ -> SqlAction+makeDeleteSqlAction relation conds =+ let (whereStmtQuery, whereStmtValues) =+ makeWhereStatement @cols conds+ query = "DELETE FROM " <> relation <> whereStmtQuery+ in+ (query, whereStmtValues)++makeWhereStatement+ :: forall (cols :: CQRS.Tab.Columns).+ CQRS.Tab.AllColumns PG.To.ToField (CQRS.Tab.Flatten cols)+ => CQRS.Tab.Tuple CQRS.Tab.Conditions cols+ -> (PG.Query, [PG.To.Action])+makeWhereStatement =+ Bifunctor.bimap+ (\cs -> if null cs+ then ""+ else mconcat (" WHERE " : intersperse " AND " cs))+ mconcat+ . unzip+ . mconcat+ . CQRS.Tab.toList @PG.To.ToField @(CQRS.Tab.Flatten cols)+ (\name value ->+ map+ (makeCond (PG.To.toField (fromString @PG.Identifier name)))+ (CQRS.Tab.getConditions value))++makeCond+ :: PG.To.ToField a+ => PG.To.Action -> CQRS.Tab.Condition a -> (PG.Query, [PG.To.Action])+makeCond name = \case+ CQRS.Tab.Equal x -> ("? = ?", [name, PG.To.toField x])+ CQRS.Tab.NotEqual x -> ("? <> ?", [name, PG.To.toField x])+ CQRS.Tab.LowerThan x -> ("? < ?", [name, PG.To.toField x])+ CQRS.Tab.LowerThanOrEqual x -> ("? <= ?", [name, PG.To.toField x])+ CQRS.Tab.GreaterThan x -> ("? > ?", [name, PG.To.toField x])+ CQRS.Tab.GreaterThanOrEqual x -> ("? >= ?", [name, PG.To.toField x])
+ src/Database/CQRS/PostgreSQL/SQLQuery.hs view
@@ -0,0 +1,41 @@+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE TypeFamilies #-}++module Database.CQRS.PostgreSQL.SQLQuery+ ( SQLQuery(..)+ ) where++import Control.Monad.Trans (MonadIO(..))++import qualified Database.PostgreSQL.Simple as PG+import qualified Database.PostgreSQL.Simple.FromRow as PG.From+import qualified Database.PostgreSQL.Simple.ToRow as PG.To++import qualified Database.CQRS as CQRS++-- | A wrapper around a SELECT query that instantiates 'ReadModel' so that it+-- can be used by the application layer without said layer to be aware of SQL.+-- The implementation can then be swapped for something else, e.g. for tests.+data SQLQuery req resp = SQLQuery+ { connectionPool :: forall a. (PG.Connection -> IO a) -> IO a+ , queryTemplate :: PG.Query+ }++instance+ {-# OVERLAPPABLE #-}+ (MonadIO m, PG.From.FromRow resp, PG.To.ToRow req)+ => CQRS.ReadModel m (SQLQuery req resp) where+ type ReadModelQuery (SQLQuery req resp) = req+ type ReadModelResponse (SQLQuery req resp) = [resp]++ query = sqlQueryQuery++sqlQueryQuery+ :: (MonadIO m, PG.From.FromRow resp, PG.To.ToRow req)+ => SQLQuery req resp -> req -> m [resp]+sqlQueryQuery SQLQuery{..} req =+ liftIO . connectionPool $ \conn -> do+ PG.query conn queryTemplate req
+ src/Database/CQRS/PostgreSQL/Stream.hs view
@@ -0,0 +1,301 @@+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE UndecidableInstances #-}++module Database.CQRS.PostgreSQL.Stream+ ( Stream+ , makeStream+ , makeStream'+ ) where++import Control.Exception (catches)+import Control.Monad (when)+import Control.Monad.Trans (MonadIO(..))+import Data.Functor ((<&>))+import Data.List (intersperse)+import Data.Maybe (catMaybes, listToMaybe)+import Data.Proxy (Proxy(..))+import Data.String (fromString)+import Database.PostgreSQL.Simple ((:.)(..))++import qualified Control.Monad.Except as Exc+import qualified Database.PostgreSQL.Simple as PG+import qualified Database.PostgreSQL.Simple.FromField as PG.From+import qualified Database.PostgreSQL.Simple.FromRow as PG.From+import qualified Database.PostgreSQL.Simple.ToField as PG.To+import qualified Database.PostgreSQL.Simple.ToRow as PG.To+import qualified Pipes++import Database.CQRS.PostgreSQL.Internal (SomeParams(..), handleError)++import qualified Database.CQRS as CQRS++-- | Stream of events stored in a PostgreSQL relation.+--+-- The job of sharding streams in different tables is left to the database. If+-- this is something you want to do, you can create a view and a trigger on+-- insert into that view.+data Stream identifier metadata event =+ forall r r'. (PG.To.ToRow r, PG.To.ToRow r') => Stream+ { connectionPool :: forall a. (PG.Connection -> IO a) -> IO a+ , selectQuery :: (PG.Query, r)+ -- ^ Select all events in correct order.+ , insertQuery+ :: forall encEvent.+ (PG.To.ToField encEvent, PG.To.ToRow metadata)+ => encEvent -> metadata -> CQRS.ConsistencyCheck identifier+ -> (PG.Query, r')+ -- ^ Build a query to insert a new event. The constraints are there to+ -- postpone them to calls to 'writeEventWithMetadata'. If the user never+ -- writes new events, these constraints won't be required by the compiler.+ , identifierColumn :: PG.Query+ }++-- | Make a 'Stream' from basic information about the relation name and columns.+makeStream+ :: forall identifier metadata event.+ ( CQRS.WritableEvent event+ , PG.To.ToField (CQRS.EncodingFormat event)+ , PG.To.ToField identifier+ , PG.To.ToRow metadata+ )+ => (forall a. (PG.Connection -> IO a) -> IO a)+ -- ^ Connection pool as a function.+ -> PG.Query -- ^ Relation name.+ -> PG.Query -- ^ Identifier column name. If there are several, use a tuple.+ -> [PG.Query] -- ^ Column names for metadata.+ -> PG.Query -- ^ Event column name.+ -> Stream identifier metadata event+makeStream connectionPool relation+ identifierColumn metadataColumns eventColumn =+ let selectQuery = (selectQuery', ())+ in Stream{..}++ where+ selectQuery' :: PG.Query+ selectQuery' =+ "SELECT "+ <> identifierColumn <> ", "+ <> metadataList <> ", "+ <> eventColumn+ <> " FROM " <> relation+ <> " ORDER BY " <> identifierColumn <> " ASC"++ insertQuery+ :: (PG.To.ToField encEvent, PG.To.ToRow metadata)+ => encEvent -> metadata -> CQRS.ConsistencyCheck identifier+ -> (PG.Query, SomeParams)+ insertQuery encEvent metadata cc =+ let baseParams = metadata :. PG.Only encEvent+ (cond, params) = case cc of+ CQRS.NoConsistencyCheck -> ("", SomeParams baseParams)+ CQRS.CheckNoEvents ->+ ( " WHERE NOT EXISTS (SELECT 1 FROM " <> relation <> ")"+ , SomeParams baseParams+ )+ CQRS.CheckLastEvent identifier ->+ ( " WHERE NOT EXISTS (SELECT 1 FROM " <> relation <> " WHERE "+ <> identifierColumn <> " > ?)"+ , SomeParams (baseParams :. PG.Only identifier)+ )+ query =+ "INSERT INTO " <> relation <> "("+ <> metadataList <> ", " <> eventColumn+ <> ") SELECT " <> metadataMarks <> ", ?"+ <> cond+ <> " RETURNING " <> identifierColumn+ in+ (query, params)++ metadataList :: PG.Query+ metadataList =+ mconcat . intersperse "," $ metadataColumns++ metadataMarks :: PG.Query+ metadataMarks =+ mconcat . intersperse "," . map (const "?") $ metadataColumns++-- | Make a stream from queries.+--+-- This function is less safe than 'makeStream' and should only be used when+-- 'makeStream' is not flexible enough. It is also closer to the implementation+-- and more subject to changes.+makeStream'+ :: (PG.To.ToRow r, PG.To.ToRow r')+ => (forall a. (PG.Connection -> IO a) -> IO a)+ -- ^ Connection pool as a function.+ -> (PG.Query, r) -- ^ Select query.+ -> (forall encEvent. (PG.To.ToField encEvent, PG.To.ToRow metadata)+ => encEvent -> metadata -> CQRS.ConsistencyCheck identifier+ -> (PG.Query, r'))+ -- ^ Insert query builder.+ -> PG.Query -- ^ Identifier column (use tuple if several.)+ -> Stream identifier metadata event+makeStream' connectionPool selectQuery insertQuery identifierColumn =+ Stream{..}++instance+ ( CQRS.Event event+ , Exc.MonadError CQRS.Error m+ , MonadIO m+ , Ord identifier+ , PG.From.FromField identifier+ , PG.To.ToField identifier+ , PG.From.FromRow metadata+ , PG.From.FromField (CQRS.EncodingFormat event)+ ) => CQRS.Stream m (Stream identifier metadata event) where++ type EventType (Stream identifier metadata event) = event+ type EventIdentifier (Stream identifier metadata event) = identifier+ type EventMetadata (Stream identifier metadata event) = metadata++ streamEvents = streamStreamEvents++instance+ ( CQRS.WritableEvent event+ , Exc.MonadError CQRS.Error m+ , MonadIO m+ , Ord identifier+ , PG.From.FromField identifier+ , PG.To.ToField identifier+ , PG.From.FromField (CQRS.EncodingFormat event)+ , PG.To.ToField (CQRS.EncodingFormat event)+ , PG.From.FromRow metadata+ , PG.To.ToRow metadata+ ) => CQRS.WritableStream m (Stream identifier metadata event) where++ writeEventWithMetadata = streamWriteEventWithMetadata++streamWriteEventWithMetadata+ :: ( CQRS.WritableEvent event+ , Exc.MonadError CQRS.Error m+ , MonadIO m+ , PG.From.FromField identifier+ , PG.From.FromField (CQRS.EncodingFormat event)+ , PG.To.ToField (CQRS.EncodingFormat event)+ , PG.To.ToRow metadata+ )+ => Stream identifier metadata event+ -> event+ -> metadata+ -> CQRS.ConsistencyCheck identifier+ -> m identifier+streamWriteEventWithMetadata Stream{..} event metadata cc = do+ eIds <- liftIO . connectionPool $ \conn -> do+ let (req, params) = insertQuery (CQRS.encodeEvent event) metadata cc+ (Right <$> PG.query conn req params)+ `catches`+ [ handleError (Proxy @PG.FormatError) CQRS.EventWriteError+ , handleError (Proxy @PG.QueryError) CQRS.EventWriteError+ , handleError (Proxy @PG.ResultError) CQRS.EventWriteError+ , handleError (Proxy @PG.SqlError) CQRS.EventWriteError+ ]++ case eIds of+ Left err -> Exc.throwError err+ Right [PG.Only identifier] -> pure identifier+ Right [] -> Exc.throwError $ CQRS.ConsistencyCheckError "no events inserted"+ Right ids -> Exc.throwError $ CQRS.EventWriteError $+ show (length ids) ++ " events were inserted"++streamStreamEvents+ :: forall identifier metadata event m.+ ( CQRS.Event event+ , Exc.MonadError CQRS.Error m+ , MonadIO m+ , Ord identifier+ , PG.From.FromField identifier+ , PG.To.ToField identifier+ , PG.From.FromRow metadata+ , PG.From.FromField (CQRS.EncodingFormat event)+ )+ => Stream identifier metadata event+ -> CQRS.StreamBounds identifier+ -> Pipes.Producer+ [ Either+ (identifier, String) (CQRS.EventWithContext identifier metadata event)+ ] m ()+streamStreamEvents Stream{..} bounds =+ go Nothing++ where+ go+ :: Maybe identifier+ -> Pipes.Producer+ [ Either+ (identifier, String)+ (CQRS.EventWithContext identifier metadata event)+ ] m ()+ go lastFetchedIdentifier = do+ -- Fetch 'batchSize' parsed events from the DB.+ eRows <- liftIO . connectionPool $ \conn ->+ (Right <$> fetchBatch conn lastFetchedIdentifier)+ `catches`+ [ handleError (Proxy @PG.FormatError) CQRS.EventRetrievalError+ , handleError (Proxy @PG.QueryError) CQRS.EventRetrievalError+ , handleError (Proxy @PG.ResultError) CQRS.EventRetrievalError+ , handleError (Proxy @PG.SqlError) CQRS.EventRetrievalError+ ]++ rows <- either Exc.throwError pure eRows++ Pipes.yield $+ map+ (\(PG.Only identifier :. metadata :. PG.Only encEvent) ->+ case CQRS.decodeEvent encEvent of+ Left err -> Left (identifier, err)+ Right event ->+ Right $ CQRS.EventWithContext identifier metadata event+ ) rows++ let mLfi =+ fmap (\(PG.Only identifier :. _) -> identifier)+ . listToMaybe+ . reverse+ $ rows+ when (length rows == batchSize) $ go mLfi++ fetchBatch+ :: PG.Connection+ -> Maybe identifier+ -> IO [ PG.Only identifier+ :. metadata+ :. PG.Only (CQRS.EncodingFormat event) ]+ fetchBatch conn lastFetchedIdentifier = do+ let bounds' =+ bounds <> maybe mempty CQRS.afterEvent lastFetchedIdentifier+ conditions = catMaybes+ [ CQRS._afterEvent bounds' <&> \i ->+ (identifierColumn <> " > ?", i)+ , CQRS._untilEvent bounds' <&> \i ->+ (identifierColumn <> " <= ?", i)+ ]+ whereClause+ | null conditions = mempty+ | otherwise =+ (" WHERE " <>)+ . mconcat+ . intersperse " AND "+ . map fst+ $ conditions+ params = snd selectQuery :. map snd conditions+ query =+ "SELECT * FROM (" <> fst selectQuery <> ") AS _"+ <> whereClause+ <> " LIMIT "+ <> fromString (show batchSize)++ PG.query conn query params++ batchSize :: Int+ batchSize = 100
+ src/Database/CQRS/PostgreSQL/StreamFamily.hs view
@@ -0,0 +1,408 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE UndecidableInstances #-}++module Database.CQRS.PostgreSQL.StreamFamily+ ( StreamFamily(..)+ , makeStreamFamily+ ) where++import Control.Concurrent+import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar)+import Control.Exception+import Control.Monad (void)+import Control.Monad.Trans (MonadIO(..))+import Data.List (intersperse)+import Data.Proxy (Proxy(..))+import Database.PostgreSQL.Simple ((:.)(..))+import System.Mem.Weak (Weak, deRefWeak, mkWeak)++import qualified Control.Concurrent.STM as STM+import qualified Control.Monad.Except as Exc+import qualified Data.ByteString as BS+import qualified Database.PostgreSQL.Simple as PG+import qualified Database.PostgreSQL.Simple.FromField as PG.From+import qualified Database.PostgreSQL.Simple.FromRow as PG.From+import qualified Database.PostgreSQL.Simple.Notification as PG+import qualified Database.PostgreSQL.Simple.ToField as PG.To+import qualified Database.PostgreSQL.Simple.ToRow as PG.To+import qualified Pipes++import Database.CQRS.PostgreSQL.Internal (SomeParams(..), handleError)+import Database.CQRS.PostgreSQL.Stream++import qualified Database.CQRS as CQRS++-- | Family of event streams stored in a PostgreSQL relation.+--+-- Each stream should have a unique stream identifier and event identifiers must+-- be unique within a stream, but not necessarily across them.+--+-- 'allNewEvents' starts a new thread which reads notifications on the given+-- channel and writes them to a transactional bounded queue (a 'TBQueue') which+-- is then consumed by the returned 'Producer'. The maximum size of this queue+-- is hard-coded to 100. Should an exception be raised in the listening thread,+-- it is thrown back by the producer.+data StreamFamily streamId eventId metadata event = StreamFamily+ { connectionPool :: forall a. (PG.Connection -> IO a) -> IO a+ , relation :: PG.Query+ , notificationChannel :: PG.Query+ , parseNotification :: BS.ByteString -> Either String (streamId, eventId)+ , streamIdentifierColumn :: PG.Query+ , eventIdentifierColumn :: PG.Query+ , metadataColumns :: [PG.Query]+ , eventColumn :: PG.Query+ }++makeStreamFamily+ :: (forall a. (PG.Connection -> IO a) -> IO a)+ -> PG.Query+ -> PG.Query+ -> (BS.ByteString -> Either String (streamId, eventId))+ -> PG.Query+ -> PG.Query+ -> [PG.Query]+ -> PG.Query+ -> StreamFamily streamId eventId metadata event+makeStreamFamily = StreamFamily++instance+ ( CQRS.Event event+ , Exc.MonadError CQRS.Error m+ , MonadIO m+ , PG.From.FromField eventId+ , PG.From.FromField streamId+ , PG.From.FromField (CQRS.EncodingFormat event)+ , PG.From.FromRow metadata+ , PG.To.ToField eventId+ , PG.To.ToField streamId+ )+ => CQRS.StreamFamily m (StreamFamily streamId eventId metadata event) where++ type StreamType (StreamFamily streamId eventId metadata event) =+ Stream eventId metadata event++ type StreamIdentifier (StreamFamily streamId eventId metadata event) =+ streamId++ getStream = streamFamilyGetStream+ allNewEvents = streamFamilyAllNewEvents+ latestEventIdentifiers = streamFamilyLastEventIdentifiers++streamFamilyGetStream+ :: forall streamId eventId metadata event m.+ ( MonadIO m+ , PG.To.ToField streamId+ , PG.To.ToField eventId+ )+ => StreamFamily streamId eventId metadata event+ -> streamId+ -> m (Stream eventId metadata event)+streamFamilyGetStream StreamFamily{..} streamId =+ pure $+ makeStream' connectionPool selectQuery insertQuery eventIdentifierColumn++ where+ selectQuery :: (PG.Query, PG.Only streamId)+ selectQuery = (selectQueryTpl, PG.Only streamId)++ selectQueryTpl :: PG.Query+ selectQueryTpl =+ "SELECT "+ <> eventIdentifierColumn <> ", " <> metadataList <> ", " <> eventColumn+ <> " FROM " <> relation+ <> " WHERE " <> streamIdentifierColumn+ <> " = ? ORDER BY "+ <> eventIdentifierColumn <> " ASC"++ insertQuery+ :: (PG.To.ToField encEvent, PG.To.ToRow metadata)+ => encEvent+ -> metadata+ -> CQRS.ConsistencyCheck eventId+ -> (PG.Query, SomeParams)+ insertQuery encEvent metadata cc =+ let baseParams =+ PG.Only streamId :. metadata :. PG.Only encEvent+ (cond, params) = case cc of+ CQRS.NoConsistencyCheck -> ("", SomeParams baseParams)+ CQRS.CheckNoEvents ->+ ( " WHERE NOT EXISTS (SELECT 1 FROM " <> relation <> " WHERE "+ <> streamIdentifierColumn <> " = ?)"+ , SomeParams (baseParams :. PG.Only streamId)+ )+ CQRS.CheckLastEvent eventId ->+ ( " WHERE NOT EXISTS (SELECT 1 FROM " <> relation <> " WHERE "+ <> streamIdentifierColumn <> " = ? AND "+ <> eventIdentifierColumn <> " > ?)"+ , SomeParams (baseParams :. (streamId, eventId))+ )+ query =+ "INSERT INTO "+ <> relation <> "("+ <> streamIdentifierColumn <> ", "+ <> metadataList <> ", "+ <> eventColumn <> ") SELECT ?, "+ <> metadataMarks <> ", ?" <> cond <> " RETURNING "+ <> eventIdentifierColumn+ in+ (query, params)++ metadataList :: PG.Query+ metadataList =+ mconcat . intersperse "," $ metadataColumns++ metadataMarks :: PG.Query+ metadataMarks =+ mconcat . intersperse "," . map (const "?") $ metadataColumns++data GCKey = GCKey++streamFamilyAllNewEvents+ :: forall streamId eventId metadata event m a.+ ( CQRS.Event event+ , Exc.MonadError CQRS.Error m+ , MonadIO m+ , PG.From.FromField eventId+ , PG.From.FromField streamId+ , PG.From.FromField (CQRS.EncodingFormat event)+ , PG.From.FromRow metadata+ , PG.To.ToField eventId+ , PG.To.ToField streamId+ )+ => StreamFamily streamId eventId metadata event+ -> m (Pipes.Producer+ [ ( streamId+ , Either+ (eventId, String) (CQRS.EventWithContext eventId metadata event)+ ) ]+ m a)+streamFamilyAllNewEvents StreamFamily{..} = liftIO $ do+ let gcKey = GCKey+ queue <- STM.newTBQueueIO 100+ queueWeak <- mkWeak gcKey queue Nothing+ mvar <- startListeningThread queueWeak+ pure $ producer gcKey mvar queue++ where+ -- Start the listening thread and return an 'MVar' of potential error of the+ -- thread once it has started.+ startListeningThread+ :: Weak (STM.TBQueue (streamId, eventId))+ -> IO (MVar CQRS.Error)+ startListeningThread queueWeak = do+ errorMVar <- newEmptyMVar+ startedMVar <- newEmptyMVar+ _ <- forkFinally (listen startedMVar queueWeak) $ \eErr -> do+ putMVar startedMVar () -- Unblock it.+ putMVar errorMVar $ case eErr of+ Left err -> CQRS.NewEventsStreamingError . show $ err+ Right err -> err+ takeMVar startedMVar -- Wait to be sure it started to listen.+ pure errorMVar++ -- Entry point of the thread. It gets a connection, starts listening for+ -- notifications and signals it has started to the parent thread.+ listen+ :: MVar ()+ -> Weak (STM.TBQueue (streamId, eventId))+ -> IO CQRS.Error+ listen startedMVar queueWeak =+ connectionPool $ \conn -> do+ eThreadRes <- Exc.runExceptT $ do+ eRes <- liftIO $+ (Right <$> PG.execute_ conn ("LISTEN " <> notificationChannel))+ `catches`+ [ handleError (Proxy @PG.FormatError)+ CQRS.NewEventsStreamingError+ , handleError (Proxy @PG.SqlError) CQRS.NewEventsStreamingError+ ]+ either Exc.throwError (\_ -> pure ()) eRes+ liftIO $ putMVar startedMVar ()+ handleNotifications conn queueWeak++ case eThreadRes of+ Left err -> do+ _ <- try @SomeException $ PG.execute_ conn "UNLISTEN *"+ pure err+ Right () ->+ pure $ CQRS.NewEventsStreamingError "listening thread terminated"++ -- Once the listening thread has initialised itself, it runs this code over+ -- and over again.+ handleNotifications+ :: PG.Connection+ -> Weak (STM.TBQueue (streamId, eventId))+ -> Exc.ExceptT CQRS.Error IO ()+ handleNotifications conn queueWeak = do+ -- Get notifications first even if the queue might not exist anymore.+ -- Otherwise, this thread would *not* have a pointer to the queue only+ -- between the recursion call and the first line of the function, i.e.+ -- maybe not enough time for the garbage collector to run.+ notif <- liftIO $ PG.getNotification conn+ mQueue <- liftIO $ deRefWeak queueWeak+ case mQueue of+ Nothing -> pure () -- The queue has been garbage collected.+ Just queue -> do+ case parseNotification (PG.notificationData notif) of+ Left err ->+ Exc.throwError . CQRS.NewEventsStreamingError $+ "error decoding notification: " ++ err+ Right pair ->+ liftIO . STM.atomically . STM.writeTBQueue queue $ pair+ handleNotifications conn queueWeak++ -- Producer that repeatedly checks the listening thread is still alive and+ -- fetch corresponding events to the notifications.+ producer+ :: GCKey+ -> MVar CQRS.Error -- Error from listening thread.+ -> STM.TBQueue (streamId, eventId)+ -> Pipes.Producer+ [ ( streamId+ , Either+ (eventId, String) (CQRS.EventWithContext eventId metadata event)+ ) ]+ m a+ producer gcKey mvar queue = do+ -- Check the listening thread is still running.+ mErr <- liftIO $ tryTakeMVar mvar+ maybe (pure ()) Exc.throwError mErr++ -- Get some events or none after timeout just in case the listening+ -- thread died in the meantime (we don't want to block forever.)+ events <- liftIO $ race+ (\tmvar -> STM.atomically $ do+ events <- (:) <$> STM.readTBQueue queue <*> STM.flushTBQueue queue+ STM.putTMVar tmvar . Right $ events)+ (\tmvar -> do+ threadDelay 1000000 -- 1 second+ STM.atomically . STM.putTMVar tmvar . Right $ [])++ fetchEvents events+ producer gcKey mvar queue++ fetchEvents+ :: [(streamId, eventId)]+ -> Pipes.Producer+ [ ( streamId+ , Either+ (eventId, String) (CQRS.EventWithContext eventId metadata event)+ ) ]+ m ()+ fetchEvents [] = Pipes.yield []+ fetchEvents pairs = do+ let pairs' = map (uncurry Pair) pairs+ eRows <- liftIO . connectionPool $ \conn ->+ (Right <$> PG.query conn fetchQuery (PG.Only (PG.In pairs')))+ `catches`+ [ handleError (Proxy @PG.FormatError) CQRS.EventRetrievalError+ , handleError (Proxy @PG.QueryError) CQRS.EventRetrievalError+ , handleError (Proxy @PG.ResultError) CQRS.EventRetrievalError+ , handleError (Proxy @PG.SqlError) CQRS.EventRetrievalError+ ]++ rows <- either Exc.throwError pure eRows++ Pipes.yield $+ map+ (\(PG.Only streamId :. PG.Only identifier+ :. metadata :. PG.Only encEvent) ->+ case CQRS.decodeEvent encEvent of+ Left err -> (streamId, Left (identifier, err))+ Right event ->+ ( streamId+ , Right (CQRS.EventWithContext identifier metadata event)+ )+ ) rows++ fetchQuery :: PG.Query+ fetchQuery =+ "SELECT "+ <> streamIdentifierColumn <> ", "+ <> eventIdentifierColumn <> ", "+ <> metadataList <> ", "+ <> eventColumn+ <> " FROM " <> relation+ <> " WHERE (" <> streamIdentifierColumn <> ", "+ <> eventIdentifierColumn+ <> ") IN ? ORDER BY " <> eventIdentifierColumn <> " ASC"++ metadataList :: PG.Query+ metadataList =+ mconcat . intersperse "," $ metadataColumns++streamFamilyLastEventIdentifiers+ :: ( Exc.MonadError CQRS.Error m+ , MonadIO m+ , PG.From.FromField eventId+ , PG.From.FromField streamId+ )+ => StreamFamily streamId eventId metadata event+ -> Pipes.Producer (streamId, eventId) m ()+streamFamilyLastEventIdentifiers StreamFamily{..} = do+ eIds <- liftIO . connectionPool $ \conn -> do+ (Right <$> PG.query_ conn query)+ `catches`+ [ handleError (Proxy @PG.FormatError) CQRS.EventRetrievalError+ , handleError (Proxy @PG.QueryError) CQRS.EventRetrievalError+ , handleError (Proxy @PG.ResultError) CQRS.EventRetrievalError+ , handleError (Proxy @PG.SqlError) CQRS.EventRetrievalError+ ]++ either Exc.throwError Pipes.each eIds++ where+ query :: PG.Query+ query =+ "SELECT "+ <> streamIdentifierColumn <> ", "+ <> "max(" <> eventIdentifierColumn+ <> ") FROM " <> relation+ <> " GROUP BY " <> streamIdentifierColumn++-- | Run two threads concurrently and return the result of the first one to+-- write to the givem 'TMVar'. If the first thread to finish does so because+-- it throws an exception, the exception is rethrown in the main process.+race+ :: (STM.TMVar (Either SomeException a) -> IO ())+ -> (STM.TMVar (Either SomeException a) -> IO ())+ -> IO a+race f g = do+ tmvar <- STM.newEmptyTMVarIO+ (tid, tid') <- mask $ \restore ->+ (,) <$> run f tmvar restore <*> run g tmvar restore+ eRes <- STM.atomically . STM.readTMVar $ tmvar+ killThread tid+ killThread tid'+ either throw pure eRes++ where+ run+ :: (STM.TMVar (Either SomeException a) -> IO ())+ -> STM.TMVar (Either SomeException a)+ -> (forall b. IO b -> IO b)+ -> IO ThreadId+ run h tmvar restore =+ forkIO $+ restore (h tmvar)+ `catch` (void . STM.atomically . STM.tryPutTMVar tmvar . Left)++data Pair a b = Pair a b++instance (PG.To.ToField a, PG.To.ToField b) => PG.To.ToField (Pair a b) where+ toField (Pair x y) =+ PG.To.Many+ [ PG.To.Plain "ROW("+ , PG.To.toField x+ , PG.To.Plain ","+ , PG.To.toField y+ , PG.To.Plain ")"+ ]
+ src/Database/CQRS/PostgreSQL/TrackingTable.hs view
@@ -0,0 +1,196 @@+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE InstanceSigs #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE TypeApplications #-}++module Database.CQRS.PostgreSQL.TrackingTable+ ( TrackingTable+ , createTrackingTable+ , upsertTrackingTable+ , doUpsertTrackingTable+ ) where++import Control.Applicative ((<|>))+import Control.Exception (catches)+import Control.Monad ((<=<), void)+import Control.Monad.Trans (MonadIO(..))+import Data.Proxy (Proxy(..))++import qualified Control.Monad.Except as Exc+import qualified Database.PostgreSQL.Simple as PG+import qualified Database.PostgreSQL.Simple.Types as PG+import qualified Database.PostgreSQL.Simple.FromField as PG.From+import qualified Database.PostgreSQL.Simple.ToField as PG.To+import qualified Database.PostgreSQL.Simple.ToRow as PG.To++import Database.CQRS.PostgreSQL.Internal (makeSqlAction, SqlAction, handleError)++import qualified Database.CQRS as CQRS++data TrackingTable streamId eventId st = TrackingTable+ { connectionPool :: forall r. (PG.Connection -> IO r) -> IO r+ , relation :: PG.Query+ }++instance+ ( Exc.MonadError CQRS.Error m+ , MonadIO m+ , PG.From.FromField eventId+ , PG.From.FromField st+ , PG.From.FromField streamId+ , PG.To.ToField eventId+ , PG.To.ToField st+ , PG.To.ToField streamId+ )+ => CQRS.TrackingTable m (TrackingTable streamId eventId st)+ streamId eventId st where++ getTrackedState+ :: TrackingTable streamId eventId st+ -> streamId+ -> m (CQRS.TrackedState eventId st)+ getTrackedState TrackingTable{..} streamId =+ handlePgErrors . connectionPool $ \conn -> do+ let query =+ "SELECT event_id, failed_event_id, failed_message, state FROM "+ <> relation <> " WHERE stream_id = ?"+ rows <- PG.query conn query (PG.Only streamId)+ pure $ case rows of+ [(Just eventId, Nothing, Nothing, SomeState state)] ->+ CQRS.SuccessAt eventId state+ [(mEventId, Just failedAt, Just err, oState)] ->+ CQRS.FailureAt+ ((,) <$> mEventId <*> fromOptionalState oState) failedAt err+ _ -> CQRS.NeverRan++ upsertError+ :: TrackingTable streamId eventId st+ -> streamId+ -> eventId+ -> String+ -> m ()+ upsertError trackingTable streamId eventId err =+ doUpsertTrackingTable trackingTable streamId eventId (Left err)++-- | Return SQL query to upsert a row in a tracking table.+upsertTrackingTable+ :: ( PG.To.ToField streamId+ , PG.To.ToField eventId+ , PG.To.ToField st+ )+ => TrackingTable streamId eventId st+ -> streamId+ -> eventId+ -> Either String st -- ^ The new state or the error message if it failed.+ -> SqlAction+upsertTrackingTable TrackingTable{..} streamId eventId eState =+ let (updates, updateValues, insertValues) =+ case eState of+ Right state ->+ ( "event_id = ?, failed_event_id = NULL,\+ \ failed_message = NULL, state = ?"+ , [PG.To.toField eventId, PG.To.toField state]+ , PG.To.toRow (streamId, eventId, PG.Null, PG.Null, state)+ )+ Left err ->+ ( "failed_event_id = ?, failed_message = ?"+ , PG.To.toRow (eventId, err)+ , PG.To.toRow (streamId, PG.Null, eventId, err, PG.Null)+ )+ query =+ "INSERT INTO " <> relation+ <> " (stream_id, event_id, failed_event_id, failed_message, state)"+ <> " VALUES (?, ?, ?, ?, ?) ON CONFLICT (stream_id) DO UPDATE SET "+ <> updates+ in+ makeSqlAction query $ insertValues ++ updateValues++-- | Update the tracking table for the given stream.+doUpsertTrackingTable+ :: ( Exc.MonadError CQRS.Error m+ , MonadIO m+ , PG.To.ToField eventId+ , PG.To.ToField streamId+ , PG.To.ToField st+ )+ => TrackingTable streamId eventId st+ -> streamId+ -> eventId+ -> Either String st -- ^ The new state or the error message if it failed.+ -> m ()+doUpsertTrackingTable tt@TrackingTable{..} streamId eventId eState = do+ handlePgErrors . connectionPool $ \conn -> do+ let (query, values) = upsertTrackingTable tt streamId eventId eState+ void $ PG.execute conn query values++-- | Create tracking table if it doesn't exist already.+--+-- A tracking table is a table used to track the last events processed by a+-- projection for each stream in a stream family. It allows them to restart from+-- where they have left off.+createTrackingTable+ :: ( Exc.MonadError CQRS.Error m+ , MonadIO m+ )+ => (forall r. (PG.Connection -> IO r) -> IO r)+ -> PG.Query -- ^ Name of the tracking table.+ -> PG.Query -- ^ Type of stream identifiers.+ -> PG.Query -- ^ Type of event identifiers.+ -> PG.Query -- ^ Type of the state.+ -> m (TrackingTable streamId eventId st)+createTrackingTable+ connectionPool relation streamIdType eventIdType stateType = do++ handlePgErrors . connectionPool $ \conn -> do+ let query =+ "CREATE TABLE IF NOT EXISTS " <> relation+ <> " ( stream_id " <> streamIdType <> " PRIMARY KEY"+ <> " , event_id " <> eventIdType+ <> " , failed_event_id " <> eventIdType+ <> " , failed_message varchar"+ <> " , state " <> stateType <> ")"+ void $ PG.execute_ conn query++ pure TrackingTable{..}++handlePgErrors+ :: ( Exc.MonadError CQRS.Error m+ , MonadIO m+ )+ => IO a -> m a+handlePgErrors f =+ Exc.liftEither <=< liftIO $ do+ (Right <$> f)+ `catches`+ [ handleError (Proxy @PG.FormatError) CQRS.TrackingTableError+ , handleError (Proxy @PG.SqlError) CQRS.TrackingTableError+ , handleError (Proxy @PG.QueryError) CQRS.TrackingTableError+ , handleError (Proxy @PG.ResultError) CQRS.TrackingTableError+ ]++-- | An alternative to 'Maybe st'.+--+-- If we use 'Maybe st' and 'st ~ Maybe a' (or something else where @NULL@ is a+-- valid state,) then 'getTrackedState''s @SELECT@ statement would return a+-- value of type 'Maybe (Maybe a)' which would be 'Nothing' instead of+-- 'Just Nothing' if the column is 'NULL.+--+-- This type works differently: it tries to use 'PG.From.fromField' on the field+-- and return 'NoState' if it didn't work AND it is @NULL@. Otherwise, it fails.+data OptionalState st = SomeState st | NoState++instance PG.From.FromField st => PG.From.FromField (OptionalState st) where+ fromField f mBS =+ case mBS of+ Nothing -> (SomeState <$> PG.From.fromField f mBS) <|> pure NoState+ Just _ -> SomeState <$> PG.From.fromField f mBS++fromOptionalState :: OptionalState a -> Maybe a+fromOptionalState = \case+ SomeState x -> Just x+ NoState -> Nothing