{-# LANGUAGE OverloadedStrings #-}

-- | Schema setup, teardown, and connection helpers for the arbiter test suites.
module Arbiter.Test.Setup
  ( cleanupData
  , cleanupOnce
  , withConn
  , terminateBackendsMatching
  , terminatePid
  , listenerConnectionCount
  , execute_
  , toQuery
  , execStatement
  , execQuery
  , setupOnce
  , addQueueTable
  , createSharedPool
  , createPoolOf
  , createPoolWith
  , sharedPoolSize
  , concurrencyPoolSize
  , stateMachinePoolSize
  , truncateToMicros
  , mkTime
  , seedConcurrencyPolicySQL
  , drainWith
  , quietHubLog
  ) where

import Arbiter.Core.Codec (RowCodec)
import Arbiter.Core.Concurrency.Schema qualified as CC
import Arbiter.Core.Concurrency.Spec (ConcurrencyPolicy (..))
import Arbiter.Core.FailureGate (defaultFailureRepeatInterval)
import Arbiter.Core.Job.Schema qualified as Schema
import Arbiter.Core.Listen (HubLog (..))
import Arbiter.Core.MonadArbiter (MonadArbiter, Params)
import Arbiter.Core.MonadArbiter qualified as MA
import Arbiter.Core.RateLimit.Schema qualified as RL
import Arbiter.Core.SchemaTables (allSchemaTables)
import Arbiter.Core.Sql.Query (Piece (..))
import Arbiter.Core.SqlLiterals (textLiteral)
import Arbiter.Migrations
  ( allTableAdmission
  , jobQueueMigrationsForTable
  , schemaLevelMigrations
  )
import Control.Concurrent (threadDelay)
import Control.Exception (bracket, throwIO, try)
import Control.Monad (void, when)
import Data.ByteString (ByteString)
import Data.Foldable (traverse_)
import Data.Int (Int32, Int64)
import Data.List (intersperse)
import Data.Maybe (listToMaybe)
import Data.Pool (Pool, defaultPoolConfig, newPool, setNumStripes)
import Data.String (fromString)
import Data.Text (Text)
import Data.Text qualified as T
import Data.Time (UTCTime (..), fromGregorian, picosecondsToDiffTime, secondsToDiffTime)
import Database.PostgreSQL.LibPQ qualified as LibPQ
import Database.PostgreSQL.Simple (Connection, Only (..), SqlError (..), close, connectPostgreSQL, execute, query)
import Database.PostgreSQL.Simple.Internal qualified as PGS
import Database.PostgreSQL.Simple.Migration (MigrationCommand (..))
import Database.PostgreSQL.Simple.Types (Query (..))

-- | Rebuild the schema with the shipped migration scripts.
setupDDLWithConfig :: Bool -> Text -> Text -> Connection -> IO ()
setupDDLWithConfig :: Bool -> Text -> Text -> Connection -> IO ()
setupDDLWithConfig Bool
withNotify Text
schemaName Text
tableName Connection
conn = do
  Connection -> Text -> IO ()
execute_ Connection
conn (Text -> IO ()) -> Text -> IO ()
forall a b. (a -> b) -> a -> b
$ Text
"DROP SCHEMA IF EXISTS " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
schemaName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" CASCADE"
  Connection -> Text -> IO ()
execute_ Connection
conn (Text -> IO ()) -> Text -> IO ()
forall a b. (a -> b) -> a -> b
$ Text -> Text
Schema.createSchemaSQL Text
schemaName
  (MigrationCommand -> IO ()) -> [MigrationCommand] -> IO ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ (Connection -> MigrationCommand -> IO ()
runScript Connection
conn) ([MigrationCommand] -> IO ()) -> [MigrationCommand] -> IO ()
forall a b. (a -> b) -> a -> b
$
    Text -> [MigrationCommand]
schemaLevelMigrations Text
schemaName
      [MigrationCommand] -> [MigrationCommand] -> [MigrationCommand]
forall a. Semigroup a => a -> a -> a
<> Text -> Text -> TableAdmission -> [MigrationCommand]
jobQueueMigrationsForTable Text
schemaName Text
tableName TableAdmission
allTableAdmission
  Bool -> Text -> Text -> Connection -> IO ()
createNotifyObjects Bool
withNotify Text
schemaName Text
tableName Connection
conn

runScript :: Connection -> MigrationCommand -> IO ()
runScript :: Connection -> MigrationCommand -> IO ()
runScript Connection
conn (MigrationScript String
_ ByteString
sql) = IO Int64 -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO Int64 -> IO ()) -> IO Int64 -> IO ()
forall a b. (a -> b) -> a -> b
$ Connection -> Query -> () -> IO Int64
forall q. ToRow q => Connection -> Query -> q -> IO Int64
execute Connection
conn (ByteString -> Query
Query ByteString
sql) ()
runScript Connection
_ MigrationCommand
_ = () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()

-- | Truncate the queue's tables and the schema-wide tables, except rate-limit policies.
cleanupData :: Text -> Text -> Connection -> IO ()
cleanupData :: Text -> Text -> Connection -> IO ()
cleanupData Text
schemaName Text
tableName Connection
conn = do
  Connection -> Text -> IO ()
execute_ Connection
conn Text
"SET client_min_messages = WARNING"
  -- A draining worker pool can still hold locks on the job and groups tables.
  -- Bound the wait and retry on a deadlock or lock timeout.
  Connection -> Text -> IO ()
execute_ Connection
conn (Text
"SET lock_timeout = " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
textLiteral Text
cleanupLockTimeout)
  -- Rate-limit policies are seeded once per suite.
  let truncated :: [Text]
truncated = (Text -> Bool) -> [Text] -> [Text]
forall a. (a -> Bool) -> [a] -> [a]
filter (Text -> Text -> Bool
forall a. Eq a => a -> a -> Bool
/= Text
RL.arbiterRateLimitPoliciesTableName) ([Text] -> [Text]
allSchemaTables [Text
tableName])
      truncateSql :: Text
truncateSql =
        Text
"TRUNCATE "
          Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> [Text] -> Text
T.intercalate Text
", " ((Text -> Text) -> [Text] -> [Text]
forall a b. (a -> b) -> [a] -> [b]
map (Text -> Text -> Text
Schema.qualifiedName Text
schemaName) [Text]
truncated)
          Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" CASCADE"
      go :: Int -> IO ()
go Int
remaining = do
        outcome <- IO () -> IO (Either SqlError ())
forall e a. Exception e => IO a -> IO (Either e a)
try (Connection -> Text -> IO ()
execute_ Connection
conn Text
truncateSql) :: IO (Either SqlError ())
        case outcome of
          Right () -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
          -- 40P01 deadlock_detected, 55P03 lock_not_available
          Left SqlError
sqlErr
            | SqlError -> ByteString
sqlState SqlError
sqlErr ByteString -> [ByteString] -> Bool
forall a. Eq a => a -> [a] -> Bool
forall (t :: * -> *) a. (Foldable t, Eq a) => a -> t a -> Bool
`elem` [ByteString
"40P01", ByteString
"55P03"] Bool -> Bool -> Bool
&& Int
remaining Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> (Int
0 :: Int) ->
                Int -> IO ()
threadDelay Int
cleanupRetryDelayMicros IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Int -> IO ()
go (Int
remaining Int -> Int -> Int
forall a. Num a => a -> a -> a
- Int
1)
            | Bool
otherwise -> SqlError -> IO ()
forall e a. (HasCallStack, Exception e) => e -> IO a
throwIO SqlError
sqlErr
  Int -> IO ()
go Int
cleanupRetries
  Connection -> Text -> IO ()
execute_ Connection
conn Text
"RESET lock_timeout"
  Connection -> Text -> IO ()
execute_ Connection
conn Text
"SET client_min_messages = NOTICE"

-- | Lock wait bound for the cleanup truncate.
cleanupLockTimeout :: Text
cleanupLockTimeout :: Text
cleanupLockTimeout = Text
"5s"

-- | Wait before the cleanup truncate tries again.
cleanupRetryDelayMicros :: Int
cleanupRetryDelayMicros :: Int
cleanupRetryDelayMicros = Int
100_000

-- | Retries of the cleanup truncate after a deadlock or lock timeout.
cleanupRetries :: Int
cleanupRetries :: Int
cleanupRetries = Int
10

-- | 'cleanupData' on a fresh connection.
cleanupOnce :: ByteString -> Text -> Text -> IO ()
cleanupOnce :: ByteString -> Text -> Text -> IO ()
cleanupOnce ByteString
connStr Text
schemaName Text
tableName = ByteString -> (Connection -> IO ()) -> IO ()
forall a. ByteString -> (Connection -> IO a) -> IO a
withConn ByteString
connStr (Text -> Text -> Connection -> IO ()
cleanupData Text
schemaName Text
tableName)

-- | Run an action on a fresh connection that is closed afterwards.
withConn :: ByteString -> (Connection -> IO a) -> IO a
withConn :: forall a. ByteString -> (Connection -> IO a) -> IO a
withConn ByteString
connStr = IO Connection
-> (Connection -> IO ()) -> (Connection -> IO a) -> IO a
forall a b c. IO a -> (a -> IO b) -> (a -> IO c) -> IO c
bracket (ByteString -> IO Connection
connectPostgreSQL ByteString
connStr) Connection -> IO ()
close

-- | Terminate every other backend on this database whose last query matches the LIKE pattern.
terminateBackendsMatching :: ByteString -> Text -> IO ()
terminateBackendsMatching :: ByteString -> Text -> IO ()
terminateBackendsMatching ByteString
connStr Text
queryLike =
  ByteString -> (Connection -> IO ()) -> IO ()
forall a. ByteString -> (Connection -> IO a) -> IO a
withConn ByteString
connStr ((Connection -> IO ()) -> IO ()) -> (Connection -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Connection
conn ->
    IO [Only Bool] -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO [Only Bool] -> IO ()) -> IO [Only Bool] -> IO ()
forall a b. (a -> b) -> a -> b
$
      forall q r.
(ToRow q, FromRow r) =>
Connection -> Query -> q -> IO [r]
query @_ @(Only Bool)
        Connection
conn
        Query
"SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE pid <> pg_backend_pid() AND datname = current_database() AND query LIKE ?"
        (Text -> Only Text
forall a. a -> Only a
Only Text
queryLike)

-- | Terminate one backend by pid.
terminatePid :: ByteString -> Int32 -> IO ()
terminatePid :: ByteString -> Int32 -> IO ()
terminatePid ByteString
connStr Int32
pid =
  ByteString -> (Connection -> IO ()) -> IO ()
forall a. ByteString -> (Connection -> IO a) -> IO a
withConn ByteString
connStr ((Connection -> IO ()) -> IO ()) -> (Connection -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Connection
conn -> IO [Only Bool] -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO [Only Bool] -> IO ()) -> IO [Only Bool] -> IO ()
forall a b. (a -> b) -> a -> b
$ forall q r.
(ToRow q, FromRow r) =>
Connection -> Query -> q -> IO [r]
query @_ @(Only Bool) Connection
conn Query
"SELECT pg_terminate_backend(?)" (Int32 -> Only Int32
forall a. a -> Only a
Only Int32
pid)

-- | Count distinct backends whose last query is a LISTEN that names this schema.
listenerConnectionCount :: ByteString -> Text -> IO Int
listenerConnectionCount :: ByteString -> Text -> IO Int
listenerConnectionCount ByteString
connStr Text
schema =
  ByteString -> (Connection -> IO Int) -> IO Int
forall a. ByteString -> (Connection -> IO a) -> IO a
withConn ByteString
connStr ((Connection -> IO Int) -> IO Int)
-> (Connection -> IO Int) -> IO Int
forall a b. (a -> b) -> a -> b
$ \Connection
conn -> do
    rows <-
      forall q r.
(ToRow q, FromRow r) =>
Connection -> Query -> q -> IO [r]
query @_ @(Only Int)
        Connection
conn
        Query
"SELECT count(DISTINCT pid)::int FROM pg_stat_activity WHERE datname = current_database() AND query LIKE 'LISTEN%' AND query LIKE ?"
        (Text -> Only Text
forall a. a -> Only a
Only (Text
"%" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
schema Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"%" :: Text))
    pure (maybe 0 fromOnly (listToMaybe rows))

-- | Run a statement with no parameters (test setup only).
execute_ :: Connection -> Text -> IO ()
execute_ :: Connection -> Text -> IO ()
execute_ Connection
conn Text
sql = IO Int64 -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO Int64 -> IO ()) -> IO Int64 -> IO ()
forall a b. (a -> b) -> a -> b
$ Connection -> Query -> () -> IO Int64
forall q. ToRow q => Connection -> Query -> q -> IO Int64
execute Connection
conn (Text -> Query
toQuery Text
sql) ()

-- | A postgresql-simple query from rendered SQL text.
toQuery :: Text -> Query
toQuery :: Text -> Query
toQuery = String -> Query
forall a. IsString a => String -> a
fromString (String -> Query) -> (Text -> String) -> Text -> Query
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Text -> String
T.unpack

-- | Run a pre-rendered statement with @?@ placeholders and positional parameters
-- (test setup only).
execStatement :: (MonadArbiter m) => Text -> Params -> m Int64
execStatement :: forall (m :: * -> *). MonadArbiter m => Text -> Params -> m Int64
execStatement Text
sql Params
params = Query () -> m Int64
forall a. Query a -> m Int64
forall (m :: * -> *) a. MonadArbiter m => Query a -> m Int64
MA.executeStatement ([Piece] -> Params -> RowCodec () -> Query ()
forall a. [Piece] -> Params -> RowCodec a -> Query a
MA.mkQuery (Text -> [Piece]
placeholderPieces Text
sql) Params
params (() -> RowCodec ()
forall a. a -> Ap NullCol a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()))

-- | Run a pre-rendered query with @?@ placeholders, positional parameters and a
-- decoder (test setup only).
execQuery :: (MonadArbiter m) => Text -> Params -> RowCodec a -> m [a]
execQuery :: forall (m :: * -> *) a.
MonadArbiter m =>
Text -> Params -> RowCodec a -> m [a]
execQuery Text
sql Params
params RowCodec a
codec = Query a -> m [a]
forall a. Query a -> m [a]
forall (m :: * -> *) a. MonadArbiter m => Query a -> m [a]
MA.executeQuery ([Piece] -> Params -> RowCodec a -> Query a
forall a. [Piece] -> Params -> RowCodec a -> Query a
MA.mkQuery (Text -> [Piece]
placeholderPieces Text
sql) Params
params RowCodec a
codec)

-- | Every @?@ in the text is a hole.
placeholderPieces :: Text -> [Piece]
placeholderPieces :: Text -> [Piece]
placeholderPieces = Piece -> [Piece] -> [Piece]
forall a. a -> [a] -> [a]
intersperse Piece
Hole ([Piece] -> [Piece]) -> (Text -> [Piece]) -> Text -> [Piece]
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Text -> Piece) -> [Text] -> [Piece]
forall a b. (a -> b) -> [a] -> [b]
map Text -> Piece
Lit ([Text] -> [Piece]) -> (Text -> [Text]) -> Text -> [Piece]
forall b c a. (b -> c) -> (a -> b) -> a -> c
. HasCallStack => Text -> Text -> [Text]
Text -> Text -> [Text]
T.splitOn Text
"?"

-- | Connect and rebuild the schema once, for a suite's outer bracket.
-- It drops the schema with CASCADE and creates it again.
setupOnce
  :: ByteString
  -> Text
  -> Text
  -> Bool
  -- ^ Install the job-arrival notify trigger
  -> IO ()
setupOnce :: ByteString -> Text -> Text -> Bool -> IO ()
setupOnce ByteString
connStr Text
schemaName Text
tableName Bool
withNotify = ByteString -> (Connection -> IO ()) -> IO ()
forall a. ByteString -> (Connection -> IO a) -> IO a
withConn ByteString
connStr ((Connection -> IO ()) -> IO ()) -> (Connection -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Connection
conn -> do
  Connection -> IO ()
disableNoticeReporting Connection
conn
  Bool -> Text -> Text -> Connection -> IO ()
setupDDLWithConfig Bool
withNotify Text
schemaName Text
tableName Connection
conn

-- | Add one more job-queue table to an existing schema.
addQueueTable
  :: ByteString
  -> Text
  -> Text
  -> Bool
  -- ^ Install the job-arrival notify trigger
  -> IO ()
addQueueTable :: ByteString -> Text -> Text -> Bool -> IO ()
addQueueTable ByteString
connStr Text
schemaName Text
tableName Bool
withNotify = ByteString -> (Connection -> IO ()) -> IO ()
forall a. ByteString -> (Connection -> IO a) -> IO a
withConn ByteString
connStr ((Connection -> IO ()) -> IO ()) -> (Connection -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Connection
conn -> do
  Connection -> IO ()
disableNoticeReporting Connection
conn
  (MigrationCommand -> IO ()) -> [MigrationCommand] -> IO ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ (Connection -> MigrationCommand -> IO ()
runScript Connection
conn) (Text -> Text -> TableAdmission -> [MigrationCommand]
jobQueueMigrationsForTable Text
schemaName Text
tableName TableAdmission
allTableAdmission)
  Bool -> Text -> Text -> Connection -> IO ()
createNotifyObjects Bool
withNotify Text
schemaName Text
tableName Connection
conn

-- | Install the notification function and trigger the reconciler would.
createNotifyObjects :: Bool -> Text -> Text -> Connection -> IO ()
createNotifyObjects :: Bool -> Text -> Text -> Connection -> IO ()
createNotifyObjects Bool
withNotify Text
schemaName Text
tableName Connection
conn =
  Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
withNotify (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
    Connection -> Text -> IO ()
execute_ Connection
conn (Text -> Text -> Text
Schema.createNotifyFunctionSQL Text
schemaName Text
tableName)
    Connection -> Text -> IO ()
execute_ Connection
conn (Text -> Text -> Text
Schema.createNotifyTriggerSQL Text
schemaName Text
tableName)

-- | Silence libpq notice output for the connection.
disableNoticeReporting :: Connection -> IO ()
disableNoticeReporting :: Connection -> IO ()
disableNoticeReporting Connection
conn =
  Connection -> (Connection -> IO ()) -> IO ()
forall a. Connection -> (Connection -> IO a) -> IO a
PGS.withConnection Connection
conn Connection -> IO ()
LibPQ.disableNoticeReporting

-- | A single-stripe pool of the default size for tests.
createSharedPool :: ByteString -> IO (Pool Connection)
createSharedPool :: ByteString -> IO (Pool Connection)
createSharedPool = Int -> ByteString -> IO (Pool Connection)
createPoolOf Int
sharedPoolSize

-- | A single-stripe pool of the given size for tests.
createPoolOf :: Int -> ByteString -> IO (Pool Connection)
createPoolOf :: Int -> ByteString -> IO (Pool Connection)
createPoolOf Int
size ByteString
connStr = Int
-> IO Connection -> (Connection -> IO ()) -> IO (Pool Connection)
forall conn. Int -> IO conn -> (conn -> IO ()) -> IO (Pool conn)
createPoolWith Int
size (ByteString -> IO Connection
connectPostgreSQL ByteString
connStr) Connection -> IO ()
close

-- | A single-stripe pool of the given size over any connect and release actions.
createPoolWith :: Int -> IO conn -> (conn -> IO ()) -> IO (Pool conn)
createPoolWith :: forall conn. Int -> IO conn -> (conn -> IO ()) -> IO (Pool conn)
createPoolWith Int
size IO conn
connect conn -> IO ()
release =
  PoolConfig conn -> IO (Pool conn)
forall a. PoolConfig a -> IO (Pool a)
newPool (PoolConfig conn -> IO (Pool conn))
-> PoolConfig conn -> IO (Pool conn)
forall a b. (a -> b) -> a -> b
$ Maybe Int -> PoolConfig conn -> PoolConfig conn
forall a. Maybe Int -> PoolConfig a -> PoolConfig a
setNumStripes (Int -> Maybe Int
forall a. a -> Maybe a
Just Int
1) (PoolConfig conn -> PoolConfig conn)
-> PoolConfig conn -> PoolConfig conn
forall a b. (a -> b) -> a -> b
$ IO conn -> (conn -> IO ()) -> Double -> Int -> PoolConfig conn
forall a. IO a -> (a -> IO ()) -> Double -> Int -> PoolConfig a
defaultPoolConfig IO conn
connect conn -> IO ()
release Double
sharedPoolIdleSeconds Int
size

-- | The default test pool size.
sharedPoolSize :: Int
sharedPoolSize :: Int
sharedPoolSize = Int
5

-- | The pool size for suites that race many claimers.
concurrencyPoolSize :: Int
concurrencyPoolSize :: Int
concurrencyPoolSize = Int
10

-- | The pool size for the state-machine suite.
stateMachinePoolSize :: Int
stateMachinePoolSize :: Int
stateMachinePoolSize = Int
40

sharedPoolIdleSeconds :: Double
sharedPoolIdleSeconds :: Double
sharedPoolIdleSeconds = Double
60

-- | Seed a concurrency policy's default limit and clear any override, as SQL statements.
seedConcurrencyPolicySQL :: Text -> Text -> Int32 -> [Text]
seedConcurrencyPolicySQL :: Text -> Text -> Int32 -> [Text]
seedConcurrencyPolicySQL Text
schema Text
prefix Int32
lim =
  [ Text -> ConcurrencyPolicy -> Text
CC.upsertConcurrencyPolicyRowSQL Text
schema ConcurrencyPolicy {cpPrefix :: Text
cpPrefix = Text
prefix, cpLimit :: Int32
cpLimit = Int32
lim}
  , Text
"UPDATE "
      Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
CC.arbiterConcurrencyPoliciesTable Text
schema
      Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" SET override_limit = NULL WHERE prefix_id = "
      Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
textLiteral Text
prefix
  ]

-- | Repeat a batch action until it returns empty, accumulating the results.
drainWith :: IO [a] -> IO [a]
drainWith :: forall a. IO [a] -> IO [a]
drainWith IO [a]
fetch = [[a]] -> IO [a]
go []
  where
    go :: [[a]] -> IO [a]
go [[a]]
batches = do
      batch <- IO [a]
fetch
      if null batch then pure (concat (reverse batches)) else go (batch : batches)

-- | A UTCTime from calendar and clock components.
mkTime :: Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime :: Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
year Int
month Int
day Int
hour Int
minute Int
second =
  Day -> DiffTime -> UTCTime
UTCTime (Integer -> Int -> Int -> Day
fromGregorian Integer
year Int
month Int
day) (Integer -> DiffTime
secondsToDiffTime (Int -> Integer
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Int
hour Int -> Int -> Int
forall a. Num a => a -> a -> a
* Int
3600 Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
minute Int -> Int -> Int
forall a. Num a => a -> a -> a
* Int
60 Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
second)))

-- | Truncate to microsecond precision to match PostgreSQL @timestamptz@.
truncateToMicros :: UTCTime -> UTCTime
truncateToMicros :: UTCTime -> UTCTime
truncateToMicros (UTCTime Day
day DiffTime
dayTime) =
  let micros :: Integer
micros = DiffTime -> Integer
forall b. Integral b => DiffTime -> b
forall a b. (RealFrac a, Integral b) => a -> b
floor (DiffTime
dayTime DiffTime -> DiffTime -> DiffTime
forall a. Num a => a -> a -> a
* DiffTime
1e6) :: Integer
   in Day -> DiffTime -> UTCTime
UTCTime Day
day (Integer -> DiffTime
picosecondsToDiffTime (Integer
micros Integer -> Integer -> Integer
forall a. Num a => a -> a -> a
* Integer
1000000))

-- | A hub logger that swallows warn and error output.
quietHubLog :: HubLog
quietHubLog :: HubLog
quietHubLog =
  HubLog
    { hubRecovered :: Text -> IO ()
hubRecovered = IO () -> Text -> IO ()
forall a b. a -> b -> a
const (() -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())
    , hubWarn :: Text -> IO ()
hubWarn = IO () -> Text -> IO ()
forall a b. a -> b -> a
const (() -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())
    , hubError :: Text -> IO ()
hubError = IO () -> Text -> IO ()
forall a b. a -> b -> a
const (() -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())
    , hubRepeatInterval :: NominalDiffTime
hubRepeatInterval = NominalDiffTime
defaultFailureRepeatInterval
    }