{-# LANGUAGE DeriveAnyClass #-}
{-# LANGUAGE TypeFamilies #-}

-- | The hasql database monad with a built-in 'MonadArbiter' instance:
--
-- @
-- import Arbiter.Core
-- import Arbiter.Hasql
--
-- myFunction :: HasqlDb MyRegistry IO ()
-- myFunction = insertJob (defaultJob myPayload)
-- @
module Arbiter.Hasql.HasqlDb
  ( -- * Database Monad
    HasqlDb (..)
  , HasqlEnv (..)
  , runHasqlDb
  , inTransaction

    -- * Environment Creation
  , createHasqlEnv
  , createHasqlEnvWithConfig
  , createHasqlEnvWithPool
  , destroyHasqlEnv
  , disableListener
  , useDedicatedListener
  , setPreparedStatements

    -- * Hasql Settings
  , hasqlSettings

    -- * Exceptions
  , HasqlConnectionError (..)
  ) where

import Arbiter.Core.Job.Schema (SchemaName)
import Arbiter.Core.Listen (Listener, dedicatedListener, newDedicatedListen, newPoolListener)
import Arbiter.Core.MonadArbiter (MonadArbiter (..))
import Arbiter.Core.PoolConfig (PoolConfig (..))
import Arbiter.Core.PoolConfig qualified as PC
import Arbiter.Core.QueueRegistry (JobPayloadRegistry)
import Control.Exception (Exception, throwIO)
import Control.Monad.Catch (MonadCatch, MonadMask, MonadThrow)
import Control.Monad.IO.Class (MonadIO, liftIO)
import Control.Monad.Reader (MonadReader, asks, local)
import Control.Monad.Trans.Reader (ReaderT (..), runReaderT)
import Data.ByteString (ByteString)
import Data.Foldable (traverse_)
import Data.Pool (Pool, defaultPoolConfig, destroyAllResources, newPool, setNumStripes, withResource)
import Data.Proxy (Proxy (..))
import Hasql.Connection qualified as Hasql
import UnliftIO (MonadUnliftIO)

import Arbiter.Hasql.Compat qualified as Compat
import Arbiter.Hasql.MonadArbiter
  ( HasHasqlPool (..)
  , HasqlConnectionPool (..)
  , hasqlExecuteQuery
  , hasqlExecuteQueryPrepared
  , hasqlExecuteStatement
  , hasqlRunHandlerWithConnection
  , hasqlWithDbTransaction
  )

-- | Thrown when a hasql connection cannot be acquired from the pool.
newtype HasqlConnectionError = HasqlConnectionError String
  deriving stock (Int -> HasqlConnectionError -> ShowS
[HasqlConnectionError] -> ShowS
HasqlConnectionError -> String
(Int -> HasqlConnectionError -> ShowS)
-> (HasqlConnectionError -> String)
-> ([HasqlConnectionError] -> ShowS)
-> Show HasqlConnectionError
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> HasqlConnectionError -> ShowS
showsPrec :: Int -> HasqlConnectionError -> ShowS
$cshow :: HasqlConnectionError -> String
show :: HasqlConnectionError -> String
$cshowList :: [HasqlConnectionError] -> ShowS
showList :: [HasqlConnectionError] -> ShowS
Show)
  deriving anyclass (Show HasqlConnectionError
Typeable HasqlConnectionError
(Typeable HasqlConnectionError, Show HasqlConnectionError) =>
(HasqlConnectionError -> SomeException)
-> (SomeException -> Maybe HasqlConnectionError)
-> (HasqlConnectionError -> String)
-> (HasqlConnectionError -> Bool)
-> Exception HasqlConnectionError
SomeException -> Maybe HasqlConnectionError
HasqlConnectionError -> Bool
HasqlConnectionError -> String
HasqlConnectionError -> SomeException
forall e.
(Typeable e, Show e) =>
(e -> SomeException)
-> (SomeException -> Maybe e)
-> (e -> String)
-> (e -> Bool)
-> Exception e
$ctoException :: HasqlConnectionError -> SomeException
toException :: HasqlConnectionError -> SomeException
$cfromException :: SomeException -> Maybe HasqlConnectionError
fromException :: SomeException -> Maybe HasqlConnectionError
$cdisplayException :: HasqlConnectionError -> String
displayException :: HasqlConnectionError -> String
$cbacktraceDesired :: HasqlConnectionError -> Bool
backtraceDesired :: HasqlConnectionError -> Bool
Exception)

-- | Schema name and connection pool for 'HasqlDb'.
data HasqlEnv (registry :: JobPayloadRegistry) = HasqlEnv
  { forall (registry :: JobPayloadRegistry).
HasqlEnv registry -> SchemaName
schema :: SchemaName
  -- ^ Schema name
  , forall (registry :: JobPayloadRegistry).
HasqlEnv registry -> HasqlConnectionPool
hasqlPool :: HasqlConnectionPool
  -- ^ The connection pool state
  , forall (registry :: JobPayloadRegistry).
HasqlEnv registry -> Maybe Listener
listener :: Maybe Listener
  -- ^ Resolved LISTEN source. 'Nothing' runs poll-only.
  }

-- | The hasql database monad.
newtype HasqlDb (registry :: JobPayloadRegistry) m a = HasqlDb {forall (registry :: JobPayloadRegistry) (m :: * -> *) a.
HasqlDb registry m a -> ReaderT (HasqlEnv registry) m a
unHasqlDb :: ReaderT (HasqlEnv registry) m a}
  deriving newtype
    ( Functor (HasqlDb registry m)
Functor (HasqlDb registry m) =>
(forall a. a -> HasqlDb registry m a)
-> (forall a b.
    HasqlDb registry m (a -> b)
    -> HasqlDb registry m a -> HasqlDb registry m b)
-> (forall a b c.
    (a -> b -> c)
    -> HasqlDb registry m a
    -> HasqlDb registry m b
    -> HasqlDb registry m c)
-> (forall a b.
    HasqlDb registry m a
    -> HasqlDb registry m b -> HasqlDb registry m b)
-> (forall a b.
    HasqlDb registry m a
    -> HasqlDb registry m b -> HasqlDb registry m a)
-> Applicative (HasqlDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *).
Applicative m =>
Functor (HasqlDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *) a.
Applicative m =>
a -> HasqlDb registry m a
forall (registry :: JobPayloadRegistry) (m :: * -> *) a b.
Applicative m =>
HasqlDb registry m a
-> HasqlDb registry m b -> HasqlDb registry m a
forall (registry :: JobPayloadRegistry) (m :: * -> *) a b.
Applicative m =>
HasqlDb registry m a
-> HasqlDb registry m b -> HasqlDb registry m b
forall (registry :: JobPayloadRegistry) (m :: * -> *) a b.
Applicative m =>
HasqlDb registry m (a -> b)
-> HasqlDb registry m a -> HasqlDb registry m b
forall (registry :: JobPayloadRegistry) (m :: * -> *) a b c.
Applicative m =>
(a -> b -> c)
-> HasqlDb registry m a
-> HasqlDb registry m b
-> HasqlDb registry m c
forall a. a -> HasqlDb registry m a
forall a b.
HasqlDb registry m a
-> HasqlDb registry m b -> HasqlDb registry m a
forall a b.
HasqlDb registry m a
-> HasqlDb registry m b -> HasqlDb registry m b
forall a b.
HasqlDb registry m (a -> b)
-> HasqlDb registry m a -> HasqlDb registry m b
forall a b c.
(a -> b -> c)
-> HasqlDb registry m a
-> HasqlDb registry m b
-> HasqlDb registry m c
forall (f :: * -> *).
Functor f =>
(forall a. a -> f a)
-> (forall a b. f (a -> b) -> f a -> f b)
-> (forall a b c. (a -> b -> c) -> f a -> f b -> f c)
-> (forall a b. f a -> f b -> f b)
-> (forall a b. f a -> f b -> f a)
-> Applicative f
$cpure :: forall (registry :: JobPayloadRegistry) (m :: * -> *) a.
Applicative m =>
a -> HasqlDb registry m a
pure :: forall a. a -> HasqlDb registry m a
$c<*> :: forall (registry :: JobPayloadRegistry) (m :: * -> *) a b.
Applicative m =>
HasqlDb registry m (a -> b)
-> HasqlDb registry m a -> HasqlDb registry m b
<*> :: forall a b.
HasqlDb registry m (a -> b)
-> HasqlDb registry m a -> HasqlDb registry m b
$cliftA2 :: forall (registry :: JobPayloadRegistry) (m :: * -> *) a b c.
Applicative m =>
(a -> b -> c)
-> HasqlDb registry m a
-> HasqlDb registry m b
-> HasqlDb registry m c
liftA2 :: forall a b c.
(a -> b -> c)
-> HasqlDb registry m a
-> HasqlDb registry m b
-> HasqlDb registry m c
$c*> :: forall (registry :: JobPayloadRegistry) (m :: * -> *) a b.
Applicative m =>
HasqlDb registry m a
-> HasqlDb registry m b -> HasqlDb registry m b
*> :: forall a b.
HasqlDb registry m a
-> HasqlDb registry m b -> HasqlDb registry m b
$c<* :: forall (registry :: JobPayloadRegistry) (m :: * -> *) a b.
Applicative m =>
HasqlDb registry m a
-> HasqlDb registry m b -> HasqlDb registry m a
<* :: forall a b.
HasqlDb registry m a
-> HasqlDb registry m b -> HasqlDb registry m a
Applicative
    , (forall a b.
 (a -> b) -> HasqlDb registry m a -> HasqlDb registry m b)
-> (forall a b. a -> HasqlDb registry m b -> HasqlDb registry m a)
-> Functor (HasqlDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *) a b.
Functor m =>
a -> HasqlDb registry m b -> HasqlDb registry m a
forall (registry :: JobPayloadRegistry) (m :: * -> *) a b.
Functor m =>
(a -> b) -> HasqlDb registry m a -> HasqlDb registry m b
forall a b. a -> HasqlDb registry m b -> HasqlDb registry m a
forall a b.
(a -> b) -> HasqlDb registry m a -> HasqlDb registry m b
forall (f :: * -> *).
(forall a b. (a -> b) -> f a -> f b)
-> (forall a b. a -> f b -> f a) -> Functor f
$cfmap :: forall (registry :: JobPayloadRegistry) (m :: * -> *) a b.
Functor m =>
(a -> b) -> HasqlDb registry m a -> HasqlDb registry m b
fmap :: forall a b.
(a -> b) -> HasqlDb registry m a -> HasqlDb registry m b
$c<$ :: forall (registry :: JobPayloadRegistry) (m :: * -> *) a b.
Functor m =>
a -> HasqlDb registry m b -> HasqlDb registry m a
<$ :: forall a b. a -> HasqlDb registry m b -> HasqlDb registry m a
Functor
    , Applicative (HasqlDb registry m)
Applicative (HasqlDb registry m) =>
(forall a b.
 HasqlDb registry m a
 -> (a -> HasqlDb registry m b) -> HasqlDb registry m b)
-> (forall a b.
    HasqlDb registry m a
    -> HasqlDb registry m b -> HasqlDb registry m b)
-> (forall a. a -> HasqlDb registry m a)
-> Monad (HasqlDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *).
Monad m =>
Applicative (HasqlDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *) a.
Monad m =>
a -> HasqlDb registry m a
forall (registry :: JobPayloadRegistry) (m :: * -> *) a b.
Monad m =>
HasqlDb registry m a
-> HasqlDb registry m b -> HasqlDb registry m b
forall (registry :: JobPayloadRegistry) (m :: * -> *) a b.
Monad m =>
HasqlDb registry m a
-> (a -> HasqlDb registry m b) -> HasqlDb registry m b
forall a. a -> HasqlDb registry m a
forall a b.
HasqlDb registry m a
-> HasqlDb registry m b -> HasqlDb registry m b
forall a b.
HasqlDb registry m a
-> (a -> HasqlDb registry m b) -> HasqlDb registry m b
forall (m :: * -> *).
Applicative m =>
(forall a b. m a -> (a -> m b) -> m b)
-> (forall a b. m a -> m b -> m b)
-> (forall a. a -> m a)
-> Monad m
$c>>= :: forall (registry :: JobPayloadRegistry) (m :: * -> *) a b.
Monad m =>
HasqlDb registry m a
-> (a -> HasqlDb registry m b) -> HasqlDb registry m b
>>= :: forall a b.
HasqlDb registry m a
-> (a -> HasqlDb registry m b) -> HasqlDb registry m b
$c>> :: forall (registry :: JobPayloadRegistry) (m :: * -> *) a b.
Monad m =>
HasqlDb registry m a
-> HasqlDb registry m b -> HasqlDb registry m b
>> :: forall a b.
HasqlDb registry m a
-> HasqlDb registry m b -> HasqlDb registry m b
$creturn :: forall (registry :: JobPayloadRegistry) (m :: * -> *) a.
Monad m =>
a -> HasqlDb registry m a
return :: forall a. a -> HasqlDb registry m a
Monad
    , MonadThrow (HasqlDb registry m)
MonadThrow (HasqlDb registry m) =>
(forall e a.
 (HasCallStack, Exception e) =>
 HasqlDb registry m a
 -> (e -> HasqlDb registry m a) -> HasqlDb registry m a)
-> (forall e a.
    Exception e =>
    HasqlDb registry m a
    -> (ExceptionWithContext e -> HasqlDb registry m a)
    -> HasqlDb registry m a)
-> MonadCatch (HasqlDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *).
MonadCatch m =>
MonadThrow (HasqlDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *) e a.
(MonadCatch m, HasCallStack, Exception e) =>
HasqlDb registry m a
-> (e -> HasqlDb registry m a) -> HasqlDb registry m a
forall (registry :: JobPayloadRegistry) (m :: * -> *) e a.
(MonadCatch m, Exception e) =>
HasqlDb registry m a
-> (ExceptionWithContext e -> HasqlDb registry m a)
-> HasqlDb registry m a
forall e a.
(HasCallStack, Exception e) =>
HasqlDb registry m a
-> (e -> HasqlDb registry m a) -> HasqlDb registry m a
forall e a.
Exception e =>
HasqlDb registry m a
-> (ExceptionWithContext e -> HasqlDb registry m a)
-> HasqlDb registry m a
forall (m :: * -> *).
MonadThrow m =>
(forall e a.
 (HasCallStack, Exception e) =>
 m a -> (e -> m a) -> m a)
-> (forall e a.
    Exception e =>
    m a -> (ExceptionWithContext e -> m a) -> m a)
-> MonadCatch m
$ccatch :: forall (registry :: JobPayloadRegistry) (m :: * -> *) e a.
(MonadCatch m, HasCallStack, Exception e) =>
HasqlDb registry m a
-> (e -> HasqlDb registry m a) -> HasqlDb registry m a
catch :: forall e a.
(HasCallStack, Exception e) =>
HasqlDb registry m a
-> (e -> HasqlDb registry m a) -> HasqlDb registry m a
$ccatchNoPropagate :: forall (registry :: JobPayloadRegistry) (m :: * -> *) e a.
(MonadCatch m, Exception e) =>
HasqlDb registry m a
-> (ExceptionWithContext e -> HasqlDb registry m a)
-> HasqlDb registry m a
catchNoPropagate :: forall e a.
Exception e =>
HasqlDb registry m a
-> (ExceptionWithContext e -> HasqlDb registry m a)
-> HasqlDb registry m a
MonadCatch
    , Monad (HasqlDb registry m)
Monad (HasqlDb registry m) =>
(forall a. HasCallStack => String -> HasqlDb registry m a)
-> MonadFail (HasqlDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *).
MonadFail m =>
Monad (HasqlDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *) a.
(MonadFail m, HasCallStack) =>
String -> HasqlDb registry m a
forall a. HasCallStack => String -> HasqlDb registry m a
forall (m :: * -> *).
Monad m =>
(forall a. HasCallStack => String -> m a) -> MonadFail m
$cfail :: forall (registry :: JobPayloadRegistry) (m :: * -> *) a.
(MonadFail m, HasCallStack) =>
String -> HasqlDb registry m a
fail :: forall a. HasCallStack => String -> HasqlDb registry m a
MonadFail
    , Monad (HasqlDb registry m)
Monad (HasqlDb registry m) =>
(forall a. IO a -> HasqlDb registry m a)
-> MonadIO (HasqlDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *).
MonadIO m =>
Monad (HasqlDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *) a.
MonadIO m =>
IO a -> HasqlDb registry m a
forall a. IO a -> HasqlDb registry m a
forall (m :: * -> *).
Monad m =>
(forall a. IO a -> m a) -> MonadIO m
$cliftIO :: forall (registry :: JobPayloadRegistry) (m :: * -> *) a.
MonadIO m =>
IO a -> HasqlDb registry m a
liftIO :: forall a. IO a -> HasqlDb registry m a
MonadIO
    , MonadCatch (HasqlDb registry m)
MonadCatch (HasqlDb registry m) =>
(forall b.
 HasCallStack =>
 ((forall a. HasqlDb registry m a -> HasqlDb registry m a)
  -> HasqlDb registry m b)
 -> HasqlDb registry m b)
-> (forall b.
    HasCallStack =>
    ((forall a. HasqlDb registry m a -> HasqlDb registry m a)
     -> HasqlDb registry m b)
    -> HasqlDb registry m b)
-> (forall a b c.
    HasCallStack =>
    HasqlDb registry m a
    -> (a -> ExitCase b -> HasqlDb registry m c)
    -> (a -> HasqlDb registry m b)
    -> HasqlDb registry m (b, c))
-> MonadMask (HasqlDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *).
MonadMask m =>
MonadCatch (HasqlDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *) b.
(MonadMask m, HasCallStack) =>
((forall a. HasqlDb registry m a -> HasqlDb registry m a)
 -> HasqlDb registry m b)
-> HasqlDb registry m b
forall (registry :: JobPayloadRegistry) (m :: * -> *) a b c.
(MonadMask m, HasCallStack) =>
HasqlDb registry m a
-> (a -> ExitCase b -> HasqlDb registry m c)
-> (a -> HasqlDb registry m b)
-> HasqlDb registry m (b, c)
forall b.
HasCallStack =>
((forall a. HasqlDb registry m a -> HasqlDb registry m a)
 -> HasqlDb registry m b)
-> HasqlDb registry m b
forall a b c.
HasCallStack =>
HasqlDb registry m a
-> (a -> ExitCase b -> HasqlDb registry m c)
-> (a -> HasqlDb registry m b)
-> HasqlDb registry m (b, c)
forall (m :: * -> *).
MonadCatch m =>
(forall b. HasCallStack => ((forall a. m a -> m a) -> m b) -> m b)
-> (forall b.
    HasCallStack =>
    ((forall a. m a -> m a) -> m b) -> m b)
-> (forall a b c.
    HasCallStack =>
    m a -> (a -> ExitCase b -> m c) -> (a -> m b) -> m (b, c))
-> MonadMask m
$cmask :: forall (registry :: JobPayloadRegistry) (m :: * -> *) b.
(MonadMask m, HasCallStack) =>
((forall a. HasqlDb registry m a -> HasqlDb registry m a)
 -> HasqlDb registry m b)
-> HasqlDb registry m b
mask :: forall b.
HasCallStack =>
((forall a. HasqlDb registry m a -> HasqlDb registry m a)
 -> HasqlDb registry m b)
-> HasqlDb registry m b
$cuninterruptibleMask :: forall (registry :: JobPayloadRegistry) (m :: * -> *) b.
(MonadMask m, HasCallStack) =>
((forall a. HasqlDb registry m a -> HasqlDb registry m a)
 -> HasqlDb registry m b)
-> HasqlDb registry m b
uninterruptibleMask :: forall b.
HasCallStack =>
((forall a. HasqlDb registry m a -> HasqlDb registry m a)
 -> HasqlDb registry m b)
-> HasqlDb registry m b
$cgeneralBracket :: forall (registry :: JobPayloadRegistry) (m :: * -> *) a b c.
(MonadMask m, HasCallStack) =>
HasqlDb registry m a
-> (a -> ExitCase b -> HasqlDb registry m c)
-> (a -> HasqlDb registry m b)
-> HasqlDb registry m (b, c)
generalBracket :: forall a b c.
HasCallStack =>
HasqlDb registry m a
-> (a -> ExitCase b -> HasqlDb registry m c)
-> (a -> HasqlDb registry m b)
-> HasqlDb registry m (b, c)
MonadMask
    , MonadReader (HasqlEnv registry)
    , Monad (HasqlDb registry m)
Monad (HasqlDb registry m) =>
(forall e a.
 (HasCallStack, Exception e) =>
 e -> HasqlDb registry m a)
-> (forall e a.
    Exception e =>
    ExceptionWithContext e -> HasqlDb registry m a)
-> MonadThrow (HasqlDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *).
MonadThrow m =>
Monad (HasqlDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *) e a.
(MonadThrow m, HasCallStack, Exception e) =>
e -> HasqlDb registry m a
forall (registry :: JobPayloadRegistry) (m :: * -> *) e a.
(MonadThrow m, Exception e) =>
ExceptionWithContext e -> HasqlDb registry m a
forall e a.
(HasCallStack, Exception e) =>
e -> HasqlDb registry m a
forall e a.
Exception e =>
ExceptionWithContext e -> HasqlDb registry m a
forall (m :: * -> *).
Monad m =>
(forall e a. (HasCallStack, Exception e) => e -> m a)
-> (forall e a. Exception e => ExceptionWithContext e -> m a)
-> MonadThrow m
$cthrowM :: forall (registry :: JobPayloadRegistry) (m :: * -> *) e a.
(MonadThrow m, HasCallStack, Exception e) =>
e -> HasqlDb registry m a
throwM :: forall e a.
(HasCallStack, Exception e) =>
e -> HasqlDb registry m a
$crethrowM :: forall (registry :: JobPayloadRegistry) (m :: * -> *) e a.
(MonadThrow m, Exception e) =>
ExceptionWithContext e -> HasqlDb registry m a
rethrowM :: forall e a.
Exception e =>
ExceptionWithContext e -> HasqlDb registry m a
MonadThrow
    , MonadIO (HasqlDb registry m)
MonadIO (HasqlDb registry m) =>
(forall b.
 ((forall a. HasqlDb registry m a -> IO a) -> IO b)
 -> HasqlDb registry m b)
-> MonadUnliftIO (HasqlDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *).
MonadUnliftIO m =>
MonadIO (HasqlDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *) b.
MonadUnliftIO m =>
((forall a. HasqlDb registry m a -> IO a) -> IO b)
-> HasqlDb registry m b
forall b.
((forall a. HasqlDb registry m a -> IO a) -> IO b)
-> HasqlDb registry m b
forall (m :: * -> *).
MonadIO m =>
(forall b. ((forall a. m a -> IO a) -> IO b) -> m b)
-> MonadUnliftIO m
$cwithRunInIO :: forall (registry :: JobPayloadRegistry) (m :: * -> *) b.
MonadUnliftIO m =>
((forall a. HasqlDb registry m a -> IO a) -> IO b)
-> HasqlDb registry m b
withRunInIO :: forall b.
((forall a. HasqlDb registry m a -> IO a) -> IO b)
-> HasqlDb registry m b
MonadUnliftIO
    )

instance (Monad m) => HasHasqlPool (HasqlDb registry m) where
  getHasqlPool :: HasqlDb registry m HasqlConnectionPool
getHasqlPool = (HasqlEnv registry -> HasqlConnectionPool)
-> HasqlDb registry m HasqlConnectionPool
forall r (m :: * -> *) a. MonadReader r m => (r -> a) -> m a
asks HasqlEnv registry -> HasqlConnectionPool
forall (registry :: JobPayloadRegistry).
HasqlEnv registry -> HasqlConnectionPool
hasqlPool
  localHasqlPool :: forall a.
(HasqlConnectionPool -> HasqlConnectionPool)
-> HasqlDb registry m a -> HasqlDb registry m a
localHasqlPool HasqlConnectionPool -> HasqlConnectionPool
adjust = (HasqlEnv registry -> HasqlEnv registry)
-> HasqlDb registry m a -> HasqlDb registry m a
forall a.
(HasqlEnv registry -> HasqlEnv registry)
-> HasqlDb registry m a -> HasqlDb registry m a
forall r (m :: * -> *) a. MonadReader r m => (r -> r) -> m a -> m a
local (\HasqlEnv registry
env -> HasqlEnv registry
env {hasqlPool = adjust (hasqlPool env)})

instance (MonadUnliftIO m) => MonadArbiter (HasqlDb registry m) where
  type RegistryOf (HasqlDb registry m) = registry
  type Handler (HasqlDb registry m) job result = Hasql.Connection -> job -> HasqlDb registry m result
  getSchema :: HasqlDb registry m SchemaName
getSchema = (HasqlEnv registry -> SchemaName) -> HasqlDb registry m SchemaName
forall r (m :: * -> *) a. MonadReader r m => (r -> a) -> m a
asks HasqlEnv registry -> SchemaName
forall (registry :: JobPayloadRegistry).
HasqlEnv registry -> SchemaName
schema
  executeQuery :: forall a. Query a -> HasqlDb registry m [a]
executeQuery = Query a -> HasqlDb registry m [a]
forall (m :: * -> *) a.
(HasHasqlPool m, MonadIO m) =>
Query a -> m [a]
hasqlExecuteQuery
  executeQueryPrepared :: forall a. Query a -> HasqlDb registry m [a]
executeQueryPrepared = Query a -> HasqlDb registry m [a]
forall (m :: * -> *) a.
(HasHasqlPool m, MonadIO m) =>
Query a -> m [a]
hasqlExecuteQueryPrepared
  executeStatement :: forall a. Query a -> HasqlDb registry m Int64
executeStatement = Query a -> HasqlDb registry m Int64
forall (m :: * -> *) a.
(HasHasqlPool m, MonadIO m) =>
Query a -> m Int64
hasqlExecuteStatement
  withDbTransaction :: forall a. HasqlDb registry m a -> HasqlDb registry m a
withDbTransaction = HasqlDb registry m a -> HasqlDb registry m a
forall (m :: * -> *) a.
(HasHasqlPool m, MonadUnliftIO m) =>
m a -> m a
hasqlWithDbTransaction
  runHandlerWithConnection :: forall payload.
JobHandler
  (HasqlDb registry m)
  payload
  (ResultOf (HasqlDb registry m) payload)
-> JobRead payload
-> HasqlDb registry m (ResultOf (HasqlDb registry m) payload)
runHandlerWithConnection = Handler
  (HasqlDb registry m)
  (JobRecord payload Int64 SchemaName UTCTime PayloadKeys)
  (ResultFor payload (RegistryOf (HasqlDb registry m)))
-> JobRecord payload Int64 SchemaName UTCTime PayloadKeys
-> HasqlDb
     registry m (ResultFor payload (RegistryOf (HasqlDb registry m)))
(Connection
 -> JobRecord payload Int64 SchemaName UTCTime PayloadKeys
 -> HasqlDb registry m (SpecResult (MatchIn payload '[] registry)))
-> JobRecord payload Int64 SchemaName UTCTime PayloadKeys
-> HasqlDb registry m (SpecResult (MatchIn payload '[] registry))
forall (m :: * -> *) job result.
(HasHasqlPool m, MonadIO m) =>
(Connection -> job -> m result) -> job -> m result
hasqlRunHandlerWithConnection
  getListener :: HasqlDb registry m (Maybe Listener)
getListener = (HasqlEnv registry -> Maybe Listener)
-> HasqlDb registry m (Maybe Listener)
forall r (m :: * -> *) a. MonadReader r m => (r -> a) -> m a
asks HasqlEnv registry -> Maybe Listener
forall (registry :: JobPayloadRegistry).
HasqlEnv registry -> Maybe Listener
listener

-- | Release the env's connection pool, closing its open connections.
destroyHasqlEnv :: (MonadIO m) => HasqlEnv registry -> m ()
destroyHasqlEnv :: forall (m :: * -> *) (registry :: JobPayloadRegistry).
MonadIO m =>
HasqlEnv registry -> m ()
destroyHasqlEnv HasqlEnv registry
env =
  IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ (Pool Connection -> IO ()) -> Maybe (Pool Connection) -> IO ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ Pool Connection -> IO ()
forall a. Pool a -> IO ()
destroyAllResources (HasqlConnectionPool -> Maybe (Pool Connection)
connectionPool (HasqlEnv registry -> HasqlConnectionPool
forall (registry :: JobPayloadRegistry).
HasqlEnv registry -> HasqlConnectionPool
hasqlPool HasqlEnv registry
env))

-- | Turn off the shared LISTEN listener for an env, running poll-only.
disableListener :: HasqlEnv registry -> HasqlEnv registry
disableListener :: forall (registry :: JobPayloadRegistry).
HasqlEnv registry -> HasqlEnv registry
disableListener HasqlEnv registry
env = HasqlEnv registry
env {listener = Nothing}

-- | Give the env a dedicated LISTEN connection opened from a connection string.
-- The listener takes no pool slot.
useDedicatedListener :: (MonadIO m) => ByteString -> HasqlEnv registry -> m (HasqlEnv registry)
useDedicatedListener :: forall (m :: * -> *) (registry :: JobPayloadRegistry).
MonadIO m =>
ByteString -> HasqlEnv registry -> m (HasqlEnv registry)
useDedicatedListener ByteString
connStr HasqlEnv registry
env = do
  dedicated <- ByteString -> m DedicatedListen
forall (m :: * -> *). MonadIO m => ByteString -> m DedicatedListen
newDedicatedListen ByteString
connStr
  pure env {listener = Just (dedicatedListener dedicated)}

-- | A listener that borrows one pool connection for the hub's lifetime.
poolListener :: Pool Hasql.Connection -> IO Listener
poolListener :: Pool Connection -> IO Listener
poolListener Pool Connection
pool = ((Connection -> IO ()) -> IO ()) -> IO Listener
newPoolListener (\Connection -> IO ()
action -> Pool Connection -> (Connection -> IO ()) -> IO ()
forall a r. Pool a -> (a -> IO r) -> IO r
withResource Pool Connection
pool (Connection -> (Connection -> IO ()) -> IO ()
forall a. Connection -> (Connection -> IO a) -> IO a
`Compat.withHasqlLibPQConnection` Connection -> IO ()
action))

-- | Run a 'HasqlDb' action in its env.
runHasqlDb :: HasqlEnv registry -> HasqlDb registry m a -> m a
runHasqlDb :: forall (registry :: JobPayloadRegistry) (m :: * -> *) a.
HasqlEnv registry -> HasqlDb registry m a -> m a
runHasqlDb HasqlEnv registry
env HasqlDb registry m a
action = ReaderT (HasqlEnv registry) m a -> HasqlEnv registry -> m a
forall r (m :: * -> *) a. ReaderT r m a -> r -> m a
runReaderT (HasqlDb registry m a -> ReaderT (HasqlEnv registry) m a
forall (registry :: JobPayloadRegistry) (m :: * -> *) a.
HasqlDb registry m a -> ReaderT (HasqlEnv registry) m a
unHasqlDb HasqlDb registry m a
action) HasqlEnv registry
env

-- | Run a 'HasqlDb' action on one connection without a pool. The connection is pinned
-- as an open transaction. 'Arbiter.Core.MonadArbiter.withDbTransaction' nests through
-- savepoints. The caller owns the transaction.
--
-- @
-- _ <- Hasql.use conn (Session.script "BEGIN")
-- inTransaction conn "arbiter" $ do
--   Arb.insertJob (Arb.defaultJob myPayload)
-- _ <- Hasql.use conn (Session.script "COMMIT")
-- @
inTransaction
  :: forall registry m a
   . Hasql.Connection
  -> SchemaName
  -- ^ Schema name
  -> HasqlDb registry m a
  -> m a
inTransaction :: forall (registry :: JobPayloadRegistry) (m :: * -> *) a.
Connection -> SchemaName -> HasqlDb registry m a -> m a
inTransaction Connection
conn SchemaName
schemaName HasqlDb registry m a
action =
  let env :: HasqlEnv registry
env =
        HasqlEnv
          { schema :: SchemaName
schema = SchemaName
schemaName
          , hasqlPool :: HasqlConnectionPool
hasqlPool =
              HasqlConnectionPool
                { connectionPool :: Maybe (Pool Connection)
connectionPool = Maybe (Pool Connection)
forall a. Maybe a
Nothing
                , activeConn :: Maybe Connection
activeConn = Connection -> Maybe Connection
forall a. a -> Maybe a
Just Connection
conn
                , transactionDepth :: Int
transactionDepth = Int
1
                , preparedStatements :: Bool
preparedStatements = Bool
True
                }
          , listener :: Maybe Listener
listener = Maybe Listener
forall a. Maybe a
Nothing
          }
   in HasqlEnv registry -> HasqlDb registry m a -> m a
forall (registry :: JobPayloadRegistry) (m :: * -> *) a.
HasqlEnv registry -> HasqlDb registry m a -> m a
runHasqlDb HasqlEnv registry
env HasqlDb registry m a
action

-- | Create a 'HasqlEnv' with conservative pool defaults. Size worker pools with
-- 'createHasqlEnvWithConfig' and @poolConfigForWorkers@.
createHasqlEnv
  :: forall registry m
   . (MonadIO m)
  => Proxy registry
  -> ByteString
  -- ^ PostgreSQL connection string
  -> SchemaName
  -- ^ Schema name
  -> m (HasqlEnv registry)
createHasqlEnv :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
MonadIO m =>
Proxy registry -> ByteString -> SchemaName -> m (HasqlEnv registry)
createHasqlEnv Proxy registry
proxy ByteString
connStr SchemaName
schemaName =
  Proxy registry
-> ByteString -> SchemaName -> PoolConfig -> m (HasqlEnv registry)
forall (registry :: JobPayloadRegistry) (m :: * -> *).
MonadIO m =>
Proxy registry
-> ByteString -> SchemaName -> PoolConfig -> m (HasqlEnv registry)
createHasqlEnvWithConfig Proxy registry
proxy ByteString
connStr SchemaName
schemaName PoolConfig
PC.defaultPoolConfig

-- | Create a 'HasqlEnv' with custom pool settings.
createHasqlEnvWithConfig
  :: forall registry m
   . (MonadIO m)
  => Proxy registry
  -> ByteString
  -- ^ PostgreSQL connection string
  -> SchemaName
  -- ^ Schema name
  -> PoolConfig
  -> m (HasqlEnv registry)
createHasqlEnvWithConfig :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
MonadIO m =>
Proxy registry
-> ByteString -> SchemaName -> PoolConfig -> m (HasqlEnv registry)
createHasqlEnvWithConfig Proxy registry
_proxy ByteString
connStr SchemaName
schemaName PoolConfig
config = IO (HasqlEnv registry) -> m (HasqlEnv registry)
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO (HasqlEnv registry) -> m (HasqlEnv registry))
-> IO (HasqlEnv registry) -> m (HasqlEnv registry)
forall a b. (a -> b) -> a -> b
$ do
  connPool <-
    PoolConfig Connection -> IO (Pool Connection)
forall a. PoolConfig a -> IO (Pool a)
newPool
      (PoolConfig Connection -> IO (Pool Connection))
-> PoolConfig Connection -> IO (Pool Connection)
forall a b. (a -> b) -> a -> b
$ Maybe Int -> PoolConfig Connection -> PoolConfig Connection
forall a. Maybe Int -> PoolConfig a -> PoolConfig a
setNumStripes (PoolConfig -> Maybe Int
poolStripes PoolConfig
config)
      (PoolConfig Connection -> PoolConfig Connection)
-> PoolConfig Connection -> PoolConfig Connection
forall a b. (a -> b) -> a -> b
$ IO Connection
-> (Connection -> IO ()) -> Double -> Int -> PoolConfig Connection
forall a. IO a -> (a -> IO ()) -> Double -> Int -> PoolConfig a
defaultPoolConfig
        ( do
            result <- Settings -> IO (Either ConnectionError Connection)
Hasql.acquire (ByteString -> Settings
hasqlSettings ByteString
connStr)
            case result of
              Right Connection
conn -> Connection -> IO Connection
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Connection
conn
              Left ConnectionError
err -> HasqlConnectionError -> IO Connection
forall e a. (HasCallStack, Exception e) => e -> IO a
throwIO (HasqlConnectionError -> IO Connection)
-> HasqlConnectionError -> IO Connection
forall a b. (a -> b) -> a -> b
$ String -> HasqlConnectionError
HasqlConnectionError (ConnectionError -> String
forall a. Show a => a -> String
show ConnectionError
err)
        )
        Connection -> IO ()
Hasql.release
        (Int -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Int -> Double) -> Int -> Double
forall a b. (a -> b) -> a -> b
$ PoolConfig -> Int
poolIdleTimeout PoolConfig
config)
        (PoolConfig -> Int
poolSize PoolConfig
config)
  lstn <- poolListener connPool
  pure
    HasqlEnv
      { schema = schemaName
      , hasqlPool =
          HasqlConnectionPool
            { connectionPool = Just connPool
            , activeConn = Nothing
            , transactionDepth = 0
            , preparedStatements = True
            }
      , listener = Just lstn
      }

-- | Create a 'HasqlEnv' over a caller's own connection pool. The shared listener
-- holds one pool connection for the env's lifetime. Size the pool for the worker
-- load plus one. 'disableListener' runs poll-only and frees that slot.
-- 'useDedicatedListener' gives the listener its own connection.
createHasqlEnvWithPool
  :: forall registry m
   . (MonadIO m)
  => Proxy registry
  -> Pool Hasql.Connection
  -> SchemaName
  -- ^ Schema name
  -> m (HasqlEnv registry)
createHasqlEnvWithPool :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
MonadIO m =>
Proxy registry
-> Pool Connection -> SchemaName -> m (HasqlEnv registry)
createHasqlEnvWithPool Proxy registry
_proxy Pool Connection
connPool SchemaName
schemaName = IO (HasqlEnv registry) -> m (HasqlEnv registry)
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO (HasqlEnv registry) -> m (HasqlEnv registry))
-> IO (HasqlEnv registry) -> m (HasqlEnv registry)
forall a b. (a -> b) -> a -> b
$ do
  lstn <- Pool Connection -> IO Listener
poolListener Pool Connection
connPool
  pure
    HasqlEnv
      { schema = schemaName
      , hasqlPool =
          HasqlConnectionPool
            { connectionPool = Just connPool
            , activeConn = Nothing
            , transactionDepth = 0
            , preparedStatements = True
            }
      , listener = Just lstn
      }

-- | Enable or disable prepared hot statements (the claim). Each pooled connection
-- prepares once and reuses the plan. Requires direct connections or a pooler that
-- supports server-side prepared statements.
setPreparedStatements :: Bool -> HasqlEnv registry -> HasqlEnv registry
setPreparedStatements :: forall (registry :: JobPayloadRegistry).
Bool -> HasqlEnv registry -> HasqlEnv registry
setPreparedStatements Bool
flag HasqlEnv registry
env = HasqlEnv registry
env {hasqlPool = (hasqlPool env) {preparedStatements = flag}}

-- | Re-exported from "Arbiter.Hasql.Compat".
hasqlSettings :: ByteString -> Compat.HasqlSettings
hasqlSettings :: ByteString -> Settings
hasqlSettings = ByteString -> Settings
Compat.hasqlSettings