make a start to stetting up hasql with db pools

Author: Cmdv

cabal.project

@@ -10,8 +10,8 @@ repository cardano-haskell-packages
     d4a35cd3121aa00d18544bb0ac01c3e1691d618f462c46129271bccf39f7e8ee
 
 index-state:
-  , hackage.haskell.org 2024-10-10T00:52:24Z
-  , cardano-haskell-packages 2024-11-26T16:00:26Z
+  , hackage.haskell.org 2025-02-05T12:01:20Z
+  , cardano-haskell-packages 2025-02-04T11:56:25Z
 
 packages:
   cardano-db

cardano-db-sync/cardano-db-sync.cabal

@@ -184,6 +184,8 @@ library
                       , extra
                       , filepath
                       , groups
+                      , hasql
+                      , hasql-pool
                       , http-client
                       , http-client-tls
                       , http-types

cardano-db-sync/src/Cardano/DbSync.hs

@@ -164,19 +164,20 @@ runSyncNode metricsSetters trce iomgr dbConnString runMigrationFnc syncNodeConfi
   logInfo trce $ "Using alonzo genesis file from: " <> (show . unGenesisFile $ dncAlonzoGenesisFile syncNodeConfigFromFile)
 
   let useLedger = shouldUseLedger (sioLedger $ dncInsertOptions syncNodeConfigFromFile)
-
-  Db.runIohkLogging trce $
-    withPostgresqlConn dbConnString $
-      \backend -> liftIO $ do
+  cPool <- createPool dbConnString
+  bracket
+    cPool
+    Pool.release
+    (\pool -> do
         runOrThrowIO $ runExceptT $ do
           genCfg <- readCardanoGenesisConfig syncNodeConfigFromFile
-          isJsonbInSchema <- queryIsJsonbInSchema backend
+          isJsonbInSchema <- queryIsJsonbInSchema pool
           logProtocolMagicId trce $ genesisProtocolMagicId genCfg
           syncEnv <-
             ExceptT $
               mkSyncEnvFromConfig
                 trce
-                backend
+                pool
                 dbConnString
                 syncOptions
                 genCfg
@@ -196,7 +197,7 @@ runSyncNode metricsSetters trce iomgr dbConnString runMigrationFnc syncNodeConfi
           liftIO $ runExtraMigrationsMaybe syncEnv
           unless useLedger $ liftIO $ do
             logInfo trce "Migrating to a no ledger schema"
-            Db.noLedgerMigrations backend trce
+            Db.noLedgerMigrations pool trce
           insertValidateGenesisDist syncEnv (dncNetworkName syncNodeConfigFromFile) genCfg (useShelleyInit syncNodeConfigFromFile)
 
           -- communication channel between datalayer thread and chainsync-client thread

cardano-db-sync/src/Cardano/DbSync/Api.hs

@@ -309,7 +309,7 @@ getCurrentTipBlockNo env = do
 
 mkSyncEnv ::
   Trace IO Text ->
-  SqlBackend ->
+  Pool ->
   ConnectionString ->
   SyncOptions ->
   ProtocolInfo CardanoBlock ->
@@ -367,7 +367,7 @@ mkSyncEnv trce backend connectionString syncOptions protoInfo nw nwMagic systemS
 
   pure $
     SyncEnv
-      { envBackend = backend
+      { envPool = dbPool
       , envBootstrap = bootstrapVar
       , envCache = cache
       , envConnectionString = connectionString
@@ -393,7 +393,7 @@ mkSyncEnv trce backend connectionString syncOptions protoInfo nw nwMagic systemS
 
 mkSyncEnvFromConfig ::
   Trace IO Text ->
-  SqlBackend ->
+  Pool ->
   ConnectionString ->
   SyncOptions ->
   GenesisConfig ->
@@ -402,7 +402,11 @@ mkSyncEnvFromConfig ::
   -- | run migration function
   RunMigration ->
   IO (Either SyncNodeError SyncEnv)
+<<<<<<< HEAD
 mkSyncEnvFromConfig trce backend connectionString syncOptions genCfg syncNodeConfigFromFile syncNodeParams runMigrationFnc =
+=======
+mkSyncEnvFromConfig trce dbPool connectionString syncOptions genCfg syncNodeConfigFromFile syncNodeParams ranMigration runMigrationFnc =
+>>>>>>> 77f1a2c6 (make a start to stetting up hasql with db pools)
   case genCfg of
     GenesisCardano _ bCfg sCfg _ _
       | unProtocolMagicId (Byron.configProtocolMagicId bCfg) /= Shelley.sgNetworkMagic (scConfig sCfg) ->
@@ -429,7 +433,7 @@ mkSyncEnvFromConfig trce backend connectionString syncOptions genCfg syncNodeCon
           Right
             <$> mkSyncEnv
               trce
-              backend
+              dbPool
               connectionString
               syncOptions
               (fst $ mkProtocolInfoCardano genCfg [])

cardano-db-sync/src/Cardano/DbSync/Api/Types.hs

@@ -32,12 +32,11 @@ import Control.Concurrent.Class.MonadSTM.Strict.TBQueue (StrictTBQueue)
 import qualified Data.Strict.Maybe as Strict
 import Data.Time.Clock (UTCTime)
 import Database.Persist.Postgresql (ConnectionString)
-import Database.Persist.Sql (SqlBackend)
 import Ouroboros.Consensus.BlockchainTime.WallClock.Types (SystemStart (..))
 import Ouroboros.Network.Magic (NetworkMagic (..))
 
 data SyncEnv = SyncEnv
-  { envBackend :: !SqlBackend
+  { envPool :: !!Pool.Pool
   , envCache :: !CacheStatus
   , envConnectionString :: !ConnectionString
   , envConsistentLevel :: !(StrictTVar IO ConsistentLevel)

cardano-db-sync/src/Cardano/DbSync/Util/Constraint.hs

@@ -41,9 +41,9 @@ dbConstraintNamesExists :: MonadIO m => SqlBackend -> m ManualDbConstraints
 dbConstraintNamesExists sqlBackend = do
   runReaderT queryRewardAndEpochStakeConstraints sqlBackend
 
-queryIsJsonbInSchema :: MonadIO m => SqlBackend -> m Bool
-queryIsJsonbInSchema sqlBackend = do
-  runReaderT DB.queryJsonbInSchemaExists sqlBackend
+queryIsJsonbInSchema :: MonadIO m => Pool -> m Bool
+queryIsJsonbInSchema pool = do
+  runReaderT DB.queryJsonbInSchemaExists pool
 
 queryRewardAndEpochStakeConstraints ::
   MonadIO m =>

cardano-db/cardano-db.cabal

@@ -86,6 +86,7 @@ library
                       , filepath
                       , file-embed
                       , hasql
+                      , hasql-pool
                       , hasql-transaction
                       , iohk-monitoring
                       , lifted-base

cardano-db/src/Cardano/Db/Error.hs

@@ -10,15 +10,15 @@ module Cardano.Db.Error (
 ) where
 
 import Cardano.BM.Trace (Trace, logError)
-import Cardano.Db.Schema.Core
+import Cardano.Db.Schema.Ids
 import Cardano.Prelude (throwIO)
 import Control.Exception (Exception)
-import qualified Data.ByteString.Base16 as Base16
 import Data.ByteString.Char8 (ByteString)
 import Data.Text (Text)
-import qualified Data.Text.Encoding as Text
 import Data.Word (Word16, Word64)
 import GHC.Generics (Generic)
+import qualified Data.ByteString.Base16 as Base16
+import qualified Data.Text.Encoding as Text
 
 data LookupFail
   = DbLookupBlockHash !ByteString

cardano-db/src/Cardano/Db/PGConfig.hs

@@ -21,48 +21,65 @@ import qualified Control.Exception as Exception
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as BS
 import qualified Data.Text as Text
-import Database.Persist.Postgresql (ConnectionString)
 import System.Environment (lookupEnv, setEnv)
 import System.Posix.User (getEffectiveUserName)
+import qualified Hasql.Connection.Setting.Connection as HCS
+import qualified Hasql.Connection.Setting.Connection.Param as HCSP
+import qualified Hasql.Connection.Setting as HC
+import Cardano.Prelude (decodeUtf8)
+import Data.Word (Word16)
+import qualified Data.Text.Read as Text (decimal)
+import Control.Monad.Extra (unless)
+
 
 data PGPassSource
   = PGPassDefaultEnv
   | PGPassEnv String
   | PGPassCached PGConfig
   deriving (Show)
 
--- | PGConfig as specified by https://www.postgresql.org/docs/11/libpq-pgpass.html
--- However, this module expects the config data to be on the first line.
+-- | Preconstructed connection string according to <https://www.postgresql.org/docs/17/libpq-connect.html#LIBPQ-CONNSTRING the PostgreSQL format>.
 data PGConfig = PGConfig
-  { pgcHost :: ByteString
-  , pgcPort :: ByteString
-  , pgcDbname :: ByteString
-  , pgcUser :: ByteString
-  , pgcPassword :: ByteString
+  { pgcHost :: Text.Text
+  , pgcPort :: Text.Text
+  , pgcDbname :: Text.Text
+  , pgcUser :: Text.Text
+  , pgcPassword :: Text.Text
   }
   deriving (Show)
 
 newtype PGPassFile
   = PGPassFile FilePath
 
-toConnectionString :: PGConfig -> ConnectionString
-toConnectionString pgc =
-  BS.concat
-    [ "host="
-    , pgcHost pgc
-    , " "
-    , "port="
-    , pgcPort pgc
-    , " "
-    , "user="
-    , pgcUser pgc
-    , " "
-    , "dbname="
-    , pgcDbname pgc
-    , " "
-    , "password="
-    , pgcPassword pgc
-    ]
+-- | Convert PGConfig to Hasql connection settings, or return an error message.
+toConnectionString :: PGConfig -> Either String HC.Setting
+toConnectionString pgc = do
+  -- Convert the port from Text to Word16
+  portWord16 <- textToWord16 (pgcPort pgc)
+  -- Build the connection settings
+  pure $ HC.connection (HCS.params [host, port portWord16 , user, dbname, password])
+  where
+    host = HCSP.host (pgcHost pgc)
+    port = HCSP.port
+    user = HCSP.user (pgcUser pgc)
+    dbname = HCSP.dbname (pgcDbname pgc)
+    password = HCSP.password (pgcPassword pgc)
+
+-- | Convert a Text port to Word16, or return an error message.
+textToWord16 :: Text.Text -> Either String Word16
+textToWord16 portText =
+  case Text.decimal portText of
+    Left err ->
+      Left $ "Invalid port: '" <> Text.unpack portText <> "'. " <> err
+    Right (portInt, remainder) -> do
+      -- Check for leftover characters (e.g., "123abc" is invalid)
+      unless (Text.null remainder) $
+        Left $ "Invalid port: '" <> Text.unpack portText <> "'. Contains non-numeric characters."
+      -- Check if the port is within the valid Word16 range (0-65535)
+      unless (portInt >= (0 :: Integer) && portInt <= 65535) $
+        Left $ "Invalid port: '" <> Text.unpack portText <> "'. Port must be between 0 and 65535."
+      -- Convert to Word16
+      Right (fromIntegral portInt)
 
 readPGPassDefault :: IO (Either PGPassError PGConfig)
 readPGPassDefault = readPGPass PGPassDefaultEnv
@@ -94,24 +111,31 @@ readPGPassFile (PGPassFile fpath) = do
     extract bs =
       case BS.lines bs of
         (b : _) -> parsePGConfig b
-        _ -> pure $ Left (FailedToParsePGPassConfig bs)
+        _otherwise -> pure $ Left (FailedToParsePGPassConfig bs)
 
 parsePGConfig :: ByteString -> IO (Either PGPassError PGConfig)
 parsePGConfig bs =
   case BS.split ':' bs of
-    [h, pt, d, u, pwd] -> replaceUser (PGConfig h pt d u pwd)
-    _ -> pure $ Left (FailedToParsePGPassConfig bs)
+    [h, pt, d, u, pwd] ->
+      replaceUser (PGConfig
+                    (decodeUtf8 h)
+                    (decodeUtf8 pt)
+                    (decodeUtf8 d)
+                    (decodeUtf8 u)
+                    (decodeUtf8 pwd)
+                  )
+    _otherwise -> pure $ Left (FailedToParsePGPassConfig bs)
   where
     replaceUser :: PGConfig -> IO (Either PGPassError PGConfig)
     replaceUser pgc
-      | pgcUser pgc /= "*" = pure $ Right pgc
+      | pgcUser pgc /= Text.pack "*" = pure $ Right pgc
       | otherwise = do
           euser <- Exception.try getEffectiveUserName
           case euser of
             Left (err :: IOException) ->
               pure $ Left (UserFailed err)
             Right user ->
-              pure $ Right (pgc {pgcUser = BS.pack user})
+              pure $ Right (pgc {pgcUser = Text.pack user})
 
 -- | Read 'PGPassFile' into 'PGConfig'.
 -- If it fails it will raise an error.

cardano-db/src/Cardano/Db/Run.hs

@@ -4,6 +4,7 @@
 {-# LANGUAGE OverloadedStrings #-}
 
 module Cardano.Db.Run (
+  createPool,
   getBackendGhci,
   ghciDebugQuery,
   runDbHandleLogger,
@@ -76,6 +77,22 @@ import Database.PostgreSQL.Simple (connectPostgreSQL)
 import Language.Haskell.TH.Syntax (Loc)
 import System.IO (Handle, stdout)
 import System.Log.FastLogger (LogStr, fromLogStr)
+import qualified Hasql.Pool as HP
+import qualified Hasql.Pool.Config as HPC
+
+-- | Create a connection pool.
+createPool :: PGConfig -> IO HP.Pool
+createPool pgc =
+  case toConnectionString pgc of
+    Left err -> error $ "createPool: " ++ err
+    Right connStr ->
+      HP.acquire $ HPC.settings
+            [ HPC.size 10                        -- number of connections
+            , HPC.acquisitionTimeout 10          -- seconds
+            , HPC.agingTimeout 1800              -- 30 minutes
+            , HPC.idlenessTimeout 1800           -- 30 minutes
+            , HPC.staticConnectionSettings [connStr]
+            ]
 
 -- | Run a DB action logging via the provided Handle.
 runDbHandleLogger :: Handle -> PGPassSource -> ReaderT SqlBackend (LoggingT IO) a -> IO a