feat: transaction, session monads to handle resource acq + rel

spago.lock

@@ -13,6 +13,7 @@ workspace:
         - exceptions
         - foldable-traversable
         - foreign
+        - fork
         - integers
         - js-bigints
         - lists

spago.yaml

@@ -16,6 +16,7 @@ package:
     - exceptions
     - foldable-traversable
     - foreign
+    - fork
     - integers
     - js-bigints
     - lists

src/Control.Monad.Postgres.purs

@@ -1 +1,64 @@
 module Control.Monad.Postgres where
+
+import Prelude
+
+import Control.Monad.Error.Class (catchError, throwError)
+import Control.Monad.Fork.Class (class MonadBracket, bracket)
+import Control.Monad.Reader (class MonadReader, ReaderT, ask, runReaderT)
+import Control.Monad.Trans.Class (lift)
+import Data.Postgres.Query (class AsQuery)
+import Data.Postgres.Result (class FromRows)
+import Effect.Aff (Fiber)
+import Effect.Aff.Class (class MonadAff, liftAff)
+import Effect.Aff.Postgres.Client (Client)
+import Effect.Aff.Postgres.Client as Client
+import Effect.Aff.Postgres.Pool (Pool)
+import Effect.Aff.Postgres.Pool as Pool
+import Effect.Class (liftEffect)
+import Effect.Exception (Error)
+import Prim.Row (class Union)
+
+type PoolT :: forall k. (k -> Type) -> k -> Type
+type PoolT = ReaderT Pool
+
+type SessionT :: forall k. (k -> Type) -> k -> Type
+type SessionT = ReaderT Client
+
+class (MonadBracket Error Fiber m, MonadAff m, MonadReader Client m) <= MonadSession m where
+  query :: forall q r. AsQuery q => FromRows r => q -> m r
+  exec :: forall q. AsQuery q => q -> m Int
+  exec_ :: forall q. AsQuery q => q -> m Unit
+
+instance (MonadBracket Error Fiber m, MonadAff m, MonadReader Client m) => MonadSession m where
+  query q = do
+    client <- ask
+    liftAff $ Client.query q client
+  exec q = do
+    client <- ask
+    liftAff $ Client.exec q client
+  exec_ = void <<< exec
+
+session :: forall m a. MonadBracket Error Fiber m => MonadAff m => MonadSession (SessionT m) => SessionT m a -> PoolT m a
+session m = do
+  pool <- ask
+  let
+    acq = liftAff $ Pool.connect pool
+    rel _ c = liftEffect $ Pool.release pool c
+  lift $ bracket acq rel (runReaderT m)
+
+transaction :: forall m a. MonadBracket Error Fiber m => MonadAff m => MonadSession (SessionT m) => SessionT m a -> PoolT m a
+transaction m =
+  let
+    begin = void $ exec "begin;"
+    commit = m <* exec "commit;"
+    rollback e = exec "rollback;" *> throwError e
+  in
+    session $ begin *> catchError commit rollback
+
+runPool :: forall m a missing trash r e f. MonadBracket e f m => MonadAff m => Union r missing (Pool.Config trash) => Record r -> PoolT m a -> m a
+runPool cfg m =
+  let
+    acq = liftEffect $ Pool.make @r @missing @trash cfg
+    rel _ p = liftAff $ Pool.end p
+  in
+    bracket acq rel (runReaderT m)

src/Data.Postgres.Result.purs

@@ -10,6 +10,7 @@ import Data.Nullable (Nullable)
 import Data.Nullable as Nullable
 import Data.Postgres (class Rep, RepT, deserialize)
 import Data.Postgres.Raw (Raw)
+import Data.Traversable (traverse)
 import Data.Tuple (Tuple)
 import Data.Tuple.Nested (type (/\), (/\))
 import Foreign (ForeignError(..))
@@ -26,6 +27,18 @@ foreign import data Result :: Type
 rowsAffected :: Result -> Maybe Int
 rowsAffected = Int.fromNumber <=< Nullable.toMaybe <<< __rowsAffected
 
+class FromRows a where
+  fromRows :: Array (Array Raw) -> RepT a
+
+instance (FromRow a) => FromRows (Array a) where
+  fromRows = traverse fromRow
+else instance (FromRow a) => FromRows a where
+  fromRows =
+    let
+      e = pure $ ForeignError $ "Expected at least 1 row"
+    in
+      liftMaybe e <=< map Array.head <<< traverse fromRow
+
 -- | Can be unmarshalled from a queried row
 -- |
 -- | Implementations are provided for:

src/Effect.Aff.Postgres.Client.purs

@@ -8,13 +8,12 @@ import Data.Functor (voidRight)
 import Data.Maybe (fromMaybe)
 import Data.Postgres (smash)
 import Data.Postgres.Query (class AsQuery, QueryRaw, asQuery, __queryToRaw)
-import Data.Postgres.Result (class FromRow, Result, fromRow, rows, rowsAffected)
+import Data.Postgres.Result (class FromRows, Result, fromRows, rows, rowsAffected)
-import Data.Traversable (traverse)
 import Effect (Effect)
 import Effect.Aff (Aff)
 import Effect.Class (liftEffect)
 import Effect.Postgres.Client (Client, Config, make)
-import Effect.Postgres.Client as X
+import Effect.Postgres.Client (Client, ClientConfigRaw, Config, Notification, NotificationRaw, __make, __uncfg, endE, errorE, make, noticeE, notificationE) as X
 import Prim.Row (class Union)
 
 -- | Create a client and immediately connect it to the database
@@ -56,8 +55,8 @@ exec q = map (fromMaybe 0 <<< rowsAffected) <<< queryRaw q
 -- | returning them unmarshalled into destination type `r`.
 -- |
 -- | <https://node-postgres.com/apis/client#clientquery>
-query :: forall q r. AsQuery q => FromRow r => q -> Client -> Aff (Array r)
+query :: forall q r. AsQuery q => FromRows r => q -> Client -> Aff r
-query q = traverse (liftEffect <<< smash <<< fromRow) <=< map rows <<< queryRaw q
+query q = (liftEffect <<< smash <<< fromRows) <=< map rows <<< queryRaw q
 
 -- | FFI binding to `Client#connect`
 foreign import __connect :: Client -> Effect (Promise Unit)

src/Effect.Postgres.Pool.purs

@@ -39,7 +39,7 @@ foreign import clientWaitingCount :: Pool -> Int
 -- | The config parameter `r` is `Config` with all keys optional.
 -- |
 -- | <https://node-postgres.com/apis/pool#new-pool>
-make :: forall r missing trash. Union r missing (Config trash) => Record r -> Effect Pool
+make :: forall @r @missing @trash. Union r missing (Config trash) => Record r -> Effect Pool
 make r = do
   modifyPgTypes
   let asClientConfig = Client.__uncfg { unwrapMillis: unwrap } $ unsafeToForeign r

test/Test.Common.purs

@@ -2,6 +2,10 @@ module Test.Common where
 
 import Prelude
 
+import Data.Either (Either(..))
+import Data.String.Regex (Regex)
+import Data.String.Regex as Regex
+import Data.String.Regex.Flags (RegexFlags(..))
 import Effect (Effect)
 import Effect.Aff (Aff, bracket)
 import Effect.Aff.Postgres.Client (Client)
@@ -12,22 +16,27 @@ import Effect.Class (liftEffect)
 import Effect.Unsafe (unsafePerformEffect)
 import Node.Path as Path
 import Node.Process (cwd)
+import Partial.Unsafe (unsafePartial)
 import Record (insert)
 import Type.Prelude (Proxy(..))
 
-config
+type Config =
-  :: Effect
+  { database :: String
-       { database :: String
+  , host :: String
-       , host :: String
+  , password :: String
-       , password :: String
+  , user :: String
-       , user :: String
+  , max :: Int
-       , max :: Int
+  }
-       }
+
+config :: Effect Config
 config = do
   cwd' <- liftEffect cwd
   host <- liftEffect $ Path.resolve [ cwd' ] "./pg"
   pure { host, user: "postgres", password: "password", database: "postgres", max: 3 }
 
+withConfig :: (Config -> Aff Unit) -> Aff Unit
+withConfig f = f =<< liftEffect config
+
 withClient :: (Client -> Aff Unit) -> Aff Unit
 withClient = bracket (Client.connected =<< liftEffect config) Client.end
 
@@ -39,3 +48,9 @@ withPool = bracket (liftEffect $ Pool.make =<< config) Pool.end
 
 withPoolClient :: (Client -> Aff Unit) -> Aff Unit
 withPoolClient = bracket (Pool.connect pool) (liftEffect <<< Pool.release pool)
+
+unsafeFromRight :: forall a b. Either a b -> b
+unsafeFromRight e = unsafePartial $ case e of Right b -> b
+
+re :: String -> RegexFlags -> Regex
+re s f = unsafeFromRight $ Regex.regex s f

test/Test.Control.Monad.Postgres.purs

@@ -0,0 +1,51 @@
+module Test.Control.Monad.Postgres where
+
+import Prelude
+
+import Control.Monad.Error.Class (throwError)
+import Control.Monad.Fork.Class (class MonadBracket, bracket)
+import Control.Monad.Postgres (PoolT, exec_, query, runPool, session, transaction)
+import Data.Array as Array
+import Data.Array.NonEmpty as Array.NonEmpty
+import Data.Maybe (fromJust, maybe)
+import Data.String.Regex as Regex
+import Data.String.Regex.Flags as Regex.Flag
+import Effect.Aff (Fiber)
+import Effect.Aff.Class (class MonadAff)
+import Effect.Exception (Error, error)
+import Partial.Unsafe (unsafePartial)
+import Test.Common (re, withConfig)
+import Test.Spec (Spec, around, describe, it)
+import Test.Spec.Assertions (expectError, shouldEqual)
+
+withTable :: forall a m. MonadBracket Error Fiber m => MonadAff m => String -> PoolT m a -> PoolT m a
+withTable s m =
+  let
+    tabname = unsafePartial fromJust $ join $ Array.index (maybe [] Array.NonEmpty.toArray $ Regex.match (re "create table (\\w+)" Regex.Flag.ignoreCase) s) 1
+  in
+    bracket (session $ exec_ s) (\_ _ -> session $ exec_ $ "drop table " <> tabname <> ";") (const m)
+
+spec :: Spec Unit
+spec =
+  around withConfig $ describe "Control.Monad.Postgres" do
+    it "empty works" \cfg -> runPool cfg $ pure unit
+    it "connects" \cfg -> runPool cfg do
+      act <- session $ query "select 1"
+      act `shouldEqual` 1
+    it "connects multiple" \cfg -> runPool cfg do
+      a <- session $ query "select 1"
+      b <- session $ query "select 2"
+      a `shouldEqual` 1
+      b `shouldEqual` 2
+    it "transaction commits" \cfg -> runPool cfg do
+      withTable "create table test_txn_commits (id int);" do
+        transaction $ exec_ "insert into test_txn_commits values (1);"
+        act <- session $ query "select * from test_txn_commits"
+        act `shouldEqual` [ 1 ]
+    it "transaction rolls back" \cfg -> runPool cfg do
+      withTable "create table test_txn_rolls_back (id int);" do
+        expectError $ transaction do
+          exec_ "insert into test_txn_rolls_back values (1);"
+          throwError $ error "foo"
+        act :: Array Int <- session $ query "select * from test_txn_rolls_back"
+        act `shouldEqual` []

test/Test.Effect.Postgres.Client.purs

@@ -3,18 +3,15 @@ module Test.Effect.Postgres.Client where
 import Prelude
 
 import Control.Monad.Error.Class (try)
-import Data.Either (Either(..), isLeft)
+import Data.Either (Either, isLeft)
 import Data.Maybe (Maybe(..))
 import Data.Newtype (wrap)
 import Data.Postgres (JSON(..))
 import Data.PreciseDateTime (fromRFC3339String, toDateTimeLossy)
-import Effect (Effect)
+import Effect.Aff (forkAff, joinFiber)
-import Effect.Aff (Aff, forkAff, joinFiber, makeAff)
 import Effect.Aff.Postgres.Client (query)
 import Effect.Aff.Postgres.Client as Client
 import Effect.Exception as Error
-import Effect.Uncurried (EffectFn1)
-import Node.EventEmitter (EventHandle, once)
 import Test.Common (withClient)
 import Test.Event (onceAff)
 import Test.Spec (Spec, around, describe, it)
@@ -57,10 +54,10 @@ spec =
         describe "timestamp" do
           it "unmarshals" \c -> do
             let exp = toDateTimeLossy <$> fromRFC3339String (wrap "2020-01-01T00:00:00Z")
-            shouldEqual [ exp ] =<< query "select '2020-01-01T00:00:00Z' :: timestamptz" c
+            shouldEqual exp =<< query "select '2020-01-01T00:00:00Z' :: timestamptz" c
-          it "is string" \c -> shouldEqual [ "2020-01-01 00:00:00+00" ] =<< query "select '2020-01-01T00:00:00Z' :: timestamptz" c
+          it "is string" \c -> shouldEqual "2020-01-01 00:00:00+00" =<< query "select '2020-01-01T00:00:00Z' :: timestamptz" c
           it "array is string" \c -> shouldEqual [ [ "2020-01-01 00:00:00+00" ] ] =<< query "select array['2020-01-01T00:00:00Z' :: timestamptz]" c
         describe "json" do
-          it "unmarshals" \c -> shouldEqual [ JSON { foo: "bar" } ] =<< query "select '{\"foo\": \"bar\"}' :: json" c
+          it "unmarshals" \c -> shouldEqual (JSON { foo: "bar" }) =<< query "select '{\"foo\": \"bar\"}' :: json" c
-          it "is string" \c -> shouldEqual [ "{\"foo\": \"bar\"}" ] =<< query "select '{\"foo\": \"bar\"}' :: json" c
+          it "is string" \c -> shouldEqual "{\"foo\": \"bar\"}" =<< query "select '{\"foo\": \"bar\"}' :: json" c
           it "array is string" \c -> shouldEqual [ [ "{\"foo\": \"bar\"}" ] ] =<< query "select array['{\"foo\": \"bar\"}' :: json]" c

test/Test.Effect.Postgres.Pool.purs

@@ -73,7 +73,7 @@ spec = describe "Pool" do
     it "connect" \p -> do
       c <- Pool.connect p
       let rel = liftEffect $ Pool.release p c
-      finally rel $ shouldEqual [ 1 ] =<< Client.query "select 1" c
+      finally rel $ shouldEqual 1 =<< Client.query "select 1" c
     describe "destroy" do
       it "throws on query after destroy" \p -> do
         c <- Pool.connect p

test/Test.Main.purs

@@ -20,6 +20,7 @@ import Node.ChildProcess.Aff as ChildProcess.Aff
 import Node.ChildProcess.Types (Exit(..), stringSignal)
 import Node.Encoding (Encoding(..))
 import Node.EventEmitter as Event
+import Test.Control.Monad.Postgres as Test.Control.Monad.Postgres
 import Test.Data.Postgres as Test.Data.Postgres
 import Test.Effect.Postgres.Client as Test.Effect.Postgres.Client
 import Test.Effect.Postgres.Pool as Test.Effect.Postgres.Pool
@@ -64,3 +65,4 @@ main = launchAff_ do
         Test.Data.Postgres.spec
         Test.Effect.Postgres.Client.spec
         Test.Effect.Postgres.Pool.spec
+        Test.Control.Monad.Postgres.spec