Merge branch 'dev' into 467-dev-api-refactorings

gargantext.cabal

@@ -299,15 +299,17 @@ library
       Gargantext.Database.Query.Table.NgramsPostag
       Gargantext.Database.Query.Table.Node
       Gargantext.Database.Query.Table.Node.Error
-      Gargantext.Database.Query.Table.NodeNode
       Gargantext.Database.Query.Table.Node.UpdateOpaleye
       Gargantext.Database.Query.Table.Node.User
+      Gargantext.Database.Query.Table.NodeNode
       Gargantext.Database.Query.Table.User
       Gargantext.Database.Query.Tree.Root
       Gargantext.Database.Schema.Context
       Gargantext.Database.Schema.Ngrams
       Gargantext.Database.Schema.Node
       Gargantext.Database.Schema.User
+      Gargantext.Database.Transactional
+      Gargantext.Database.Transactional.Example
       Gargantext.Defaults
       Gargantext.MicroServices.ReverseProxy
       Gargantext.Orphans
@@ -540,6 +542,7 @@ library
     , filepath ^>= 1.4.2.2
     , fmt
     , formatting ^>= 7.2.0
+    , free >= 0.5.0
     , fullstop ^>= 0.1.4
     , gargantext-graph-core >= 0.2.0.0
     , gargantext-prelude
@@ -558,8 +561,8 @@ library
     , http-client-tls >= 0.3.6.1 && < 0.4
     , http-conduit >= 2.3.8 && < 2.3.9
     , http-media ^>= 0.8.0.0
-    , http-types ^>= 0.12.3
     , http-reverse-proxy >= 0.6.1.0
+    , http-types ^>= 0.12.3
     , ini ^>= 0.4.1
     , insert-ordered-containers ^>= 0.2.5.1
     , iso639 ^>= 0.1.0.3
@@ -612,8 +615,8 @@ library
     , servant-client-core >= 0.20 && < 0.21
     , servant-conduit >= 0.15 && < 0.17
     , servant-ekg ^>= 0.3.1
-    , servant-routes < 0.2
     , servant-openapi3 >= 2.0.1.6
+    , servant-routes < 0.2
     , servant-server >= 0.18.3 && < 0.21
     , servant-swagger ^>= 1.2
     , servant-swagger-ui ^>= 0.3.5.3.5.0
@@ -637,13 +640,13 @@ library
     , text ^>= 2.0.2
     , text-metrics ^>= 0.3.2
     , time ^>= 1.12.2
+    , toml-parser >= 2.0.1.0 && < 3
     , transformers
     , transformers-base ^>= 0.4.6
     , tree-diff
-    , toml-parser >= 2.0.1.0 && < 3
     , tuple ^>= 0.3.0.2
-    , unordered-containers ^>= 0.2.16.0
     , unicode-collation >= 0.1.3.5
+    , unordered-containers ^>= 0.2.16.0
     -- needed for Worker / System.Posix.Signals
     , unix >= 2.7.3 && < 2.9
     , uri-encode ^>= 1.5.0.7
@@ -757,6 +760,7 @@ common commonTestDependencies
     , postgres-options >= 0.2 && < 0.3
     , postgresql-simple >= 0.6.4 && <= 0.7.0.0
     , quickcheck-instances ^>= 0.3.25.2
+    , random
     , raw-strings-qq
     , resource-pool >= 0.4.0.0 && < 0.5
     , safe-exceptions >= 0.1.7.4 && < 0.2
@@ -835,6 +839,7 @@ test-suite garg-test-tasty
       Test.Database.Operations.NodeStory
       Test.Database.Operations.PublishNode
       Test.Database.Setup
+      Test.Database.Transactions
       Test.Database.Types
       Test.Graph.Clustering
       Test.Graph.Distance
@@ -902,6 +907,7 @@ test-suite garg-test-hspec
       Test.Database.Operations.NodeStory
       Test.Database.Operations.PublishNode
       Test.Database.Setup
+      Test.Database.Transactions
       Test.Database.Types
       Test.Instances
       Test.Server.ReverseProxy

src/Gargantext/Database/Action/Share.hs

@@ -125,7 +125,7 @@ shareNodeWith (ShareNodeWith_Node NodeFolderPublic nId) n = do
 shareNodeWith _ _ = errorWith "[G.D.A.S.shareNodeWith] Not implemented for this NodeType"
 
 ------------------------------------------------------------------------
-getFolderId :: HasNodeError err => User -> NodeType -> DBCmdExtra err NodeId
+getFolderId :: HasNodeError err => User -> NodeType -> DBCmd err NodeId
 getFolderId u nt = do
   rootId <- getRootId u
   s <- getNodesWith rootId HyperdataAny (Just nt) Nothing Nothing

src/Gargantext/Database/Query/Table/Node.hs

@@ -50,6 +50,7 @@ module Gargantext.Database.Query.Table.Node
   , getNodesWith
   , getNodesWithParentId
   , getNodesWithType
+  , selectNodesWith
 
   -- * Creating one or more nodes
   , insertDefaultNode

src/Gargantext/Database/Query/Table/Node/Error.hs

@@ -24,6 +24,7 @@ module Gargantext.Database.Query.Table.Node.Error (
   , nodeCreationError
   , nodeLookupError
   , catchNodeError
+  , dbFailWith
   ) where
 
 import Control.Lens (Prism', (#), (^?))
@@ -34,6 +35,7 @@ import Gargantext.Database.Admin.Types.Node (ListId, NodeId(..), ContextId, User
 import Gargantext.Prelude hiding (sum, head)
 import Prelude hiding (null, id, map, sum, show)
 import Prelude qualified
+import Gargantext.Database.Transactional
 
 data NodeCreationError
   = UserParentAlreadyExists UserId ParentId
@@ -155,3 +157,7 @@ nodeLookupError ne = throwError $ _NodeError # NodeLookupFailed ne
 
 catchNodeError :: (MonadError e m, HasNodeError e) => m a -> (NodeError -> m a) -> m a
 catchNodeError f g = catchError f (\e -> maybe (throwError e) g (e ^? _NodeError))
+
+
+dbFailWith :: HasNodeError err => T.Text -> DBTx err r b
+dbFailWith x = dbFail $ _NodeError # (NodeError $ toException $ userError $ T.unpack x)

src/Gargantext/Database/Query/Table/NodeNode.hs

@@ -45,6 +45,9 @@ module Gargantext.Database.Query.Table.NodeNode
   , unpublishNode
   , queryNodeNodeTable
   , shareNode
+
+  -- * Internals (use with caution)
+  , insertNodeNode
   )
   where
 

src/Gargantext/Database/Transactional.hs

+{-# LANGUAGE KindSignatures      #-}
+{-# LANGUAGE LambdaCase          #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+module Gargantext.Database.Transactional (
+    DBOperation
+  , DBTransactionOp -- opaque
+  , DBTx -- opaque
+  , DBUpdate
+  , DBQuery
+  , DBTxCmd
+  -- * Executing queries and updates
+  , runDBQuery
+  , runDBTx
+
+  -- * Smart constructors
+  , mkPGQuery
+  , mkPGUpdate
+  , mkPGUpdateReturning
+  , mkOpaQuery
+  , mkOpaUpdate
+  , mkOpaInsert
+
+  -- * Throwing errors (which allow rollbacks)
+  , dbFail
+  ) where
+
+import Control.Lens
+import Control.Monad.Base
+import Control.Monad.Error.Class
+import Control.Monad.Free
+import Control.Monad.Trans.Control (MonadBaseControl, control)
+import Data.Pool (withResource, Pool)
+import Data.Profunctor.Product.Default
+import Database.PostgreSQL.Simple qualified as PG
+import Database.PostgreSQL.Simple.Transaction qualified as PG
+import Gargantext.Database.Prelude
+import Opaleye
+import Prelude
+import qualified Control.Exception.Safe as Safe
+
+data DBOperation = DBRead | DBWrite
+
+-- | A functor describing a single operation on the database. Each constructor takes a continuation
+-- argument which can be used to derive a sound 'Functor' instance, making this viable to be applied
+-- in a monadic/free context.
+data DBTransactionOp err (r :: DBOperation) next where
+  -- | A Postgres /read/, returning a list of results. The 'r' in the result is polymorphic
+  -- so that reads can be embedded in updates transactions.
+  PGQuery   :: (PG.ToRow q, PG.FromRow a) => PG.Query -> q -> ([a] -> next) -> DBTransactionOp err r next
+  -- | A Postgres /write/, returning the number of affected rows. It can be used only in
+  -- 'DBWrite' transactions.
+  PGUpdate  :: PG.ToRow a => PG.Query -> a -> (Int -> next) -> DBTransactionOp err DBWrite next
+  -- | Unlike a 'PGUpdate' that returns the list of affected rows, this can be used
+  -- to write updates that returns a value via the \"RETURNING\" directive. It's the programmer's
+  -- responsibility to ensure that the SQL fragment contains it.
+  PGUpdateReturning  :: (PG.ToRow q, PG.FromRow a) => PG.Query -> q -> (a -> next) -> DBTransactionOp err DBWrite next
+  -- | An Opaleye /read/, returning a list of results. The 'r' in the result is polymorphic
+  -- so that reads can be embedded in updates transactions.
+  OpaQuery  :: Default FromFields fields a => Select fields -> ([a] -> next) -> DBTransactionOp err r next
+  -- | An Opaleye /write/, returning a result depending on the input 'Insert'. It can be used only in
+  -- 'DBWrite' transactions.
+  OpaInsert :: Insert a -> (a -> next) -> DBTransactionOp err DBWrite next
+  -- | An Opaleye /write/, returning a result depending on the input 'Update'. It can be used only in
+  -- 'DBWrite' transactions.
+  OpaUpdate :: Update a -> (a -> next) -> DBTransactionOp err DBWrite next
+  -- | Monadic failure for DB transactions.
+  DBFail    :: err -> DBTransactionOp err r next
+
+newtype DBTx err r a = DBTx { _DBTx :: Free (DBTransactionOp err r) a }
+  deriving (Functor, Applicative, Monad)
+
+type DBQuery err r a    = DBTx err r a
+type DBUpdate err a     = DBTx err DBWrite a
+type DBReadOnly err r a = DBTx err DBRead a
+
+-- Strict constraints to perform transactional read and writes.
+-- Isomorphic to a DBCmd but it doesn't impose a 'HasConfig' constraint, as
+-- values can always be passed as parameters of a query or update.
+type DBTxCmd err a = forall m env. (IsCmd env err m, HasConnectionPool env) => m a
+
+instance Functor (DBTransactionOp err r) where
+  fmap f = \case
+    PGQuery   q params cont     -> PGQuery   q params (f . cont)
+    PGUpdate  q a cont          -> PGUpdate  q a      (f . cont)
+    PGUpdateReturning  q a cont -> PGUpdateReturning q a      (f . cont)
+    OpaQuery  sel cont          -> OpaQuery  sel      (f . cont)
+    OpaInsert ins cont          -> OpaInsert ins      (f . cont)
+    OpaUpdate upd cont          -> OpaUpdate upd      (f . cont)
+    DBFail    err               -> DBFail err
+
+-- | Generalised version of 'withResource' to work over any unlifted monad.
+-- For some reason 'resource-pool' removed this from version 0.3.0.0 onwards.
+withResourceM :: MonadBaseControl IO m => Pool a -> (a -> m b) -> m b
+withResourceM pool func = control $ \run -> withResource pool (run . func)
+
+withTransactionM
+  :: forall m a.
+     MonadBaseControl IO m
+  => PG.Connection
+  -> m a
+  -> m a
+withTransactionM conn action = control $ \runInIO -> PG.withTransaction conn $ runInIO action
+
+-- | Run a PostgreSQL "read-only" transaction, suitable for read-only queries.
+withReadOnlyTransactionM
+  :: forall m a.
+     MonadBaseControl IO m
+  => PG.Connection
+  -> m a
+  -> m a
+withReadOnlyTransactionM conn action =
+  control $ \runInIO ->
+    PG.withTransactionMode tmode conn (runInIO action)
+  where
+    tmode :: PG.TransactionMode
+    tmode = PG.TransactionMode PG.DefaultIsolationLevel PG.ReadOnly
+
+runDBTx :: DBUpdate err a -> DBTxCmd err a
+runDBTx (DBTx m) = do
+  pool <- view connPool
+  withResourceM pool $ \conn -> withTransactionM conn $ foldFree (evalOp conn) m
+
+-- | /NOTE/ the input type is 'DBReadOnly', i.e. a transaction where /all/
+-- the operations are 'DBRead'. This makes impossible to sneak in updates
+-- into otherwise read-only queries.
+runDBQuery :: DBReadOnly err r a -> DBTxCmd err a
+runDBQuery (DBTx m) = do
+  pool <- view connPool
+  withResourceM pool $ \conn -> withReadOnlyTransactionM conn $ foldFree (evalOp conn) m
+
+-- | The main evaluator, turns our pure operations into side-effects that run into the
+-- 'DBCmd'.
+evalOp :: PG.Connection -> DBTransactionOp err r a -> DBTxCmd err a
+evalOp conn = \case
+  PGQuery   qr q cc          -> cc <$> liftBase (PG.query conn qr q)
+  PGUpdate  qr a cc          -> cc <$> liftBase (fromIntegral <$> PG.execute conn qr a)
+  PGUpdateReturning  qr a cc -> cc <$> liftBase (queryOne conn qr a)
+  OpaQuery  sel  cc          -> cc <$> liftBase (runSelect conn sel)
+  OpaInsert ins  cc          -> cc <$> liftBase (runInsert conn ins)
+  OpaUpdate upd  cc          -> cc <$> liftBase (runUpdate conn upd)
+  DBFail    err              -> throwError err
+
+queryOne :: (PG.ToRow q, PG.FromRow r) => PG.Connection -> PG.Query -> q -> IO r
+queryOne conn q v = do
+  rs <- PG.query conn q v
+  case rs of
+    [x] -> pure x
+    [ ] -> Safe.throwIO $ userError "queryOne: no result returned. Check your SQL!"
+    _   -> Safe.throwIO $ userError "queryOne: more than one result returned. Have you used the 'RETURNING' directive?"
+
+--
+-- Smart constructors
+--
+
+dbFail :: err -> DBTx err r b
+dbFail = DBTx . liftF . DBFail
+
+mkPGQuery :: (PG.ToRow q, PG.FromRow a)
+           => PG.Query
+           -> q
+           -> DBQuery err r [a]
+mkPGQuery q a = DBTx $ liftF (PGQuery q a id)
+
+mkPGUpdate :: PG.ToRow a => PG.Query -> a -> DBUpdate err Int
+mkPGUpdate q a = DBTx $ liftF (PGUpdate q a id)
+
+mkPGUpdateReturning :: (PG.ToRow q, PG.FromRow a) => PG.Query -> q -> DBUpdate err a
+mkPGUpdateReturning q a = DBTx $ liftF (PGUpdateReturning q a id)
+
+mkOpaQuery :: Default FromFields fields a
+           => Select fields
+           -> DBQuery err x [a]
+mkOpaQuery s = DBTx $ liftF (OpaQuery s id)
+
+mkOpaUpdate :: Update a -> DBUpdate err a
+mkOpaUpdate a = DBTx $ liftF (OpaUpdate a id)
+
+mkOpaInsert :: Insert a -> DBUpdate err a
+mkOpaInsert a = DBTx $ liftF (OpaInsert a id)

src/Gargantext/Database/Transactional/Example.hs

+{-# LANGUAGE TypeApplications    #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+module Gargantext.Database.Transactional.Example where
+
+import Control.Lens
+import Data.Aeson qualified as JSON
+import Gargantext.Core (toDBid)
+import Gargantext.Core.Types
+import Gargantext.Core.Types.Individu
+import Gargantext.Database.Action.Share hiding (getFolderId)
+import Gargantext.Database.Admin.Config
+import Gargantext.Database.Admin.Types.Hyperdata (HyperdataUser, HyperdataFolder)
+import Gargantext.Database.Prelude
+import Gargantext.Database.Query.Table.Node (selectNode, selectNodesWith)
+import Gargantext.Database.Query.Table.Node.Error
+import Gargantext.Database.Query.Table.NodeNode (TargetId(..), SourceId (..), NodeNodePoly (..), nodeNodeTable)
+import Gargantext.Database.Query.Table.NodeNode qualified as GGTX hiding (insertNodeNode)
+import Gargantext.Database.Query.Tree.Root (selectRoot)
+import Gargantext.Database.Schema.Node (_node_id)
+import Gargantext.Database.Schema.Node (node_user_id)
+import Gargantext.Database.Transactional
+import Gargantext.Prelude (panicTrace, headMay)
+import Opaleye
+import Prelude
+
+-- | In this example we can compose two QUERY operations
+-- and the result it's still a pure query.
+pureQueryExample :: HasNodeError err
+                => NodeId
+                -> User
+                -> DBCmd err (Node JSON.Value, UserId)
+pureQueryExample n u = runDBQuery $ do
+  nodeToCheck <- getNode n
+  userIdCheck <- getUserId u
+  pure (nodeToCheck, userIdCheck)
+
+-- | In this example we can compose a QUERY operation
+-- with an UPDATE operation, and the overall \"flavor\"
+-- of the DbTx is an UPDATE, so we have to use 'runDBTx'.
+simpleTxExample :: HasNodeError err
+                => NodeId
+                -> DBCmd err Int
+simpleTxExample n = runDBTx $ do
+  nodeToCheck <- getNode n
+  shareNode (SourceId $ _node_id nodeToCheck) (TargetId $ _node_id nodeToCheck)
+
+shareNodeWithTx :: HasNodeError err
+                => ShareNodeWith
+                -> NodeId
+                -- ^ The target node we would like to share, it has
+                -- to be a 'NodeFolderShared'.
+                -> DBCmd err Int
+shareNodeWithTx (ShareNodeWith_User NodeFolderShared u) n = runDBTx $ do
+  nodeToCheck <- getNode   n
+  userIdCheck <- getUserId u
+  if Prelude.not (hasNodeType nodeToCheck NodeTeam)
+    then dbFailWith "[G.D.A.S.shareNodeWith] Can share node Team only"
+    else
+      if (view node_user_id nodeToCheck == userIdCheck)
+        then dbFailWith "[G.D.A.S.shareNodeWith] Can share to others only"
+        else do
+          folderSharedId  <- getFolderId u NodeFolderShared
+          shareNode (SourceId folderSharedId) (TargetId n)
+shareNodeWithTx _ _ = panicTrace "unimplemented (just testing)"
+
+shareNode :: SourceId -> TargetId -> DBUpdate err Int
+shareNode (SourceId sourceId) (TargetId targetId) =
+  insertNodeNode [ NodeNode sourceId targetId Nothing Nothing ]
+
+--
+-- Mock functions for testing
+--
+
+getNode :: forall err x. HasNodeError err => NodeId -> DBQuery err x (Node JSON.Value)
+getNode nId = do
+  xs <- mkOpaQuery (selectNode (pgNodeId nId))
+  case headMay xs of
+    Nothing -> dbFail $ _NodeError # (DoesNotExist nId)
+    Just  r -> pure r
+
+getUserId :: User -> DBQuery err x UserId
+getUserId = undefined
+
+getFolderId :: HasNodeError err => User -> NodeType -> DBQuery err x NodeId
+getFolderId u nt = do
+  rootId <- getRootId u
+  (s :: [ Node HyperdataFolder]) <- mkOpaQuery (selectNodesWith rootId (Just nt) Nothing Nothing)
+  case headMay s of
+    Nothing -> dbFailWith "[G.D.A.S.getFolderId] No folder shared found"
+    Just  f -> pure (_node_id f)
+
+getRootId :: HasNodeError err => User -> DBQuery err x NodeId
+getRootId u = do
+  (xs :: [ Node HyperdataUser]) <- mkOpaQuery (selectRoot u)
+  case headMay xs of
+    Nothing -> dbFailWith "[G.D.Q.T.R.getRootId] No root id"
+    Just  r -> pure (_node_id r)
+
+insertNodeNode :: [GGTX.NodeNode] -> DBUpdate err Int
+insertNodeNode ns = fromIntegral <$> mkOpaInsert (Insert nodeNodeTable ns' rCount (Just doNothing))
+  where
+    ns' :: [GGTX.NodeNodeWrite]
+    ns' = map (\(NodeNode n1 n2 x y)
+                -> NodeNode (pgNodeId n1)
+                            (pgNodeId n2)
+                            (sqlDouble <$> x)
+                            (sqlInt4 . toDBid <$> y)
+              ) ns

test/Test/Database/Transactions.hs

+{-# LANGUAGE QuasiQuotes          #-}
+{-# LANGUAGE UndecidableInstances #-}
+{-# LANGUAGE OverloadedStrings    #-}
+{-# LANGUAGE TupleSections        #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+
+{-| Tests for the transactional DB API -}
+
+module Test.Database.Transactions (
+     tests
+  ) where
+
+import System.Random.Stateful
+import Control.Exception.Safe
+import Control.Exception.Safe qualified as Safe
+import Control.Monad.Reader
+import Control.Monad.Trans.Control
+import Data.Pool
+import Data.String
+import Data.Text qualified as T
+import Data.Text.Encoding qualified as TE
+import Database.PostgreSQL.Simple qualified as PG
+import Database.PostgreSQL.Simple.FromField
+import Database.PostgreSQL.Simple.FromRow
+import Database.PostgreSQL.Simple.Options qualified as Client
+import Database.PostgreSQL.Simple.SqlQQ (sql)
+import Database.PostgreSQL.Simple.ToField
+import Database.Postgres.Temp qualified as Tmp
+import Gargantext.Database.Transactional
+import Gargantext.Prelude
+import Prelude qualified
+import Shelly as SH
+import Test.Database.Types hiding (Counter)
+import Test.Hspec
+import Test.Tasty.HUnit hiding (assert)
+import Text.RawString.QQ
+import Control.Concurrent.Async (forConcurrently)
+
+--
+-- For these tests we do not want to test the normal GGTX database queries, but rather
+-- the foundational approach for the DBTx monad. Therefore we don't use the usual
+-- 'withTestDB' code, but we rely on something very simple, a single table representing
+-- counters with IDs, like so:
+--
+-- | ID | Counter_value |
+-- | 1  | 0
+-- | 2  | ...
+--
+
+newtype TestDBTxMonad a = TestDBTxMonad { _TestDBTxMonad :: TestMonadM DBHandle a }
+  deriving ( Functor, Applicative, Monad
+           , MonadReader DBHandle, MonadError IOException
+           , MonadBase IO
+           , MonadBaseControl IO
+           , MonadFail
+           , MonadIO
+           , MonadMask
+           , MonadCatch
+           , MonadThrow
+           )
+
+setup :: IO DBHandle
+setup = do
+  res <- Tmp.startConfig tmpPgConfig
+  case res of
+    Left err -> Prelude.fail $ show err
+    Right db -> do
+      let idleTime = 60.0
+      let maxResources = 2
+      let poolConfig = defaultPoolConfig (PG.connectPostgreSQL (Tmp.toConnectionString db))
+                                         PG.close
+                                         idleTime
+                                         maxResources
+      pool <- newPool (setNumStripes (Just 2) poolConfig)
+      bootstrapCounterDB db pool
+      pure $ DBHandle pool db
+  where
+    tmpPgConfig :: Tmp.Config
+    tmpPgConfig = Tmp.defaultConfig <>
+      Tmp.optionsToDefaultConfig mempty
+        { Client.dbname   = pure dbName
+        , Client.user     = pure dbUser
+        , Client.password = pure dbPassword
+        }
+
+dbUser, dbPassword, dbName, dbTable :: String
+dbUser = "ggtx_test_counter_db_user"
+dbPassword = "ggtx_test_counter_db_pwd"
+dbName = "ggtx_test_counter_db"
+dbTable = "public.ggtx_test_counter_table"
+
+bootstrapCounterDB :: Tmp.DB -> Pool PG.Connection -> IO ()
+bootstrapCounterDB tmpDB pool = withResource pool $ \conn -> do
+  void $ PG.execute_ conn (fromString $ "ALTER USER \"" <> dbUser <> "\" with PASSWORD '" <> dbPassword <> "'")
+  let schemaContent = counterDBSchema
+  let connString = Tmp.toConnectionString tmpDB
+  (res,ec) <- shelly $ silently $ escaping False $ do
+    withTmpDir $ \tdir -> do
+      let schemaPath = tdir <> "/schema.sql"
+      writefile schemaPath (T.pack schemaContent)
+      result <- SH.run "psql" ["-d", "\"" <> TE.decodeUtf8 connString <> "\"", "<", fromString schemaPath]
+      (result,) <$> lastExitCode
+  unless (ec == 0) $ Safe.throwIO (Prelude.userError $ show ec <> ": " <> T.unpack res)
+
+counterDBSchema :: String
+counterDBSchema = [r|
+  CREATE TABLE |] <> dbTable <> [r| (
+      id SERIAL,
+      counter_value INT NOT NULL DEFAULT 0,
+      PRIMARY KEY (id)
+  );
+  ALTER TABLE public.ggtx_test_counter_table OWNER TO |] <> dbUser <> ";" <> [r|
+  INSERT INTO public.ggtx_test_counter_table(counter_value) VALUES(42);
+|]
+
+withTestCounterDB :: (DBHandle -> IO ()) -> IO ()
+withTestCounterDB = Safe.bracket setup teardown
+
+teardown :: DBHandle -> IO ()
+teardown test_db = do
+  destroyAllResources $ _DBHandle test_db
+  Tmp.stop $ _DBTmp test_db
+
+--
+-- Helpers and transactions to work with counters
+--
+
+newtype CounterId = CounterId { _CounterId :: Int }
+  deriving (Show, Eq, ToField, FromField)
+
+data Counter = Counter
+  { counterId    :: !CounterId
+  , counterValue :: Int
+  }
+  deriving (Show, Eq)
+
+instance PG.FromRow Counter where
+  fromRow = Counter <$> field <*> field
+
+getCounterById :: CounterId -> DBQuery IOException r Counter
+getCounterById (CounterId cid) = do
+  xs <- mkPGQuery [sql| SELECT * FROM public.ggtx_test_counter_table WHERE id = ?; |] (PG.Only cid)
+  case xs of
+    [c] -> pure c
+    rst -> dbFail $ Prelude.userError ("getCounterId returned more than one result: " <> show rst)
+
+insertCounter :: DBUpdate IOException Counter
+insertCounter = do
+  mkPGUpdateReturning [sql| INSERT INTO public.ggtx_test_counter_table(counter_value) VALUES(0) RETURNING id, counter_value|] ()
+
+updateCounter :: CounterId -> Int -> DBUpdate IOException Counter
+updateCounter cid x = do
+  mkPGUpdateReturning [sql| UPDATE public.ggtx_test_counter_table SET counter_value = ? WHERE id = ? RETURNING *|] (x, cid)
+
+-- | We deliberately write this as a composite operation.
+stepCounter :: CounterId -> DBUpdate IOException Counter
+stepCounter cid = do
+  Counter{..} <- getCounterById cid
+  mkPGUpdateReturning [sql| UPDATE public.ggtx_test_counter_table SET counter_value = ? WHERE id = ? RETURNING *|] (counterValue + 1, cid)
+
+--
+-- MAIN TESTS
+--
+
+tests :: Spec
+tests = parallel $ around withTestCounterDB $
+  describe "Database Transactions" $ do
+    describe "Pure PG Queries" $ do
+      it "Simple query works" simplePGQueryWorks
+    describe "Pure PG Inserts" $ do
+      it "Simple insert works" simplePGInsertWorks
+    describe "Pure PG Updates" $ do
+      it "Simple updates works" simplePGUpdateWorks
+    describe "PG Queries and Updates" $ do
+      it "Supports mixing queries and updates" mixQueriesAndUpdates
+    describe "Rollback support" $ do
+      it "can rollback in case of errors" testRollback
+    describe "Read/Write Consistency" $ do
+      it "should return a consistent state to different actors" testConsistency
+
+simplePGQueryWorks :: DBHandle -> Assertion
+simplePGQueryWorks env = flip runReaderT env $ runTestMonad $ do
+  x <- runDBQuery $ getCounterById (CounterId 1)
+  liftIO $ counterValue x `shouldBe` 42
+
+simplePGInsertWorks :: DBHandle -> Assertion
+simplePGInsertWorks env = flip runReaderT env $ runTestMonad $ do
+  x <- runDBTx $ insertCounter
+  liftIO $ x `shouldBe` (Counter (CounterId 2) 0)
+
+simplePGUpdateWorks :: DBHandle -> Assertion
+simplePGUpdateWorks env = flip runReaderT env $ runTestMonad $ do
+  x <- runDBTx $ updateCounter (CounterId 1) 99
+  liftIO $ x `shouldBe` (Counter (CounterId 1) 99)
+
+mixQueriesAndUpdates :: DBHandle -> Assertion
+mixQueriesAndUpdates env = flip runReaderT env $ runTestMonad $ do
+  (final_1, final_2) <- runDBTx $ do
+    c1 <- insertCounter
+    c2 <- insertCounter
+    c1' <- getCounterById (counterId c1)
+    c2' <- stepCounter (counterId c2)
+    pure (c1', c2')
+  liftIO $ do
+    final_1 `shouldBe` (Counter (CounterId 2) 0)
+    final_2 `shouldBe` (Counter (CounterId 3) 1)
+
+testRollback :: DBHandle -> Assertion
+testRollback env = flip runReaderT env $ runTestMonad $ do
+  initialCounter <- runDBTx $ insertCounter >>= stepCounter . counterId
+  liftIO $ counterValue initialCounter `shouldBe` 1
+  -- Let's do another transaction where at the very last instruction we
+  -- fail.
+  Safe.handle (\(_ :: SomeException) -> pure ()) $ runDBTx $ do
+    _x' <- stepCounter (counterId initialCounter)
+    dbFail (Prelude.userError "urgh")
+
+  -- Let's check that the second 'stepCounter' didn't actually modified the counter's value.
+  finalCounter <- runDBTx $ getCounterById (counterId initialCounter)
+  liftIO $ counterValue finalCounter `shouldBe` 1
+
+-- | In this test we create concurrent actors all writing to the /same/ counter.
+-- Each one should observe only the state it's updating.
+testConsistency :: DBHandle -> Assertion
+testConsistency env = do
+
+  let competing_actors = 10
+  initialCounter <- flip runReaderT env $ runTestMonad $ runDBTx insertCounter
+
+  results <- forConcurrently [ 1 .. competing_actors ] $ \x -> flip runReaderT env $ runTestMonad $ do
+    -- random delay
+    liftIO $ do
+      delay_us <- uniformRM (100, 2_000_000) globalStdGen
+      threadDelay delay_us
+    runDBTx $ do
+      _ <- updateCounter (counterId initialCounter) x
+      getCounterById (counterId initialCounter)
+
+  -- Each actor should observe a consistent state.
+  liftIO $ results `shouldBe` map (Counter (CounterId 2)) [ 1 .. competing_actors ]

test/Test/Database/Types.hs

@@ -62,9 +62,9 @@ data TestEnv = TestEnv {
   , test_worker_tid          :: !ThreadId
   }
 
-newtype TestMonad a = TestMonad { runTestMonad :: ReaderT TestEnv IO a }
+newtype TestMonadM e a = TestMonad { runTestMonad :: ReaderT e IO a }
   deriving ( Functor, Applicative, Monad
-           , MonadReader TestEnv, MonadError IOException
+           , MonadReader e, MonadError IOException
            , MonadBase IO
            , MonadBaseControl IO
            , MonadFail
@@ -74,6 +74,7 @@ newtype TestMonad a = TestMonad { runTestMonad :: ReaderT TestEnv IO a }
            , MonadThrow
            )
 
+type TestMonad = TestMonadM TestEnv
 data TestJobHandle = TestNoJobHandle
 
 instance MonadJobStatus TestMonad where
@@ -97,6 +98,9 @@ data DBHandle = DBHandle {
   , _DBTmp    :: Tmp.DB
   }
 
+instance HasConnectionPool DBHandle where
+  connPool = to _DBHandle
+
 instance HasConnectionPool TestEnv where
   connPool = to (_DBHandle . test_db)
 

test/drivers/hspec/Main.hs

@@ -12,6 +12,7 @@ import System.Posix.Process
 import System.Posix.Signals
 import Test.API qualified as API
 import Test.Database.Operations qualified as DB
+import Test.Database.Transactions qualified as DBT
 import Test.Hspec
 import Test.Server.ReverseProxy qualified as ReverseProxy
 
@@ -67,5 +68,6 @@ main = do
       API.tests
       ReverseProxy.tests
       DB.tests
+      DBT.tests
       DB.nodeStoryTests
       runIO $ putText "tests finished"