{-# LANGUAGE TypeFamilies #-}

-- | The postgresql-simple database monad with a built-in 'MonadArbiter' instance:
--
-- @
-- import Arbiter.Core
-- import Arbiter.Simple
--
-- myFunction :: SimpleDb MyRegistry IO ()
-- myFunction = insertJob (defaultJob myPayload)
-- @
module Arbiter.Simple.SimpleDb
  ( -- * Database Monad
    SimpleDb (..)
  , SimpleEnv (..)
  , runSimpleDb
  , inTransaction

    -- * Environment Creation
  , createSimpleEnv
  , createSimpleEnvWithConfig
  , createSimpleEnvWithPool
  , destroySimpleEnv
  , disableListener
  , useDedicatedListener
  ) 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.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 Database.PostgreSQL.Simple (Connection, close, connectPostgreSQL)
import Database.PostgreSQL.Simple.Internal (withConnection)
import UnliftIO (MonadUnliftIO)

import Arbiter.Simple.MonadArbiter
  ( HasSimplePool (..)
  , SimpleConnectionPool (..)
  , simpleExecuteQuery
  , simpleExecuteStatement
  , simpleRunHandlerWithConnection
  , simpleWithDbTransaction
  )

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

-- | The postgresql-simple database monad.
newtype SimpleDb (registry :: JobPayloadRegistry) m a = SimpleDb {forall (registry :: JobPayloadRegistry) (m :: * -> *) a.
SimpleDb registry m a -> ReaderT (SimpleEnv registry) m a
unSimpleDb :: ReaderT (SimpleEnv registry) m a}
  deriving newtype
    ( Functor (SimpleDb registry m)
Functor (SimpleDb registry m) =>
(forall a. a -> SimpleDb registry m a)
-> (forall a b.
    SimpleDb registry m (a -> b)
    -> SimpleDb registry m a -> SimpleDb registry m b)
-> (forall a b c.
    (a -> b -> c)
    -> SimpleDb registry m a
    -> SimpleDb registry m b
    -> SimpleDb registry m c)
-> (forall a b.
    SimpleDb registry m a
    -> SimpleDb registry m b -> SimpleDb registry m b)
-> (forall a b.
    SimpleDb registry m a
    -> SimpleDb registry m b -> SimpleDb registry m a)
-> Applicative (SimpleDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *).
Applicative m =>
Functor (SimpleDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *) a.
Applicative m =>
a -> SimpleDb registry m a
forall (registry :: JobPayloadRegistry) (m :: * -> *) a b.
Applicative m =>
SimpleDb registry m a
-> SimpleDb registry m b -> SimpleDb registry m a
forall (registry :: JobPayloadRegistry) (m :: * -> *) a b.
Applicative m =>
SimpleDb registry m a
-> SimpleDb registry m b -> SimpleDb registry m b
forall (registry :: JobPayloadRegistry) (m :: * -> *) a b.
Applicative m =>
SimpleDb registry m (a -> b)
-> SimpleDb registry m a -> SimpleDb registry m b
forall (registry :: JobPayloadRegistry) (m :: * -> *) a b c.
Applicative m =>
(a -> b -> c)
-> SimpleDb registry m a
-> SimpleDb registry m b
-> SimpleDb registry m c
forall a. a -> SimpleDb registry m a
forall a b.
SimpleDb registry m a
-> SimpleDb registry m b -> SimpleDb registry m a
forall a b.
SimpleDb registry m a
-> SimpleDb registry m b -> SimpleDb registry m b
forall a b.
SimpleDb registry m (a -> b)
-> SimpleDb registry m a -> SimpleDb registry m b
forall a b c.
(a -> b -> c)
-> SimpleDb registry m a
-> SimpleDb registry m b
-> SimpleDb 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 -> SimpleDb registry m a
pure :: forall a. a -> SimpleDb registry m a
$c<*> :: forall (registry :: JobPayloadRegistry) (m :: * -> *) a b.
Applicative m =>
SimpleDb registry m (a -> b)
-> SimpleDb registry m a -> SimpleDb registry m b
<*> :: forall a b.
SimpleDb registry m (a -> b)
-> SimpleDb registry m a -> SimpleDb registry m b
$cliftA2 :: forall (registry :: JobPayloadRegistry) (m :: * -> *) a b c.
Applicative m =>
(a -> b -> c)
-> SimpleDb registry m a
-> SimpleDb registry m b
-> SimpleDb registry m c
liftA2 :: forall a b c.
(a -> b -> c)
-> SimpleDb registry m a
-> SimpleDb registry m b
-> SimpleDb registry m c
$c*> :: forall (registry :: JobPayloadRegistry) (m :: * -> *) a b.
Applicative m =>
SimpleDb registry m a
-> SimpleDb registry m b -> SimpleDb registry m b
*> :: forall a b.
SimpleDb registry m a
-> SimpleDb registry m b -> SimpleDb registry m b
$c<* :: forall (registry :: JobPayloadRegistry) (m :: * -> *) a b.
Applicative m =>
SimpleDb registry m a
-> SimpleDb registry m b -> SimpleDb registry m a
<* :: forall a b.
SimpleDb registry m a
-> SimpleDb registry m b -> SimpleDb registry m a
Applicative
    , (forall a b.
 (a -> b) -> SimpleDb registry m a -> SimpleDb registry m b)
-> (forall a b.
    a -> SimpleDb registry m b -> SimpleDb registry m a)
-> Functor (SimpleDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *) a b.
Functor m =>
a -> SimpleDb registry m b -> SimpleDb registry m a
forall (registry :: JobPayloadRegistry) (m :: * -> *) a b.
Functor m =>
(a -> b) -> SimpleDb registry m a -> SimpleDb registry m b
forall a b. a -> SimpleDb registry m b -> SimpleDb registry m a
forall a b.
(a -> b) -> SimpleDb registry m a -> SimpleDb 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) -> SimpleDb registry m a -> SimpleDb registry m b
fmap :: forall a b.
(a -> b) -> SimpleDb registry m a -> SimpleDb registry m b
$c<$ :: forall (registry :: JobPayloadRegistry) (m :: * -> *) a b.
Functor m =>
a -> SimpleDb registry m b -> SimpleDb registry m a
<$ :: forall a b. a -> SimpleDb registry m b -> SimpleDb registry m a
Functor
    , Applicative (SimpleDb registry m)
Applicative (SimpleDb registry m) =>
(forall a b.
 SimpleDb registry m a
 -> (a -> SimpleDb registry m b) -> SimpleDb registry m b)
-> (forall a b.
    SimpleDb registry m a
    -> SimpleDb registry m b -> SimpleDb registry m b)
-> (forall a. a -> SimpleDb registry m a)
-> Monad (SimpleDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *).
Monad m =>
Applicative (SimpleDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *) a.
Monad m =>
a -> SimpleDb registry m a
forall (registry :: JobPayloadRegistry) (m :: * -> *) a b.
Monad m =>
SimpleDb registry m a
-> SimpleDb registry m b -> SimpleDb registry m b
forall (registry :: JobPayloadRegistry) (m :: * -> *) a b.
Monad m =>
SimpleDb registry m a
-> (a -> SimpleDb registry m b) -> SimpleDb registry m b
forall a. a -> SimpleDb registry m a
forall a b.
SimpleDb registry m a
-> SimpleDb registry m b -> SimpleDb registry m b
forall a b.
SimpleDb registry m a
-> (a -> SimpleDb registry m b) -> SimpleDb 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 =>
SimpleDb registry m a
-> (a -> SimpleDb registry m b) -> SimpleDb registry m b
>>= :: forall a b.
SimpleDb registry m a
-> (a -> SimpleDb registry m b) -> SimpleDb registry m b
$c>> :: forall (registry :: JobPayloadRegistry) (m :: * -> *) a b.
Monad m =>
SimpleDb registry m a
-> SimpleDb registry m b -> SimpleDb registry m b
>> :: forall a b.
SimpleDb registry m a
-> SimpleDb registry m b -> SimpleDb registry m b
$creturn :: forall (registry :: JobPayloadRegistry) (m :: * -> *) a.
Monad m =>
a -> SimpleDb registry m a
return :: forall a. a -> SimpleDb registry m a
Monad
    , MonadThrow (SimpleDb registry m)
MonadThrow (SimpleDb registry m) =>
(forall e a.
 (HasCallStack, Exception e) =>
 SimpleDb registry m a
 -> (e -> SimpleDb registry m a) -> SimpleDb registry m a)
-> (forall e a.
    Exception e =>
    SimpleDb registry m a
    -> (ExceptionWithContext e -> SimpleDb registry m a)
    -> SimpleDb registry m a)
-> MonadCatch (SimpleDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *).
MonadCatch m =>
MonadThrow (SimpleDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *) e a.
(MonadCatch m, HasCallStack, Exception e) =>
SimpleDb registry m a
-> (e -> SimpleDb registry m a) -> SimpleDb registry m a
forall (registry :: JobPayloadRegistry) (m :: * -> *) e a.
(MonadCatch m, Exception e) =>
SimpleDb registry m a
-> (ExceptionWithContext e -> SimpleDb registry m a)
-> SimpleDb registry m a
forall e a.
(HasCallStack, Exception e) =>
SimpleDb registry m a
-> (e -> SimpleDb registry m a) -> SimpleDb registry m a
forall e a.
Exception e =>
SimpleDb registry m a
-> (ExceptionWithContext e -> SimpleDb registry m a)
-> SimpleDb 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) =>
SimpleDb registry m a
-> (e -> SimpleDb registry m a) -> SimpleDb registry m a
catch :: forall e a.
(HasCallStack, Exception e) =>
SimpleDb registry m a
-> (e -> SimpleDb registry m a) -> SimpleDb registry m a
$ccatchNoPropagate :: forall (registry :: JobPayloadRegistry) (m :: * -> *) e a.
(MonadCatch m, Exception e) =>
SimpleDb registry m a
-> (ExceptionWithContext e -> SimpleDb registry m a)
-> SimpleDb registry m a
catchNoPropagate :: forall e a.
Exception e =>
SimpleDb registry m a
-> (ExceptionWithContext e -> SimpleDb registry m a)
-> SimpleDb registry m a
MonadCatch
    , Monad (SimpleDb registry m)
Monad (SimpleDb registry m) =>
(forall a. HasCallStack => String -> SimpleDb registry m a)
-> MonadFail (SimpleDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *).
MonadFail m =>
Monad (SimpleDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *) a.
(MonadFail m, HasCallStack) =>
String -> SimpleDb registry m a
forall a. HasCallStack => String -> SimpleDb 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 -> SimpleDb registry m a
fail :: forall a. HasCallStack => String -> SimpleDb registry m a
MonadFail
    , Monad (SimpleDb registry m)
Monad (SimpleDb registry m) =>
(forall a. IO a -> SimpleDb registry m a)
-> MonadIO (SimpleDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *).
MonadIO m =>
Monad (SimpleDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *) a.
MonadIO m =>
IO a -> SimpleDb registry m a
forall a. IO a -> SimpleDb 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 -> SimpleDb registry m a
liftIO :: forall a. IO a -> SimpleDb registry m a
MonadIO
    , MonadCatch (SimpleDb registry m)
MonadCatch (SimpleDb registry m) =>
(forall b.
 HasCallStack =>
 ((forall a. SimpleDb registry m a -> SimpleDb registry m a)
  -> SimpleDb registry m b)
 -> SimpleDb registry m b)
-> (forall b.
    HasCallStack =>
    ((forall a. SimpleDb registry m a -> SimpleDb registry m a)
     -> SimpleDb registry m b)
    -> SimpleDb registry m b)
-> (forall a b c.
    HasCallStack =>
    SimpleDb registry m a
    -> (a -> ExitCase b -> SimpleDb registry m c)
    -> (a -> SimpleDb registry m b)
    -> SimpleDb registry m (b, c))
-> MonadMask (SimpleDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *).
MonadMask m =>
MonadCatch (SimpleDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *) b.
(MonadMask m, HasCallStack) =>
((forall a. SimpleDb registry m a -> SimpleDb registry m a)
 -> SimpleDb registry m b)
-> SimpleDb registry m b
forall (registry :: JobPayloadRegistry) (m :: * -> *) a b c.
(MonadMask m, HasCallStack) =>
SimpleDb registry m a
-> (a -> ExitCase b -> SimpleDb registry m c)
-> (a -> SimpleDb registry m b)
-> SimpleDb registry m (b, c)
forall b.
HasCallStack =>
((forall a. SimpleDb registry m a -> SimpleDb registry m a)
 -> SimpleDb registry m b)
-> SimpleDb registry m b
forall a b c.
HasCallStack =>
SimpleDb registry m a
-> (a -> ExitCase b -> SimpleDb registry m c)
-> (a -> SimpleDb registry m b)
-> SimpleDb 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. SimpleDb registry m a -> SimpleDb registry m a)
 -> SimpleDb registry m b)
-> SimpleDb registry m b
mask :: forall b.
HasCallStack =>
((forall a. SimpleDb registry m a -> SimpleDb registry m a)
 -> SimpleDb registry m b)
-> SimpleDb registry m b
$cuninterruptibleMask :: forall (registry :: JobPayloadRegistry) (m :: * -> *) b.
(MonadMask m, HasCallStack) =>
((forall a. SimpleDb registry m a -> SimpleDb registry m a)
 -> SimpleDb registry m b)
-> SimpleDb registry m b
uninterruptibleMask :: forall b.
HasCallStack =>
((forall a. SimpleDb registry m a -> SimpleDb registry m a)
 -> SimpleDb registry m b)
-> SimpleDb registry m b
$cgeneralBracket :: forall (registry :: JobPayloadRegistry) (m :: * -> *) a b c.
(MonadMask m, HasCallStack) =>
SimpleDb registry m a
-> (a -> ExitCase b -> SimpleDb registry m c)
-> (a -> SimpleDb registry m b)
-> SimpleDb registry m (b, c)
generalBracket :: forall a b c.
HasCallStack =>
SimpleDb registry m a
-> (a -> ExitCase b -> SimpleDb registry m c)
-> (a -> SimpleDb registry m b)
-> SimpleDb registry m (b, c)
MonadMask
    , MonadReader (SimpleEnv registry)
    , Monad (SimpleDb registry m)
Monad (SimpleDb registry m) =>
(forall e a.
 (HasCallStack, Exception e) =>
 e -> SimpleDb registry m a)
-> (forall e a.
    Exception e =>
    ExceptionWithContext e -> SimpleDb registry m a)
-> MonadThrow (SimpleDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *).
MonadThrow m =>
Monad (SimpleDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *) e a.
(MonadThrow m, HasCallStack, Exception e) =>
e -> SimpleDb registry m a
forall (registry :: JobPayloadRegistry) (m :: * -> *) e a.
(MonadThrow m, Exception e) =>
ExceptionWithContext e -> SimpleDb registry m a
forall e a.
(HasCallStack, Exception e) =>
e -> SimpleDb registry m a
forall e a.
Exception e =>
ExceptionWithContext e -> SimpleDb 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 -> SimpleDb registry m a
throwM :: forall e a.
(HasCallStack, Exception e) =>
e -> SimpleDb registry m a
$crethrowM :: forall (registry :: JobPayloadRegistry) (m :: * -> *) e a.
(MonadThrow m, Exception e) =>
ExceptionWithContext e -> SimpleDb registry m a
rethrowM :: forall e a.
Exception e =>
ExceptionWithContext e -> SimpleDb registry m a
MonadThrow
    , MonadIO (SimpleDb registry m)
MonadIO (SimpleDb registry m) =>
(forall b.
 ((forall a. SimpleDb registry m a -> IO a) -> IO b)
 -> SimpleDb registry m b)
-> MonadUnliftIO (SimpleDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *).
MonadUnliftIO m =>
MonadIO (SimpleDb registry m)
forall (registry :: JobPayloadRegistry) (m :: * -> *) b.
MonadUnliftIO m =>
((forall a. SimpleDb registry m a -> IO a) -> IO b)
-> SimpleDb registry m b
forall b.
((forall a. SimpleDb registry m a -> IO a) -> IO b)
-> SimpleDb 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. SimpleDb registry m a -> IO a) -> IO b)
-> SimpleDb registry m b
withRunInIO :: forall b.
((forall a. SimpleDb registry m a -> IO a) -> IO b)
-> SimpleDb registry m b
MonadUnliftIO
    )

instance (Monad m) => HasSimplePool (SimpleDb registry m) where
  getSimplePool :: SimpleDb registry m SimpleConnectionPool
getSimplePool = (SimpleEnv registry -> SimpleConnectionPool)
-> SimpleDb registry m SimpleConnectionPool
forall r (m :: * -> *) a. MonadReader r m => (r -> a) -> m a
asks SimpleEnv registry -> SimpleConnectionPool
forall (registry :: JobPayloadRegistry).
SimpleEnv registry -> SimpleConnectionPool
simplePool
  localSimplePool :: forall a.
(SimpleConnectionPool -> SimpleConnectionPool)
-> SimpleDb registry m a -> SimpleDb registry m a
localSimplePool SimpleConnectionPool -> SimpleConnectionPool
adjust = (SimpleEnv registry -> SimpleEnv registry)
-> SimpleDb registry m a -> SimpleDb registry m a
forall a.
(SimpleEnv registry -> SimpleEnv registry)
-> SimpleDb registry m a -> SimpleDb registry m a
forall r (m :: * -> *) a. MonadReader r m => (r -> r) -> m a -> m a
local (\SimpleEnv registry
env -> SimpleEnv registry
env {simplePool = adjust (simplePool env)})

instance (MonadUnliftIO m) => MonadArbiter (SimpleDb registry m) where
  type RegistryOf (SimpleDb registry m) = registry
  type Handler (SimpleDb registry m) job result = Connection -> job -> SimpleDb registry m result
  getSchema :: SimpleDb registry m SchemaName
getSchema = (SimpleEnv registry -> SchemaName)
-> SimpleDb registry m SchemaName
forall r (m :: * -> *) a. MonadReader r m => (r -> a) -> m a
asks SimpleEnv registry -> SchemaName
forall (registry :: JobPayloadRegistry).
SimpleEnv registry -> SchemaName
schema
  executeQuery :: forall a. Query a -> SimpleDb registry m [a]
executeQuery = Query a -> SimpleDb registry m [a]
forall (m :: * -> *) a.
(HasSimplePool m, MonadUnliftIO m) =>
Query a -> m [a]
simpleExecuteQuery
  executeStatement :: forall a. Query a -> SimpleDb registry m Int64
executeStatement = Query a -> SimpleDb registry m Int64
forall (m :: * -> *) a.
(HasSimplePool m, MonadUnliftIO m) =>
Query a -> m Int64
simpleExecuteStatement
  withDbTransaction :: forall a. SimpleDb registry m a -> SimpleDb registry m a
withDbTransaction = SimpleDb registry m a -> SimpleDb registry m a
forall (m :: * -> *) a.
(HasSimplePool m, MonadUnliftIO m) =>
m a -> m a
simpleWithDbTransaction
  runHandlerWithConnection :: forall payload.
JobHandler
  (SimpleDb registry m)
  payload
  (ResultOf (SimpleDb registry m) payload)
-> JobRead payload
-> SimpleDb registry m (ResultOf (SimpleDb registry m) payload)
runHandlerWithConnection = Handler
  (SimpleDb registry m)
  (JobRecord payload Int64 SchemaName UTCTime PayloadKeys)
  (ResultFor payload (RegistryOf (SimpleDb registry m)))
-> JobRecord payload Int64 SchemaName UTCTime PayloadKeys
-> SimpleDb
     registry m (ResultFor payload (RegistryOf (SimpleDb registry m)))
(Connection
 -> JobRecord payload Int64 SchemaName UTCTime PayloadKeys
 -> SimpleDb registry m (SpecResult (MatchIn payload '[] registry)))
-> JobRecord payload Int64 SchemaName UTCTime PayloadKeys
-> SimpleDb registry m (SpecResult (MatchIn payload '[] registry))
forall (m :: * -> *) job result.
(HasSimplePool m, MonadUnliftIO m) =>
(Connection -> job -> m result) -> job -> m result
simpleRunHandlerWithConnection
  getListener :: SimpleDb registry m (Maybe Listener)
getListener = (SimpleEnv registry -> Maybe Listener)
-> SimpleDb registry m (Maybe Listener)
forall r (m :: * -> *) a. MonadReader r m => (r -> a) -> m a
asks SimpleEnv registry -> Maybe Listener
forall (registry :: JobPayloadRegistry).
SimpleEnv registry -> Maybe Listener
listener

-- | Release the env's connection pool, closing its open connections.
destroySimpleEnv :: (MonadIO m) => SimpleEnv registry -> m ()
destroySimpleEnv :: forall (m :: * -> *) (registry :: JobPayloadRegistry).
MonadIO m =>
SimpleEnv registry -> m ()
destroySimpleEnv SimpleEnv 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 (SimpleConnectionPool -> Maybe (Pool Connection)
connectionPool (SimpleEnv registry -> SimpleConnectionPool
forall (registry :: JobPayloadRegistry).
SimpleEnv registry -> SimpleConnectionPool
simplePool SimpleEnv registry
env))

-- | Turn off the shared LISTEN listener for an env, running poll-only.
disableListener :: SimpleEnv registry -> SimpleEnv registry
disableListener :: forall (registry :: JobPayloadRegistry).
SimpleEnv registry -> SimpleEnv registry
disableListener SimpleEnv registry
env = SimpleEnv 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 -> SimpleEnv registry -> m (SimpleEnv registry)
useDedicatedListener :: forall (m :: * -> *) (registry :: JobPayloadRegistry).
MonadIO m =>
ByteString -> SimpleEnv registry -> m (SimpleEnv registry)
useDedicatedListener ByteString
connStr SimpleEnv 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 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
`withConnection` Connection -> IO ()
action))

-- | Run a 'SimpleDb' action in its env.
runSimpleDb :: SimpleEnv registry -> SimpleDb registry m a -> m a
runSimpleDb :: forall (registry :: JobPayloadRegistry) (m :: * -> *) a.
SimpleEnv registry -> SimpleDb registry m a -> m a
runSimpleDb SimpleEnv registry
env SimpleDb registry m a
action = ReaderT (SimpleEnv registry) m a -> SimpleEnv registry -> m a
forall r (m :: * -> *) a. ReaderT r m a -> r -> m a
runReaderT (SimpleDb registry m a -> ReaderT (SimpleEnv registry) m a
forall (registry :: JobPayloadRegistry) (m :: * -> *) a.
SimpleDb registry m a -> ReaderT (SimpleEnv registry) m a
unSimpleDb SimpleDb registry m a
action) SimpleEnv registry
env

-- | Run a 'SimpleDb' action on one connection without a pool or env. The connection is
-- pinned as an open transaction. 'Arbiter.Core.MonadArbiter.withDbTransaction' nests
-- through savepoints. The caller owns the transaction.
--
-- @
-- PG.withTransaction conn $ do
--   PG.execute conn "INSERT INTO orders ..." params
--   inTransaction conn "arbiter" $
--     Arb.insertJob (Arb.defaultJob (ProcessOrder orderId))
-- @
inTransaction
  :: forall registry m a
   . Connection
  -> SchemaName
  -- ^ Schema name
  -> SimpleDb registry m a
  -> m a
inTransaction :: forall (registry :: JobPayloadRegistry) (m :: * -> *) a.
Connection -> SchemaName -> SimpleDb registry m a -> m a
inTransaction Connection
conn SchemaName
schemaName SimpleDb registry m a
action =
  let env :: SimpleEnv registry
env =
        SimpleEnv
          { schema :: SchemaName
schema = SchemaName
schemaName
          , simplePool :: SimpleConnectionPool
simplePool =
              SimpleConnectionPool
                { 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
                }
          , listener :: Maybe Listener
listener = Maybe Listener
forall a. Maybe a
Nothing
          }
   in SimpleEnv registry -> SimpleDb registry m a -> m a
forall (registry :: JobPayloadRegistry) (m :: * -> *) a.
SimpleEnv registry -> SimpleDb registry m a -> m a
runSimpleDb SimpleEnv registry
env SimpleDb registry m a
action

-- | Create a 'SimpleEnv' with default pool settings. Size worker pools with
-- 'createSimpleEnvWithConfig' and @poolConfigForWorkers@.
createSimpleEnv
  :: forall registry m
   . (MonadIO m)
  => Proxy registry
  -- ^ Type-level job payload registry
  -> ByteString
  -- ^ PostgreSQL connection string
  -> SchemaName
  -- ^ Schema name
  -> m (SimpleEnv registry)
createSimpleEnv :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
MonadIO m =>
Proxy registry
-> ByteString -> SchemaName -> m (SimpleEnv registry)
createSimpleEnv Proxy registry
proxy ByteString
connStr SchemaName
schemaName =
  Proxy registry
-> ByteString -> SchemaName -> PoolConfig -> m (SimpleEnv registry)
forall (registry :: JobPayloadRegistry) (m :: * -> *).
MonadIO m =>
Proxy registry
-> ByteString -> SchemaName -> PoolConfig -> m (SimpleEnv registry)
createSimpleEnvWithConfig Proxy registry
proxy ByteString
connStr SchemaName
schemaName PoolConfig
PC.defaultPoolConfig

-- | Create a 'SimpleEnv' with custom pool settings.
--
-- @
-- let config = PoolConfig
--       { poolSize = 50
--       , poolIdleTimeout = 120
--       , poolStripes = Just 4
--       }
-- env <- createSimpleEnvWithConfig (Proxy @MyRegistry) "host=localhost dbname=mydb" "arbiter" config
-- @
createSimpleEnvWithConfig
  :: forall registry m
   . (MonadIO m)
  => Proxy registry
  -- ^ Type-level job payload registry
  -> ByteString
  -- ^ PostgreSQL connection string
  -> SchemaName
  -- ^ Schema name
  -> PoolConfig
  -- ^ Pool configuration
  -> m (SimpleEnv registry)
createSimpleEnvWithConfig :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
MonadIO m =>
Proxy registry
-> ByteString -> SchemaName -> PoolConfig -> m (SimpleEnv registry)
createSimpleEnvWithConfig Proxy registry
_proxy ByteString
connStr SchemaName
schemaName PoolConfig
config = IO (SimpleEnv registry) -> m (SimpleEnv registry)
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO (SimpleEnv registry) -> m (SimpleEnv registry))
-> IO (SimpleEnv registry) -> m (SimpleEnv registry)
forall a b. (a -> b) -> a -> b
$ do
  let stripes :: Maybe Int
stripes = PoolConfig -> Maybe Int
poolStripes PoolConfig
config
  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 Maybe Int
stripes
      (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
        (ByteString -> IO Connection
connectPostgreSQL ByteString
connStr)
        Connection -> IO ()
close
        (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) -- idle time (seconds)
        (PoolConfig -> Int
poolSize PoolConfig
config)
  lstn <- poolListener connPool
  pure
    SimpleEnv
      { schema = schemaName
      , simplePool = SimpleConnectionPool {connectionPool = Just connPool, activeConn = Nothing, transactionDepth = 0}
      , listener = Just lstn
      }

-- | Create a 'SimpleEnv' 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.
createSimpleEnvWithPool
  :: forall registry m
   . (MonadIO m)
  => Proxy registry
  -- ^ Type-level job payload registry
  -> Pool Connection
  -- ^ User-provided connection pool
  -> SchemaName
  -- ^ Schema name
  -> m (SimpleEnv registry)
createSimpleEnvWithPool :: forall (registry :: JobPayloadRegistry) (m :: * -> *).
MonadIO m =>
Proxy registry
-> Pool Connection -> SchemaName -> m (SimpleEnv registry)
createSimpleEnvWithPool Proxy registry
_proxy Pool Connection
connPool SchemaName
schemaName = IO (SimpleEnv registry) -> m (SimpleEnv registry)
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO (SimpleEnv registry) -> m (SimpleEnv registry))
-> IO (SimpleEnv registry) -> m (SimpleEnv registry)
forall a b. (a -> b) -> a -> b
$ do
  lstn <- Pool Connection -> IO Listener
poolListener Pool Connection
connPool
  pure
    SimpleEnv
      { schema = schemaName
      , simplePool = SimpleConnectionPool {connectionPool = Just connPool, activeConn = Nothing, transactionDepth = 0}
      , listener = Just lstn
      }